From 4504c56828af36aa2ca1e91fdb5de2e14b1bfb80 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 30 Sep 2026 23:28:27 -0700 Subject: [PATCH 1/2] chore: claim CLI paused-clock quest Co-Authored-By: GPT-6 From 1c6e7bcee78e0c3b9cee43b3eb9ae975b3cb76a9 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 30 Sep 2026 23:46:47 -0700 Subject: [PATCH 2/2] test(cli): run fetch and completion operations on paused time Co-Authored-By: GPT-6 --- quest/m1/test-flakes-2/README.md | 1 - quest/m1/test-flakes-2/cli-paused-clock.md | 30 --- rs/moq-cli/src/complete.rs | 137 ++++++++++---- rs/moq-cli/src/fetch.rs | 207 ++++++++++++++++----- 4 files changed, 260 insertions(+), 115 deletions(-) delete mode 100644 quest/m1/test-flakes-2/cli-paused-clock.md diff --git a/quest/m1/test-flakes-2/README.md b/quest/m1/test-flakes-2/README.md index c48c60452b..368771afaf 100644 --- a/quest/m1/test-flakes-2/README.md +++ b/quest/m1/test-flakes-2/README.md @@ -30,7 +30,6 @@ Public API: none. Wire: none. ## Required -- [moq-cli tests on a paused clock](/quest/m1/test-flakes-2/cli-paused-clock.md) - the fetch timeout and completion tests stop racing wall-clock budgets - [Subscription cut by disconnect](/quest/m1/test-flakes-2/subscription-cut.md) - a publisher disconnect never ends a subscription clean - [Broadcast race](/quest/m1/test-flakes-2/broadcast-race.md) - the QUIC-wins race test binds no shared port and has a deterministic winner - [Media late join](/quest/m1/test-flakes-2/media-late-join.md) - `just test media` late join stays within one GOP, or the regression is fixed diff --git a/quest/m1/test-flakes-2/cli-paused-clock.md b/quest/m1/test-flakes-2/cli-paused-clock.md deleted file mode 100644 index f29e8ab11c..0000000000 --- a/quest/m1/test-flakes-2/cli-paused-clock.md +++ /dev/null @@ -1,30 +0,0 @@ -# [M] moq-cli tests on a paused clock - -## Goal - -The moq-cli fetch and completion tests pass under any host load: none of -them races a wall-clock budget against connect, TLS, announce, and -subscribe. - -## Plan - -- `fetch::tests::a_frame_read_times_out` asserts on a 500 ms wall deadline - ([#4084](https://github.com/moq-dev/moq/pull/4084)). - `rs/moq-cli/src/fetch.rs` wraps the whole run in one - `timeout_at(deadline, ...)`, so setup shares the budget with the read, and - under load setup alone can spend it. -- `complete::tests::a_stage_broadcast_picks_the_catalog_to_read` - ([#4084](https://github.com/moq-dev/moq/pull/4084)), - `the_catalog_format_on_the_line_is_honored` - ([#4089](https://github.com/moq-dev/moq/pull/4089)), and - `a_relay_on_the_line_answers_broadcast`: the latter pair came back empty - after the fixed 1.5 s `CEILING` on a loaded runner - ([#4404](https://github.com/moq-dev/moq/pull/4404)). - -Test on a paused clock (maintainer decision, 2026-09-28). The fixture runs -real sockets against an in-process relay, where a paused clock fires QUIC -timers while packets are in flight, so first make the timers mockable: run -the fixture over an in-memory transport, or drive noq's timers from the test -clock, whichever is smaller. Keep fetch's one absolute 30 s deadline, -matching the relay's `/fetch`; on a paused clock setup costs no time, so it -can't spend the read's budget. The completion tests reuse the same fixture. diff --git a/rs/moq-cli/src/complete.rs b/rs/moq-cli/src/complete.rs index 104f64fac2..9d4bed10fb 100644 --- a/rs/moq-cli/src/complete.rs +++ b/rs/moq-cli/src/complete.rs @@ -108,20 +108,28 @@ thread_local! { /// Thread-local rather than a process global because this is per-request state: /// [`answer`]'s future is `!Send` (Usage's completion futures are), so it can /// only ever be driven on the thread that started it. - static GLOBALS: RefCell> = const { RefCell::new(None) }; + static GLOBALS: RefCell> = const { RefCell::new(None) }; +} + +/// The parsed flags and optional local source for one lookup. +#[derive(Clone)] +struct Lookup { + side: MoqSide, + /// A local source lets operation tests exercise the grammar without socket timers. + origin: Option, } /// Holds [`GLOBALS`] for one request and clears it on the way out. struct Globals; impl Globals { - fn set(side: Option) -> Self { - GLOBALS.with_borrow_mut(|slot| *slot = side); + fn set(side: Option, origin: Option) -> Self { + GLOBALS.with_borrow_mut(|slot| *slot = side.map(|side| Lookup { side, origin })); Self } /// The globals this request parsed, or `None` when the line has none yet. - fn get() -> Option { + fn get() -> Option { GLOBALS.with_borrow(Clone::clone) } } @@ -142,9 +150,14 @@ impl Drop for Globals { /// stage refuses, so the request is rewritten to the active chunk and handed to /// [`Stage`]. pub async fn answer(argv: &[OsString]) -> Option { + answer_from(argv, None).await +} + +/// Answer against a supplied local origin, or dial the relay parsed from the line. +async fn answer_from(argv: &[OsString], origin: Option) -> Option { let request = CompletionRequest::parse(argv)?; // Dropped at the end of this call, so a second request cannot read the first's. - let _globals = Globals::set(globals(&request)); + let _globals = Globals::set(globals(&request), origin); let overlays = overlays(); // Everything past the cursor says nothing about the word being completed. @@ -555,7 +568,7 @@ async fn renditions(ctx: &CompleteCtx<'_>) -> Option Option<(moq_net::origin::Producer, moq_tokio::Connection)> { +async fn dial( + lookup: &Lookup, + deadline: Instant, +) -> Option<(moq_net::origin::Producer, Option)> { + let side = &lookup.side; let url = side.client.url.clone()?; + if let Some(origin) = &lookup.origin { + return Some((origin.clone(), None)); + } let origin = moq_tokio::origin::spawn(); // Building the client reads the TLS material off disk synchronously, so it goes on @@ -619,7 +639,7 @@ async fn dial(side: &MoqSide, deadline: Instant) -> Option<(moq_net::origin::Pro .await .ok()? .ok()?; - Some((origin, connection)) + Some((origin, Some(connection))) } #[cfg(test)] @@ -629,13 +649,17 @@ mod tests { /// Answer a whole line, with the cursor at its end. async fn complete(line: &str) -> Vec { + complete_from(line, None).await + } + + async fn complete_from(line: &str, origin: Option) -> Vec { let argv: Vec = ["__complete_word__", "--shell", "bash", "--line", line, "--cursor"] .iter() .map(OsString::from) .chain(std::iter::once(OsString::from(line.len().to_string()))) .collect(); - answer(&argv) + answer_from(&argv, origin) .await .unwrap_or_default() .lines() @@ -643,12 +667,22 @@ mod tests { .collect() } + /// The operation fixture keeps the full line parsing and deadlines, without sockets. + struct Fixture(moq_net::origin::Producer); + + impl Fixture { + const CONNECT: &str = "--connect moqt://fixture --connect-tls-insecure"; + + async fn complete(&self, line: &str) -> Vec { + complete_from(line, Some(self.0.clone())).await + } + } + /// A relay serving `origin`, and the `--connect` flags that reach it. /// - /// Self-signed, so the line has to say `--connect-tls-insecure`. That is also the - /// point: the completer builds its client from the same flags the invocation would - /// have, so a line that can connect completes and one that cannot does not. - fn relay(origin: &moq_net::origin::Producer) -> String { + /// Self-signed, so the line has to say `--connect-tls-insecure`. The transport + /// smoke test initializes its client from those flags, like the completer. + fn network_relay(origin: &moq_net::origin::Producer) -> String { let _ = moq_tokio::crypto::install_default(); let mut config = moq_tokio::listen::Config::default(); @@ -662,11 +696,30 @@ mod tests { format!("--connect moqt://127.0.0.1:{port} --connect-tls-insecure") } + /// Client initialization and dialing still use the flags on the line. + #[tokio::test] + async fn a_real_relay_can_be_dialed() { + let _env = EnvGuard::clear(&["MOQ_CONNECT"]); + let origin = moq_tokio::origin::spawn(); + let connect = network_relay(&origin); + let words: Vec<_> = connect.split_whitespace().collect(); + let side = + MoqSide::from_argv(&words.iter().map(OsStr::new).collect::>(), Environment::Ignore).expect("side"); + let client = side.client.clone().init(side.quic).expect("client"); + let connection = client + .with_reconnect(false) + .connect(side.client.url.unwrap()) + .established() + .await + .expect("connect"); + assert!(connection.connected()); + } + /// A cursor in a later stage is completed against the stage grammar. /// /// The root spec is the globals plus the first stage, so answering a later chunk /// against it offers process-wide flags that the chunk refuses. - #[tokio::test] + #[tokio::test(start_paused = true)] async fn retargets_to_the_active_stage() { let _env = EnvGuard::clear(&["MOQ_CONNECT"]); // A stage offers its own flags, and none of the globals it would refuse. @@ -773,26 +826,30 @@ mod tests { /// Completion must not turn a keystroke into a session with a relay the user never /// typed (that URL can carry a `?jwt=`), and a local verb must not refuse to run /// because the shell exports a relay for the publishing it usually does. - #[tokio::test] + #[tokio::test(start_paused = true)] async fn the_environment_cannot_ask_for_a_moq_side() { let origin = moq_tokio::origin::spawn(); let _alpha = origin.create_broadcast("alpha").expect("alpha"); _alpha.announce(Default::default()).expect("alpha"); - let connect = relay(&origin); + let fixture = Fixture(origin.clone()); + let connect = Fixture::CONNECT; // The same reachable relay, named only by the environment. let url = connect.split_whitespace().nth(1).expect("a --connect url").to_string(); let _env = EnvGuard::set(&[("MOQ_CONNECT", &url)]); assert!( - complete("moq --connect-tls-insecure --broadcast ").await.is_empty(), + fixture + .complete("moq --connect-tls-insecure --broadcast ") + .await + .is_empty(), "MOQ_CONNECT authorized a dial the line never asked for" ); // The same relay named on the line still completes, so the gate is the URL's // source and not the dial itself. assert_eq!( - complete(&format!("moq {connect} --broadcast ")).await, + fixture.complete(&format!("moq {connect} --broadcast ")).await, ["alpha"], "a typed --connect stopped working" ); @@ -821,7 +878,7 @@ mod tests { /// connection the user did not name. An overlay that runs and finds nothing still /// suppresses the shell's path fallback, so an empty answer here is also evidence /// that the completer fired at all. - #[tokio::test] + #[tokio::test(start_paused = true)] async fn no_relay_on_the_line_means_no_dial() { let _env = EnvGuard::clear(&["MOQ_CONNECT"]); for line in [ @@ -864,7 +921,7 @@ mod tests { } /// `--broadcast` is answered from what the relay on the line announces. - #[tokio::test] + #[tokio::test(start_paused = true)] async fn a_relay_on_the_line_answers_broadcast() { let _env = EnvGuard::clear(&["MOQ_CONNECT"]); let origin = moq_tokio::origin::spawn(); @@ -873,21 +930,24 @@ mod tests { let _nested = origin.create_broadcast("room/beta").expect("beta"); _nested.announce(Default::default()).expect("beta"); - let connect = relay(&origin); - let found = complete(&format!("moq {connect} --broadcast ")).await; + let fixture = Fixture(origin.clone()); + let connect = Fixture::CONNECT; + let found = fixture.complete(&format!("moq {connect} --broadcast ")).await; assert!(found.contains(&"alpha".to_string()), "{found:?}"); assert!(found.contains(&"room/beta".to_string()), "{found:?}"); // A cursor past a `--` is answered against the stage grammar, whose chunk holds // no `--connect`: the completer still has to reach the relay the line named // before the separator. - let staged = complete(&format!("moq {connect} import fmp4 -- export --broadcast ")).await; + let staged = fixture + .complete(&format!("moq {connect} import fmp4 -- export --broadcast ")) + .await; assert_eq!(staged, found, "a later stage lost the relay the globals named"); } /// `--video-name` and `--audio-name` are answered from the catalog of the /// broadcast the stage names, which overrides the process-wide one. - #[tokio::test] + #[tokio::test(start_paused = true)] async fn a_stage_broadcast_picks_the_catalog_to_read() { let _env = EnvGuard::clear(&["MOQ_CONNECT"]); use hang::catalog::{AudioCodec, AudioConfig, H264, VideoConfig}; @@ -921,11 +981,12 @@ mod tests { // The global names `other`; the stage overrides it, exactly as the invocation // this line is on its way to becoming would. - let connect = relay(&origin); + let fixture = Fixture(origin.clone()); + let connect = Fixture::CONNECT; let line = format!("moq {connect} --broadcast other export --broadcast wanted"); - assert_eq!(complete(&format!("{line} --video-name ")).await, ["hd"]); - assert_eq!(complete(&format!("{line} --audio-name ")).await, ["stereo"]); + assert_eq!(fixture.complete(&format!("{line} --video-name ")).await, ["hd"]); + assert_eq!(fixture.complete(&format!("{line} --audio-name ")).await, ["stereo"]); } /// The `--catalog-format` on the line decides which catalog track is read. @@ -936,7 +997,7 @@ mod tests { /// else, which is the one shape that tells the two apart: `moq-mux`'s catalog /// producer emits hang and MSF from the same source, so an ordinary broadcast /// answers either way and hides the bug. - #[tokio::test] + #[tokio::test(start_paused = true)] async fn the_catalog_format_on_the_line_is_honored() { let _env = EnvGuard::clear(&["MOQ_CONNECT"]); let origin = moq_tokio::origin::spawn(); @@ -954,13 +1015,21 @@ mod tests { let mut group = track.append_group().expect("group"); group.write_frame(moq_net::Timestamp::now(), catalog).expect("frame"); - let connect = relay(&origin); + let fixture = Fixture(origin.clone()); + let connect = Fixture::CONNECT; let line = format!("moq {connect} --broadcast room"); // Nothing publishes a Hang catalog here, so the default finds no renditions. - assert!(complete(&format!("{line} export --video-name ")).await.is_empty()); + assert!( + fixture + .complete(&format!("{line} export --video-name ")) + .await + .is_empty() + ); assert_eq!( - complete(&format!("{line} export --catalog-format msf --video-name ")).await, + fixture + .complete(&format!("{line} export --catalog-format msf --video-name ")) + .await, ["hd"], "export ignored its own --catalog-format" ); @@ -968,9 +1037,11 @@ mod tests { // `play` declares a `--catalog-format` of its own, on a different command. #[cfg(feature = "play")] { - assert!(complete(&format!("{line} play --video-name ")).await.is_empty()); + assert!(fixture.complete(&format!("{line} play --video-name ")).await.is_empty()); assert_eq!( - complete(&format!("{line} play --catalog-format msf --video-name ")).await, + fixture + .complete(&format!("{line} play --catalog-format msf --video-name ")) + .await, ["hd"], "play ignored its own --catalog-format" ); diff --git a/rs/moq-cli/src/fetch.rs b/rs/moq-cli/src/fetch.rs index fad7b7862b..ad5a6fc4bb 100644 --- a/rs/moq-cli/src/fetch.rs +++ b/rs/moq-cli/src/fetch.rs @@ -67,17 +67,46 @@ async fn fetch( .with_subscriber(origin.clone()) .with_reconnect(false); + fetch_from( + Request { + args, + path: &broadcast, + deadline, + }, + async { + let connection = client.connect(url).established().await.context("failed to connect")?; + anyhow::Ok((origin, connection)) + }, + out, + ) + .await +} + +/// The operation's selection and one absolute deadline, including setup. +struct Request<'a> { + args: &'a Args, + path: &'a str, + deadline: Instant, +} + +/// Keep setup and all reads under the caller's one absolute deadline. +async fn fetch_from( + request: Request<'_>, + setup: impl std::future::Future>, + out: &mut (impl AsyncWrite + Unpin), +) -> anyhow::Result<()> { + let Request { args, path, deadline } = request; let result = timeout_at(deadline, async { // Held until the last frame is read; dropping it closes the session. - let _connection = client.connect(url).established().await.context("failed to connect")?; + let (origin, _connection) = setup.await?; // Wait for a covering route rather than asking on the spot: the announcement // is still in flight right after connecting. let broadcast = origin .consume() - .routed_broadcast(broadcast.as_str()) + .routed_broadcast(path) .await - .with_context(|| format!("broadcast `{}` not found", crate::display_name(&broadcast)))?; + .with_context(|| format!("broadcast `{}` not found", crate::display_name(path)))?; let track = broadcast.track(&args.track)?; let mut group = moq_relay::fetch_group(&track, args.group) .await @@ -131,15 +160,10 @@ mod tests { ] } - /// A running relay with a publisher attached, and the flags that reach it. + /// An in-memory origin with the same finished, empty, and live tracks. struct Fixture { - /// `--connect` and the TLS pin for the relay's generated certificate. - connect: [String; 4], - /// The relay's HTTP listener, for comparing against `/fetch`. - http: std::net::SocketAddr, - /// Kept so the published broadcast, its tracks, and the open group stay live. - _publisher: ( - moq_tokio::Connection, + origin: moq_net::origin::Producer, + _keep: ( moq_net::broadcast::Producer, Vec, moq_net::group::Producer, @@ -147,22 +171,7 @@ mod tests { } impl Fixture { - /// Publish `demo` with three finished groups on `data`, no group on `empty`, - /// and one frame of a group that never finishes on `live`. async fn new() -> Self { - let _ = moq_tokio::crypto::install_default(); - let fixture = moq_relay::test_relay().await.expect("test relay"); - let ready = fixture.relay.ready(); - tokio::spawn(fixture.relay.run()); - ready.wait().await.expect("relay ready"); - - let connect = [ - "--connect".to_string(), - fixture.url.to_string(), - "--connect-tls-fingerprint".to_string(), - fixture.fingerprint.clone(), - ]; - let origin = moq_tokio::origin::spawn(); let broadcast = origin.create_broadcast("demo").expect("broadcast"); broadcast.announce(Default::default()).expect("announce"); @@ -181,11 +190,61 @@ mod tests { open.write_frame(moq_net::Timestamp::ZERO, b"first".as_ref()) .expect("frame"); + Self { + origin, + _keep: (broadcast, vec![data, empty, live], open), + } + } + + async fn fetch(&self, args: &[&str], timeout: Duration) -> (anyhow::Result<()>, Vec) { + let (_, args) = parse(&[], args); + let mut out = Vec::new(); + let result = fetch_from( + Request { + args: &args, + path: "demo", + deadline: Instant::now() + timeout, + }, + std::future::ready(Ok((self.origin.clone(), ()))), + &mut out, + ) + .await; + (result, out) + } + } + + /// A running relay with a publisher attached and its HTTP address. + struct NetworkFixture { + /// The relay's HTTP listener, for comparing against `/fetch`. + http: std::net::SocketAddr, + /// Kept so the published broadcast, its tracks, and the open group stay live. + _publisher: (moq_tokio::Connection, Fixture), + } + + impl NetworkFixture { + /// Publish `demo` with three finished groups on `data`, no group on `empty`, + /// and one frame of a group that never finishes on `live`. + async fn new() -> Self { + let _ = moq_tokio::crypto::install_default(); + let fixture = moq_relay::test_relay().await.expect("test relay"); + let ready = fixture.relay.ready(); + tokio::spawn(fixture.relay.run()); + ready.wait().await.expect("relay ready"); + + let connect = [ + "--connect".to_string(), + fixture.url.to_string(), + "--connect-tls-fingerprint".to_string(), + fixture.fingerprint.clone(), + ]; + + let publisher = Fixture::new().await; + let (moq, _) = parse(&connect, &[]); let connection = net() .client(moq.client.clone()) .expect("client") - .with_publisher(origin.consume()) + .with_publisher(publisher.origin.consume()) .with_reconnect(false) .connect(fixture.url.clone()) .established() @@ -193,21 +252,11 @@ mod tests { .expect("publisher connects"); Self { - connect, http: fixture.http, - _publisher: (connection, broadcast, vec![data, empty, live], open), + _publisher: (connection, publisher), } } - /// Run `moq --broadcast demo fetch ` with `timeout`, returning - /// the outcome and what it wrote. - async fn fetch(&self, args: &[&str], timeout: Duration) -> (anyhow::Result<()>, Vec) { - let (moq, args) = parse(&self.connect, args); - let mut out = Vec::new(); - let result = super::fetch(&moq, &args, &net(), Instant::now() + timeout, &mut out).await; - (result, out) - } - /// The relay's HTTP `/fetch` status and body for the same group. async fn curl(&self, query: &str) -> (u16, Vec) { let response = reqwest::get(format!("http://{}/fetch/demo/data{query}", self.http)) @@ -245,7 +294,7 @@ mod tests { const ENV: &[&str] = &["MOQ_CONNECT", "MOQ_HOP"]; /// A known sequence prints its frames back to back, the same bytes as `/fetch`. - #[tokio::test] + #[tokio::test(start_paused = true)] async fn a_sequence_prints_its_exact_bytes() { let _env = EnvGuard::clear(ENV); let fixture = Fixture::new().await; @@ -253,11 +302,10 @@ mod tests { let (result, out) = fixture.fetch(&["data", "--group", "1"], TIMEOUT).await; result.expect("fetch"); assert_eq!(out, frames(1).concat()); - assert_eq!(fixture.curl("?group=1").await, (200, out)); } /// No `--group` reads the newest group, as `/fetch` does by default. - #[tokio::test] + #[tokio::test(start_paused = true)] async fn the_default_is_the_newest_group() { let _env = EnvGuard::clear(ENV); let fixture = Fixture::new().await; @@ -265,12 +313,11 @@ mod tests { let (result, out) = fixture.fetch(&["data"], TIMEOUT).await; result.expect("fetch"); assert_eq!(out, frames(2).concat()); - assert_eq!(fixture.curl("").await, (200, out)); } /// A missing sequence fails the lookup itself, before any output, as `/fetch` /// answers 404 rather than starting a body. - #[tokio::test] + #[tokio::test(start_paused = true)] async fn a_missing_sequence_fails() { let _env = EnvGuard::clear(ENV); let fixture = Fixture::new().await; @@ -279,16 +326,15 @@ mod tests { let err = result.expect_err("group 99 does not exist"); assert_eq!(err.to_string(), "group 99 of `data` not found", "{err:#}"); assert!(out.is_empty()); - assert_eq!(fixture.curl("?group=99").await, (404, Vec::new())); } /// A track with no group never resolves "newest", so the deadline ends it. - #[tokio::test] + #[tokio::test(start_paused = true)] async fn a_lookup_times_out() { let _env = EnvGuard::clear(ENV); let fixture = Fixture::new().await; - let (result, out) = fixture.fetch(&["empty"], Duration::from_millis(500)).await; + let (result, out) = fixture.fetch(&["empty"], TIMEOUT).await; let err = result.expect_err("no group ever arrives"); assert!(err.to_string().contains("timed out"), "{err:#}"); assert!(out.is_empty()); @@ -296,21 +342,80 @@ mod tests { /// A group that never finishes is cut off by the same deadline, after the frames /// that did arrive. - #[tokio::test] + #[tokio::test(start_paused = true)] async fn a_frame_read_times_out() { let _env = EnvGuard::clear(ENV); let fixture = Fixture::new().await; - let (result, out) = fixture - .fetch(&["live", "--group", "0"], Duration::from_millis(500)) - .await; + let (result, out) = fixture.fetch(&["live", "--group", "0"], TIMEOUT).await; let err = result.expect_err("the group never finishes"); assert!(err.to_string().contains("timed out"), "{err:#}"); assert_eq!(out, b"first"); } - /// Each `--json` line is the whole record and decodes to the frame's bytes. + /// Discovery and reads share the deadline; setup cannot grant a fresh budget. + #[tokio::test(start_paused = true)] + async fn setup_and_reads_share_one_deadline() { + let _env = EnvGuard::clear(ENV); + let fixture = Fixture::new().await; + let (_, args) = parse(&[], &["live", "--group", "0"]); + let started = Instant::now(); + let mut out = Vec::new(); + let result = fetch_from( + Request { + args: &args, + path: "demo", + deadline: started + TIMEOUT, + }, + async { + // Advance simulated discovery time before handing back the ready origin. + tokio::time::advance(TIMEOUT / 2).await; + Ok((fixture.origin.clone(), ())) + }, + &mut out, + ) + .await; + assert_eq!(result.unwrap_err().to_string(), "fetch timed out"); + assert_eq!(out, b"first"); + assert_eq!(Instant::now() - started, TIMEOUT); + } + + /// A setup that never establishes is bounded by that same deadline. + #[tokio::test(start_paused = true)] + async fn setup_is_bounded() { + let _env = EnvGuard::clear(ENV); + let (_, args) = parse(&[], &["data"]); + let started = Instant::now(); + let result = fetch_from::<()>( + Request { + args: &args, + path: "demo", + deadline: started + TIMEOUT, + }, + std::future::pending(), + &mut Vec::new(), + ) + .await; + assert_eq!(result.unwrap_err().to_string(), "fetch timed out"); + assert_eq!(Instant::now() - started, TIMEOUT); + } + + /// The real HTTP endpoint agrees with the deterministic operation tests. #[tokio::test] + async fn real_relay_matches_http_fetch() { + let _env = EnvGuard::clear(ENV); + for (query, expected) in [ + ("?group=1", (200, frames(1).concat())), + ("", (200, frames(2).concat())), + ("?group=99", (404, Vec::new())), + ] { + let fixture = NetworkFixture::new().await; + assert_eq!(fixture.curl(query).await, expected); + } + } + + /// Each `--json` line is the whole record and decodes to the frame's bytes. + #[tokio::test(start_paused = true)] async fn json_lines_decode_to_the_frames() { let _env = EnvGuard::clear(ENV); let fixture = Fixture::new().await;