open-standards/wai/wai-rs/tests/cli_http.rs

//! The `wai http` commands run as a user runs them (`--features http --test cli_http`):
//!
//! - `wai http publish` writes `<cid>.wai`, its head proof and its ETag sidecar, named by
//!   the content hash, and refuses an envelope with a byte after it (§5.1);
//! - `wai http codecs` prints a sequence's union and floor (§4.2);
//! - `wai http origin` serves an object with §5.3's headers, byte ranges and `If-Range`,
//!   and `wai http bench` fetches it through the byte-range procedure (§7) and writes a
//!   record with the pick, the bytes by category and energy stated as not measured;
//! - the origin reads at most 16 KiB of a request head, gives a whole head one deadline
//!   from the accept and a whole connection one lifetime, caps the connections it serves
//!   at once (and those of one peer only when asked), percent-decodes `/__hits` as bytes,
//!   and refuses a path segment with `\`, `:`, NUL or a control byte. Every origin a test
//!   starts is stopped when the test ends, whether or not it passed.

use std::io::{BufRead, BufReader, Read, Write};
use std::net::TcpStream;
use std::path::{Path, PathBuf};
use std::process::{Command, Output, Stdio};

use wai::http_binding::head_proof::HeadProof;
use wai::http_binding::{content_hash, hex};

fn wai(args: &[&str]) -> Output {
    Command::new(env!("CARGO_BIN_EXE_wai")).args(args).output().expect("run wai")
}

fn corpus() -> PathBuf {
    Path::new(env!("CARGO_MANIFEST_DIR")).join("../http-conformance")
}

fn object(id: &str) -> PathBuf {
    let listing: serde_json::Value = serde_json::from_slice(&std::fs::read(corpus().join("objects.json")).unwrap()).unwrap();
    corpus().join(listing[id]["path"].as_str().unwrap())
}

fn scratch(name: &str) -> PathBuf {
    let d = Path::new(env!("CARGO_TARGET_TMPDIR")).join("cli_http").join(name);
    let _ = std::fs::remove_dir_all(&d);
    std::fs::create_dir_all(&d).unwrap();
    d
}

#[test]
fn publish_names_by_content_hash() {
    let out = scratch("publish");
    let src = object("O3");
    let run = wai(&["http", "publish", src.to_str().unwrap(), "--out", out.to_str().unwrap()]);
    assert!(run.status.success(), "{}", String::from_utf8_lossy(&run.stderr));
    let env = std::fs::read(&src).unwrap();
    let cid = hex(&content_hash(&env));
    assert_eq!(std::fs::read(out.join(format!("{cid}.wai"))).unwrap(), env);
    assert_eq!(std::fs::read(out.join(format!("{cid}.wai.hp"))).unwrap(), HeadProof::build(&env).unwrap().to_bytes());
    assert_eq!(std::fs::read_to_string(out.join(format!("{cid}.wai.etag"))).unwrap(), format!("\"{cid}\"\n"));
    let line = String::from_utf8_lossy(&run.stdout);
    assert!(line.starts_with(&cid) && line.contains(&format!("n={}", env.len())) && line.contains("wai2.video.av1+meta.hdr10plus"), "{line}");
    // One byte after the envelope: not an object.
    let bad = corpus().join("bad/O6.wai");
    let run = wai(&["http", "publish", bad.to_str().unwrap(), "--out", out.to_str().unwrap()]);
    assert!(!run.status.success());
    assert!(String::from_utf8_lossy(&run.stderr).contains("object_length_mismatch"));
}

#[test]
fn codecs_prints_union_and_floor() {
    let run = wai(&["http", "codecs", object("O9a").to_str().unwrap(), object("O9b").to_str().unwrap(), object("O9c").to_str().unwrap()]);
    assert!(run.status.success(), "{}", String::from_utf8_lossy(&run.stderr));
    assert_eq!(String::from_utf8_lossy(&run.stdout), "union\twai2.video.av1|video.int_hyper\nfloor\twai2.video.av1\n");
}

/// A running `wai http origin`, stopped when dropped, so a failing assertion never leaves
/// it running.
struct Origin {
    child: std::process::Child,
    port: u16,
}

impl Origin {
    fn start(root: &Path, extra: &[&str]) -> Origin {
        let mut child = Command::new(env!("CARGO_BIN_EXE_wai"))
            .args(["http", "origin", "--root", root.to_str().unwrap(), "--port", "0"])
            .args(extra)
            .stdout(Stdio::piped())
            .spawn()
            .unwrap();
        let mut first = String::new();
        BufReader::new(child.stdout.take().unwrap()).read_line(&mut first).unwrap();
        let port: u16 = first.trim().rsplit(':').next().unwrap().trim_end_matches('/').parse().unwrap();
        Origin { child, port }
    }
}

impl Drop for Origin {
    fn drop(&mut self) {
        let _ = self.child.kill();
        let _ = self.child.wait();
    }
}

/// Send raw request bytes; return what the origin answered before closing.
fn raw(port: u16, request: &[u8]) -> String {
    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
    s.set_read_timeout(Some(std::time::Duration::from_secs(10))).unwrap();
    s.write_all(request).unwrap();
    let mut out = Vec::new();
    let _ = s.read_to_end(&mut out);
    String::from_utf8_lossy(&out).into_owned()
}

/// One request to the origin: status line and headers, and the body.
fn get(port: u16, path: &str, extra: &[(&str, &str)]) -> (String, Vec<u8>) {
    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
    let mut req = format!("GET {path} HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\n");
    for (k, v) in extra {
        req.push_str(&format!("{k}: {v}\r\n"));
    }
    req.push_str("\r\n");
    s.write_all(req.as_bytes()).unwrap();
    let mut raw = Vec::new();
    s.read_to_end(&mut raw).unwrap();
    let end = raw.windows(4).position(|w| w == b"\r\n\r\n").unwrap() + 4;
    (String::from_utf8_lossy(&raw[..end]).into_owned(), raw[end..].to_vec())
}

#[test]
fn origin_serves_and_bench_fetches() {
    let root = scratch("origin");
    let src = object("OD");
    let env = std::fs::read(&src).unwrap();
    let pubd = wai(&["http", "publish", src.to_str().unwrap(), "--out", root.to_str().unwrap()]);
    assert!(pubd.status.success());
    let cid = hex(&content_hash(&env));
    let origin = Origin::start(&root, &[]);
    let port = origin.port;
    let path = format!("/{cid}.wai");

    let (h, body) = get(port, &path, &[]);
    assert!(h.starts_with("HTTP/1.1 200") && body == env, "{h}");
    for want in ["Cache-Control: public, max-age=31536000, immutable, no-transform", &format!("ETag: \"{cid}\""), "Accept-Ranges: bytes", "Access-Control-Allow-Origin: *"] {
        assert!(h.contains(want), "{want} missing from {h}");
    }
    assert!(!h.contains("Vary") && !h.contains("Content-Encoding"), "{h}");
    let (h, body) = get(port, &path, &[("Range", "bytes=1009-3593")]);
    assert!(h.starts_with("HTTP/1.1 206") && h.contains("Content-Range: bytes 1009-3593/53794") && body == env[1009..3594], "{h}");
    let (h, _) = get(port, &path, &[("Range", "bytes=1009-3593"), ("If-Range", "\"not-this\"")]);
    assert!(h.starts_with("HTTP/1.1 200"), "{h}");
    let (h, _) = get(port, &path, &[("Range", "bytes=53794-")]);
    assert!(h.starts_with("HTTP/1.1 416") && h.contains("Content-Range: bytes */53794"), "{h}");

    let rec = root.join("bench.json");
    let url = format!("http://127.0.0.1:{port}{path}#h=1009&n=53794");
    let run = wai(&["http", "bench", "--url", &url, "--runs", "3", "--advertise", "wai.video.av1", "--out", rec.to_str().unwrap()]);
    drop(origin);
    assert!(run.status.success(), "{}", String::from_utf8_lossy(&run.stderr));
    let r: serde_json::Value = serde_json::from_slice(&std::fs::read(&rec).unwrap()).unwrap();
    // Appendix D's plan 1: the head, the proof and `low`'s rest.
    assert_eq!(r["pick"], 0);
    assert_eq!(r["body_bytes_media"]["median"], 1009.0 + 2585.0);
    assert_eq!(r["body_bytes_proof"]["median"], 245.0);
    assert_eq!(r["requests"], 3);
    assert_eq!(r["energy"], "omitted: not measured");
    assert_eq!(r["envelope_len"], 53_794);
}


/// The origin reads at most 16 KiB of a request head: a 400 MiB header is answered 431
/// and the connection closed long before it is sent.
#[test]
fn origin_refuses_an_oversized_head() {
    let root = scratch("origin-big");
    std::fs::write(root.join("x.json"), b"{}").unwrap();
    let origin = Origin::start(&root, &[]);
    let mut s = TcpStream::connect(("127.0.0.1", origin.port)).unwrap();
    s.set_read_timeout(Some(std::time::Duration::from_secs(20))).unwrap();
    let mut w = s.try_clone().unwrap();
    let t0 = std::time::Instant::now();
    let writer = std::thread::spawn(move || {
        let _ = w.write_all(b"GET /x.json HTTP/1.1\r\nHost: a\r\nX-Fill: ");
        let chunk = vec![b'a'; 1 << 20];
        let mut sent = 0usize;
        while sent < 400 << 20 {
            if w.write_all(&chunk).is_err() {
                break;
            }
            sent += chunk.len();
        }
        sent
    });
    let mut head = [0u8; 64];
    let n = s.read(&mut head).unwrap_or(0);
    let answer = String::from_utf8_lossy(&head[..n]).into_owned();
    drop(s);
    let sent = writer.join().unwrap();
    assert!(answer.starts_with("HTTP/1.1 431"), "{answer:?}");
    assert!(sent < 400 << 20, "the origin read the whole 400 MiB");
    assert!(t0.elapsed() < std::time::Duration::from_secs(30));
    // And it still serves.
    assert!(raw(origin.port, b"GET /x.json HTTP/1.1\r\nHost: a\r\n\r\n").starts_with("HTTP/1.1 200"));
}

/// `/__hits?path=` is percent-decoded as bytes: an escape before a non-ASCII character, or
/// an invalid UTF-8 sequence, is answered, not a panic.
#[test]
fn origin_decodes_hits_as_bytes() {
    let root = scratch("origin-hits");
    std::fs::write(root.join("x.json"), b"{}").unwrap();
    let origin = Origin::start(&root, &["--count-hits"]);
    for q in ["%aé", "%ff%fe", "%", "%4", "%e9%"] {
        let r = raw(origin.port, format!("GET /__hits?path={q} HTTP/1.1\r\nHost: a\r\n\r\n").as_bytes());
        assert!(r.starts_with("HTTP/1.1 200"), "{q}: {r:?}");
    }
    let _ = raw(origin.port, b"GET /x.json HTTP/1.1\r\nHost: a\r\n\r\n");
    let r = raw(origin.port, b"GET /__hits?path=%2Fx.json HTTP/1.1\r\nHost: a\r\n\r\n");
    assert!(r.ends_with("\r\n\r\n1"), "{r:?}");
}

/// A path segment holding `\`, `:`, NUL or a control byte, or `.`/`..`, names nothing:
/// 400.
#[test]
fn origin_refuses_backslash_drive_and_nul_paths() {
    let root = scratch("origin-paths");
    std::fs::write(root.join("x.json"), b"{}").unwrap();
    let origin = Origin::start(&root, &[]);
    for path in [&b"/..\\..\\etc\\passwd"[..], b"/C:/Windows/win.ini", b"/x.json:stream", b"/a\x00b", b"/a\x01b", b"/./x.json", b"/../x.json", b"//x.json", b"/x.json/"] {
        let mut req = b"GET ".to_vec();
        req.extend_from_slice(path);
        req.extend_from_slice(b" HTTP/1.1\r\nHost: a\r\n\r\n");
        let r = raw(origin.port, &req);
        assert!(r.starts_with("HTTP/1.1 400"), "{}: {r:?}", String::from_utf8_lossy(path));
    }
    assert!(raw(origin.port, b"GET /x.json HTTP/1.1\r\nHost: a\r\n\r\n").starts_with("HTTP/1.1 200"));
}

/// At most `--max-connections` at once (one more is answered 503), and a connection that
/// sends nothing is answered 408 and closed `--read-timeout-ms` after its accept.
#[test]
fn origin_caps_connections_and_times_out() {
    let root = scratch("origin-cap");
    std::fs::write(root.join("x.json"), b"{}").unwrap();
    let origin = Origin::start(&root, &["--max-connections", "2", "--read-timeout-ms", "700"]);
    let idle: Vec<TcpStream> = (0..2).map(|_| TcpStream::connect(("127.0.0.1", origin.port)).unwrap()).collect();
    std::thread::sleep(std::time::Duration::from_millis(200));
    let r = raw(origin.port, b"GET /x.json HTTP/1.1\r\nHost: a\r\n\r\n");
    assert!(r.starts_with("HTTP/1.1 503"), "{r:?}");
    let t0 = std::time::Instant::now();
    for mut c in idle {
        // Told 408, and closed.
        c.set_read_timeout(Some(std::time::Duration::from_secs(10))).unwrap();
        let mut out = Vec::new();
        let _ = c.read_to_end(&mut out);
        assert!(out.starts_with(b"HTTP/1.1 408"), "{:?}", String::from_utf8_lossy(&out));
    }
    assert!(t0.elapsed() < std::time::Duration::from_secs(5));
    assert!(raw(origin.port, b"GET /x.json HTTP/1.1\r\nHost: a\r\n\r\n").starts_with("HTTP/1.1 200"));
}

/// N1: a whole request head has one deadline from the accept, so connections that send a
/// byte at a time cannot hold the connection cap: a GET succeeds within about one head
/// deadline while they keep trickling.
#[test]
fn origin_survives_trickling_heads() {
    let root = scratch("origin-trickle");
    std::fs::write(root.join("x.json"), b"{}").unwrap();
    let origin = Origin::start(&root, &["--max-connections", "4", "--read-timeout-ms", "1000"]);
    let port = origin.port;
    let stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
    let tricklers: Vec<_> = (0..4).map(|_| {
        let stop = stop.clone();
        std::thread::spawn(move || {
            let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
            let mut sent = 0;
            // One byte every 100 ms, never ending the head, until the origin closes.
            for b in b"GET /x.json HTTP/1.1\r\nX-Slow: ".iter().chain(std::iter::repeat(&b'a')) {
                if stop.load(std::sync::atomic::Ordering::SeqCst) || s.write_all(&[*b]).is_err() {
                    break;
                }
                sent += 1;
                std::thread::sleep(std::time::Duration::from_millis(100));
            }
            sent
        })
    }).collect();
    std::thread::sleep(std::time::Duration::from_millis(200));
    let t0 = std::time::Instant::now();
    let mut answer = String::new();
    while t0.elapsed() < std::time::Duration::from_secs(5) {
        answer = raw(port, b"GET /x.json HTTP/1.1\r\nHost: a\r\n\r\n");
        if answer.starts_with("HTTP/1.1 200") {
            break;
        }
        std::thread::sleep(std::time::Duration::from_millis(50));
    }
    let waited = t0.elapsed();
    stop.store(true, std::sync::atomic::Ordering::SeqCst);
    for t in tricklers {
        t.join().unwrap();
    }
    assert!(answer.starts_with("HTTP/1.1 200"), "{answer:?}");
    assert!(waited < std::time::Duration::from_millis(2500), "the GET waited {waited:?}");
}

/// R3-3: with its defaults (no cap per peer), eight connections that trickle a head and
/// reconnect the moment they are closed never shut a legitimate GET out: on a loopback
/// listener every client has their address.
#[test]
fn origin_serves_past_reconnecting_tricklers() {
    let root = scratch("origin-reconnect");
    std::fs::write(root.join("x.json"), b"{}").unwrap();
    let origin = Origin::start(&root, &["--read-timeout-ms", "1000"]);
    let port = origin.port;
    let stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
    let tricklers: Vec<_> = (0..8).map(|_| {
        let stop = stop.clone();
        std::thread::spawn(move || {
            while !stop.load(std::sync::atomic::Ordering::SeqCst) {
                let Ok(mut s) = TcpStream::connect(("127.0.0.1", port)) else { continue };
                for b in b"GET /x.json HTTP/1.1\r\nX-Slow: ".iter().chain(std::iter::repeat(&b'a')) {
                    if stop.load(std::sync::atomic::Ordering::SeqCst) || s.write_all(&[*b]).is_err() {
                        break;
                    }
                    std::thread::sleep(std::time::Duration::from_millis(50));
                }
            }
        })
    }).collect();
    std::thread::sleep(std::time::Duration::from_millis(300));
    let t0 = std::time::Instant::now();
    let mut answers = Vec::new();
    while t0.elapsed() < std::time::Duration::from_secs(3) {
        answers.push(raw(port, b"GET /x.json HTTP/1.1\r\nHost: a\r\n\r\n").lines().next().unwrap_or("").to_string());
        std::thread::sleep(std::time::Duration::from_millis(100));
    }
    stop.store(true, std::sync::atomic::Ordering::SeqCst);
    for t in tricklers {
        t.join().unwrap();
    }
    assert!(answers.len() >= 10 && answers.iter().all(|a| a.starts_with("HTTP/1.1 200")), "{answers:?}");
}

/// R3-3: a connection lives at most `--max-connection-ms` from its accept, so a peer that
/// reads a large response slowly is cut off then, not served for as long as it keeps
/// reading.
#[test]
fn origin_cuts_a_slow_reader_at_the_lifetime() {
    let root = scratch("origin-slow-reader");
    let size = 32usize << 20;
    std::fs::write(root.join("big.bin"), vec![7u8; size]).unwrap();
    let origin = Origin::start(&root, &["--max-connection-ms", "1500"]);
    let mut s = TcpStream::connect(("127.0.0.1", origin.port)).unwrap();
    s.write_all(b"GET /big.bin HTTP/1.1\r\nHost: a\r\n\r\n").unwrap();
    s.set_read_timeout(Some(std::time::Duration::from_secs(10))).unwrap();
    let (t0, mut got, mut buf) = (std::time::Instant::now(), Vec::new(), [0u8; 4096]);
    // Slowly, past the lifetime: 4 KiB every 100 ms.
    while t0.elapsed() < std::time::Duration::from_millis(2500) {
        match s.read(&mut buf) {
            Ok(0) | Err(_) => break,
            Ok(n) => got.extend_from_slice(&buf[..n]),
        }
        std::thread::sleep(std::time::Duration::from_millis(100));
    }
    // Then as fast as it comes: what was already sent, and the end of the connection.
    let ended = loop {
        match s.read(&mut buf) {
            Ok(0) => break true,
            Ok(n) => got.extend_from_slice(&buf[..n]),
            Err(e) if matches!(e.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut) => break false,
            Err(_) => break true,
        }
    };
    assert!(got.starts_with(b"HTTP/1.1 200"), "{:?}", String::from_utf8_lossy(&got[..got.len().min(64)]));
    assert!(ended, "the connection was still open");
    assert!(got.len() < size, "the whole response arrived: the connection outlived its lifetime");
}

/// N1: at most `--max-per-peer` connections from one address, when it is given; one more
/// is answered 503.
#[test]
fn origin_per_peer_limit() {
    let root = scratch("origin-peer");
    std::fs::write(root.join("x.json"), b"{}").unwrap();
    let origin = Origin::start(&root, &["--max-per-peer", "2", "--read-timeout-ms", "60000"]);
    let request = b"GET /x.json HTTP/1.1\r\nHost: a\r\n\r\n";
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
    // The origin counts each connection on a thread of its own, so a third can overtake the idle
    // two and one of them is refused instead. An attempt that went that way starts again.
    let (idle, r) = loop {
        let idle: Vec<TcpStream> = (0..2).map(|_| TcpStream::connect(("127.0.0.1", origin.port)).unwrap()).collect();
        std::thread::sleep(std::time::Duration::from_millis(100));
        let r = raw(origin.port, request);
        if r.starts_with("HTTP/1.1 503") || std::time::Instant::now() > deadline {
            break (idle, r);
        }
    };
    assert!(r.starts_with("HTTP/1.1 503"), "{r:?}");
    // A refusal ends cleanly: closing with the request unread would end the connection in a reset,
    // which a client reading to the close takes for a failure, after a valid answer. So the read
    // itself must succeed here, and no attempt is retried.
    for i in 0..200 {
        let mut s = TcpStream::connect(("127.0.0.1", origin.port)).unwrap();
        s.set_read_timeout(Some(std::time::Duration::from_secs(10))).unwrap();
        let mut out = Vec::new();
        let read = s.write_all(request).and_then(|()| s.read_to_end(&mut out));
        let r = String::from_utf8_lossy(&out);
        assert!(read.is_ok() && r.starts_with("HTTP/1.1 503") && r.contains("Retry-After: 1"), "refusal {i}: {read:?} {r:?}");
    }
    drop(idle);
    let r = loop {
        let r = raw(origin.port, request);
        if r.starts_with("HTTP/1.1 200") || std::time::Instant::now() > deadline {
            break r;
        }
        std::thread::sleep(std::time::Duration::from_millis(50));
    };
    assert!(r.starts_with("HTTP/1.1 200"), "{r:?}");
}

/// N3: `wai http bench` times the base from the session's `base_ready`: with the origin
/// holding the enhancement's bytes back, the base's time does not include the delay, and
/// the time to complete does.
#[test]
fn bench_times_the_base_apart_from_its_refinement() {
    use wai::http_binding::head::{parse_head, HeadParse};
    let root = scratch("bench-delay");
    let src = object("O8");
    let env = std::fs::read(&src).unwrap();
    assert!(wai(&["http", "publish", src.to_str().unwrap(), "--out", root.to_str().unwrap()]).status.success());
    let HeadParse::Head(h) = parse_head(&env, env.len() as u64).unwrap() else { panic!("a head") };
    let enh = h.components[1];
    let delay = format!("{}-{}:1500", enh.start, enh.start + enh.len - 1);
    let origin = Origin::start(&root, &["--delay-range", &delay]);
    let rec = root.join("bench.json");
    let url = format!("http://127.0.0.1:{}/{}.wai#h={}&n={}", origin.port, hex(&content_hash(&env)), h.covered_len(), env.len());
    let run = wai(&["http", "bench", "--url", &url, "--runs", "2", "--advertise", "wai.video.avc", "--out", rec.to_str().unwrap()]);
    assert!(run.status.success(), "{}", String::from_utf8_lossy(&run.stderr));
    let r: serde_json::Value = serde_json::from_slice(&std::fs::read(&rec).unwrap()).unwrap();
    let base = r["ms_to_verified"]["median"].as_f64().unwrap();
    let all = r["ms_to_complete"]["median"].as_f64().unwrap();
    assert!(base < 1000.0, "the base's time includes the enhancement's delay: {base} ms");
    assert!(all >= 1500.0, "the enhancement was not delayed: {all} ms");
    // With a refinement deadline shorter than the delay, the enhancement is abandoned.
    let run = wai(&["http", "bench", "--url", &url, "--runs", "1", "--advertise", "wai.video.avc", "--refinement-deadline-ms", "1", "--out", rec.to_str().unwrap()]);
    assert!(run.status.success(), "{}", String::from_utf8_lossy(&run.stderr));
    let r: serde_json::Value = serde_json::from_slice(&std::fs::read(&rec).unwrap()).unwrap();
    assert_eq!(r["refinements_abandoned"], serde_json::json!([1]));
    assert!(r["ms_to_complete"]["median"].as_f64().unwrap() < 1000.0);
    drop(origin);
}

/// A classical playlist for the O9 segments (three of 640 ms), and an objects list over
/// them, written in a scratch directory with the corpus objects beside them.
fn o9_presentation(name: &str) -> PathBuf {
    let d = scratch(name);
    std::fs::write(d.join("classical.m3u8"), "#EXTM3U\n#EXT-X-VERSION:7\n#EXT-X-TARGETDURATION:1\n#EXT-X-MEDIA-SEQUENCE:0\n#EXTINF:0.640,\nc0.m4s\n#EXTINF:0.640,\nc1.m4s\n#EXTINF:0.640,\nc2.m4s\n#EXT-X-ENDLIST\n").unwrap();
    let objs: Vec<String> = ["O9a", "O9b", "O9c"].iter().map(|id| object(id).to_str().unwrap().to_string()).collect();
    let list = serde_json::json!({
        "uri_prefix": "objects/",
        "variants": [{ "uri": "wai-180.m3u8", "classical": "classical.m3u8", "resolution": [320, 180], "frame_rate_milli": 25000,
                       "segments": objs.iter().map(|o| serde_json::json!({ "object": o })).collect::<Vec<_>>() }],
    });
    std::fs::write(d.join("list.json"), serde_json::to_vec_pretty(&list).unwrap()).unwrap();
    d
}

/// `wai http hls` writes the WAI media playlist aligned with the classical one, and
/// `wai.m3u8` with the union, the floor and computed bandwidths; `wai http mpd` writes
/// `wai.mpd` on the same timeline.
#[test]
fn hls_and_mpd_write_aligned_playlists() {
    let d = o9_presentation("hls");
    let run = wai(&["http", "hls", "--objects", d.join("list.json").to_str().unwrap(), "--out", d.to_str().unwrap()]);
    assert!(run.status.success(), "{}", String::from_utf8_lossy(&run.stderr));
    let media = std::fs::read_to_string(d.join("wai-180.m3u8")).unwrap();
    let classical = std::fs::read_to_string(d.join("classical.m3u8")).unwrap();
    wai::http_binding::hls::check_adjacent(&media, &classical).unwrap();
    let cid = hex(&content_hash(&std::fs::read(object("O9b")).unwrap()));
    // A WAI1 is listed with its n alone; a WAI2 with its h too (§5.4).
    assert!(media.contains(&format!("#EXTINF:0.640,\nobjects/{cid}.wai#n=18004\n")), "{media}");
    assert!(media.contains("wai#n=21034&h=1009\n"), "{media}");
    let master = std::fs::read_to_string(d.join("wai.m3u8")).unwrap();
    assert_eq!(master, "#EXTM3U\n#EXT-X-STREAM-INF:BANDWIDTH=262925,AVERAGE-BANDWIDTH=239246,CODECS=\"wai2.video.av1|video.int_hyper\",X-WAI-FLOOR=\"wai2.video.av1\",RESOLUTION=320x180,FRAME-RATE=25.000\nwai-180.m3u8\n");
    let run = wai(&["http", "mpd", "--objects", d.join("list.json").to_str().unwrap(), "--out", d.join("wai.mpd").to_str().unwrap()]);
    assert!(run.status.success(), "{}", String::from_utf8_lossy(&run.stderr));
    let mpd = std::fs::read_to_string(d.join("wai.mpd")).unwrap();
    assert!(mpd.contains("<Representation id=\"wai-180\" bandwidth=\"262925\">") && mpd.contains("<S t=\"1280\" d=\"640\"/>"), "{mpd}");
    // An object list that does not cover the classical timeline is refused: three
    // objects for two classical segments.
    std::fs::write(d.join("classical.m3u8"), classical.replacen("#EXTINF:0.640,\nc2.m4s\n", "", 1)).unwrap();
    let run = wai(&["http", "hls", "--objects", d.join("list.json").to_str().unwrap(), "--out", d.to_str().unwrap()]);
    assert!(!run.status.success());
}

/// `wai http mpd --classical-mpd` takes the classical MPD's period, timescale, `t` and
/// `d`, leaves a gap's `S` and `SegmentURL` out, and checks the result adjacent to the
/// classical MPD; a list that does not cover the classical timeline is refused.
#[test]
fn mpd_takes_the_classical_mpd_timeline_and_leaves_gaps_out() {
    let d = o9_presentation("mpd-classical");
    let classical = "<?xml version=\"1.0\"?>\n<MPD xmlns=\"urn:mpeg:dash:schema:mpd:2011\" type=\"static\">\n  <Period id=\"main\" start=\"PT10S\">\n    <AdaptationSet id=\"7\" mimeType=\"video/mp4\">\n      <SegmentTemplate timescale=\"90000\" media=\"c$Number$.m4s\">\n        <SegmentTimeline><S t=\"0\" d=\"57600\" r=\"2\"/></SegmentTimeline>\n      </SegmentTemplate>\n      <Representation id=\"v\" bandwidth=\"1\"/>\n    </AdaptationSet>\n  </Period>\n</MPD>\n";
    std::fs::write(d.join("classical.mpd"), classical).unwrap();
    let mut list: serde_json::Value = serde_json::from_slice(&std::fs::read(d.join("list.json")).unwrap()).unwrap();
    list["variants"][0]["segments"][1]["gap"] = serde_json::json!(true);
    std::fs::write(d.join("list.json"), serde_json::to_vec(&list).unwrap()).unwrap();
    let args = |out: &str| vec!["http".to_string(), "mpd".into(), "--objects".into(), d.join("list.json").to_str().unwrap().into(), "--out".into(),
                                 d.join(out).to_str().unwrap().into(), "--classical-mpd".into(), d.join("classical.mpd").to_str().unwrap().into()];
    let run = wai(&args("wai.mpd").iter().map(String::as_str).collect::<Vec<_>>());
    assert!(run.status.success(), "{}", String::from_utf8_lossy(&run.stderr));
    let mpd = std::fs::read_to_string(d.join("wai.mpd")).unwrap();
    assert!(mpd.contains("<Period id=\"main\" start=\"PT10S\">") && mpd.contains("<SegmentList timescale=\"90000\">"), "{mpd}");
    assert!(mpd.contains("<S t=\"0\" d=\"57600\"/>") && mpd.contains("<S t=\"115200\" d=\"57600\"/>") && !mpd.contains("t=\"57600\""), "{mpd}");
    assert_eq!(mpd.matches("<SegmentURL ").count(), 2, "{mpd}");
    wai::http_binding::dash_read::check_dash_adjacent(&mpd, classical).unwrap();
    std::fs::write(d.join("classical.mpd"), classical.replace("r=\"2\"", "r=\"1\"")).unwrap();
    let run = wai(&args("short.mpd").iter().map(String::as_str).collect::<Vec<_>>());
    assert!(!run.status.success());
    assert!(String::from_utf8_lossy(&run.stderr).contains("3 segments, and the classical timeline has 2"), "{}", String::from_utf8_lossy(&run.stderr));
}

/// `wai http seal-playlist` writes `mr/<manifest_hash>.json` and
/// `mr/<receipt_hash>.json`; `check-playlist` reports it signed, refuses a changed
/// playlist, and holds a revision to the head it chains to, walking the revisions
/// between.
#[cfg(feature = "video_manifest")]
#[test]
fn seal_and_check_playlist() {
    let d = o9_presentation("seal");
    let run = wai(&["http", "hls", "--objects", d.join("list.json").to_str().unwrap(), "--out", d.to_str().unwrap()]);
    assert!(run.status.success(), "{}", String::from_utf8_lossy(&run.stderr));
    let pl = d.join("wai-180.m3u8");
    let key = "07".repeat(32);
    let run = wai(&["http", "seal-playlist", pl.to_str().unwrap(), "--key", &key, "--signer-id", "did:example:cli", "--content-id", "o9"]);
    assert!(run.status.success(), "{}", String::from_utf8_lossy(&run.stderr));
    let bytes = std::fs::read(&pl).unwrap();
    let name = format!("{}.json", blake3::hash(&bytes).to_hex());
    let receipt = d.join("mr").join(&name);
    assert!(String::from_utf8_lossy(&run.stdout).starts_with(receipt.to_str().unwrap()));
    let ok = wai(&["http", "check-playlist", pl.to_str().unwrap(), "--receipt", receipt.to_str().unwrap(), "--classical", d.join("classical.m3u8").to_str().unwrap()]);
    assert!(ok.status.success(), "{}", String::from_utf8_lossy(&ok.stdout));
    assert!(String::from_utf8_lossy(&ok.stdout).starts_with("signed\t"));
    let mut changed = bytes.clone();
    let at = changed.len() - 20;
    changed[at] ^= 1;
    std::fs::write(d.join("changed.m3u8"), &changed).unwrap();
    let bad = wai(&["http", "check-playlist", d.join("changed.m3u8").to_str().unwrap(), "--receipt", receipt.to_str().unwrap()]);
    assert_eq!(bad.status.code(), Some(1));
    assert_eq!(String::from_utf8_lossy(&bad.stdout), "refused\tplaylist_receipt_mismatch\n");
    // With no grant, one of no ceilings, written null; under the publisher's grant, its
    // ceilings; a ceiling past 2^53 - 1 is not sealed.
    assert!(String::from_utf8_lossy(&std::fs::read(&receipt).unwrap()).contains("\"joule_ceiling_micro\":null"));
    let grant = d.join("grant.json");
    std::fs::write(&grant, format!("{{\"grant_hash\":\"{}\",\"capability\":\"video.manifest\",\"joule_ceiling_micro\":5000000,\"funds_ceiling\":null}}", "ab".repeat(32))).unwrap();
    let granted = wai(&["http", "seal-playlist", pl.to_str().unwrap(), "--key", &key, "--grant", grant.to_str().unwrap(), "--out", d.join("granted").to_str().unwrap()]);
    assert!(granted.status.success(), "{}", String::from_utf8_lossy(&granted.stderr));
    let file = String::from_utf8(std::fs::read(d.join("granted/mr").join(&name)).unwrap()).unwrap();
    assert!(file.contains("\"joule_ceiling_micro\":5000000") && file.contains(&"ab".repeat(32)), "{file}");
    let ok = wai(&["http", "check-playlist", pl.to_str().unwrap(), "--receipt", d.join("granted/mr").join(&name).to_str().unwrap()]);
    assert!(ok.status.success(), "{}", String::from_utf8_lossy(&ok.stdout));
    std::fs::write(&grant, format!("{{\"grant_hash\":\"{}\",\"capability\":\"video.manifest\",\"joule_ceiling_micro\":9007199254740992,\"funds_ceiling\":null}}", "ab".repeat(32))).unwrap();
    let past = wai(&["http", "seal-playlist", pl.to_str().unwrap(), "--key", &key, "--grant", grant.to_str().unwrap(), "--out", d.join("past").to_str().unwrap()]);
    assert!(!past.status.success());
    // Each receipt is also written under its receipt hash.
    let hash_of = |out: &std::process::Output| String::from_utf8_lossy(&out.stdout).trim_end().rsplit(' ').next().unwrap().to_string();
    let r1_hash = hash_of(&run);
    assert_eq!(std::fs::read(d.join("mr").join(format!("{r1_hash}.json"))).unwrap(), std::fs::read(&receipt).unwrap());
    // A second revision sealed with the first as its parent checks against it; the
    // first fetched again is the head unchanged.
    let run = wai(&["http", "seal-playlist", pl.to_str().unwrap(), "--key", &key, "--parent", receipt.to_str().unwrap(), "--out", d.join("r2").to_str().unwrap()]);
    assert!(run.status.success(), "{}", String::from_utf8_lossy(&run.stderr));
    let r2 = d.join("r2/mr").join(&name);
    let ok = wai(&["http", "check-playlist", pl.to_str().unwrap(), "--receipt", r2.to_str().unwrap(), "--parent", receipt.to_str().unwrap()]);
    assert!(ok.status.success(), "{}", String::from_utf8_lossy(&ok.stdout));
    assert!(String::from_utf8_lossy(&ok.stdout).contains("\tchained (0 parents walked)"));
    let same = wai(&["http", "check-playlist", pl.to_str().unwrap(), "--receipt", receipt.to_str().unwrap(), "--parent", receipt.to_str().unwrap()]);
    assert!(same.status.success(), "{}", String::from_utf8_lossy(&same.stdout));
    // A third revision, chained to the second, checked against the first: the second is
    // fetched by its link from the third's directory; without it the walk stops.
    let r2_file = d.join("r2/mr").join(format!("{}.json", hash_of(&run)));
    let run = wai(&["http", "seal-playlist", pl.to_str().unwrap(), "--key", &key, "--parent", r2_file.to_str().unwrap(), "--out", d.join("r2").to_str().unwrap()]);
    assert!(run.status.success(), "{}", String::from_utf8_lossy(&run.stderr));
    let ok = wai(&["http", "check-playlist", pl.to_str().unwrap(), "--receipt", r2.to_str().unwrap(), "--parent", receipt.to_str().unwrap()]);
    assert!(String::from_utf8_lossy(&ok.stdout).contains("\tchained (1 parents walked)"), "{}", String::from_utf8_lossy(&ok.stdout));
    std::fs::remove_file(&r2_file).unwrap();
    let bad = wai(&["http", "check-playlist", pl.to_str().unwrap(), "--receipt", r2.to_str().unwrap(), "--parent", receipt.to_str().unwrap()]);
    assert_eq!(String::from_utf8_lossy(&bad.stdout), "refused\tplaylist_receipt_parent_missing\n");
}

/// The origin serves a playlist to be revalidated on every use, and its receipt
/// immutable.
#[test]
fn origin_serves_playlists_revalidated() {
    let d = o9_presentation("origin-playlists");
    let run = wai(&["http", "hls", "--objects", d.join("list.json").to_str().unwrap(), "--out", d.to_str().unwrap()]);
    assert!(run.status.success());
    std::fs::write(d.join("wai.mpd"), "<MPD/>\n").unwrap();
    let o = Origin::start(&d, &[]);
    let (h, body) = get(o.port, "/wai-180.m3u8", &[]);
    assert!(h.contains("Content-Type: application/vnd.apple.mpegurl\r\n") && h.contains("Cache-Control: no-cache, no-transform\r\n"), "{h}");
    assert_eq!(body, std::fs::read(d.join("wai-180.m3u8")).unwrap());
    let (h, _) = get(o.port, "/wai.mpd", &[]);
    assert!(h.contains("Content-Type: application/dash+xml\r\n") && h.contains("Cache-Control: no-cache, no-transform\r\n"), "{h}");
}

/// The replayed live edge (§11.4): the playlist as its current revision; a blocking
/// reload for a part answered by a revision that lists it; a hinted part's alias held
/// until a revision lists it, and answered with the object whose content hash is the cid
/// that revision lists; an alias served with another part's bytes, which that cid
/// refuses; and a hint no revision ever fills, not found once the edge ends. Nothing here
/// reads a clock: each answer is checked by what it holds.
#[test]
fn origin_replays_a_live_edge() {
    let root = corpus();
    let revs: Vec<String> = (0..4).map(|k| format!("live/wai-r{k}.m3u8")).collect();
    let texts: Vec<Vec<u8>> = revs.iter().map(|r| std::fs::read(root.join(r)).unwrap()).collect();
    let edge = format!("live/wai-720.m3u8={}", revs.join(","));
    let o = Origin::start(&root, &["--live-edge", &edge, "--live-step-ms", "400", "--live-swap", "720/1201.7.wai=720/1201.8.wai"]);
    let (h, body) = get(o.port, "/live/wai-720.m3u8", &[]);
    assert!(h.starts_with("HTTP/1.1 200") && h.contains("Cache-Control: no-cache, no-transform\r\n"), "{h}");
    assert!(texts.contains(&body), "the playlist is one of its revisions");
    // A blocking reload more than two segments past the edge, or a part with no segment,
    // is refused at once (RFC 8216bis §6.2.5.2), not held.
    // (A held reload is answered 200 with a revision; 400 is the refusal.)
    for q in ["_HLS_msn=1300", "_HLS_msn=1300&_HLS_part=0", "_HLS_part=1", "_HLS_msn=x", "_HLS_msn=-1"] {
        let (h, _) = get(o.port, &format!("/live/wai-720.m3u8?{q}"), &[]);
        assert!(h.starts_with("HTTP/1.1 400"), "{q}: {h}");
    }
    // Part 2 of segment 1201 is first listed by revision 1.
    let (_, body) = get(o.port, "/live/wai-720.m3u8?_HLS_msn=1201&_HLS_part=2", &[]);
    let p = wai::http_binding::hls::parse_media_playlist(&String::from_utf8(body.clone()).unwrap()).unwrap();
    assert!(wai::http_binding::hls::listed_cid(&p, "720/1201.2.wai").is_some(), "the blocking reload lists the part");
    assert!(texts[1..].contains(&body));
    // Revision 0 hints 1200.5; revision 1 lists it with its cid.
    let r1 = wai::http_binding::hls::parse_media_playlist(&String::from_utf8(texts[1].clone()).unwrap()).unwrap();
    let (h, part) = get(o.port, "/live/720/1200.5.wai", &[]);
    assert!(h.starts_with("HTTP/1.1 200"), "{h}");
    assert_eq!(Some(content_hash(&part)), wai::http_binding::hls::listed_cid(&r1, "720/1200.5.wai"));
    // Served with another part's bytes: the cid the playlist lists refuses them.
    let r2 = wai::http_binding::hls::parse_media_playlist(&String::from_utf8(texts[2].clone()).unwrap()).unwrap();
    let (_, swapped) = get(o.port, "/live/720/1201.7.wai", &[]);
    assert_ne!(Some(content_hash(&swapped)), wai::http_binding::hls::listed_cid(&r2, "720/1201.7.wai"));
    assert_eq!(Some(content_hash(&swapped)), wai::http_binding::hls::listed_cid(&r2, "720/1201.8.wai"));
    // Revision 2 hints 1202.0, which no revision fills: not found when the edge ends,
    // and the playlist is then its last revision, which ends.
    let (h, _) = get(o.port, "/live/720/1202.0.wai", &[]);
    assert!(h.starts_with("HTTP/1.1 404"), "{h}");
    let (_, body) = get(o.port, "/live/wai-720.m3u8", &[]);
    assert_eq!(body, texts[3]);
    // A byte range of a part, under its alias.
    let (h, b) = get(o.port, "/live/720/1200.5.wai", &[("Range", "bytes=0-7")]);
    assert!(h.starts_with("HTTP/1.1 206") && b == part[..8], "{h}");
}

/// `wai http receipts` (§8.1–§8.3) against the reference origin serving the corpus:
/// O3 fetched by its head proof, its group accepted on that cid, and its claims (the
/// ancestor's soft binding and a receipt claim with its parent fetched by link), each
/// signer confirmed under the pinned key log; O9c's group walked back two parents; a
/// layer not fetched inclusion-verified; and a walk past its bound refused.
#[cfg(feature = "key_discovery")]
#[test]
fn receipts_checks_an_index_over_http() {
    let o = Origin::start(&corpus(), &[]);
    let url = |id: &str| {
        let p = object(id);
        format!("http://127.0.0.1:{}/{}", o.port, p.strip_prefix(corpus()).unwrap().to_str().unwrap())
    };
    let log = corpus().join(".well-known/wai-keys");
    let run = wai(&["http", "receipts", "--url", &url("O3"), "--advertise", "wai.video.av1", "--pin-key-log", log.to_str().unwrap(),
        "--key-origin", "https://origin.example"]);
    let out = String::from_utf8_lossy(&run.stdout);
    let v: serde_json::Value = serde_json::from_str(&out).unwrap_or_else(|e| panic!("{e}: {out}{}", String::from_utf8_lossy(&run.stderr)));
    assert_eq!((v["fetch"].as_str(), v["held"].as_str()), (Some("head-proof"), Some("head-proof")), "{v}");
    let d = &v["delivery"][0];
    assert_eq!((d["evidence"].as_str(), d["established"].as_str()), (Some("delivered-verified"), Some("head-proof")), "{v}");
    assert!(d["confirm"].as_str().unwrap().starts_with("confirmed origin=https://origin.example key_id=d1"), "{v}");
    // O3's claims: a soft binding of O7, with no recovery run here, is about another
    // object; a receipt claim with its parent; an energy claim checked at step 6 against
    // the object's figure, and one that disagrees with it.
    assert!(v["claims"][0]["refused"].as_str().unwrap().starts_with("receipt_not_about_object"), "{v}");
    assert_eq!(v["claims"][1]["attests"], "receipt");
    assert!(v["claims"][1]["parent"].is_string(), "{v}");
    assert_eq!(v["claims"][2]["attests"], "object");
    assert!(v["claims"][3]["refused"].as_str().unwrap().starts_with("energy_disagrees_with_delivery"), "{v}");
    assert!(!run.status.success(), "a refused claim makes the exit status 1");
    let run = wai(&["http", "receipts", "--url", &url("O9c"), "--advertise", "wai.video.av1"]);
    let v: serde_json::Value = serde_json::from_slice(&run.stdout).unwrap();
    assert!(run.status.success(), "{v}");
    assert_eq!((v["delivery"][0]["walked"].as_u64(), v["delivery"][0]["admission"].as_str()), (Some(2), Some("accept")), "{v}");
    let run = wai(&["http", "receipts", "--url", &url("O10L2"), "--held", "none"]);
    let v: serde_json::Value = serde_json::from_slice(&run.stdout).unwrap();
    assert_eq!((v["held"].as_str(), v["delivery"][0]["evidence"].as_str()), (Some("nothing"), Some("inclusion-verified")), "{v}");
    let run = wai(&["http", "receipts", "--url", &url("O9c"), "--held", "none", "--max-walk", "1"]);
    assert!(!run.status.success());
    let v: serde_json::Value = serde_json::from_slice(&run.stdout).unwrap();
    assert!(v["delivery"][0]["refused"].as_str().unwrap().starts_with("parent_not_held"), "{v}");
    // The origin serves the index for revalidation and the status for a quarter of its
    // freshness at most.
    let cid = hex(&content_hash(&std::fs::read(object("O3")).unwrap()));
    let (h, _) = get(o.port, &format!("/objects/{cid}.wai.receipts"), &[]);
    assert!(h.contains("Cache-Control: no-cache, no-transform"), "{h}");
    let (h, _) = get(o.port, "/.well-known/wai-key-status", &[]);
    assert!(h.contains("Cache-Control: max-age=900, no-transform"), "{h}");
}

/// A test server for `wai http receipts`: each connection's request head is kept, and it
/// answers `mode`: `log` (the corpus's key log, 200), `redirect` (302 to another host),
/// `hang` (a status line and headers, then nothing), `stream` (a body that does not end).
#[cfg(feature = "key_discovery")]
fn test_server(mode: &'static str) -> (u16, std::sync::Arc<std::sync::Mutex<Vec<String>>>) {
    let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
    let port = l.local_addr().unwrap().port();
    let seen = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
    let kept = seen.clone();
    std::thread::spawn(move || {
        for s in l.incoming().flatten() {
            let kept = kept.clone();
            std::thread::spawn(move || {
                let mut s = s;
                if mode == "plain" {
                    // What a client sends first, recorded as it arrives, answered in plain
                    // HTTP with the key log.
                    let _ = s.set_read_timeout(Some(std::time::Duration::from_secs(10)));
                    let mut first = [0u8; 512];
                    let n = s.read(&mut first).unwrap_or(0);
                    kept.lock().unwrap().push(String::from_utf8_lossy(&first[..n]).into_owned());
                    let body = std::fs::read(corpus().join(".well-known/wai-keys")).unwrap();
                    let _ = s.write_all(format!("HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n", body.len()).as_bytes());
                    let _ = s.write_all(&body);
                    return;
                }
                let mut head = Vec::new();
                let mut b = [0u8; 1];
                while !head.ends_with(b"\r\n\r\n") && s.read(&mut b).map_or(false, |n| n == 1) {
                    head.push(b[0]);
                }
                kept.lock().unwrap().push(String::from_utf8_lossy(&head).into_owned());
                match mode {
                    "log" => {
                        let body = std::fs::read(corpus().join(".well-known/wai-keys")).unwrap();
                        let _ = s.write_all(format!("HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n", body.len()).as_bytes());
                        let _ = s.write_all(&body);
                    }
                    "biglog" => {
                        // The log padded past its cap (1 MiB), framed by Content-Length.
                        let mut body = std::fs::read(corpus().join(".well-known/wai-keys")).unwrap();
                        body.resize(1_048_621, b' ');
                        let _ = s.write_all(format!("HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n", body.len()).as_bytes());
                        let _ = s.write_all(&body);
                    }
                    "redirect" => {
                        let _ = s.write_all(b"HTTP/1.1 302 Found\r\nLocation: https://mirror.example/.well-known/wai-keys\r\nContent-Length: 0\r\n\r\n");
                    }
                    "hang" => {
                        let _ = s.write_all(b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n\r\n");
                        std::thread::sleep(std::time::Duration::from_secs(30));
                    }
                    _ => {
                        // A body that does not end before any deadline the test gives: 64 KiB
                        // every 5 ms, for longer than a minute, until the client hangs up.
                        let _ = s.write_all(b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n\r\n");
                        let chunk = vec![b' '; 64 << 10];
                        for _ in 0..(1 << 14) {
                            if s.write_all(&chunk).is_err() {
                                break;
                            }
                            std::thread::sleep(std::time::Duration::from_millis(5));
                        }
                    }
                }
            });
        }
    });
    (port, seen)
}

/// §8.4's test aid: `--test-plaintext-connect` fetches the key log over plain HTTP from a
/// stated address, as the TLS path asks for it (revalidating, no redirect followed). What
/// it retrieves confirms this run's signers and is labelled `test-plaintext`, never
/// `fetched`, as anchoring nothing a verifier may rely on, and the command fails; a
/// redirect anchors nothing. `--connect` is gone.
#[cfg(feature = "key_discovery")]
#[test]
fn receipts_labels_a_plaintext_test_anchor_as_such() {
    let o = Origin::start(&corpus(), &[]);
    let p = object("O3");
    let url = format!("http://127.0.0.1:{}/{}", o.port, p.strip_prefix(corpus()).unwrap().to_str().unwrap());
    let (kp, seen) = test_server("log");
    let run = wai(&["http", "receipts", "--url", &url, "--held", "none", "--key-origin", "https://origin.example",
        "--test-plaintext-connect", &format!("127.0.0.1:{kp}")]);
    let v: serde_json::Value = serde_json::from_slice(&run.stdout).unwrap_or_else(|e| panic!("{e}: {}", String::from_utf8_lossy(&run.stderr)));
    assert_eq!(v["anchor"]["path"], "test-plaintext", "{v}");
    assert_eq!(v["anchor"]["relied_on"], false, "{v}");
    assert!(v["anchor"]["note"].as_str().unwrap().contains("anchors nothing a verifier may rely on"), "{v}");
    assert!(v["delivery"][0]["confirm"].as_str().unwrap().starts_with("confirmed origin=https://origin.example key_id=d1"), "{v}");
    // Every confirmation names the test anchor, and no anchor anywhere is `fetched`.
    let confirms: Vec<&str> = ["delivery", "claims"].iter().flat_map(|k| v[*k].as_array().unwrap().iter())
        .filter_map(|x| x["confirm"].as_str()).collect();
    assert!(confirms.iter().filter(|c| c.starts_with("confirmed")).count() >= 2, "{v}");
    for c in confirms.iter().filter(|c| c.starts_with("confirmed")) {
        assert!(c.contains(" anchor=test-plaintext"), "{c}");
    }
    let out = String::from_utf8_lossy(&run.stdout);
    assert!(!out.contains("anchor=fetched") && !out.contains("\"fetched\""), "{v}");
    let req = seen.lock().unwrap().join("");
    assert!(req.starts_with("GET /.well-known/wai-keys HTTP/1.1\r\n"), "{req}");
    assert!(req.contains("\r\nHost: origin.example\r\n") && req.contains("\r\nCache-Control: no-cache\r\n"), "{req}");
    // A redirect is not followed, and the answer anchors nothing.
    let (rp, seen) = test_server("redirect");
    let run = wai(&["http", "receipts", "--url", &url, "--held", "none", "--key-origin", "https://origin.example",
        "--test-plaintext-connect", &format!("127.0.0.1:{rp}")]);
    assert!(!run.status.success());
    let v: serde_json::Value = serde_json::from_slice(&run.stdout).unwrap();
    assert_eq!(v["anchor"]["refused"], "key_log_not_from_origin", "{v}");
    assert!(v["delivery"][0].get("confirm").is_none(), "{v}");
    assert_eq!(seen.lock().unwrap().len(), 1, "the redirect was followed");
    // A log past its cap is refused, read no further than one byte past it.
    let (bp, _) = test_server("biglog");
    let run = wai(&["http", "receipts", "--url", &url, "--held", "none", "--key-origin", "https://origin.example",
        "--test-plaintext-connect", &format!("127.0.0.1:{bp}")]);
    let v: serde_json::Value = serde_json::from_slice(&run.stdout).unwrap();
    assert_eq!(v["anchor"]["refused"], "key_log_too_large", "{v}");
    // The flag that let any server stand in for the origin is gone.
    let run = wai(&["http", "receipts", "--url", &url, "--held", "none", "--key-origin", "https://origin.example", "--connect", &format!("127.0.0.1:{kp}")]);
    assert!(!run.status.success() && String::from_utf8_lossy(&run.stderr).contains("--connect is gone"));
}

/// The test aid's anchor never lets the command succeed: on an object whose index passes
/// whole (O7), the same check exits 0 with the log pinned and 1 with it fetched over
/// plaintext from a stated address.
#[cfg(feature = "key_discovery")]
#[test]
fn receipts_never_succeeds_on_a_test_anchor() {
    let o = Origin::start(&corpus(), &[]);
    let p = object("O7");
    let url = format!("http://127.0.0.1:{}/{}", o.port, p.strip_prefix(corpus()).unwrap().to_str().unwrap());
    let log = corpus().join(".well-known/wai-keys");
    let pinned = wai(&["http", "receipts", "--url", &url, "--held", "none", "--key-origin", "https://origin.example",
        "--pin-key-log", log.to_str().unwrap()]);
    let v: serde_json::Value = serde_json::from_slice(&pinned.stdout).unwrap_or_else(|e| panic!("{e}: {}", String::from_utf8_lossy(&pinned.stderr)));
    assert!(pinned.status.success(), "a pinned anchor on a whole index succeeds: {v}");
    assert_eq!(v["anchor"]["path"], "pinned", "{v}");
    let (kp, _) = test_server("log");
    let test = wai(&["http", "receipts", "--url", &url, "--held", "none", "--key-origin", "https://origin.example",
        "--test-plaintext-connect", &format!("127.0.0.1:{kp}")]);
    let w: serde_json::Value = serde_json::from_slice(&test.stdout).unwrap();
    assert_eq!(w["anchor"]["path"], "test-plaintext", "{w}");
    let strip = |x: &serde_json::Value| x.as_array().map(|d| json_without_confirm(d)).unwrap_or_default();
    assert_eq!((strip(&w["delivery"]), &w["index"]), (strip(&v["delivery"]), &v["index"]), "the same check but for its anchor");
    assert!(!test.status.success(), "a plaintext anchor never makes the command succeed: {w}");
}

/// Delivery verdicts with their confirmations dropped: the part an anchor does not
/// decide.
fn json_without_confirm(d: &[serde_json::Value]) -> Vec<serde_json::Value> {
    d.iter().map(|x| { let mut x = x.clone(); x.as_object_mut().map(|m| m.remove("confirm")); x }).collect()
}

/// §8.4: without the test aid, the key log is retrieved over TLS from the origin itself,
/// so a server that speaks plain HTTP at the origin's address is never asked for it in
/// plaintext and anchors nothing; a build without TLS asks nothing at all.
#[cfg(feature = "key_discovery")]
#[test]
fn receipts_never_anchors_over_plaintext() {
    let o = Origin::start(&corpus(), &[]);
    let p = object("O3");
    let url = format!("http://127.0.0.1:{}/{}", o.port, p.strip_prefix(corpus()).unwrap().to_str().unwrap());
    let (kp, seen) = test_server("plain");
    let run = wai(&["http", "receipts", "--url", &url, "--held", "none", "--key-origin", &format!("https://127.0.0.1:{kp}")]);
    assert!(!run.status.success());
    let v: serde_json::Value = serde_json::from_slice(&run.stdout).unwrap_or_else(|e| panic!("{e}: {}", String::from_utf8_lossy(&run.stderr)));
    assert_eq!(v["anchor"]["refused"], "key_log_not_from_origin", "{v}");
    assert!(v["anchor"].get("path").is_none(), "{v}");
    assert!(v["delivery"][0].get("confirm").is_none(), "{v}");
    let sent = seen.lock().unwrap().join("");
    assert!(!sent.contains("GET "), "a plaintext request reached the server: {sent:?}");
    // With TLS the client opens with a handshake, unless the host has no trust store to
    // validate against, when it refuses before it connects.
    #[cfg(feature = "cli_tls")]
    assert!(sent.starts_with('\u{16}') || (sent.is_empty() && v["anchor"]["why"].as_str().unwrap().contains("TLS")),
        "the client opened with a TLS handshake: {sent:?} {v}");
    #[cfg(not(feature = "cli_tls"))]
    {
        assert!(sent.is_empty(), "a build without TLS connects to nothing: {sent:?}");
        assert!(v["anchor"]["why"].as_str().unwrap().contains("TLS"), "{v}");
    }
}

/// §8.2 over HTTP: a server that stops after its headers, or never ends its body, costs
/// `wai http receipts` no more than its deadline and the index's cap. The hanging server
/// holds its connection open for 30 s and the stream runs past a minute, so a client
/// without the deadline, or without the cap, reports something else.
#[cfg(feature = "key_discovery")]
#[test]
fn receipts_is_bounded_by_its_deadline_and_caps() {
    let cid = "31abc1da2cdf927191f3f447f68508ead30d078690cb7bea2435ae3d1df5f4f3";
    let (hp, _) = test_server("hang");
    let run = wai(&["http", "receipts", "--held", "none", "--deadline-ms", "1500", "--url", &format!("http://127.0.0.1:{hp}/o/{cid}.wai")]);
    assert!(!run.status.success());
    assert!(String::from_utf8_lossy(&run.stderr).contains("deadline"), "{}", String::from_utf8_lossy(&run.stderr));
    let (sp, _) = test_server("stream");
    let run = wai(&["http", "receipts", "--held", "none", "--deadline-ms", "30000", "--url", &format!("http://127.0.0.1:{sp}/o/{cid}.wai")]);
    assert!(!run.status.success());
    let v: serde_json::Value = serde_json::from_slice(&run.stdout).unwrap_or_else(|e| panic!("{e}: {}", String::from_utf8_lossy(&run.stderr)));
    assert!(v["index"].as_str().unwrap().starts_with("receipt_index_too_large"), "{v}");
}