1use std::sync::Arc;
9use std::time::{Duration, Instant};
10
11use anyhow::{anyhow, Result};
12use regex::Regex;
13use serde_json::{json, Value};
14
15use crate::bidi::BidiClient;
16use crate::cdp::CdpClient;
17use crate::detect::Engine;
18use crate::errors::SessionError;
19use crate::session::freshness;
20use crate::session::targets::{open_bidi, open_cdp, BidiContext, CdpTarget};
21
22pub enum PageSession {
25 Cdp(CdpPage),
26 Bidi(BidiPage),
30}
31
32pub struct CdpPage {
33 pub client: CdpClient,
34 pub session_id: String,
35 pub target_id: String,
36}
37
38pub struct BidiPage {
39 pub client: Arc<BidiClient>,
40 pub context: String,
41 owns_session: bool,
46}
47
48impl PageSession {
49 pub async fn attach(endpoint: &str, engine: Engine, url_regex: Option<&str>) -> Result<Self> {
55 let pattern = url_regex.map(Regex::new).transpose()?;
56 match engine {
57 Engine::Cdp => {
58 let client = open_cdp(endpoint).await?;
59 let target_id = pick_cdp_page(&client, pattern.as_ref()).await?;
60 let session_id = client.attach_to_target(&target_id).await?;
61 let _ = client
66 .send_with_session("Inspector.enable", json!({}), Some(&session_id))
67 .await;
68 Ok(PageSession::Cdp(CdpPage {
69 client,
70 session_id,
71 target_id,
72 }))
73 }
74 Engine::Bidi => {
75 let client = Arc::new(open_bidi(endpoint).await?);
76 client.session_new().await?;
77 let context = pick_bidi_context(&client, pattern.as_ref()).await?;
78 Ok(PageSession::Bidi(BidiPage {
79 client,
80 context,
81 owns_session: true,
82 }))
83 }
84 }
85 }
86
87 pub async fn from_bidi_cache(client: Arc<BidiClient>, url_regex: Option<&str>) -> Result<Self> {
93 let pattern = url_regex.map(Regex::new).transpose()?;
94 let context = pick_bidi_context(&client, pattern.as_ref()).await?;
95 Ok(PageSession::Bidi(BidiPage {
96 client,
97 context,
98 owns_session: false,
99 }))
100 }
101
102 pub async fn attach_for_origin(endpoint: &str, engine: Engine, origin: &str) -> Result<Self> {
113 let want =
114 url::Url::parse(origin).map_err(|e| anyhow!("invalid origin URL `{origin}`: {e}"))?;
115 let origin_root = origin_root_url(&want);
116 match engine {
117 Engine::Cdp => {
118 let client = open_cdp(endpoint).await?;
119 let target_id = match find_cdp_target_for_origin(&client, &want).await? {
120 Some(id) => id,
121 None => create_cdp_tab(&client, &origin_root).await?,
122 };
123 let session_id = client.attach_to_target(&target_id).await?;
124 let _ = client
125 .send_with_session("Inspector.enable", json!({}), Some(&session_id))
126 .await;
127 Ok(PageSession::Cdp(CdpPage {
128 client,
129 session_id,
130 target_id,
131 }))
132 }
133 Engine::Bidi => {
134 let client = Arc::new(open_bidi(endpoint).await?);
135 client.session_new().await?;
136 let context = match find_bidi_context_for_origin(&client, &want).await? {
137 Some(c) => c,
138 None => create_bidi_tab(&client, &origin_root).await?,
139 };
140 Ok(PageSession::Bidi(BidiPage {
141 client,
142 context,
143 owns_session: true,
144 }))
145 }
146 }
147 }
148
149 pub async fn evaluate(&self, expression: &str, await_promise: bool) -> Result<Value> {
160 self.evaluate_with_timeout(expression, await_promise, None)
161 .await
162 }
163
164 pub async fn evaluate_with_timeout(
186 &self,
187 expression: &str,
188 await_promise: bool,
189 timeout: Option<Duration>,
190 ) -> Result<Value> {
191 let target_id = self.target_id();
192 let url = None;
193 match self {
194 PageSession::Cdp(p) => {
195 let inner = async {
196 let v = p
197 .client
198 .send_with_session(
199 "Runtime.evaluate",
200 json!({
201 "expression": expression,
202 "returnByValue": true,
203 "awaitPromise": await_promise,
204 }),
205 Some(&p.session_id),
206 )
207 .await?;
208 Ok::<Value, anyhow::Error>(v["result"]["value"].clone())
209 };
210 crate::session::crash::evaluate_with_crash_detection(
211 &p.client,
212 &p.target_id,
213 Some(&p.session_id),
214 inner,
215 timeout,
216 )
217 .await
218 }
219 PageSession::Bidi(p) => {
220 let inner = async {
221 let _ = await_promise; let v = p.client.script_evaluate(&p.context, expression).await?;
223 Ok::<Value, anyhow::Error>(v["result"]["value"].clone())
224 };
225 match timeout {
226 None => inner.await,
227 Some(d) => match tokio::time::timeout(d, inner).await {
228 Ok(r) => r,
229 Err(_) => Err(SessionError::TabHung {
230 target_id,
231 url,
232 timeout_ms: d.as_millis() as u64,
233 hint: "op-timeout",
234 }
235 .into()),
236 },
237 }
238 }
239 }
240 }
241
242 pub fn target_id(&self) -> Option<String> {
245 match self {
246 PageSession::Cdp(p) => Some(p.target_id.clone()),
247 PageSession::Bidi(p) => Some(p.context.clone()),
248 }
249 }
250
251 pub async fn navigate(&self, url: &str) -> Result<()> {
253 match self {
254 PageSession::Cdp(p) => {
255 p.client
256 .send_with_session("Page.navigate", json!({"url": url}), Some(&p.session_id))
257 .await?;
258 Ok(())
259 }
260 PageSession::Bidi(p) => {
261 p.client.browsing_context_navigate(&p.context, url).await?;
262 Ok(())
263 }
264 }
265 }
266
267 pub async fn ensure_fresh(&self, max_age: Duration) -> Result<()> {
273 let info_value = self
274 .evaluate_with_timeout(
275 freshness::PAGE_FRESHNESS_EXPR,
276 false,
277 Some(freshness::CHECK_TIMEOUT),
278 )
279 .await?;
280 let info = freshness::parse_page_freshness(info_value)?;
281 if !info.should_reload(max_age) {
282 return Ok(());
283 }
284
285 tracing::info!(
286 target = "session",
287 url = %info.href,
288 age_ms = info.age_ms,
289 max_age_ms = max_age.as_millis(),
290 "reloading stale page before reading page context"
291 );
292 tokio::time::timeout(freshness::RELOAD_READY_TIMEOUT, self.navigate(&info.href)).await??;
293 self.wait_until_ready().await
294 }
295
296 async fn wait_until_ready(&self) -> Result<()> {
297 let deadline = Instant::now() + freshness::RELOAD_READY_TIMEOUT;
298 loop {
299 let value = self
300 .evaluate_with_timeout(
301 freshness::READY_STATE_EXPR,
302 false,
303 Some(freshness::CHECK_TIMEOUT),
304 )
305 .await?;
306 if freshness::is_ready(&value) {
307 return Ok(());
308 }
309 if Instant::now() >= deadline {
310 tracing::warn!(
311 target = "session",
312 "page reload did not reach document.readyState=complete before continuing"
313 );
314 return Ok(());
315 }
316 tokio::time::sleep(freshness::READY_POLL_INTERVAL).await;
317 }
318 }
319
320 pub async fn screenshot(&self, full_page: bool) -> Result<String> {
322 match self {
323 PageSession::Cdp(p) => {
324 let v = p
325 .client
326 .send_with_session(
327 "Page.captureScreenshot",
328 json!({
329 "format": "png",
330 "captureBeyondViewport": full_page,
331 }),
332 Some(&p.session_id),
333 )
334 .await?;
335 v["data"]
336 .as_str()
337 .map(|s| s.to_string())
338 .ok_or_else(|| anyhow!("no screenshot data"))
339 }
340 PageSession::Bidi(p) => {
341 let _ = full_page; p.client
343 .browsing_context_capture_screenshot(&p.context, None)
344 .await
345 }
346 }
347 }
348
349 pub fn engine(&self) -> Engine {
351 match self {
352 PageSession::Cdp(_) => Engine::Cdp,
353 PageSession::Bidi(_) => Engine::Bidi,
354 }
355 }
356
357 pub async fn close(self) {
362 match self {
363 PageSession::Cdp(p) => p.client.close().await,
364 PageSession::Bidi(p) => {
365 if p.owns_session {
366 let _ = p.client.session_end().await;
367 }
368 }
369 }
370 }
371}
372
373pub async fn evaluate_for_origin_with_recover_once(
380 endpoint: &str,
381 engine: Engine,
382 origin_url: &str,
383 expression: &str,
384 await_promise: bool,
385 timeout: Duration,
386 max_age: Duration,
387) -> Result<Value> {
388 let first = evaluate_for_origin_once(
389 endpoint,
390 engine,
391 origin_url,
392 expression,
393 await_promise,
394 timeout,
395 max_age,
396 )
397 .await;
398 match first {
399 Ok(v) => Ok(v),
400 Err(e) if crate::errors::is_recoverable_tab_failure(&e) => {
401 tracing::warn!(
402 target = "session",
403 "origin-bound evaluate failed with recoverable error; re-attaching and retrying once: {e:#}"
404 );
405 evaluate_for_origin_once(
406 endpoint,
407 engine,
408 origin_url,
409 expression,
410 await_promise,
411 timeout,
412 max_age,
413 )
414 .await
415 }
416 Err(e) => Err(e),
417 }
418}
419
420async fn evaluate_for_origin_once(
421 endpoint: &str,
422 engine: Engine,
423 origin_url: &str,
424 expression: &str,
425 await_promise: bool,
426 timeout: Duration,
427 max_age: Duration,
428) -> Result<Value> {
429 let session = PageSession::attach_for_origin(endpoint, engine, origin_url).await?;
430 let result = async {
431 session.ensure_fresh(max_age).await?;
432 session
433 .evaluate_with_timeout(expression, await_promise, Some(timeout))
434 .await
435 }
436 .await;
437 session.close().await;
438 result
439}
440
441const PICK_PROBE_TIMEOUT: Duration = Duration::from_millis(500);
449
450async fn pick_cdp_page(client: &CdpClient, pattern: Option<&Regex>) -> Result<String> {
451 let targets = client.list_targets().await?;
452 let pages: Vec<CdpTarget> = CdpTarget::pages(&targets).collect();
453
454 let Some(re) = pattern else {
460 return pages
461 .into_iter()
462 .next()
463 .map(|t| t.id)
464 .ok_or_else(|| anyhow!("no page target found"));
465 };
466
467 let matches: Vec<CdpTarget> = pages.into_iter().filter(|t| re.is_match(&t.url)).collect();
468 if matches.is_empty() {
469 return Err(anyhow!("no CDP page target matched URL regex"));
470 }
471
472 let mut hung_count = 0usize;
476 let mut last_target: Option<String> = None;
477 let mut last_url: Option<String> = None;
478 for t in &matches {
479 let target_id = t.id.clone();
480 last_target = Some(target_id.clone());
481 last_url = Some(t.url.clone());
482 if probe_cdp_target(client, &target_id, PICK_PROBE_TIMEOUT).await {
483 return Ok(target_id);
484 }
485 hung_count += 1;
486 }
487 let err: anyhow::Error = SessionError::TabHung {
488 target_id: last_target,
489 url: last_url,
490 timeout_ms: PICK_PROBE_TIMEOUT.as_millis() as u64,
491 hint: "all-matches-hung",
492 }
493 .into();
494 Err(err.context(format!(
495 "URL regex matched {hung_count} page(s) but none responded to a {}ms probe",
496 PICK_PROBE_TIMEOUT.as_millis()
497 )))
498}
499
500async fn probe_cdp_target(client: &CdpClient, target_id: &str, budget: Duration) -> bool {
505 let attach = tokio::time::timeout(
506 budget,
507 client.send(
508 "Target.attachToTarget",
509 json!({ "targetId": target_id, "flatten": true }),
510 ),
511 )
512 .await;
513 let session_id = match attach {
514 Ok(Ok(v)) => match v.get("sessionId").and_then(|s| s.as_str()) {
515 Some(s) => s.to_string(),
516 None => return false,
517 },
518 _ => return false,
519 };
520 let eval = client.send_with_session(
521 "Runtime.evaluate",
522 json!({
523 "expression": "1",
524 "returnByValue": true,
525 "awaitPromise": false,
526 }),
527 Some(&session_id),
528 );
529 let alive = matches!(tokio::time::timeout(budget, eval).await, Ok(Ok(_)));
530 let _ = client
531 .send(
532 "Target.detachFromTarget",
533 json!({ "sessionId": session_id }),
534 )
535 .await;
536 alive
537}
538
539async fn pick_bidi_context(client: &BidiClient, pattern: Option<&Regex>) -> Result<String> {
540 let tree = client.send("browsingContext.getTree", json!({})).await?;
541 let contexts = BidiContext::from_tree(&tree);
542
543 let Some(re) = pattern else {
545 return contexts
546 .into_iter()
547 .next()
548 .map(|c| c.context)
549 .ok_or_else(|| anyhow!("no top-level browsing context"));
550 };
551
552 let matches: Vec<BidiContext> = contexts
553 .into_iter()
554 .filter(|c| re.is_match(&c.url))
555 .collect();
556 if matches.is_empty() {
557 return Err(anyhow!("no BiDi context matched URL regex"));
558 }
559
560 let mut hung_count = 0usize;
561 let mut last_ctx: Option<String> = None;
562 let mut last_url: Option<String> = None;
563 for c in &matches {
564 let ctx = c.context.clone();
565 last_ctx = Some(ctx.clone());
566 last_url = Some(c.url.clone());
567 if probe_bidi_context(client, &ctx, PICK_PROBE_TIMEOUT).await {
568 return Ok(ctx);
569 }
570 hung_count += 1;
571 }
572 let err: anyhow::Error = SessionError::TabHung {
573 target_id: last_ctx,
574 url: last_url,
575 timeout_ms: PICK_PROBE_TIMEOUT.as_millis() as u64,
576 hint: "all-matches-hung",
577 }
578 .into();
579 Err(err.context(format!(
580 "URL regex matched {hung_count} context(s) but none responded to a {}ms probe",
581 PICK_PROBE_TIMEOUT.as_millis()
582 )))
583}
584
585async fn probe_bidi_context(client: &BidiClient, context: &str, budget: Duration) -> bool {
590 matches!(
591 tokio::time::timeout(budget, client.script_evaluate(context, "1")).await,
592 Ok(Ok(_))
593 )
594}
595
596pub(crate) fn same_origin(a: &url::Url, b: &url::Url) -> bool {
598 a.scheme() == b.scheme()
599 && a.host_str() == b.host_str()
600 && a.port_or_known_default() == b.port_or_known_default()
601}
602
603pub(crate) fn origin_root_url(u: &url::Url) -> String {
605 let scheme = u.scheme();
606 let host = u.host_str().unwrap_or("");
607 match (u.port(), u.port_or_known_default()) {
608 (Some(p), _) => format!("{scheme}://{host}:{p}/"),
610 (None, _) => format!("{scheme}://{host}/"),
611 }
612}
613
614async fn find_cdp_target_for_origin(client: &CdpClient, want: &url::Url) -> Result<Option<String>> {
615 let targets = client.list_targets().await?;
616 let found = CdpTarget::pages(&targets).find_map(|t| {
617 let parsed = url::Url::parse(&t.url).ok()?;
618 same_origin(&parsed, want).then_some(t.id)
619 });
620 Ok(found)
621}
622
623async fn create_cdp_tab(client: &CdpClient, url: &str) -> Result<String> {
624 let v = client
625 .send("Target.createTarget", json!({ "url": url }))
626 .await?;
627 v.get("targetId")
628 .and_then(|x| x.as_str())
629 .map(|s| s.to_string())
630 .ok_or_else(|| anyhow!("Target.createTarget did not return targetId"))
631}
632
633async fn find_bidi_context_for_origin(
634 client: &BidiClient,
635 want: &url::Url,
636) -> Result<Option<String>> {
637 let tree = client.send("browsingContext.getTree", json!({})).await?;
638 Ok(BidiContext::from_tree(&tree).into_iter().find_map(|c| {
639 let parsed = url::Url::parse(&c.url).ok()?;
640 same_origin(&parsed, want).then_some(c.context)
641 }))
642}
643
644async fn create_bidi_tab(client: &BidiClient, url: &str) -> Result<String> {
645 let v = client
646 .send("browsingContext.create", json!({ "type": "tab" }))
647 .await?;
648 let ctx = v
649 .get("context")
650 .and_then(|x| x.as_str())
651 .ok_or_else(|| anyhow!("browsingContext.create did not return context"))?
652 .to_string();
653 client.browsing_context_navigate(&ctx, url).await?;
654 Ok(ctx)
655}
656
657#[cfg(test)]
658mod tests {
659 use super::*;
660 use futures_util::{SinkExt, StreamExt};
661 use std::sync::{
662 atomic::{AtomicUsize, Ordering},
663 Arc,
664 };
665 use tokio::sync::Mutex;
666 use tokio_tungstenite::tungstenite::Message;
667
668 async fn spawn_cdp_mock(targets: Vec<Value>) -> String {
669 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
670 let addr = listener.local_addr().unwrap();
671 tokio::spawn(async move {
672 let (stream, _) = listener.accept().await.unwrap();
673 let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap();
674 while let Some(Ok(Message::Text(t))) = ws.next().await {
675 let req: Value = serde_json::from_str(&t).unwrap();
676 let id = req["id"].as_u64().unwrap();
677 let method = req["method"].as_str().unwrap_or("");
678 let result = match method {
679 "Target.getTargets" => json!({"targetInfos": targets.clone()}),
680 "Target.attachToTarget" => json!({"sessionId": "S1"}),
681 "Target.createTarget" => json!({"targetId": "NEW"}),
682 "Runtime.evaluate" => json!({"result": {"value": "ok"}}),
683 "Page.navigate" => json!({}),
684 "Page.captureScreenshot" => json!({"data": "PNGDATA"}),
685 _ => json!({}),
686 };
687 let resp = json!({"id": id, "result": result});
688 ws.send(Message::Text(resp.to_string())).await.unwrap();
689 }
690 });
691 format!("ws://{addr}")
692 }
693
694 async fn spawn_cdp_origin_eval_mock(
695 targets: Vec<Value>,
696 fail_first_eval: bool,
697 ) -> (String, Arc<Mutex<Vec<String>>>, Arc<Mutex<Vec<String>>>) {
698 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
699 let addr = listener.local_addr().unwrap();
700 let targets = Arc::new(targets);
701 let created_urls = Arc::new(Mutex::new(Vec::new()));
702 let attached_targets = Arc::new(Mutex::new(Vec::new()));
703 let eval_count = Arc::new(AtomicUsize::new(0));
704
705 tokio::spawn({
706 let targets = targets.clone();
707 let created_urls = created_urls.clone();
708 let attached_targets = attached_targets.clone();
709 let eval_count = eval_count.clone();
710 async move {
711 loop {
712 let Ok((stream, _)) = listener.accept().await else {
713 break;
714 };
715 let targets = targets.clone();
716 let created_urls = created_urls.clone();
717 let attached_targets = attached_targets.clone();
718 let eval_count = eval_count.clone();
719 tokio::spawn(async move {
720 let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap();
721 while let Some(Ok(Message::Text(t))) = ws.next().await {
722 let req: Value = serde_json::from_str(&t).unwrap();
723 let id = req["id"].as_u64().unwrap();
724 let method = req["method"].as_str().unwrap_or("");
725 if method == "Runtime.evaluate"
726 && fail_first_eval
727 && eval_count.fetch_add(1, Ordering::SeqCst) == 0
728 {
729 let resp = json!({
730 "id": id,
731 "error": {
732 "code": -32000,
733 "message": "No target with given id",
734 }
735 });
736 ws.send(Message::Text(resp.to_string())).await.unwrap();
737 continue;
738 }
739 let result = match method {
740 "Target.getTargets" => {
741 json!({"targetInfos": targets.as_ref().clone()})
742 }
743 "Target.createTarget" => {
744 let url = req
745 .pointer("/params/url")
746 .and_then(|v| v.as_str())
747 .unwrap_or("")
748 .to_string();
749 created_urls.lock().await.push(url);
750 json!({"targetId": "NEW"})
751 }
752 "Target.attachToTarget" => {
753 let target_id = req
754 .pointer("/params/targetId")
755 .and_then(|v| v.as_str())
756 .unwrap_or("")
757 .to_string();
758 let mut attached = attached_targets.lock().await;
759 attached.push(target_id);
760 json!({"sessionId": format!("S{}", attached.len())})
761 }
762 "Target.detachFromTarget" => json!({}),
763 "Inspector.enable" => json!({}),
764 "Runtime.evaluate" => {
765 let expression = req
766 .pointer("/params/expression")
767 .and_then(|v| v.as_str())
768 .unwrap_or("");
769 let value = if expression == freshness::READY_STATE_EXPR {
770 json!("complete")
771 } else if expression == freshness::PAGE_FRESHNESS_EXPR {
772 json!({
773 "href": "https://example.com/login",
774 "ageMs": 0.0,
775 "readyState": "complete"
776 })
777 } else {
778 json!("ok")
779 };
780 json!({"result": {"value": value}})
781 }
782 _ => json!({}),
783 };
784 let resp = json!({"id": id, "result": result});
785 ws.send(Message::Text(resp.to_string())).await.unwrap();
786 }
787 });
788 }
789 }
790 });
791
792 (format!("ws://{addr}"), created_urls, attached_targets)
793 }
794
795 #[test]
796 fn same_origin_basic() {
797 let a = url::Url::parse("https://example.com/path?q=1").unwrap();
798 let b = url::Url::parse("https://example.com/other").unwrap();
799 let c = url::Url::parse("https://other.test/path").unwrap();
800 let d = url::Url::parse("http://example.com/").unwrap();
801 assert!(same_origin(&a, &b));
802 assert!(!same_origin(&a, &c));
803 assert!(!same_origin(&a, &d));
804 }
805
806 #[test]
807 fn origin_root_strips_path_and_default_port() {
808 let u = url::Url::parse("https://example.com/foo/bar?x=1#z").unwrap();
809 assert_eq!(origin_root_url(&u), "https://example.com/");
810 let u2 = url::Url::parse("http://localhost:8080/foo").unwrap();
811 assert_eq!(origin_root_url(&u2), "http://localhost:8080/");
812 }
813
814 #[tokio::test]
815 async fn attach_for_origin_reuses_matching_tab() {
816 let url = spawn_cdp_mock(vec![
817 json!({"targetId":"a","type":"page","url":"https://other.test/x"}),
818 json!({"targetId":"b","type":"page","url":"https://example.com/login"}),
819 ])
820 .await;
821 let s = PageSession::attach_for_origin(&url, Engine::Cdp, "https://example.com/api/v1")
822 .await
823 .unwrap();
824 match s {
825 PageSession::Cdp(p) => assert_eq!(p.target_id, "b"),
826 _ => panic!("expected CDP"),
827 }
828 }
829
830 #[tokio::test]
831 async fn attach_for_origin_creates_tab_when_no_match() {
832 let url = spawn_cdp_mock(vec![
833 json!({"targetId":"a","type":"page","url":"https://other.test/"}),
834 ])
835 .await;
836 let s = PageSession::attach_for_origin(&url, Engine::Cdp, "https://example.com/api")
837 .await
838 .unwrap();
839 match s {
840 PageSession::Cdp(p) => assert_eq!(p.target_id, "NEW"),
841 _ => panic!("expected CDP"),
842 }
843 }
844
845 #[tokio::test]
846 async fn evaluate_for_origin_creates_origin_tab_when_no_match() {
847 let (url, created_urls, attached_targets) = spawn_cdp_origin_eval_mock(
848 vec![json!({"targetId":"a","type":"page","url":"https://other.test/"})],
849 false,
850 )
851 .await;
852 let value = evaluate_for_origin_with_recover_once(
853 &url,
854 Engine::Cdp,
855 "https://example.com/api",
856 "1+1",
857 true,
858 Duration::from_secs(1),
859 freshness::DEFAULT_MAX_AGE,
860 )
861 .await
862 .unwrap();
863 assert_eq!(value, json!("ok"));
864 assert_eq!(
865 *created_urls.lock().await,
866 vec!["https://example.com/".to_string()]
867 );
868 assert_eq!(*attached_targets.lock().await, vec!["NEW".to_string()]);
869 }
870
871 #[tokio::test]
872 async fn evaluate_for_origin_reattaches_and_retries_once() {
873 let (url, created_urls, attached_targets) = spawn_cdp_origin_eval_mock(
874 vec![json!({"targetId":"A","type":"page","url":"https://example.com/login"})],
875 true,
876 )
877 .await;
878 let value = evaluate_for_origin_with_recover_once(
879 &url,
880 Engine::Cdp,
881 "https://example.com/api",
882 "1+1",
883 true,
884 Duration::from_secs(1),
885 freshness::DEFAULT_MAX_AGE,
886 )
887 .await
888 .unwrap();
889 assert_eq!(value, json!("ok"));
890 assert!(created_urls.lock().await.is_empty());
891 assert_eq!(
892 *attached_targets.lock().await,
893 vec!["A".to_string(), "A".to_string()]
894 );
895 }
896
897 #[tokio::test]
898 async fn attach_cdp_picks_first_page_when_no_regex() {
899 let url = spawn_cdp_mock(vec![
900 json!({"targetId":"a","type":"page","url":"https://example.com/"}),
901 json!({"targetId":"b","type":"page","url":"https://other.test/"}),
902 ])
903 .await;
904 let s = PageSession::attach(&url, Engine::Cdp, None).await.unwrap();
905 match s {
906 PageSession::Cdp(p) => {
907 assert_eq!(p.target_id, "a");
908 assert_eq!(p.session_id, "S1");
909 }
910 _ => panic!("expected CDP"),
911 }
912 }
913
914 #[tokio::test]
915 async fn attach_cdp_url_regex_selects_matching() {
916 let url = spawn_cdp_mock(vec![
917 json!({"targetId":"a","type":"page","url":"https://example.com/"}),
918 json!({"targetId":"b","type":"page","url":"https://other.test/"}),
919 ])
920 .await;
921 let s = PageSession::attach(&url, Engine::Cdp, Some(r"other"))
922 .await
923 .unwrap();
924 match s {
925 PageSession::Cdp(p) => assert_eq!(p.target_id, "b"),
926 _ => panic!("expected CDP"),
927 }
928 }
929
930 #[tokio::test]
931 async fn attach_cdp_url_regex_no_match_errors() {
932 let url = spawn_cdp_mock(vec![
933 json!({"targetId":"a","type":"page","url":"https://example.com/"}),
934 ])
935 .await;
936 let err = match PageSession::attach(&url, Engine::Cdp, Some("nomatch")).await {
937 Ok(_) => panic!("expected error"),
938 Err(e) => e,
939 };
940 assert!(err.to_string().contains("no CDP page target matched"));
941 }
942
943 #[tokio::test]
944 async fn evaluate_round_trip_cdp() {
945 let url = spawn_cdp_mock(vec![
946 json!({"targetId":"a","type":"page","url":"https://example.com/"}),
947 ])
948 .await;
949 let s = PageSession::attach(&url, Engine::Cdp, None).await.unwrap();
950 let v = s.evaluate("1+1", false).await.unwrap();
951 assert_eq!(v, json!("ok"));
952 s.close().await;
953 }
954
955 #[tokio::test]
956 async fn screenshot_round_trip_cdp() {
957 let url = spawn_cdp_mock(vec![
958 json!({"targetId":"a","type":"page","url":"https://example.com/"}),
959 ])
960 .await;
961 let s = PageSession::attach(&url, Engine::Cdp, None).await.unwrap();
962 let b64 = s.screenshot(false).await.unwrap();
963 assert_eq!(b64, "PNGDATA");
964 s.close().await;
965 }
966
967 async fn spawn_cdp_mock_eval_hangs(targets: Vec<Value>) -> String {
971 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
972 let addr = listener.local_addr().unwrap();
973 tokio::spawn(async move {
974 let (stream, _) = listener.accept().await.unwrap();
975 let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap();
976 while let Some(Ok(Message::Text(t))) = ws.next().await {
977 let req: Value = serde_json::from_str(&t).unwrap();
978 let id = req["id"].as_u64().unwrap();
979 let method = req["method"].as_str().unwrap_or("");
980 if method == "Runtime.evaluate" {
981 continue;
983 }
984 let result = match method {
985 "Target.getTargets" => json!({"targetInfos": targets.clone()}),
986 "Target.attachToTarget" => json!({"sessionId": "S1"}),
987 _ => json!({}),
988 };
989 let resp = json!({"id": id, "result": result});
990 ws.send(Message::Text(resp.to_string())).await.unwrap();
991 }
992 });
993 format!("ws://{addr}")
994 }
995
996 #[tokio::test]
999 async fn evaluate_with_timeout_returns_tab_hung_on_no_reply() {
1000 let url = spawn_cdp_mock_eval_hangs(vec![
1001 json!({"targetId":"iLO","type":"page","url":"https://192.168.2.28/"}),
1002 ])
1003 .await;
1004 let s = PageSession::attach(&url, Engine::Cdp, None).await.unwrap();
1005 let start = std::time::Instant::now();
1006 let err = s
1007 .evaluate_with_timeout("1+1", false, Some(Duration::from_millis(300)))
1008 .await
1009 .expect_err("must return TabHung");
1010 let elapsed = start.elapsed();
1011 assert!(
1012 elapsed < Duration::from_secs(1),
1013 "did not honour 300ms bound, took {elapsed:?}"
1014 );
1015 let downcast = err.downcast_ref::<SessionError>().expect("typed error");
1016 match downcast {
1017 SessionError::TabHung {
1018 target_id,
1019 timeout_ms,
1020 hint,
1021 ..
1022 } => {
1023 assert_eq!(target_id.as_deref(), Some("iLO"));
1024 assert_eq!(*timeout_ms, 300);
1025 assert_eq!(*hint, "op-timeout");
1026 }
1027 other => panic!("expected TabHung, got {other:?}"),
1028 }
1029 s.close().await;
1030 }
1031
1032 #[tokio::test]
1038 async fn stuck_eval_does_not_block_sibling_session() {
1039 let bad = spawn_cdp_mock_eval_hangs(vec![
1040 json!({"targetId":"BAD","type":"page","url":"https://192.168.2.28/"}),
1041 ])
1042 .await;
1043 let good = spawn_cdp_mock(vec![
1044 json!({"targetId":"GOOD","type":"page","url":"https://example.com/"}),
1045 ])
1046 .await;
1047
1048 let s_bad = PageSession::attach(&bad, Engine::Cdp, None).await.unwrap();
1049 let s_good = PageSession::attach(&good, Engine::Cdp, None).await.unwrap();
1050
1051 let bad_fut = s_bad.evaluate_with_timeout("1+1", false, Some(Duration::from_millis(200)));
1054 let good_fut = s_good.evaluate_with_timeout("1+1", false, Some(Duration::from_secs(5)));
1055 let (bad_res, good_res) = tokio::join!(bad_fut, good_fut);
1056
1057 assert!(bad_res.is_err(), "bad session must surface TabHung");
1058 assert_eq!(good_res.unwrap(), json!("ok"));
1059
1060 s_bad.close().await;
1061 s_good.close().await;
1062 }
1063
1064 async fn spawn_cdp_mock_per_target_eval(
1069 targets: Vec<Value>,
1070 wedged_targets: Vec<&'static str>,
1071 ) -> String {
1072 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
1073 let addr = listener.local_addr().unwrap();
1074 tokio::spawn(async move {
1075 let (stream, _) = listener.accept().await.unwrap();
1076 let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap();
1077 let mut session_wedge: std::collections::HashMap<String, bool> =
1079 std::collections::HashMap::new();
1080 let mut next_session: u32 = 0;
1081 while let Some(Ok(Message::Text(t))) = ws.next().await {
1082 let req: Value = serde_json::from_str(&t).unwrap();
1083 let id = req["id"].as_u64().unwrap();
1084 let method = req["method"].as_str().unwrap_or("");
1085 if method == "Runtime.evaluate" {
1086 if let Some(sid) = req.get("sessionId").and_then(|v| v.as_str()) {
1087 if session_wedge.get(sid).copied().unwrap_or(false) {
1088 continue;
1090 }
1091 }
1092 }
1093 let result = match method {
1094 "Target.getTargets" => json!({"targetInfos": targets.clone()}),
1095 "Target.attachToTarget" => {
1096 let target_id = req
1097 .get("params")
1098 .and_then(|p| p.get("targetId"))
1099 .and_then(|v| v.as_str())
1100 .unwrap_or("")
1101 .to_string();
1102 next_session += 1;
1103 let sid = format!("S{next_session}");
1104 let wedge = wedged_targets.iter().any(|w| *w == target_id);
1105 session_wedge.insert(sid.clone(), wedge);
1106 json!({"sessionId": sid})
1107 }
1108 "Target.detachFromTarget" => json!({}),
1109 "Runtime.evaluate" => json!({"result": {"value": "ok"}}),
1110 _ => json!({}),
1111 };
1112 let resp = json!({"id": id, "result": result});
1113 ws.send(Message::Text(resp.to_string())).await.unwrap();
1114 }
1115 });
1116 format!("ws://{addr}")
1117 }
1118
1119 #[tokio::test]
1122 async fn pick_cdp_iterates_past_hung_match() {
1123 let url = spawn_cdp_mock_per_target_eval(
1124 vec![
1125 json!({"targetId":"DEAD","type":"page","url":"https://twitch.tv/gametechnology"}),
1126 json!({"targetId":"LIVE","type":"page","url":"https://gametechnology.somewhere.com"}),
1127 ],
1128 vec!["DEAD"],
1129 )
1130 .await;
1131 let s = PageSession::attach(&url, Engine::Cdp, Some(r"gametechnology"))
1132 .await
1133 .expect("must iterate past the wedged tab and pick LIVE");
1134 match s {
1135 PageSession::Cdp(p) => assert_eq!(p.target_id, "LIVE"),
1136 _ => panic!("expected CDP"),
1137 }
1138 }
1139
1140 #[tokio::test]
1144 async fn pick_cdp_all_matches_hung_returns_tab_hung() {
1145 let url = spawn_cdp_mock_per_target_eval(
1146 vec![
1147 json!({"targetId":"A","type":"page","url":"https://example.com/foo"}),
1148 json!({"targetId":"B","type":"page","url":"https://example.com/bar"}),
1149 ],
1150 vec!["A", "B"],
1151 )
1152 .await;
1153 let start = std::time::Instant::now();
1154 let err = match PageSession::attach(&url, Engine::Cdp, Some(r"example\.com")).await {
1155 Ok(_) => panic!("all matches wedged → must error"),
1156 Err(e) => e,
1157 };
1158 let elapsed = start.elapsed();
1159 assert!(
1160 elapsed < PICK_PROBE_TIMEOUT * 2 + Duration::from_millis(500),
1161 "took too long: {elapsed:?}"
1162 );
1163 let typed = err.downcast_ref::<SessionError>().expect("typed error");
1164 match typed {
1165 SessionError::TabHung { hint, .. } => {
1166 assert_eq!(*hint, "all-matches-hung");
1167 }
1168 other => panic!("expected TabHung, got {other:?}"),
1169 }
1170 let text = format!("{err:#}");
1171 assert!(
1172 text.contains("URL regex matched 2 page(s)"),
1173 "context missing count: {text}"
1174 );
1175 }
1176
1177 #[tokio::test]
1180 async fn pick_cdp_single_healthy_match_is_picked() {
1181 let url = spawn_cdp_mock_per_target_eval(
1182 vec![json!({"targetId":"OK","type":"page","url":"https://example.com/x"})],
1183 vec![],
1184 )
1185 .await;
1186 let s = PageSession::attach(&url, Engine::Cdp, Some(r"example"))
1187 .await
1188 .unwrap();
1189 match s {
1190 PageSession::Cdp(p) => assert_eq!(p.target_id, "OK"),
1191 _ => panic!("expected CDP"),
1192 }
1193 }
1194}