open-standards/joule-code/joule-code-rs/crates/joule-moq-interop/tests/moq_test_relay.rs

//! `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<String>) -> (i32, String) {
    tokio::task::spawn_blocking(move || client(&url, &args.iter().map(String::as_str).collect::<Vec<_>>())).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}");
        }
    }
}