open-standards/joule-code/joule-code-rs/crates/joule-moq-room/tests/openapi_contract.rs

//! The service's OpenAPI description (`openapi.json`) held to the code, in both directions:
//!
//! - **Names.** Every operation's `operationId` is its name in the toolchain's catalogue, the
//!   one `HANDLERS` gives it: an operation of `wai::catalogue` that runs on this service, at the
//!   route the catalogue gives it, each name once; and every operation the catalogue routes to
//!   this service is described here.
//! - **Routes.** The (method, path) pairs the routers match (`src/http.rs`,
//!   `src/messaging.rs`, `src/call.rs`) are exactly the description's operations.
//! - **Statuses, from the source.** Each operation names the functions that answer it below
//!   (`HANDLERS`). Every status the description lists for an operation is one those functions
//!   return, and every status a function returns is listed for an operation it answers (405,
//!   the answer to a method a path does not take, is stated once in the description).
//! - **Statuses, from the running service.** The service runs on loopback beside a relay, in
//!   each of its modes (a plain room, end-to-end encryption, admission by age, co-performance,
//!   a watch room, a remote desktop room, a conversation, calls, and without its relay), and
//!   every documented operation is asked at least once in the mode that serves it: every
//!   answer's status is one the description lists for that operation, each probe's the one it
//!   was written to reach, and a method a documented path does not take is answered 405 (404
//!   under `/msg` and the call paths) and never 200.

#[path = "common/rfc_test_keys.rs"]
mod rfc_test_keys;

use ed25519_dalek::SigningKey;
use joule_moq::grant_proof::{self as gp, GrantProof};
use joule_moq_room::dev::DevRoom;
use joule_moq_room::keys::Keys;
use joule_moq_room::{http, Config, Service};
use serde_json::{json, Value};
use std::collections::{BTreeMap, BTreeSet};
use std::net::SocketAddr;
use std::path::Path;
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::time::timeout;

const STEP: Duration = Duration::from_secs(20);
const HOST: &str = "room.test";

// ------------------------------------------------------------------------- the description

type Op = (String, String);

fn description() -> Value {
    let text = std::fs::read_to_string(Path::new(env!("CARGO_MANIFEST_DIR")).join("openapi.json")).expect("openapi.json");
    serde_json::from_str(&text).expect("openapi.json is JSON")
}

/// Every operation of `openapi.json` and the statuses it lists.
fn documented() -> BTreeMap<Op, BTreeSet<u16>> {
    let doc = description();
    assert_eq!(doc["openapi"].as_str().map(|v| v.starts_with("3.1")), Some(true), "an OpenAPI 3.1 document");
    let mut ops = BTreeMap::new();
    for (path, item) in doc["paths"].as_object().expect("paths") {
        for (method, o) in item.as_object().expect("a path item") {
            let statuses: BTreeSet<u16> = o["responses"].as_object().expect("responses").keys().map(|s| s.parse().expect("a numeric status")).collect();
            assert!(!statuses.is_empty(), "{method} {path} lists no status");
            ops.insert((method.to_ascii_uppercase(), path.clone()), statuses);
        }
    }
    ops
}

// ------------------------------------------------------------------------------- the source

fn source(file: &str) -> String {
    std::fs::read_to_string(Path::new(env!("CARGO_MANIFEST_DIR")).join("src").join(file)).expect("a source file")
}

/// The (method, path) pairs a router's arms match: every `(Method::X, "/path")` in its file.
fn routes_in(file: &str) -> BTreeSet<Op> {
    let s = source(file);
    let mut out = BTreeSet::new();
    let mut rest = s.as_str();
    while let Some(i) = rest.find("(Method::") {
        rest = &rest[i + "(Method::".len()..];
        let Some(comma) = rest.find(", \"") else { break };
        let method = &rest[..comma];
        if !method.chars().all(|c| c.is_ascii_uppercase()) {
            continue;
        }
        let after = &rest[comma + 3..];
        let Some(end) = after.find('"') else { break };
        out.insert((method.to_string(), after[..end].to_string()));
    }
    out
}

const STATUS: &[(&str, u16)] = &[
    ("OK", 200), ("BAD_REQUEST", 400), ("UNAUTHORIZED", 401), ("FORBIDDEN", 403), ("NOT_FOUND", 404), ("METHOD_NOT_ALLOWED", 405),
    ("CONFLICT", 409), ("GONE", 410), ("PAYLOAD_TOO_LARGE", 413), ("TOO_MANY_REQUESTS", 429), ("INTERNAL_SERVER_ERROR", 500), ("SERVICE_UNAVAILABLE", 503),
];

/// The statuses a top-level function of `file` names (`StatusCode::X` in its body: from its
/// `fn` line to the next line that is `}` alone).
fn fn_statuses(file: &str, name: &str) -> BTreeSet<u16> {
    let s = source(file);
    let lines: Vec<&str> = s.lines().collect();
    let starts: Vec<usize> = lines
        .iter()
        .enumerate()
        .filter(|(_, l)| {
            let t = l.trim_start_matches("pub(crate) ").trim_start_matches("pub ").trim_start_matches("async ");
            !l.starts_with(' ') && t.starts_with(&format!("fn {name}("))
        })
        .map(|(i, _)| i)
        .collect();
    assert_eq!(starts.len(), 1, "{file} has one top-level fn {name}");
    let body: Vec<&str> = lines[starts[0]..].iter().take_while(|l| **l != "}").copied().collect();
    let body = body.join("\n");
    let mut out = BTreeSet::new();
    let mut rest = body.as_str();
    while let Some(i) = rest.find("StatusCode::") {
        rest = &rest[i + "StatusCode::".len()..];
        let name: String = rest.chars().take_while(|c| c.is_ascii_uppercase() || *c == '_').collect();
        let code = STATUS.iter().find(|(n, _)| *n == name).unwrap_or_else(|| panic!("{file}: {name}: a status this test does not know")).1;
        out.insert(code);
    }
    out
}

/// Each operation: its method and path, its name in the toolchain's catalogue, and the functions
/// that answer it (the router that matches it, and every function on its way that answers).
type Handlers = &'static [(&'static str, &'static str)];

const HANDLERS: &[(&str, &str, &str, Handlers)] = &[
    ("GET", "/", "service.show", &[("http.rs", "handle")]),
    ("GET", "/health", "service.check", &[("http.rs", "handle")]),
    ("GET", "/room", "room.show", &[("http.rs", "handle")]),
    ("POST", "/grant", "room.grant", &[("http.rs", "handle"), ("http.rs", "grant"), ("remote.rs", "grant"), ("remote.rs", "bad"), ("call.rs", "handle")]),
    ("GET", "/age/receipt", "admission.show", &[("http.rs", "age_receipt")]),
    ("GET", "/mls", "mls.tail", &[("http.rs", "mls_log")]),
    ("POST", "/mls/key-package", "mls.import", &[("http.rs", "handle"), ("http.rs", "mls_body"), ("http.rs", "mls_key_package")]),
    ("POST", "/mls/post", "mls.update", &[("http.rs", "handle"), ("http.rs", "mls_body"), ("http.rs", "mls_post")]),
    ("GET", "/co/objects", "performance.list", &[("http.rs", "co_objects")]),
    ("GET", "/co/state", "performance.show", &[("http.rs", "co_state")]),
    ("POST", "/co/schedule", "schedule.create", &[("http.rs", "co_request"), ("http.rs", "unauthorised"), ("http.rs", "signed_reply")]),
    ("POST", "/co/handover", "handover.create", &[("http.rs", "co_request"), ("http.rs", "unauthorised"), ("http.rs", "signed_reply")]),
    ("POST", "/co/window", "window.create", &[("http.rs", "co_request"), ("http.rs", "unauthorised"), ("http.rs", "signed_reply")]),
    ("POST", "/co/fidelity", "fidelity.create", &[("http.rs", "co_request"), ("http.rs", "unauthorised"), ("http.rs", "signed_reply")]),
    ("GET", "/watch/time", "time.show", &[("http.rs", "handle")]),
    ("GET", "/watch/source", "source.fetch", &[("http.rs", "watch_source")]),
    ("GET", "/watch/challenge", "watch.setup", &[("http.rs", "handle")]),
    ("GET", "/watch", "watch.tail", &[("http.rs", "watch_log")]),
    ("GET", "/watch/room", "watch.show", &[("http.rs", "handle")]),
    ("POST", "/watch/join", "watch.join", &[("http.rs", "handle"), ("http.rs", "watch_join")]),
    ("POST", "/watch/post", "watch.update", &[("http.rs", "handle"), ("http.rs", "watch_post")]),
    ("POST", "/remote/control", "control.grant", &[("remote.rs", "control"), ("http.rs", "unauthorised")]),
    ("GET", "/remote/room", "remote.show", &[("remote.rs", "session"), ("remote.rs", "admit_reader"), ("http.rs", "unauthorised")]),
    ("GET", "/remote/state", "remote.verify", &[("remote.rs", "state"), ("remote.rs", "admit_reader"), ("http.rs", "unauthorised")]),
    ("GET", "/msg", "conversation.show", &[("messaging.rs", "handle")]),
    ("GET", "/msg/challenge", "conversation.setup", &[("messaging.rs", "handle")]),
    ("POST", "/msg/join", "conversation.join", &[("messaging.rs", "handle"), ("messaging.rs", "join"), ("messaging.rs", "unauthorised")]),
    ("GET", "/msg/log", "conversation.tail", &[("messaging.rs", "handle"), ("messaging.rs", "log"), ("messaging.rs", "unauthorised")]),
    ("POST", "/msg/key-package", "conversation.import", &[("messaging.rs", "handle"), ("messaging.rs", "unauthorised"), ("messaging.rs", "body"), ("messaging.rs", "act")]),
    ("POST", "/msg/mls", "conversation.update", &[("messaging.rs", "handle"), ("messaging.rs", "unauthorised"), ("messaging.rs", "body"), ("messaging.rs", "act")]),
    ("POST", "/msg/post", "message.create", &[("messaging.rs", "handle"), ("messaging.rs", "unauthorised"), ("messaging.rs", "body"), ("messaging.rs", "act")]),
    ("POST", "/msg/ack", "acknowledgement.create", &[("messaging.rs", "handle"), ("messaging.rs", "unauthorised"), ("messaging.rs", "body"), ("messaging.rs", "act")]),
    ("POST", "/msg/leave", "conversation.leave", &[("messaging.rs", "handle"), ("messaging.rs", "unauthorised"), ("messaging.rs", "body"), ("messaging.rs", "act")]),
    ("POST", "/msg/remove", "member.delete", &[("messaging.rs", "handle"), ("messaging.rs", "unauthorised"), ("messaging.rs", "body"), ("messaging.rs", "act")]),
    ("GET", "/call/challenge", "call.setup", &[("http.rs", "handle"), ("call.rs", "handle")]),
    ("POST", "/call/place", "call.create", &[("http.rs", "handle"), ("call.rs", "handle"), ("call.rs", "unauthorised"), ("call.rs", "body"), ("call.rs", "place")]),
    ("POST", "/call/ring", "call.invite", &[("http.rs", "handle"), ("call.rs", "handle"), ("http.rs", "unauthorised"), ("call.rs", "body"), ("call.rs", "ring")]),
    ("GET", "/inbox", "inbox.show", &[("http.rs", "handle"), ("call.rs", "handle"), ("call.rs", "unauthorised"), ("call.rs", "inbox")]),
    ("POST", "/call/answer", "answer.create", &[("http.rs", "handle"), ("call.rs", "handle"), ("call.rs", "unauthorised"), ("call.rs", "body"), ("call.rs", "answer_of")]),
    ("GET", "/call", "call.show", &[("http.rs", "handle"), ("call.rs", "handle"), ("call.rs", "unauthorised"), ("call.rs", "current")]),
];

#[test]
fn every_route_the_routers_match_is_documented_and_no_other() {
    let doc: BTreeSet<Op> = documented().into_keys().collect();
    let code: BTreeSet<Op> = ["http.rs", "messaging.rs", "call.rs"].iter().flat_map(|f| routes_in(f)).collect();
    let undocumented: Vec<&Op> = code.difference(&doc).collect();
    let unserved: Vec<&Op> = doc.difference(&code).collect();
    assert!(undocumented.is_empty(), "routes the code serves that openapi.json does not describe: {undocumented:?}");
    assert!(unserved.is_empty(), "routes openapi.json describes that no router matches: {unserved:?}");
    let mapped: BTreeSet<Op> = HANDLERS.iter().map(|(m, p, _, _)| (m.to_string(), p.to_string())).collect();
    assert_eq!(mapped, doc, "HANDLERS names the functions of every documented operation, and of no other");
}

#[test]
fn every_operation_is_named_as_the_catalogue_names_it() {
    let doc = description();
    let mut seen = BTreeSet::new();
    for (m, p, op, _) in HANDLERS {
        let o = wai::catalogue::op(op).unwrap_or_else(|| panic!("{m} {p}: {op} is not an operation of the catalogue (wai/wai-rs/src/catalogue.rs)"));
        let r = o.route.unwrap_or_else(|| panic!("{op}: the catalogue gives it no route"));
        assert_eq!((r.service, r.method, r.path), ("wai-room", *m, *p), "{op}: the catalogue's route");
        assert!(seen.insert(*op), "{op} names two operations");
        let id = doc["paths"][*p][m.to_ascii_lowercase()]["operationId"].as_str();
        assert_eq!(id, Some(*op), "{m} {p}: its operationId is its name in the catalogue");
    }
    // Every operation the catalogue routes to this service is one of these.
    for o in wai::catalogue::OPS.iter().filter(|o| o.route.is_some_and(|r| r.service == "wai-room")) {
        let r = o.route.expect("routed");
        assert!(HANDLERS.iter().any(|h| (h.0, h.1, h.2) == (r.method, r.path, o.id)), "the catalogue routes {} to {} {}, which this service does not describe", o.id, r.method, r.path);
    }
}

#[test]
fn every_documented_status_is_one_its_handlers_return_and_every_returned_status_is_documented() {
    let doc = documented();
    let mut by_fn: BTreeMap<(&str, &str), BTreeSet<u16>> = BTreeMap::new();
    for (m, p, _, fns) in HANDLERS {
        let listed = &doc[&(m.to_string(), p.to_string())];
        let mut can: BTreeSet<u16> = BTreeSet::new();
        for (f, n) in *fns {
            can.extend(fn_statuses(f, n));
            by_fn.entry((f, n)).or_default().extend(listed.iter().copied());
        }
        let never: Vec<&u16> = listed.difference(&can).collect();
        assert!(never.is_empty(), "{m} {p}: openapi.json lists {never:?}, which none of {fns:?} returns");
    }
    for ((f, n), listed) in &by_fn {
        let returned: BTreeSet<u16> = fn_statuses(f, n).into_iter().filter(|s| *s != 405).collect();
        let missing: Vec<&u16> = returned.difference(listed).collect();
        assert!(missing.is_empty(), "{f}:{n} returns {missing:?}, which openapi.json lists for none of the operations it answers");
    }
}

// --------------------------------------------------------------------------------- requests

struct Answer {
    status: u16,
    headers: BTreeMap<String, String>,
    json: Value,
}

async fn send(addr: SocketAddr, method: &str, path: &str, headers: &[(&str, String)], body: Option<Vec<u8>>) -> Answer {
    let mut s = tokio::net::TcpStream::connect(addr).await.expect("the service answers");
    let b = body.unwrap_or_default();
    let extra: String = headers.iter().map(|(k, v)| format!("{k}: {v}\r\n")).collect();
    let head = format!("{method} {path} HTTP/1.1\r\nHost: {HOST}\r\nX-Forwarded-For: 10.20.0.1\r\n{extra}Content-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", b.len());
    s.write_all(head.as_bytes()).await.unwrap();
    // A body past what the request takes may be refused before it is read: a reset then is fine.
    let _ = s.write_all(&b).await;
    let mut out = Vec::new();
    let _ = timeout(STEP, s.read_to_end(&mut out)).await.expect("an answer in time");
    let text = String::from_utf8_lossy(&out).to_string();
    let (head, body) = text.split_once("\r\n\r\n").unwrap_or_else(|| panic!("{method} {path}: no response: {text:?}"));
    let status: u16 = head.split(' ').nth(1).unwrap().parse().unwrap();
    let headers = head.lines().skip(1).filter_map(|l| l.split_once(':')).map(|(k, v)| (k.trim().to_ascii_lowercase(), v.trim().to_string())).collect();
    Answer { status, headers, json: serde_json::from_str(body).unwrap_or(Value::Null) }
}

/// What the running service answered, held to the description as it is noted.
struct Probes {
    doc: BTreeMap<Op, BTreeSet<u16>>,
    seen: BTreeMap<Op, BTreeSet<u16>>,
    mode: &'static str,
    addr: SocketAddr,
}

impl Probes {
    /// Ask, require a status the description lists for the operation and, when given, the one
    /// the probe was written to reach.
    async fn ask(&mut self, method: &str, path: &str, headers: &[(&str, String)], body: Option<Vec<u8>>, want: u16) -> Answer {
        let a = send(self.addr, method, path, headers, body).await;
        let op = (method.to_string(), path.split('?').next().unwrap().to_string());
        let listed = self.doc.get(&op).unwrap_or_else(|| panic!("[{}] {method} {path}: not an operation of openapi.json", self.mode));
        assert!(listed.contains(&a.status), "[{}] {method} {path} answered {} {}, which openapi.json does not list ({listed:?})", self.mode, a.status, a.json);
        assert_eq!(a.status, want, "[{}] {method} {path}: {}", self.mode, a.json);
        self.seen.entry(op).or_default().insert(a.status);
        a
    }

    async fn json(&mut self, method: &str, path: &str, body: Value, want: u16) -> Answer {
        self.ask(method, path, &[], Some(body.to_string().into_bytes()), want).await
    }

    async fn get(&mut self, path: &str, want: u16) -> Answer {
        self.ask("GET", path, &[], None, want).await
    }
}

fn key(i: u8) -> SigningKey {
    SigningKey::from_bytes(&[i; 32])
}

fn pubhex(k: &SigningKey) -> String {
    hex::encode(k.verifying_key().to_bytes())
}

fn challenge_from(a: &Answer) -> [u8; 32] {
    maybe_challenge(a).unwrap_or_else(|| panic!("no challenge in an answer {} {}", a.status, a.json))
}

fn maybe_challenge(a: &Answer) -> Option<[u8; 32]> {
    a.headers.get(&gp::HEADER_CHALLENGE.to_ascii_lowercase()).and_then(|c| gp::from_base64url(c)).and_then(|c| c.try_into().ok())
}

async fn call_challenge(addr: SocketAddr) -> [u8; 32] {
    hex_challenge(&send(addr, "GET", "/call/challenge", &[], None).await)
}

fn hex_challenge(a: &Answer) -> [u8; 32] {
    hex::decode(a.json["challenge"].as_str().expect("a challenge")).unwrap().try_into().unwrap()
}

/// A grant's `Authorization` and the proof of its key over `challenge` for the request.
fn authorised(grant: Option<&[u8]>, k: &SigningKey, challenge: [u8; 32], method: &str, path: &str) -> Vec<(&'static str, String)> {
    let digest = grant.map_or([0; 32], |g| gp::grant_digest_and_principal(g).unwrap().0);
    let proof = GrantProof::sign(digest, k, challenge, &gp::context(method, HOST, path)).unwrap();
    let mut hs = vec![(gp::HEADER_PROOF, gp::base64url(&proof.to_bytes()))];
    if let Some(g) = grant {
        hs.push(("Authorization", format!("Bearer {}", gp::base64url(g))));
    }
    hs
}

fn big(n: usize) -> Vec<u8> {
    let mut v = br#"{"pad":""#.to_vec();
    v.extend(std::iter::repeat_n(b'a', n));
    v.extend(br#""}"#);
    v
}

const OTHER_ROOM: &str = "abababababababababababababababababababababababababababababababab";

// ------------------------------------------------------------------------------------- modes

async fn plain(p: &mut Probes) {
    let room = p.get("/room", 200).await.json["room"].as_str().unwrap().to_string();
    p.get("/", 200).await;
    p.get("/health", 200).await;
    let a = p.json("POST", "/grant", json!({"principal": pubhex(&key(1)), "role": "publisher"}), 200).await;
    assert!(a.json["grant"].as_str().is_some() && a.json["room"]["room"].as_str() == Some(room.as_str()));
    p.json("POST", "/grant", json!({"principal": pubhex(&key(2)), "role": "publisher"}), 200).await;
    p.json("POST", "/grant", json!({"principal": pubhex(&key(3)), "role": "publisher"}), 409).await;
    p.json("POST", "/grant", json!({"principal": pubhex(&key(4)), "role": "viewer"}), 200).await;
    p.json("POST", "/grant", json!({"principal": "00", "role": "viewer"}), 400).await;
    p.json("POST", "/grant", json!({"principal": pubhex(&key(4)), "role": "owner"}), 400).await;
    p.ask("POST", "/grant", &[], Some(b"not json".to_vec()), 400).await;
    p.ask("POST", "/grant", &[], Some(big(2048)), 413).await;
    p.get("/age/receipt", 404).await;
    // The modes this room does not keep.
    p.get(&format!("/mls?room={room}"), 404).await;
    p.get("/mls", 400).await;
    p.get(&format!("/mls?room={OTHER_ROOM}"), 410).await;
    for path in ["/mls/key-package", "/mls/post"] {
        p.json("POST", path, json!({"room": room, "key_package": "00", "frames": ["00"]}), 404).await;
        p.json("POST", path, json!({"room": OTHER_ROOM}), 410).await;
        p.json("POST", path, json!({}), 400).await;
        p.ask("POST", path, &[], Some(big(140 * 1024)), 413).await;
    }
    p.get("/co/objects", 404).await;
    p.get("/co/state", 404).await;
    for path in ["/co/schedule", "/co/handover", "/co/window", "/co/fidelity"] {
        p.json("POST", path, json!({}), 404).await;
    }
    p.get("/watch/time", 200).await;
    p.get("/watch/source", 404).await;
    p.get("/watch/challenge", 404).await;
    p.get(&format!("/watch?room={room}&wait_ms=0"), 404).await;
    p.get("/watch", 400).await;
    p.get(&format!("/watch?room={OTHER_ROOM}"), 410).await;
    p.get("/watch/room", 404).await;
    p.json("POST", "/watch/join", json!({"room": room, "grant": "00", "proof": "00"}), 404).await;
    p.json("POST", "/watch/join", json!({"room": OTHER_ROOM, "grant": "00", "proof": "00"}), 410).await;
    p.json("POST", "/watch/join", json!({"room": room}), 400).await;
    p.ask("POST", "/watch/join", &[], Some(big(5000)), 413).await;
    p.json("POST", "/watch/post", json!({"room": room, "events": ["00"]}), 404).await;
    p.json("POST", "/watch/post", json!({"room": OTHER_ROOM, "events": ["00"]}), 410).await;
    p.json("POST", "/watch/post", json!({"room": room, "events": []}), 400).await;
    p.ask("POST", "/watch/post", &[], Some(big(17 * 1024)), 413).await;
    p.json("POST", "/remote/control", json!({}), 404).await;
    p.get("/remote/room", 404).await;
    p.get("/remote/state", 404).await;
    for path in ["/msg", "/msg/challenge", "/msg/log"] {
        p.get(path, 404).await;
    }
    for path in ["/msg/join", "/msg/key-package", "/msg/mls", "/msg/post", "/msg/ack", "/msg/leave", "/msg/remove"] {
        p.json("POST", path, json!({}), 404).await;
    }
    p.get("/call/challenge", 404).await;
    p.get("/inbox", 404).await;
    p.get("/call", 404).await;
    for path in ["/call/place", "/call/ring", "/call/answer"] {
        p.json("POST", path, json!({}), 404).await;
    }
    // A method a documented path does not take.
    for (m, path) in [("GET", "/grant"), ("POST", "/room"), ("POST", "/health"), ("GET", "/mls/post"), ("POST", "/co/state"), ("GET", "/watch/join"), ("POST", "/remote/room"), ("PUT", "/grant"), ("DELETE", "/room")] {
        let a = send(p.addr, m, path, &[], None).await;
        assert_eq!(a.status, 405, "{m} {path}");
    }
    let a = send(p.addr, "GET", "/no/such/path", &[], None).await;
    assert_eq!(a.status, 404);
}

async fn without_relay(p: &mut Probes) {
    p.json("POST", "/grant", json!({"principal": pubhex(&key(5)), "role": "viewer"}), 503).await;
    let h = p.get("/health", 200).await;
    assert_eq!(h.json["relay"], json!(false));
}

async fn rate(p: &mut Probes) {
    p.json("POST", "/grant", json!({"principal": pubhex(&key(6)), "role": "viewer"}), 200).await;
    p.json("POST", "/grant", json!({"principal": pubhex(&key(7)), "role": "viewer"}), 429).await;
    p.get("/age/receipt", 429).await;
}

async fn e2ee(p: &mut Probes) {
    let room = p.get("/room", 200).await.json["room"].as_str().unwrap().to_string();
    let a = p.get(&format!("/mls?room={room}&wait_ms=0"), 200).await;
    assert_eq!(a.json["room"].as_str(), Some(room.as_str()));
    p.json("POST", "/mls/key-package", json!({"room": room, "key_package": "zz"}), 400).await;
    p.json("POST", "/mls/key-package", json!({"room": room, "key_package": "00"}), 400).await;
    p.json("POST", "/mls/post", json!({"room": room, "frames": ["00"]}), 400).await;
}

async fn age(p: &mut Probes) {
    let a = p.json("POST", "/grant", json!({"principal": pubhex(&key(8)), "role": "viewer"}), 401).await;
    assert_eq!(a.json["error"], json!("token_required"));
    assert!(a.headers.contains_key("www-authenticate"), "a challenge for a token");
    let a = p.ask("POST", "/grant", &[("Authorization", "PrivateToken token=\"!!\"".into())], Some(json!({"principal": pubhex(&key(8)), "role": "viewer"}).to_string().into_bytes()), 403).await;
    assert_eq!(a.json["error"], json!("token_malformed"));
    let r = p.get("/age/receipt", 200).await;
    assert!(r.json["receipt"].is_object() || r.json["receipt"].is_string(), "{}", r.json);
}

async fn co(p: &mut Probes) {
    let objects = p.get("/co/objects?from=0", 200).await;
    assert!(objects.json["objects"].is_array());
    p.get("/co/state", 200).await;
    let lead = key(0x21);
    let g = p.json("POST", "/grant", json!({"principal": pubhex(&lead), "role": "publisher"}), 200).await;
    let grant = hex::decode(g.json["grant"].as_str().unwrap()).unwrap();
    let viewer = key(0x22);
    let v = p.json("POST", "/grant", json!({"principal": pubhex(&viewer), "role": "viewer"}), 200).await;
    let vgrant = hex::decode(v.json["grant"].as_str().unwrap()).unwrap();
    // No proof: 401 with a challenge; a proof then passes.
    let a = p.json("POST", "/co/fidelity", json!({}), 401).await;
    let mut c = challenge_from(&a);
    let call = async |p: &mut Probes, who: &SigningKey, g: &[u8], path: &str, body: Vec<u8>, want: u16, c: &mut [u8; 32]| {
        let hs = authorised(Some(g), who, *c, "POST", path);
        let a = p.ask("POST", path, &hs, Some(body), want).await;
        if let Some(n) = maybe_challenge(&a) {
            *c = n;
        } else {
            *c = challenge_from(&p.json("POST", "/co/fidelity", json!({}), 401).await);
        }
        a
    };
    let a = call(p, &lead, &grant, "/co/fidelity", b"{}".to_vec(), 200, &mut c).await;
    assert!(a.json["ok"].is_boolean(), "{}", a.json);
    call(p, &viewer, &vgrant, "/co/schedule", json!({"mode": "mark", "segments": []}).to_string().into_bytes(), 403, &mut c).await;
    call(p, &viewer, &vgrant, "/co/window", json!({"rule": "plurality", "options": 3, "seconds": 5}).to_string().into_bytes(), 403, &mut c).await;
    call(p, &viewer, &vgrant, "/co/fidelity", b"{}".to_vec(), 403, &mut c).await;
    call(p, &lead, &grant, "/co/schedule", json!({"mode": "neither", "segments": []}).to_string().into_bytes(), 400, &mut c).await;
    call(p, &lead, &grant, "/co/handover", json!({"to": "00"}).to_string().into_bytes(), 400, &mut c).await;
    call(p, &lead, &grant, "/co/window", json!({"rule": "sometimes"}).to_string().into_bytes(), 400, &mut c).await;
    call(p, &lead, &grant, "/co/fidelity", b"not json".to_vec(), 400, &mut c).await;
    call(p, &lead, &grant, "/co/window", big(5000), 413, &mut c).await;
    let w = call(p, &lead, &grant, "/co/window", json!({"rule": "plurality", "options": 3, "seconds": 30, "question": "which"}).to_string().into_bytes(), 200, &mut c).await;
    assert_eq!(w.json["ok"], json!(true), "{}", w.json);
    let s = call(p, &lead, &grant, "/co/schedule", json!({"mode": "mark", "beat_us": 500000, "beats_per_bar": 4, "segments": [{"length": 8, "turns": 2, "holders": [pubhex(&lead)]}]}).to_string().into_bytes(), 200, &mut c).await;
    assert_eq!(s.json["ok"], json!(true), "{}", s.json);
    call(p, &lead, &grant, "/co/handover", json!({"to": pubhex(&viewer)}).to_string().into_bytes(), 400, &mut c).await;
    let h = call(p, &lead, &grant, "/co/handover", json!({"to": pubhex(&lead)}).to_string().into_bytes(), 200, &mut c).await;
    assert_eq!(h.json["ok"], json!(true), "{}", h.json);
}

async fn watch(p: &mut Probes) {
    let r = p.get("/room", 200).await;
    let room = r.json["room"].as_str().unwrap().to_string();
    let room_id: [u8; 32] = hex::decode(&room).unwrap().try_into().unwrap();
    let t0 = r.json["t0_us"].as_u64().unwrap();
    let src = p.get("/watch/source", 200).await;
    assert_eq!(src.json, Value::Null, "the source's bytes, not JSON");
    p.get("/watch/room", 200).await;
    p.get(&format!("/watch?room={room}&wait_ms=0"), 200).await;
    let viewer = key(0x31);
    let g = p.json("POST", "/grant", json!({"principal": pubhex(&viewer), "role": "viewer"}), 200).await;
    let grant = hex::decode(g.json["grant"].as_str().unwrap()).unwrap();
    let digest = gp::grant_digest_and_principal(&grant).unwrap().0;
    let join = |c: [u8; 32], target: &str| json!({"room": room, "grant": hex::encode(&grant), "proof": hex::encode(wai::live_room::Proof::sign(digest, &viewer, c, &format!("POST {target}")).to_bytes())});
    let c = hex_challenge(&p.get("/watch/challenge", 200).await);
    p.json("POST", "/watch/join", join(c, "//elsewhere.test/watch/join"), 401).await;
    let c = hex_challenge(&p.get("/watch/challenge", 200).await);
    let j = p.json("POST", "/watch/join", join(c, &format!("//{HOST}/watch/join")), 200).await;
    let joined = j.json["t"].as_u64().unwrap();
    // A control in the past is refused; one ahead is carried.
    use wai::co_watch::{action, Control};
    let now_t = (std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_micros() as u64).saturating_sub(t0).max(joined);
    let ev = |seq: u64, prev: [u8; 32], payload: Vec<u8>| wai::live_events::Event::sign(&viewer, room_id, seq, prev, now_t, None, wai::live_events::kind::META, payload);
    let past = ev(0, [0; 32], Control { action: action::PLAY, at: now_t.saturating_sub(1), arg: 0 }.encode());
    p.json("POST", "/watch/post", json!({"room": room, "events": [hex::encode(past.to_bytes())]}), 409).await;
    let ahead = ev(0, [0; 32], Control { action: action::PLAY, at: now_t + 700_000, arg: 0 }.encode());
    let a = p.json("POST", "/watch/post", json!({"room": room, "events": [hex::encode(ahead.to_bytes())]}), 200).await;
    assert_eq!(a.json["carried"], json!(1));
}

async fn remote(p: &mut Probes) {
    p.json("POST", "/grant", json!({"principal": pubhex(&key(0x41)), "role": "host"}), 403).await;
    p.json("POST", "/grant", json!({"principal": pubhex(&key(0x41)), "role": "publisher"}), 400).await;
    let ctl = key(0x42);
    let g = p.json("POST", "/grant", json!({"principal": pubhex(&ctl), "role": "controller"}), 200).await;
    let grant = hex::decode(g.json["grant"].as_str().unwrap()).unwrap();
    let c = challenge_from(&p.get("/remote/room", 401).await);
    let room = p.ask("GET", "/remote/room", &authorised(Some(&grant), &ctl, c, "GET", "/remote/room"), None, 200).await;
    assert!(room.json["descriptor"].is_string(), "{}", room.json);
    let c = challenge_from(&p.get("/remote/state", 401).await);
    let state = p.ask("GET", "/remote/state", &authorised(Some(&grant), &ctl, c, "GET", "/remote/state"), None, 200).await;
    assert!(state.json["verified"].is_object(), "{}", state.json);
    let a = p.json("POST", "/remote/control", json!({}), 401).await;
    let mut c = challenge_from(&a);
    let body = json!({"to": pubhex(&ctl), "control": 3}).to_string().into_bytes();
    let a = p.ask("POST", "/remote/control", &authorised(Some(&grant), &ctl, c, "POST", "/remote/control"), Some(body), 403).await;
    c = challenge_from(&a);
    p.ask("POST", "/remote/control", &authorised(Some(&grant), &ctl, c, "POST", "/remote/control"), Some(b"not json".to_vec()), 400).await;
}

async fn messaging(p: &mut Probes) {
    p.get("/msg", 200).await;
    // Join on a proof of the key; the first member owns the conversation.
    let owner = key(0x51);
    let c = hex_challenge(&p.get("/msg/challenge", 200).await);
    p.ask("POST", "/msg/join", &[], None, 401).await;
    let j = p.ask("POST", "/msg/join", &authorised(None, &owner, c, "POST", "/msg/join"), None, 200).await;
    assert_eq!(j.json["role"], json!("owner"));
    let grant = hex::decode(j.json["grant"].as_str().unwrap()).unwrap();
    let mut c = challenge_from(&j);
    let member = async |p: &mut Probes, method: &str, path: &str, body: Option<Vec<u8>>, want: u16, c: &mut [u8; 32]| {
        let a = p.ask(method, path, &authorised(Some(&grant), &owner, *c, method, path.split('?').next().unwrap()), body, want).await;
        *c = match maybe_challenge(&a) {
            Some(n) => n,
            None => hex_challenge(&send(p.addr, "GET", "/msg/challenge", &[], None).await),
        };
        a
    };
    let log = member(p, "GET", "/msg/log?from=0&wait_ms=0", None, 200, &mut c).await;
    assert!(log.json["entries"].is_array());
    member(p, "POST", "/msg/key-package", Some(b"{}".to_vec()), 400, &mut c).await;
    member(p, "POST", "/msg/mls", Some(json!({"frames": []}).to_string().into_bytes()), 400, &mut c).await;
    member(p, "POST", "/msg/post", Some(b"{}".to_vec()), 400, &mut c).await;
    member(p, "POST", "/msg/ack", Some(json!({"ack": "00"}).to_string().into_bytes()), 409, &mut c).await;
    member(p, "POST", "/msg/remove", Some(json!({"principal": pubhex(&key(0x52))}).to_string().into_bytes()), 409, &mut c).await;
    member(p, "POST", "/msg/remove", Some(json!({"principal": "00"}).to_string().into_bytes()), 400, &mut c).await;
    member(p, "POST", "/msg/leave", Some(b"not json".to_vec()), 400, &mut c).await;
    member(p, "POST", "/msg/leave", Some(b"{}".to_vec()), 200, &mut c).await;
    // Once it has left, it is no member.
    member(p, "POST", "/msg/key-package", Some(json!({"key_package": "00"}).to_string().into_bytes()), 403, &mut c).await;
    for path in ["/msg/key-package", "/msg/mls", "/msg/post", "/msg/ack", "/msg/leave", "/msg/remove"] {
        p.json("POST", path, json!({}), 401).await;
    }
    p.get("/msg/log", 401).await;
    let a = send(p.addr, "GET", "/msg/nothing", &[], None).await;
    assert_eq!(a.status, 404);
    let a = send(p.addr, "GET", "/msg/join", &[], None).await;
    assert_eq!(a.status, 404, "a method a conversation's path does not take");
}

async fn calls(p: &mut Probes) {
    let caller = key(0x61);
    let callee = key(0x62);
    p.json("POST", "/grant", json!({"principal": pubhex(&caller), "role": "publisher"}), 403).await;
    let c = hex_challenge(&p.get("/call/challenge", 200).await);
    p.json("POST", "/call/place", json!({"invitees": [pubhex(&callee)]}), 401).await;
    p.ask("POST", "/call/place", &authorised(None, &caller, c, "POST", "/call/place"), Some(json!({"invitees": []}).to_string().into_bytes()), 400).await;
    let placed = p.ask("POST", "/call/place", &authorised(None, &caller, call_challenge(p.addr).await, "POST", "/call/place"), Some(json!({"invitees": [pubhex(&callee)]}).to_string().into_bytes()), 200).await;
    let grant = hex::decode(placed.json["grant"].as_str().unwrap()).unwrap();
    let a = p.ask("GET", "/call", &authorised(None, &caller, call_challenge(p.addr).await, "GET", "/call"), None, 200).await;
    assert!(a.json.is_object());
    p.ask("GET", "/call", &authorised(None, &key(0x63), call_challenge(p.addr).await, "GET", "/call"), None, 403).await;
    let a = p.ask("GET", "/inbox", &authorised(None, &callee, call_challenge(p.addr).await, "GET", "/inbox"), None, 200).await;
    assert!(a.json["invitations"].is_array());
    p.get("/inbox", 401).await;
    p.get("/call", 401).await;
    p.ask("POST", "/call/ring", &authorised(Some(&grant), &caller, call_challenge(p.addr).await, "POST", "/call/ring"), Some(json!({"invitations": []}).to_string().into_bytes()), 400).await;
    p.ask("POST", "/call/answer", &authorised(None, &callee, call_challenge(p.addr).await, "POST", "/call/answer"), Some(json!({"answer": "00"}).to_string().into_bytes()), 400).await;
    p.json("POST", "/call/ring", json!({}), 401).await;
    p.json("POST", "/call/answer", json!({}), 401).await;
}

// ----------------------------------------------------------------------------------- running

/// A service with no relay to reach: its clock never opens a session.
async fn start_without_relay(extra: &str) -> SocketAddr {
    let cfg = Config::from_toml(&format!("listen = \"127.0.0.1:0\"\nkeys = \"/nonexistent\"\nrelay_url = \"ws://127.0.0.1:9/moq\"\n{extra}")).unwrap();
    let svc = Service::new(Keys::generate(), &cfg);
    let listener = tokio::net::TcpListener::bind(cfg.listen).await.unwrap();
    let addr = listener.local_addr().unwrap();
    svc.start();
    tokio::spawn(http::serve(svc, listener));
    addr
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn the_running_service_answers_every_documented_route_with_a_documented_status() {
    timeout(Duration::from_secs(300), run()).await.expect("the probes finish in time");
}

async fn run() {
    let doc = documented();
    let dir = std::env::temp_dir().join(format!("joule-moq-room-openapi-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&dir);
    std::fs::create_dir_all(&dir).unwrap();
    let source = dir.join("source.wav");
    std::fs::write(&source, (0..2048u32).map(|i| (i % 251) as u8).collect::<Vec<_>>()).unwrap();
    let k16 = rfc_test_keys::k16();
    let modes: Vec<(&'static str, String)> = vec![
        ("plain", String::new()),
        ("e2ee", "[room]\nmax_publishers = 3\njoin_wait_s = 30\ne2ee = true\n".into()),
        ("age", format!("[age]\ntoken_window_s = 600\n[[age.issuers]]\nname = \"issuer.example\"\nkeys = [{{ threshold = 16, spki = \"{}\" }}]\n[[age.ratings]]\nscheme = \"https://ratings.example/scheme\"\nvalue = \"16\"\nmin_age = 16\n", hex::encode(&k16.spki))),
        ("co", "[room]\nmax_publishers = 3\njoin_wait_s = 30\n[co]\naudience_input = 2\nlatency_budget_ms = 1500\nlead = \"first\"\nai = true\nfidelity_every_s = 0\ncommit_after_ms = 800\nlead_ms = 200\n".into()),
        ("watch", format!("[room]\nwatch_source = \"{}\"\nwatch_duration_ms = 60000\nwatch_window_ms = 2000\n", source.display())),
        ("remote", "[room]\nmax_present = 8\njoin_wait_s = 30\n[remote]\nwidth = 640\nheight = 360\nfps = 20\nallow = \"all\"\ncontrol = 3\nmax_age_ms = 3000\n".into()),
        ("messaging", format!("[messaging]\nstore = \"{}\"\nmax_members = 8\n", dir.join("conversation").display())),
        ("calls", "[call]\nring_ms = 30000\n".into()),
        ("rate", "[rate]\nclient_burst = 1\nclient_every_s = 600\n".into()),
    ];
    let mut seen: BTreeMap<Op, BTreeSet<u16>> = BTreeMap::new();
    for (mode, extra) in modes {
        let room = DevRoom::start(2, &extra).await.unwrap_or_else(|e| panic!("[{mode}] the room starts: {e}"));
        let mut p = Probes { doc: doc.clone(), seen: BTreeMap::new(), mode, addr: room.http };
        match mode {
            "plain" => plain(&mut p).await,
            "e2ee" => e2ee(&mut p).await,
            "age" => age(&mut p).await,
            "co" => co(&mut p).await,
            "watch" => watch(&mut p).await,
            "remote" => remote(&mut p).await,
            "messaging" => messaging(&mut p).await,
            "calls" => calls(&mut p).await,
            "rate" => rate(&mut p).await,
            _ => unreachable!(),
        }
        for (op, s) in p.seen {
            seen.entry(op).or_default().extend(s);
        }
        room.stop().await;
    }
    let addr = start_without_relay("").await;
    let mut p = Probes { doc: doc.clone(), seen: BTreeMap::new(), mode: "without its relay", addr };
    without_relay(&mut p).await;
    for (op, s) in p.seen {
        seen.entry(op).or_default().extend(s);
    }
    let _ = std::fs::remove_dir_all(&dir);

    // Every documented operation was asked, and answered with a status it lists.
    let unasked: Vec<&Op> = doc.keys().filter(|op| !seen.contains_key(*op)).collect();
    assert!(unasked.is_empty(), "documented operations no probe asked: {unasked:?}");
    let pairs: usize = doc.values().map(BTreeSet::len).sum();
    let reached: usize = seen.values().map(BTreeSet::len).sum();
    let ok_reached = seen.values().filter(|s| s.contains(&200)).count();
    eprintln!("openapi contract: {} operations asked; {reached} of {pairs} documented (operation, status) pairs reached by a running service, {ok_reached} operations answered 200; every other listed status is one its handler returns (the source check)", doc.len());
}