//! `moq-test` (the test tracks of `draft-afrind-moq-test-01`) through `joule-moq-relay`, on //! loopback: the interop client's publisher and subscriber at each draft the relay speaks, //! by PUBLISH_NAMESPACE and by PUBLISH, over native QUIC and WebTransport, and across //! drafts; then each planted fault, which the subscriber must refuse with its own reason. //! //! The relay runs in this process on a port the system chooses; the client is the built //! binary, run once a session pair, under a guard that kills and reaps it. use joule_moq_relay::{RelayConfig, RelayServer}; use serde_json::Value; use std::io::Read; use std::process::{Child, Command, Stdio}; use std::time::{Duration, Instant}; /// A client run that cannot outlive its test: killed and reaped when dropped. struct Guard(Child); impl Drop for Guard { fn drop(&mut self) { let _ = self.0.kill(); let _ = self.0.wait(); } } /// Run the client's `moq-test` against `url`; its exit status and its TAP. fn client(url: &str, args: &[&str]) -> (i32, String) { let mut cmd = Command::new(env!("CARGO_BIN_EXE_joule-moq-interop")); cmd.args(["--relay", url, "--tls-disable-verify", "--test", "moq-test"]).args(args).stdout(Stdio::piped()).stderr(Stdio::null()); let mut g = Guard(cmd.spawn().expect("the client starts")); let mut stdout = g.0.stdout.take().expect("stdout"); let reader = std::thread::spawn(move || { let mut s = String::new(); let _ = stdout.read_to_string(&mut s); s }); let deadline = Instant::now() + Duration::from_secs(120); let status = loop { match g.0.try_wait().expect("the client's status") { Some(s) => break s.code().unwrap_or(-1), None if Instant::now() > deadline => break -2, None => std::thread::sleep(Duration::from_millis(20)), } }; drop(g); (status, reader.join().unwrap_or_default()) } /// A YAML line of the test point's diagnostics, as JSON. fn diag(tap: &str, key: &str) -> Value { let prefix = format!(" {key}: "); tap.lines().find_map(|l| l.strip_prefix(&prefix)).and_then(|v| serde_json::from_str(v).ok()).unwrap_or(Value::Null) } async fn relay() -> (RelayServer, String) { let server = RelayServer::start(RelayConfig::loopback()).await.expect("the relay starts"); let addr = server.quic_addr.expect("a UDP endpoint"); let path = server.relay().config().path.clone(); (server, format!("{addr}{path}")) } async fn run(url: String, args: Vec) -> (i32, String) { tokio::task::spawn_blocking(move || client(&url, &args.iter().map(String::as_str).collect::>())).await.expect("the client's task") } /// Every case of a passing run: every scheduled object accepted, and the whole track served /// again by FETCH. fn whole(tap: &str, what: &str) { let cases = diag(tap, "cases"); let cases = cases.as_array().unwrap_or_else(|| panic!("{what}: no cases\n{tap}")); assert_eq!(cases.len(), 6, "{what}\n{tap}"); for c in cases { assert_eq!(c["objects_accepted"], c["objects_scheduled"], "{what}: {}\n{tap}", c["what"]); assert_eq!(c["fetch"]["objects"], c["fetch"]["payload_objects_scheduled"], "{what}: FETCH of {}\n{tap}", c["what"]); if !c["delivery_timeout_carried"].is_null() { assert_eq!(c["delivery_timeout_carried"], true, "{what}: {}\n{tap}", c["what"]); } } } #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn the_test_tracks_pass_the_relay_at_every_draft() { let (_server, at) = relay().await; for draft in ["23", "22", "18", "16"] { for flow in ["namespace", "publish"] { let what = format!("draft {draft}, flow {flow}"); let (status, tap) = run(format!("moqt://{at}"), vec!["--draft".into(), draft.into(), "--flow".into(), flow.into()]).await; assert_eq!(status, 0, "{what}\n{tap}"); assert!(tap.contains("\nok 1 - moq-test\n"), "{what}\n{tap}"); whole(&tap, &what); if flow == "namespace" { assert_eq!(diag(&tap, "refusals").as_array().map(Vec::len), Some(3), "{what}\n{tap}"); } } } } #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn the_test_tracks_pass_over_webtransport_and_across_drafts() { let (_server, at) = relay().await; for draft in ["23", "16"] { let what = format!("draft {draft} over WebTransport"); let (status, tap) = run(format!("https://{at}"), vec!["--draft".into(), draft.into()]).await; assert_eq!(status, 0, "{what}\n{tap}"); whole(&tap, &what); } for (publisher, subscriber) in [("23", "22"), ("23", "16"), ("16", "23"), ("18", "23"), ("22", "18")] { let what = format!("publisher at {publisher}, subscriber at {subscriber}"); let (status, tap) = run(format!("moqt://{at}"), vec!["--draft".into(), publisher.into(), "--sub-draft".into(), subscriber.into()]).await; assert_eq!(status, 0, "{what}\n{tap}"); whole(&tap, &what); } } /// The planted faults: a publisher that sends a track wrongly is refused by the subscriber, /// each fault with its own reason, at the current draft and at the oldest. #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn a_track_sent_wrongly_is_refused_with_its_reason() { let (_server, at) = relay().await; for draft in ["23", "16"] { for (fault, reason) in [ ("sequence", "moq_test_object_unscheduled"), ("pattern", "moq_test_payload_pattern"), ("size", "moq_test_payload_size"), ("subgroup", "moq_test_subgroup"), ("end-of-group", "moq_test_end_of_group_missing"), ("extension", "moq_test_extension_absent"), ] { let what = format!("fault {fault} at draft {draft}"); let (status, tap) = run(format!("moqt://{at}"), vec!["--draft".into(), draft.into(), "--fault".into(), fault.into()]).await; assert_eq!(status, 1, "{what}: the client's own verdict, a test that failed\n{tap}"); assert!(tap.contains("\nnot ok 1 - moq-test\n"), "{what}\n{tap}"); assert_eq!(diag(&tap, "refused"), reason, "{what}\n{tap}"); } } }