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(
626 "Target.createTarget",
627 json!({ "url": url, "background": true }),
628 )
629 .await?;
630 v.get("targetId")
631 .and_then(|x| x.as_str())
632 .map(|s| s.to_string())
633 .ok_or_else(|| anyhow!("Target.createTarget did not return targetId"))
634}
635
636async fn find_bidi_context_for_origin(
637 client: &BidiClient,
638 want: &url::Url,
639) -> Result<Option<String>> {
640 let tree = client.send("browsingContext.getTree", json!({})).await?;
641 Ok(BidiContext::from_tree(&tree).into_iter().find_map(|c| {
642 let parsed = url::Url::parse(&c.url).ok()?;
643 same_origin(&parsed, want).then_some(c.context)
644 }))
645}
646
647async fn create_bidi_tab(client: &BidiClient, url: &str) -> Result<String> {
648 let v = client
649 .send("browsingContext.create", json!({ "type": "tab" }))
650 .await?;
651 let ctx = v
652 .get("context")
653 .and_then(|x| x.as_str())
654 .ok_or_else(|| anyhow!("browsingContext.create did not return context"))?
655 .to_string();
656 client.browsing_context_navigate(&ctx, url).await?;
657 Ok(ctx)
658}
659
660#[cfg(test)]
661mod tests {
662 use super::*;
663 use futures_util::{SinkExt, StreamExt};
664 use std::sync::{
665 atomic::{AtomicUsize, Ordering},
666 Arc,
667 };
668 use tokio::sync::Mutex;
669 use tokio_tungstenite::tungstenite::Message;
670
671 async fn spawn_cdp_mock(targets: Vec<Value>) -> String {
672 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
673 let addr = listener.local_addr().unwrap();
674 tokio::spawn(async move {
675 let (stream, _) = listener.accept().await.unwrap();
676 let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap();
677 while let Some(Ok(Message::Text(t))) = ws.next().await {
678 let req: Value = serde_json::from_str(&t).unwrap();
679 let id = req["id"].as_u64().unwrap();
680 let method = req["method"].as_str().unwrap_or("");
681 let result = match method {
682 "Target.getTargets" => json!({"targetInfos": targets.clone()}),
683 "Target.attachToTarget" => json!({"sessionId": "S1"}),
684 "Target.createTarget" => json!({"targetId": "NEW"}),
685 "Runtime.evaluate" => json!({"result": {"value": "ok"}}),
686 "Page.navigate" => json!({}),
687 "Page.captureScreenshot" => json!({"data": "PNGDATA"}),
688 _ => json!({}),
689 };
690 let resp = json!({"id": id, "result": result});
691 ws.send(Message::Text(resp.to_string())).await.unwrap();
692 }
693 });
694 format!("ws://{addr}")
695 }
696
697 async fn spawn_cdp_origin_eval_mock(
698 targets: Vec<Value>,
699 fail_first_eval: bool,
700 ) -> (String, Arc<Mutex<Vec<Value>>>, Arc<Mutex<Vec<String>>>) {
701 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
702 let addr = listener.local_addr().unwrap();
703 let targets = Arc::new(targets);
704 let created_params = Arc::new(Mutex::new(Vec::new()));
705 let attached_targets = Arc::new(Mutex::new(Vec::new()));
706 let eval_count = Arc::new(AtomicUsize::new(0));
707
708 tokio::spawn({
709 let targets = targets.clone();
710 let created_params = created_params.clone();
711 let attached_targets = attached_targets.clone();
712 let eval_count = eval_count.clone();
713 async move {
714 loop {
715 let Ok((stream, _)) = listener.accept().await else {
716 break;
717 };
718 let targets = targets.clone();
719 let created_params = created_params.clone();
720 let attached_targets = attached_targets.clone();
721 let eval_count = eval_count.clone();
722 tokio::spawn(async move {
723 let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap();
724 while let Some(Ok(Message::Text(t))) = ws.next().await {
725 let req: Value = serde_json::from_str(&t).unwrap();
726 let id = req["id"].as_u64().unwrap();
727 let method = req["method"].as_str().unwrap_or("");
728 if method == "Runtime.evaluate"
729 && fail_first_eval
730 && eval_count.fetch_add(1, Ordering::SeqCst) == 0
731 {
732 let resp = json!({
733 "id": id,
734 "error": {
735 "code": -32000,
736 "message": "No target with given id",
737 }
738 });
739 ws.send(Message::Text(resp.to_string())).await.unwrap();
740 continue;
741 }
742 let result = match method {
743 "Target.getTargets" => {
744 json!({"targetInfos": targets.as_ref().clone()})
745 }
746 "Target.createTarget" => {
747 created_params.lock().await.push(req["params"].clone());
748 json!({"targetId": "NEW"})
749 }
750 "Target.attachToTarget" => {
751 let target_id = req
752 .pointer("/params/targetId")
753 .and_then(|v| v.as_str())
754 .unwrap_or("")
755 .to_string();
756 let mut attached = attached_targets.lock().await;
757 attached.push(target_id);
758 json!({"sessionId": format!("S{}", attached.len())})
759 }
760 "Target.detachFromTarget" => json!({}),
761 "Inspector.enable" => json!({}),
762 "Runtime.evaluate" => {
763 let expression = req
764 .pointer("/params/expression")
765 .and_then(|v| v.as_str())
766 .unwrap_or("");
767 let value = if expression == freshness::READY_STATE_EXPR {
768 json!("complete")
769 } else if expression == freshness::PAGE_FRESHNESS_EXPR {
770 json!({
771 "href": "https://example.com/login",
772 "ageMs": 0.0,
773 "readyState": "complete"
774 })
775 } else {
776 json!("ok")
777 };
778 json!({"result": {"value": value}})
779 }
780 _ => json!({}),
781 };
782 let resp = json!({"id": id, "result": result});
783 ws.send(Message::Text(resp.to_string())).await.unwrap();
784 }
785 });
786 }
787 }
788 });
789
790 (format!("ws://{addr}"), created_params, attached_targets)
791 }
792
793 #[test]
794 fn same_origin_basic() {
795 let a = url::Url::parse("https://example.com/path?q=1").unwrap();
796 let b = url::Url::parse("https://example.com/other").unwrap();
797 let c = url::Url::parse("https://other.test/path").unwrap();
798 let d = url::Url::parse("http://example.com/").unwrap();
799 assert!(same_origin(&a, &b));
800 assert!(!same_origin(&a, &c));
801 assert!(!same_origin(&a, &d));
802 }
803
804 #[test]
805 fn origin_root_strips_path_and_default_port() {
806 let u = url::Url::parse("https://example.com/foo/bar?x=1#z").unwrap();
807 assert_eq!(origin_root_url(&u), "https://example.com/");
808 let u2 = url::Url::parse("http://localhost:8080/foo").unwrap();
809 assert_eq!(origin_root_url(&u2), "http://localhost:8080/");
810 }
811
812 #[tokio::test]
813 async fn attach_for_origin_reuses_matching_tab() {
814 let url = spawn_cdp_mock(vec![
815 json!({"targetId":"a","type":"page","url":"https://other.test/x"}),
816 json!({"targetId":"b","type":"page","url":"https://example.com/login"}),
817 ])
818 .await;
819 let s = PageSession::attach_for_origin(&url, Engine::Cdp, "https://example.com/api/v1")
820 .await
821 .unwrap();
822 match s {
823 PageSession::Cdp(p) => assert_eq!(p.target_id, "b"),
824 _ => panic!("expected CDP"),
825 }
826 }
827
828 #[tokio::test]
829 async fn attach_for_origin_creates_tab_when_no_match() {
830 let url = spawn_cdp_mock(vec![
831 json!({"targetId":"a","type":"page","url":"https://other.test/"}),
832 ])
833 .await;
834 let s = PageSession::attach_for_origin(&url, Engine::Cdp, "https://example.com/api")
835 .await
836 .unwrap();
837 match s {
838 PageSession::Cdp(p) => assert_eq!(p.target_id, "NEW"),
839 _ => panic!("expected CDP"),
840 }
841 }
842
843 #[tokio::test]
844 async fn evaluate_for_origin_creates_origin_tab_when_no_match() {
845 let (url, created_params, attached_targets) = spawn_cdp_origin_eval_mock(
846 vec![json!({"targetId":"a","type":"page","url":"https://other.test/"})],
847 false,
848 )
849 .await;
850 let value = evaluate_for_origin_with_recover_once(
851 &url,
852 Engine::Cdp,
853 "https://example.com/api",
854 "1+1",
855 true,
856 Duration::from_secs(1),
857 freshness::DEFAULT_MAX_AGE,
858 )
859 .await
860 .unwrap();
861 assert_eq!(value, json!("ok"));
862 let created = created_params.lock().await;
863 assert_eq!(created.len(), 1);
864 assert_eq!(created[0]["url"], "https://example.com/");
865 assert_eq!(created[0]["background"], true);
866 assert_eq!(*attached_targets.lock().await, vec!["NEW".to_string()]);
867 }
868
869 #[tokio::test]
870 async fn evaluate_for_origin_reattaches_and_retries_once() {
871 let (url, created_params, attached_targets) = spawn_cdp_origin_eval_mock(
872 vec![json!({"targetId":"A","type":"page","url":"https://example.com/login"})],
873 true,
874 )
875 .await;
876 let value = evaluate_for_origin_with_recover_once(
877 &url,
878 Engine::Cdp,
879 "https://example.com/api",
880 "1+1",
881 true,
882 Duration::from_secs(1),
883 freshness::DEFAULT_MAX_AGE,
884 )
885 .await
886 .unwrap();
887 assert_eq!(value, json!("ok"));
888 assert!(created_params.lock().await.is_empty());
889 assert_eq!(
890 *attached_targets.lock().await,
891 vec!["A".to_string(), "A".to_string()]
892 );
893 }
894
895 #[tokio::test]
896 async fn attach_cdp_picks_first_page_when_no_regex() {
897 let url = spawn_cdp_mock(vec![
898 json!({"targetId":"a","type":"page","url":"https://example.com/"}),
899 json!({"targetId":"b","type":"page","url":"https://other.test/"}),
900 ])
901 .await;
902 let s = PageSession::attach(&url, Engine::Cdp, None).await.unwrap();
903 match s {
904 PageSession::Cdp(p) => {
905 assert_eq!(p.target_id, "a");
906 assert_eq!(p.session_id, "S1");
907 }
908 _ => panic!("expected CDP"),
909 }
910 }
911
912 #[tokio::test]
913 async fn attach_cdp_url_regex_selects_matching() {
914 let url = spawn_cdp_mock(vec![
915 json!({"targetId":"a","type":"page","url":"https://example.com/"}),
916 json!({"targetId":"b","type":"page","url":"https://other.test/"}),
917 ])
918 .await;
919 let s = PageSession::attach(&url, Engine::Cdp, Some(r"other"))
920 .await
921 .unwrap();
922 match s {
923 PageSession::Cdp(p) => assert_eq!(p.target_id, "b"),
924 _ => panic!("expected CDP"),
925 }
926 }
927
928 #[tokio::test]
929 async fn attach_cdp_url_regex_no_match_errors() {
930 let url = spawn_cdp_mock(vec![
931 json!({"targetId":"a","type":"page","url":"https://example.com/"}),
932 ])
933 .await;
934 let err = match PageSession::attach(&url, Engine::Cdp, Some("nomatch")).await {
935 Ok(_) => panic!("expected error"),
936 Err(e) => e,
937 };
938 assert!(err.to_string().contains("no CDP page target matched"));
939 }
940
941 #[tokio::test]
942 async fn evaluate_round_trip_cdp() {
943 let url = spawn_cdp_mock(vec![
944 json!({"targetId":"a","type":"page","url":"https://example.com/"}),
945 ])
946 .await;
947 let s = PageSession::attach(&url, Engine::Cdp, None).await.unwrap();
948 let v = s.evaluate("1+1", false).await.unwrap();
949 assert_eq!(v, json!("ok"));
950 s.close().await;
951 }
952
953 #[tokio::test]
954 async fn screenshot_round_trip_cdp() {
955 let url = spawn_cdp_mock(vec![
956 json!({"targetId":"a","type":"page","url":"https://example.com/"}),
957 ])
958 .await;
959 let s = PageSession::attach(&url, Engine::Cdp, None).await.unwrap();
960 let b64 = s.screenshot(false).await.unwrap();
961 assert_eq!(b64, "PNGDATA");
962 s.close().await;
963 }
964
965 async fn spawn_cdp_mock_eval_hangs(targets: Vec<Value>) -> String {
969 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
970 let addr = listener.local_addr().unwrap();
971 tokio::spawn(async move {
972 let (stream, _) = listener.accept().await.unwrap();
973 let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap();
974 while let Some(Ok(Message::Text(t))) = ws.next().await {
975 let req: Value = serde_json::from_str(&t).unwrap();
976 let id = req["id"].as_u64().unwrap();
977 let method = req["method"].as_str().unwrap_or("");
978 if method == "Runtime.evaluate" {
979 continue;
981 }
982 let result = match method {
983 "Target.getTargets" => json!({"targetInfos": targets.clone()}),
984 "Target.attachToTarget" => json!({"sessionId": "S1"}),
985 _ => json!({}),
986 };
987 let resp = json!({"id": id, "result": result});
988 ws.send(Message::Text(resp.to_string())).await.unwrap();
989 }
990 });
991 format!("ws://{addr}")
992 }
993
994 #[tokio::test]
997 async fn evaluate_with_timeout_returns_tab_hung_on_no_reply() {
998 let url = spawn_cdp_mock_eval_hangs(vec![
999 json!({"targetId":"iLO","type":"page","url":"https://192.168.2.28/"}),
1000 ])
1001 .await;
1002 let s = PageSession::attach(&url, Engine::Cdp, None).await.unwrap();
1003 let start = std::time::Instant::now();
1004 let err = s
1005 .evaluate_with_timeout("1+1", false, Some(Duration::from_millis(300)))
1006 .await
1007 .expect_err("must return TabHung");
1008 let elapsed = start.elapsed();
1009 assert!(
1010 elapsed < Duration::from_secs(1),
1011 "did not honour 300ms bound, took {elapsed:?}"
1012 );
1013 let downcast = err.downcast_ref::<SessionError>().expect("typed error");
1014 match downcast {
1015 SessionError::TabHung {
1016 target_id,
1017 timeout_ms,
1018 hint,
1019 ..
1020 } => {
1021 assert_eq!(target_id.as_deref(), Some("iLO"));
1022 assert_eq!(*timeout_ms, 300);
1023 assert_eq!(*hint, "op-timeout");
1024 }
1025 other => panic!("expected TabHung, got {other:?}"),
1026 }
1027 s.close().await;
1028 }
1029
1030 #[tokio::test]
1036 async fn stuck_eval_does_not_block_sibling_session() {
1037 let bad = spawn_cdp_mock_eval_hangs(vec![
1038 json!({"targetId":"BAD","type":"page","url":"https://192.168.2.28/"}),
1039 ])
1040 .await;
1041 let good = spawn_cdp_mock(vec![
1042 json!({"targetId":"GOOD","type":"page","url":"https://example.com/"}),
1043 ])
1044 .await;
1045
1046 let s_bad = PageSession::attach(&bad, Engine::Cdp, None).await.unwrap();
1047 let s_good = PageSession::attach(&good, Engine::Cdp, None).await.unwrap();
1048
1049 let bad_fut = s_bad.evaluate_with_timeout("1+1", false, Some(Duration::from_millis(200)));
1052 let good_fut = s_good.evaluate_with_timeout("1+1", false, Some(Duration::from_secs(5)));
1053 let (bad_res, good_res) = tokio::join!(bad_fut, good_fut);
1054
1055 assert!(bad_res.is_err(), "bad session must surface TabHung");
1056 assert_eq!(good_res.unwrap(), json!("ok"));
1057
1058 s_bad.close().await;
1059 s_good.close().await;
1060 }
1061
1062 async fn spawn_cdp_mock_per_target_eval(
1067 targets: Vec<Value>,
1068 wedged_targets: Vec<&'static str>,
1069 ) -> String {
1070 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
1071 let addr = listener.local_addr().unwrap();
1072 tokio::spawn(async move {
1073 let (stream, _) = listener.accept().await.unwrap();
1074 let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap();
1075 let mut session_wedge: std::collections::HashMap<String, bool> =
1077 std::collections::HashMap::new();
1078 let mut next_session: u32 = 0;
1079 while let Some(Ok(Message::Text(t))) = ws.next().await {
1080 let req: Value = serde_json::from_str(&t).unwrap();
1081 let id = req["id"].as_u64().unwrap();
1082 let method = req["method"].as_str().unwrap_or("");
1083 if method == "Runtime.evaluate" {
1084 if let Some(sid) = req.get("sessionId").and_then(|v| v.as_str()) {
1085 if session_wedge.get(sid).copied().unwrap_or(false) {
1086 continue;
1088 }
1089 }
1090 }
1091 let result = match method {
1092 "Target.getTargets" => json!({"targetInfos": targets.clone()}),
1093 "Target.attachToTarget" => {
1094 let target_id = req
1095 .get("params")
1096 .and_then(|p| p.get("targetId"))
1097 .and_then(|v| v.as_str())
1098 .unwrap_or("")
1099 .to_string();
1100 next_session += 1;
1101 let sid = format!("S{next_session}");
1102 let wedge = wedged_targets.iter().any(|w| *w == target_id);
1103 session_wedge.insert(sid.clone(), wedge);
1104 json!({"sessionId": sid})
1105 }
1106 "Target.detachFromTarget" => json!({}),
1107 "Runtime.evaluate" => json!({"result": {"value": "ok"}}),
1108 _ => json!({}),
1109 };
1110 let resp = json!({"id": id, "result": result});
1111 ws.send(Message::Text(resp.to_string())).await.unwrap();
1112 }
1113 });
1114 format!("ws://{addr}")
1115 }
1116
1117 #[tokio::test]
1120 async fn pick_cdp_iterates_past_hung_match() {
1121 let url = spawn_cdp_mock_per_target_eval(
1122 vec![
1123 json!({"targetId":"DEAD","type":"page","url":"https://twitch.tv/gametechnology"}),
1124 json!({"targetId":"LIVE","type":"page","url":"https://gametechnology.somewhere.com"}),
1125 ],
1126 vec!["DEAD"],
1127 )
1128 .await;
1129 let s = PageSession::attach(&url, Engine::Cdp, Some(r"gametechnology"))
1130 .await
1131 .expect("must iterate past the wedged tab and pick LIVE");
1132 match s {
1133 PageSession::Cdp(p) => assert_eq!(p.target_id, "LIVE"),
1134 _ => panic!("expected CDP"),
1135 }
1136 }
1137
1138 #[tokio::test]
1142 async fn pick_cdp_all_matches_hung_returns_tab_hung() {
1143 let url = spawn_cdp_mock_per_target_eval(
1144 vec![
1145 json!({"targetId":"A","type":"page","url":"https://example.com/foo"}),
1146 json!({"targetId":"B","type":"page","url":"https://example.com/bar"}),
1147 ],
1148 vec!["A", "B"],
1149 )
1150 .await;
1151 let start = std::time::Instant::now();
1152 let err = match PageSession::attach(&url, Engine::Cdp, Some(r"example\.com")).await {
1153 Ok(_) => panic!("all matches wedged → must error"),
1154 Err(e) => e,
1155 };
1156 let elapsed = start.elapsed();
1157 assert!(
1158 elapsed < PICK_PROBE_TIMEOUT * 2 + Duration::from_millis(500),
1159 "took too long: {elapsed:?}"
1160 );
1161 let typed = err.downcast_ref::<SessionError>().expect("typed error");
1162 match typed {
1163 SessionError::TabHung { hint, .. } => {
1164 assert_eq!(*hint, "all-matches-hung");
1165 }
1166 other => panic!("expected TabHung, got {other:?}"),
1167 }
1168 let text = format!("{err:#}");
1169 assert!(
1170 text.contains("URL regex matched 2 page(s)"),
1171 "context missing count: {text}"
1172 );
1173 }
1174
1175 #[tokio::test]
1178 async fn pick_cdp_single_healthy_match_is_picked() {
1179 let url = spawn_cdp_mock_per_target_eval(
1180 vec![json!({"targetId":"OK","type":"page","url":"https://example.com/x"})],
1181 vec![],
1182 )
1183 .await;
1184 let s = PageSession::attach(&url, Engine::Cdp, Some(r"example"))
1185 .await
1186 .unwrap();
1187 match s {
1188 PageSession::Cdp(p) => assert_eq!(p.target_id, "OK"),
1189 _ => panic!("expected CDP"),
1190 }
1191 }
1192}