//! 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", "2000"]);
    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:?}");
    drop(idle);
    std::thread::sleep(std::time::Duration::from_millis(200));
    assert!(raw(origin.port, b"GET /x.json HTTP/1.1\r\nHost: a\r\n\r\n").starts_with("HTTP/1.1 200"));
}

/// 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}");
}