crates/hub/tests/flow.rs at v0.3.0 · andymccall/oxygenhub

◆ v0.3.0 oxygenhub / crates / hub / tests / flow.rs
80.7 KiB · Rust · last changed End the jobs a runner no longer holds at its next claim 1 month ago source blame history raw
//! OxygenFlow on the hub (hub spec §13.2), end to end: a real `oxy` push
//! starts a run, a runner claims jobs through the API in dependency order,
//! job tokens read only their repository, results become checks and pages.

use axum::body::Body;
use axum::http::{header, Request, StatusCode};
use http_body_util::BodyExt;
use oxy_core::protocol::{open_remote, RemoteOptions};
use oxy_core::repo::Repo;
use oxygenhub::config::Config;
use oxygenhub::repos::RepoName;
use oxygenhub::{db, Shared};
use std::path::Path;
use tower::ServiceExt;

const ROOTFS: &str = "36fc4e3238044e424219a2b9002bf61bbe1805fec14d52cf37e17fae77c9ff16";

fn hermetic() {
    static ONCE: std::sync::Once = std::sync::Once::new();
    static DIR: std::sync::OnceLock<tempfile::TempDir> = std::sync::OnceLock::new();
    ONCE.call_once(|| {
        let d = tempfile::tempdir().unwrap();
        std::env::set_var("OXY_HOME", d.path());
        DIR.set(d).ok();
    });
}

fn write(root: &Path, rel: &str, content: &str) {
    let p = root.join(rel);
    std::fs::create_dir_all(p.parent().unwrap()).unwrap();
    std::fs::write(p, content).unwrap();
}

struct Hub {
    app: Shared,
    base: String,
    _dir: tempfile::TempDir,
}

async fn hub() -> Hub {
    hermetic();
    let dir = tempfile::tempdir().unwrap();
    let mut config = Config::load(dir.path()).unwrap();
    config.auth.push_tokens = vec!["site".into()];
    config.flow.runner_tokens = vec!["runner-secret".into()];
    config.flow.repos = vec!["andy/*".into()];
    config.registration.mode = "open".into();
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let port = listener.local_addr().unwrap().port();
    config.server.base_url = format!("http://127.0.0.1:{port}");
    let app = oxygenhub::build(config).await.unwrap();
    let router = oxygenhub::router(app.clone());
    tokio::spawn(async move { axum::serve(listener, router).await.unwrap() });
    Hub { app, base: format!("http://127.0.0.1:{port}"), _dir: dir }
}

async fn call(app: &Shared, method: &str, path: &str, bearer: Option<&str>, json: Option<String>) -> (StatusCode, String) {
    let mut req = Request::builder().method(method).uri(path);
    if let Some(b) = bearer {
        req = req.header(header::AUTHORIZATION, format!("Bearer {b}"));
    }
    let body = match json {
        Some(j) => {
            req = req.header(header::CONTENT_TYPE, "application/json");
            Body::from(j)
        }
        None => Body::empty(),
    };
    let res = oxygenhub::router(app.clone()).oneshot(req.body(body).unwrap()).await.unwrap();
    let status = res.status();
    let bytes = res.into_body().collect().await.unwrap().to_bytes();
    (status, String::from_utf8_lossy(&bytes).into_owned())
}

/// A GET's status, `Location` and `Content-Type`.
async fn head(app: &Shared, path: &str) -> (StatusCode, String, String) {
    let res = oxygenhub::router(app.clone()).oneshot(Request::get(path).body(Body::empty()).unwrap()).await.unwrap();
    let h = |k: header::HeaderName| res.headers().get(k).and_then(|v| v.to_str().ok()).unwrap_or("").to_string();
    (res.status(), h(header::LOCATION), h(header::CONTENT_TYPE))
}

async fn claim(app: &Shared) -> Option<serde_json::Value> {
    let (st, body) =
        call(app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(r#"{"runner":"test-1","runtimes":["microvm"]}"#.into())).await;
    match st {
        StatusCode::NO_CONTENT => None,
        StatusCode::OK => Some(serde_json::from_str(&body).unwrap()),
        other => panic!("claim: {other} {body}"),
    }
}

fn receipt(job: &str, pipeline_hash: &str, exit: u32) -> serde_json::Value {
    serde_json::json!({
        "job_name": job, "pipeline_hash": pipeline_hash, "scope_tree_hash": "11".repeat(32),
        "inputs": [], "environment": "22".repeat(32), "outputs": [], "exit_code": exit, "duration_ms": 640
    })
}

async fn finish(app: &Shared, id: i64, body: serde_json::Value) -> StatusCode {
    call(app, "POST", &format!("/api/v1/flow/jobs/{id}/finish"), Some("runner-secret"), Some(body.to_string())).await.0
}

async fn create(app: &Shared, owner: &str, name: &str) {
    app.repos.create(&RepoName::parse(&format!("{owner}/{name}")).unwrap(), None).unwrap();
    db::repos::create(&app.db, owner, name, "", hub_api::Visibility::Public).await.unwrap();
}

fn client(url: &str, work: &Path) -> Repo {
    let mut c = Repo::init(work).unwrap();
    c.config.set("remote.origin.url", url).unwrap();
    c.config.set("remote.origin.token", "site").unwrap();
    c.save_config().unwrap();
    c
}

fn pipeline(extra: &str) -> String {
    format!(
        "schema-version 1\nsandbox \"alpine\" {{\n    rootfs \"cas://blob/{ROOTFS}\"\n}}\njob \"build\" {{\n    sandbox \"alpine\"\n    exec \"make\"\n}}\njob \"test\" {{\n    sandbox \"alpine\"\n    depends-on \"build\"\n    exec \"sh\" \"-eu\" \"test.sh\"\n}}\njob \"lint\" {{\n    sandbox \"alpine\"\n    exec \"true\"\n}}\n{extra}"
    )
}

async fn runs_of(app: &Shared, owner: &str, name: &str) -> Vec<db::flow::Run> {
    let row = db::repos::get(&app.db, owner, name).await.unwrap().unwrap();
    for _ in 0..100 {
        let r = db::flow::runs(&app.db, row.id, 10, None).await.unwrap();
        if !r.is_empty() {
            return r;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    vec![]
}

/// The checks on a state once there are `n` of them (they follow the run).
async fn checks_when(app: &Shared, repo_id: i64, state: &str, n: usize) -> Vec<db::hooks::Check> {
    for _ in 0..100 {
        let c = db::hooks::checks_for_state(&app.db, repo_id, state).await.unwrap();
        if c.len() >= n {
            return c;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    db::hooks::checks_for_state(&app.db, repo_id, state).await.unwrap()
}

#[tokio::test(flavor = "multi_thread")]
async fn a_push_runs_the_pipeline_through_a_runner() {
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    create(&h.app, "andy", "secret").await;
    create(&h.app, "other", "plain").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let mut c = client(&url, work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "test.sh", "echo ok\n");
    c.snapshot("test").unwrap();
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));

    // The push started a run with three queued jobs, and checks for them.
    let runs = runs_of(&h.app, "andy", "demo").await;
    assert_eq!(runs.len(), 1);
    let run = &runs[0];
    assert_eq!((run.bookmark.as_str(), run.status.as_str(), run.actor.as_str()), ("main", "queued", "site token"));
    assert_eq!(run.pipeline_hash.len(), 64);
    let jobs = db::flow::jobs(&h.app.db, run.id).await.unwrap();
    assert_eq!(jobs.iter().map(|j| (j.name.as_str(), j.depth)).collect::<Vec<_>>(), [("build", 0), ("lint", 0), ("test", 1)]);
    // The checks are reported just after the run is recorded.
    let checks = checks_when(&h.app, run.repo_id, &run.state, 3).await;
    assert_eq!(
        checks.iter().map(|c| (c.name.as_str(), c.status.as_str())).collect::<Vec<_>>(),
        [("flow/build", "queued"), ("flow/lint", "queued"), ("flow/test", "queued")]
    );

    // Runners need a runner token.
    assert_eq!(
        call(&h.app, "POST", "/api/v1/flow/claim", Some("site"), Some(r#"{"runner":"x","runtimes":["microvm"]}"#.into())).await.0,
        StatusCode::UNAUTHORIZED
    );
    assert_eq!(
        call(&h.app, "POST", "/api/v1/flow/claim", None, Some(r#"{"runner":"x","runtimes":["microvm"]}"#.into())).await.0,
        StatusCode::UNAUTHORIZED
    );
    // A runner without the tier gets nothing.
    let (st, _) =
        call(&h.app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(r#"{"runner":"x","runtimes":["wasm"]}"#.into())).await;
    assert_eq!(st, StatusCode::NO_CONTENT);

    // Claims come in dependency order; `test` waits for `build`.
    let a = claim(&h.app).await.unwrap();
    let b = claim(&h.app).await.unwrap();
    assert_eq!((a["job"]["name"].as_str(), b["job"]["name"].as_str()), (Some("build"), Some("lint")));
    assert!(claim(&h.app).await.is_none(), "test waits for build");
    assert_eq!(a["repo"], "andy/demo");
    assert_eq!(a["clone_url"], url.as_str());
    assert_eq!(a["sandbox"]["rootfs"], ROOTFS);
    assert_eq!(a["job"]["exec"], serde_json::json!(["make"]));
    assert_eq!(a["timeout_s"], 3600);
    // The job is told what it builds (FLOW_STATE, FLOW_BOOKMARK, FLOW_RUN).
    assert_eq!(
        (a["state"].as_str(), a["bookmark"].as_str(), a["run_number"].as_i64()),
        (Some(run.state.as_str()), Some("main"), Some(run.number))
    );
    let heads: Vec<String> = a["heads"].as_array().unwrap().iter().map(|v| v.as_str().unwrap().to_string()).collect();
    assert_eq!(heads.join("+"), run.heads);
    let checks = db::hooks::checks_for_state(&h.app.db, run.repo_id, &run.state).await.unwrap();
    assert_eq!(checks.iter().find(|c| c.name == "flow/build").unwrap().status, "running");

    // The job token reads this repository over the wire, and nothing else.
    let token = a["token"].as_str().unwrap().to_string();
    assert!(token.starts_with("flowjob_"));
    let opts = RemoteOptions { token: Some(token.clone()), ca: None, repo_id: None };
    open_remote(&url, &opts).unwrap().refs().unwrap();
    assert!(open_remote(&format!("{}/andy/secret", h.base), &opts).unwrap().refs().is_err(), "another repository");
    let bad = RemoteOptions { token: Some("flowjob_00".into()), ca: None, repo_id: None };
    assert!(open_remote(&url, &bad).unwrap().refs().is_err(), "an unknown job token");
    // … and cannot push.
    c.config.set("remote.origin.token", &token).unwrap();
    c.save_config().unwrap();
    write(work.path(), "more.txt", "x\n");
    c.snapshot("test").unwrap();
    c.commit("More", &[]).unwrap();
    let pushed = c.push("origin", false);
    assert!(pushed.is_err() || pushed.unwrap().iter().all(|r| !r.ok), "a job token cannot push");
    c.config.set("remote.origin.token", "site").unwrap();
    c.save_config().unwrap();

    // Logs, then results with receipts.
    let a_id = a["job_id"].as_i64().unwrap();
    let b_id = b["job_id"].as_i64().unwrap();
    let log = serde_json::json!({ "records": [
        { "t": 5, "stream": "flow", "text": "booting" },
        { "t": 120, "stream": "out", "text": "cc -o demo demo.c" },
        { "t": 130, "stream": "weird", "text": "x".repeat(20_000) },
    ]});
    assert_eq!(
        call(&h.app, "POST", &format!("/api/v1/flow/jobs/{a_id}/log"), Some("runner-secret"), Some(log.to_string())).await.0,
        StatusCode::NO_CONTENT
    );
    let lines = db::flow::log(&h.app.db, a_id, 0, 100).await.unwrap();
    assert_eq!(lines.len(), 3);
    assert_eq!((lines[2].stream.as_str(), lines[2].text.len()), ("console", 16 * 1024), "streams are known ones, text is capped");
    let ph = a["pipeline_hash"].as_str().unwrap();
    assert_eq!(
        finish(&h.app, a_id, serde_json::json!({ "exit_code": 0, "duration_ms": 700, "receipt": receipt("lint", ph, 0) })).await,
        StatusCode::BAD_REQUEST,
        "a receipt for another job"
    );
    assert_eq!(
        finish(&h.app, a_id, serde_json::json!({ "exit_code": 0, "duration_ms": 700, "receipt": receipt("build", ph, 0) })).await,
        StatusCode::NO_CONTENT
    );
    assert_eq!(finish(&h.app, a_id, serde_json::json!({ "exit_code": 0, "duration_ms": 700 })).await, StatusCode::CONFLICT, "only once");
    assert!(open_remote(&url, &opts).unwrap().refs().is_err(), "the token ends with its job");
    assert_eq!(
        finish(&h.app, b_id, serde_json::json!({ "exit_code": 0, "duration_ms": 300, "receipt": receipt("lint", ph, 0) })).await,
        StatusCode::NO_CONTENT
    );
    let t = claim(&h.app).await.unwrap();
    assert_eq!(t["job"]["name"], "test");
    let t_id = t["job_id"].as_i64().unwrap();
    assert_eq!(
        finish(&h.app, t_id, serde_json::json!({ "exit_code": 3, "duration_ms": 500, "receipt": receipt("test", ph, 3) })).await,
        StatusCode::NO_CONTENT
    );

    let run = db::flow::run_by_id(&h.app.db, run.id).await.unwrap().unwrap();
    assert_eq!(run.status, "failed");
    let jobs = db::flow::jobs(&h.app.db, run.id).await.unwrap();
    let build = jobs.iter().find(|j| j.name == "build").unwrap();
    assert_eq!((build.status.as_str(), build.exit_code, build.receipt_hash.len()), ("passed", Some(0), 64));
    assert_eq!(jobs.iter().find(|j| j.name == "test").unwrap().status, "failed");
    let checks = db::hooks::checks_for_state(&h.app.db, run.repo_id, &run.state).await.unwrap();
    assert_eq!(
        checks.iter().map(|c| (c.name.as_str(), c.status.as_str())).collect::<Vec<_>>(),
        [("flow/build", "passed"), ("flow/lint", "passed"), ("flow/test", "failed")]
    );
    assert!(checks[2].summary.contains("exit 3") && checks[2].url.ends_with(&format!("/andy/demo/flow/{}", run.number)));

    // Pages: the list, the run with its graph and logs, and the live view.
    let (st, list) = call(&h.app, "GET", "/andy/demo/flow", None, None).await;
    assert_eq!(st, StatusCode::OK);
    assert!(list.contains(&format!("/andy/demo/flow/{}", run.number)) && list.contains("st-failed"), "{list}");
    let (st, page) = call(&h.app, "GET", &format!("/andy/demo/flow/{}", run.number), None, None).await;
    assert_eq!(st, StatusCode::OK);
    assert!(page.contains("<svg class=\"flow-graph\""), "graph");
    assert_eq!(page.matches("data-job-node=").count(), 3);
    assert_eq!(page.matches("<path class=\"edge").count(), 1, "one dependency edge");
    assert!(page.contains("cc -o demo demo.c"), "log shown");
    assert!(!page.contains("data-flow-live"), "a finished run does not poll");
    let (st, live) = call(&h.app, "GET", &format!("/andy/demo/flow/{}/live?after={a_id}:1", run.number), None, None).await;
    assert_eq!(st, StatusCode::OK);
    let live: serde_json::Value = serde_json::from_str(&live).unwrap();
    assert_eq!(live["status"], "failed");
    let build_live = live["jobs"].as_array().unwrap().iter().find(|j| j["name"] == "build").unwrap().clone();
    assert_eq!(build_live["lines"].as_array().unwrap().len(), 2, "lines after seq 1");
    assert_eq!(call(&h.app, "GET", "/andy/demo/flow/999", None, None).await.0, StatusCode::NOT_FOUND);

    // A failed dependency skips what follows; an unrun job is an error.
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "test.sh", "echo changed\n");
    c.snapshot("test").unwrap();
    c.commit("Change the test", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let mut second = None;
    for _ in 0..100 {
        let r = runs_of(&h.app, "andy", "demo").await;
        if r.len() == 2 {
            second = Some(r[0].clone());
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    let second = second.expect("a second run");
    let x = claim(&h.app).await.unwrap();
    let y = claim(&h.app).await.unwrap();
    let (build_id, lint_id) = if x["job"]["name"] == "build" {
        (x["job_id"].as_i64().unwrap(), y["job_id"].as_i64().unwrap())
    } else {
        (y["job_id"].as_i64().unwrap(), x["job_id"].as_i64().unwrap())
    };
    assert_eq!(
        finish(&h.app, build_id, serde_json::json!({ "exit_code": null, "duration_ms": 10, "error": "rootfs is not on this runner" }))
            .await,
        StatusCode::NO_CONTENT
    );
    assert_eq!(finish(&h.app, lint_id, serde_json::json!({ "exit_code": 0, "duration_ms": 10 })).await, StatusCode::NO_CONTENT);
    assert!(claim(&h.app).await.is_none(), "test was skipped");
    let jobs = db::flow::jobs(&h.app.db, second.id).await.unwrap();
    assert_eq!(
        jobs.iter().map(|j| (j.name.as_str(), j.status.as_str())).collect::<Vec<_>>(),
        [("build", "error"), ("lint", "passed"), ("test", "skipped")]
    );
    assert_eq!(db::flow::run_by_id(&h.app.db, second.id).await.unwrap().unwrap().status, "failed");

    // An invalid pipeline is a run in error with the line.
    write(work.path(), ".flow/pipeline.kdl", &pipeline("job \"broken\" {\n    sandbox \"missing\"\n    exec \"x\"\n}\n"));
    c.snapshot("test").unwrap();
    c.commit("Break the pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let mut third = None;
    for _ in 0..100 {
        let r = runs_of(&h.app, "andy", "demo").await;
        if r.len() == 3 {
            third = Some(r[0].clone());
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    let third = third.expect("a third run");
    assert_eq!(third.status, "error");
    assert!(third.error.contains("line 18") && third.error.contains("\"missing\""), "{}", third.error);
    let (_, page) = call(&h.app, "GET", &format!("/andy/demo/flow/{}", third.number), None, None).await;
    assert!(page.contains("The pipeline is not valid"));

    // State stamps (hub spec §13.2) and images in READMEs.
    write(work.path(), "README.md", "# Demo\n\n[![flow](.flow/stamp)](.flow/run)\n\n![logo](img/logo.png)\n");
    std::fs::create_dir_all(work.path().join("img")).unwrap();
    std::fs::write(work.path().join("img/logo.png"), b"\x89PNG\r\n\x1a\n\0\0\0\rIHDR\0\0\0\x01\0\0\0\x01\x08\x06\0\0\0\x1f\x15\xc4\x89")
        .unwrap();
    c.snapshot("test").unwrap();
    c.commit("A README with a stamp", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let mut fourth = None;
    for _ in 0..100 {
        let r = runs_of(&h.app, "andy", "demo").await;
        if r.len() == 4 {
            fourth = Some(r[0].clone());
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    let fourth = fourth.expect("a fourth run");
    let (_, overview) = call(&h.app, "GET", "/andy/demo", None, None).await;
    assert!(overview.contains("src=\"/andy/demo/raw/main/.flow/stamp\""), "{overview}");
    assert!(overview.contains("href=\"/andy/demo/file/main/.flow/run\""));
    assert!(overview.contains("src=\"/andy/demo/raw/main/img/logo.png\""), "images use the raw route");
    let (st, _, ct) = head(&h.app, "/andy/demo/raw/main/img/logo.png").await;
    assert_eq!((st, ct.as_str()), (StatusCode::OK, "image/png"));
    let (st, loc, _) = head(&h.app, "/andy/demo/raw/main/.flow/stamp").await;
    assert!(st.is_redirection() && loc == "/andy/demo/flow/stamp/main", "{st} {loc}");
    let (st, loc, _) = head(&h.app, "/andy/demo/flow/stamp/main").await;
    assert!(st.is_redirection() && loc == format!("/andy/demo/flow/stamp/{}", fourth.heads), "a bookmark's stamp names its state: {loc}");
    let (st, _, ct) = head(&h.app, &loc).await;
    assert_eq!((st, ct.as_str()), (StatusCode::OK, "image/svg+xml; charset=utf-8"));
    let (_, svg) = call(&h.app, "GET", &loc, None, None).await;
    assert!(svg.starts_with("<svg") && svg.contains("error ·") && svg.contains(&fourth.heads[..12]), "{svg}");
    let (_, svg1) = call(&h.app, "GET", &format!("/andy/demo/flow/stamp/{}", run.heads), None, None).await;
    assert!(svg1.contains("failed ·") && svg1.contains(&format!("run #{} failed", run.id)), "an older state keeps its own result: {svg1}");
    assert!(svg1.contains(&build.receipt_hash), "receipt hashes in the metadata");
    assert_eq!(svg1.matches("text-anchor=\"middle\"").count(), 3, "one glyph per job");
    let (st, loc, _) = head(&h.app, "/andy/demo/file/main/.flow/run").await;
    assert!(st.is_redirection() && loc == "/andy/demo/flow/at/main", "{loc}");
    let (st, loc, _) = head(&h.app, "/andy/demo/flow/at/main").await;
    assert!(st.is_redirection() && loc == format!("/andy/demo/flow/{}", fourth.id), "{loc}");

    // A repository outside [flow] repos starts nothing.
    let work2 = tempfile::tempdir().unwrap();
    let c2 = client(&format!("{}/other/plain", h.base), work2.path());
    write(work2.path(), ".flow/pipeline.kdl", &pipeline(""));
    c2.snapshot("test").unwrap();
    c2.commit("Pipeline", &[]).unwrap();
    assert!(c2.push("origin", false).unwrap().iter().all(|r| r.ok));
    tokio::time::sleep(std::time::Duration::from_millis(500)).await;
    let row = db::repos::get(&h.app.db, "other", "plain").await.unwrap().unwrap();
    assert!(db::flow::runs(&h.app.db, row.id, 10, None).await.unwrap().is_empty());
    let (_, page) = call(&h.app, "GET", "/other/plain/flow", None, None).await;
    assert!(page.contains("not enabled for this repository"));
    let (_, loc, _) = head(&h.app, "/other/plain/flow/stamp/main").await;
    let (_, svg) = call(&h.app, "GET", &loc, None, None).await;
    assert!(svg.contains("no run ·"), "{svg}");
    let (_, loc, _) = head(&h.app, "/other/plain/flow/at/main").await;
    assert_eq!(loc, "/other/plain/flow");
}

#[tokio::test(flavor = "multi_thread")]
async fn claims_carry_pinned_subrepos_and_the_job_token_reads_only_those() {
    let h = hub().await;
    for n in ["base", "lib", "vault", "other", "app"] {
        create(&h.app, "andy", n).await;
    }
    let push = |name: &str, file: &str| {
        let work = tempfile::tempdir().unwrap();
        let c = client(&format!("{}/andy/{name}", h.base), work.path());
        write(work.path(), file, "x\n");
        c.snapshot("t").unwrap();
        c.commit("first", &[]).unwrap();
        assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
        let heads = c.bookmark_heads().unwrap();
        (work, heads)
    };
    let (_b, base_heads) = push("base", "base.rs");
    // lib pins base in turn: a subrepo of a subrepo.
    let lib_work = tempfile::tempdir().unwrap();
    let lib = client(&format!("{}/andy/lib", h.base), lib_work.path());
    lib.submodule_add(&format!("{}/andy/base", h.base), "deps/base", Some("main")).unwrap();
    write(lib_work.path(), "lib.rs", "x\n");
    lib.snapshot("t").unwrap();
    lib.commit("lib on base", &[]).unwrap();
    assert!(lib.push("origin", false).unwrap().iter().all(|r| r.ok));
    let lib_heads = lib.bookmark_heads().unwrap();
    let (_v, _) = push("vault", "secret.rs");
    let (_o, _) = push("other", "o.rs");

    // The app pins lib and vault as subrepos and pushes as the site token.
    let work = tempfile::tempdir().unwrap();
    let app = client(&format!("{}/andy/app", h.base), work.path());
    app.submodule_add(&format!("{}/andy/lib", h.base), "deps/lib", Some("main")).unwrap();
    // Vault is private: a checkout with no token for this hub cannot add
    // it, but the app's remote on the same hub lends its token (oxygen#32).
    // The subrepo keeps no copy: its own fetches find the parent's (oxygen#47).
    sqlx::query("UPDATE repositories SET visibility = 'private' WHERE name = 'vault'").execute(&h.app.db).await.unwrap();
    let bare_work = tempfile::tempdir().unwrap();
    let mut bare = Repo::init(bare_work.path()).unwrap();
    bare.config.set("remote.origin.url", &format!("{}/andy/bare", h.base)).unwrap();
    bare.save_config().unwrap();
    assert!(bare.submodule_add(&format!("{}/andy/vault", h.base), "deps/vault", Some("main")).is_err(), "not without a token");
    app.submodule_add(&format!("{}/andy/vault", h.base), "deps/vault", Some("main")).unwrap();
    let nested = Repo::open(&work.path().join("deps/vault")).unwrap();
    assert!(nested.config.remote_tokens.is_empty(), "no token is copied into a subrepo");
    nested.fetch("origin").unwrap();
    // The pusher below is the site token, which has no account: it cannot read vault.
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "Makefile", "all:\n");
    app.snapshot("t").unwrap();
    app.commit("app with deps", &[]).unwrap();
    assert!(app.push("origin", false).unwrap().iter().all(|r| r.ok));
    assert!(!runs_of(&h.app, "andy", "app").await.is_empty());

    let job = claim(&h.app).await.unwrap();
    let subs = job["subrepos"].as_array().cloned().unwrap_or_default();
    assert_eq!(subs.len(), 2, "lib and what lib pins; vault is private and the pusher has no account: {subs:?}");
    assert_eq!(subs[1]["path"], "deps/lib/deps/base", "nested under its parent, after it");
    assert_eq!(subs[1]["heads"], serde_json::json!(base_heads.iter().map(|h| h.to_hex()).collect::<Vec<_>>()));
    let _ = std::fs::remove_dir_all(work.path().join("deps/vault2"));
    assert_eq!(subs[0]["path"], "deps/lib");
    assert_eq!(subs[0]["repo"], "andy/lib");
    assert_eq!(subs[0]["heads"], serde_json::json!(lib_heads.iter().map(|h| h.to_hex()).collect::<Vec<_>>()));
    assert!(subs[0]["clone_url"].as_str().unwrap().ends_with("/andy/lib"));
    // The job's log says vault was left out, and why (oxygenhub#56).
    let job_id = job["job_id"].as_i64().unwrap();
    let lines = db::flow::log(&h.app.db, job_id, 0, 100).await.unwrap();
    assert!(
        lines.iter().any(|l| l.stream == "flow" && l.text.contains("subrepo deps/vault (andy/vault) is left out")),
        "{:?}",
        lines.iter().map(|l| &l.text).collect::<Vec<_>>()
    );

    // The job token reads its repository and the pinned lib, nothing else.
    let token = job["token"].as_str().unwrap().to_string();
    let read = |name: &str| {
        let opts = RemoteOptions { token: Some(token.clone()), ..Default::default() };
        open_remote(&format!("{}/andy/{name}", h.base), &opts).and_then(|r| r.refs()).is_ok()
    };
    assert!(read("app") && read("lib") && read("base"));
    assert!(!read("vault"), "not pinned for this pusher");
    assert!(!read("other"), "public, but not the job's to read");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn concurrent_log_appends_never_fail_on_a_locked_database() {
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    let work = tempfile::tempdir().unwrap();
    let c = client(&format!("{}/andy/demo", h.base), work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "Makefile", "all:\n");
    c.snapshot("t").unwrap();
    c.commit("pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let run = runs_of(&h.app, "andy", "demo").await.remove(0);
    let job = db::flow::jobs(&h.app.db, run.id).await.unwrap().remove(0);
    // Many writers at once, each reading the count and then inserting: a
    // deferred transaction would fail with "database is locked".
    let mut tasks = Vec::new();
    for w in 0..16 {
        let pool = h.app.db.clone();
        tasks.push(tokio::spawn(async move {
            let records: Vec<flow_core::api::Record> =
                (0..20).map(|i| flow_core::api::Record { t: i, stream: "out".into(), text: format!("writer {w} line {i}") }).collect();
            db::flow::append_log(&pool, job.id, &records, 100_000).await
        }));
    }
    for t in tasks {
        assert_eq!(t.await.unwrap().unwrap(), 20, "every append succeeds");
    }
    let lines = db::flow::log(&h.app.db, job.id, 0, 1_000).await.unwrap();
    assert_eq!(lines.len(), 320);
    let seqs: std::collections::BTreeSet<i64> = lines.iter().map(|l| l.seq).collect();
    assert_eq!(seqs.len(), 320, "sequence numbers are unique");
}

/// A repository's own Flow setting overrides hub.toml either way.
#[tokio::test(flavor = "multi_thread")]
async fn a_repository_setting_turns_flow_on_or_off() {
    let h = hub().await;
    create(&h.app, "other", "plain").await;
    create(&h.app, "andy", "quiet").await;
    // A run elsewhere first, so hub-wide ids and per-repository numbers differ.
    create(&h.app, "andy", "first").await;
    {
        let work = tempfile::tempdir().unwrap();
        let c = client(&format!("{}/andy/first", h.base), work.path());
        write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
        write(work.path(), "test.sh", "echo ok\n");
        c.commit("Add a pipeline", &[]).unwrap();
        assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
        assert_eq!(runs_of(&h.app, "andy", "first").await.len(), 1);
    }
    let row = db::repos::get(&h.app.db, "other", "plain").await.unwrap().unwrap();
    db::repos::set_flow(&h.app.db, row.id, "on").await.unwrap();
    let quiet = db::repos::get(&h.app.db, "andy", "quiet").await.unwrap().unwrap();
    db::repos::set_flow(&h.app.db, quiet.id, "off").await.unwrap();
    for (owner, name) in [("other", "plain"), ("andy", "quiet")] {
        let work = tempfile::tempdir().unwrap();
        let c = client(&format!("{}/{owner}/{name}", h.base), work.path());
        write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
        write(work.path(), "test.sh", "echo ok\n");
        c.commit("Add a pipeline", &[]).unwrap();
        assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    }
    let plain = runs_of(&h.app, "other", "plain").await;
    assert_eq!(plain.len(), 1, "turned on");
    // With andy/first's jobs ahead, plain's run says where it stands.
    let a = claim(&h.app).await.unwrap();
    assert_eq!(a["repo"], "andy/first");
    let (_, page) = call(&h.app, "GET", "/other/plain/flow/1", None, None).await;
    assert!(page.contains("Waiting for a runner: 3rd in line") && page.contains("Running now: andy/first #1"), "{page}");
    let (_, body) = call(&h.app, "GET", "/api/v1/repos/other/plain/flow/runs/1", None, None).await;
    assert_eq!(serde_json::from_str::<serde_json::Value>(&body).unwrap()["queue_position"], 3);
    let (_, list) = call(&h.app, "GET", "/other/plain/flow", None, None).await;
    assert!(list.contains("queued, 3rd in line"), "{list}");
    // Runs are numbered per repository: this is plain's first, whatever
    // other repositories ran before.
    assert_eq!(plain[0].number, 1);
    let checks = checks_when(&h.app, row.id, &plain[0].state, 3).await;
    assert!(checks.iter().all(|c| c.url.ends_with("/other/plain/flow/1")), "{checks:?}");
    let (st, page) = call(&h.app, "GET", "/other/plain/flow/1", None, None).await;
    assert!(st == StatusCode::OK && page.contains("Run #1"), "{st}");
    assert!(runs_of(&h.app, "andy", "quiet").await.is_empty(), "turned off despite andy/*");
}

/// "Run pipeline" and "Re-run" (hub spec §13.2): runs that move no
/// bookmark, through the API and the pages.
#[tokio::test(flavor = "multi_thread")]
async fn runs_start_without_a_push_and_finished_runs_re_run() {
    let h = hub().await;
    // The site token acts as the first site administrator.
    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    create(&h.app, "andy", "demo").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "test.sh", "echo ok\n");
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let first = runs_of(&h.app, "andy", "demo").await;
    assert_eq!((first.len(), first[0].trigger.as_str()), (1, "push"));

    // Run pipeline on the default bookmark: a new run of the same state.
    let (st, body) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs", Some("site"), Some("{}".into())).await;
    assert_eq!(st, StatusCode::CREATED, "{body}");
    let v: serde_json::Value = serde_json::from_str(&body).unwrap();
    assert_eq!(v["number"], 2);
    assert!(v["url"].as_str().unwrap().ends_with("/andy/demo/flow/2"));
    let second = db::flow::run(&h.app.db, first[0].repo_id, 2).await.unwrap().unwrap();
    assert_eq!((second.trigger.as_str(), second.actor.as_str(), second.state.as_str()), ("manual", "andy", first[0].state.as_str()));
    assert_eq!(db::flow::jobs(&h.app.db, second.id).await.unwrap().len(), 3);
    let (_, page) = call(&h.app, "GET", "/andy/demo/flow/2", None, None).await;
    assert!(page.contains("started by andy"), "the run page says what started it");

    // Run pipeline on one change: its state, with no bookmark.
    let head = first[0].head_list()[0].to_string();
    let (st, body) =
        call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs", Some("site"), Some(format!(r#"{{"change":"{head}"}}"#))).await;
    assert_eq!(st, StatusCode::CREATED, "{body}");
    let on_change = db::flow::run(&h.app.db, first[0].repo_id, 3).await.unwrap().unwrap();
    assert_eq!(
        (on_change.bookmark.as_str(), on_change.heads.as_str(), on_change.state.as_str()),
        ("", head.as_str(), first[0].state.as_str())
    );
    let (_, page) = call(&h.app, "GET", "/andy/demo/flow/3", None, None).await;
    assert!(page.contains("title=\"Run on one change, not a bookmark\">change</span>"), "run page");
    let (st, body) =
        call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs", Some("site"), Some(format!(r#"{{"change":"{}"}}"#, "ab".repeat(32))))
            .await;
    assert!(st == StatusCode::CONFLICT && body.contains("no change"), "{st} {body}");
    // Viewers who cannot start runs are told why, and get no button.
    let (_, page) = call(&h.app, "GET", "/andy/demo/flow", None, None).await;
    assert!(page.contains("Sign in with write access to run the pipeline.") && !page.contains("primary-btn"));
    let (_, page) = call(&h.app, "GET", "/andy/demo/bookmarks", None, None).await;
    assert!(!page.contains("/flow/start"), "no run buttons for anonymous viewers");

    // A live run is not re-run; unknown bookmarks and runs are refused.
    let (st, body) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs/1/rerun", Some("site"), None).await;
    assert!(st == StatusCode::CONFLICT && body.contains("still queued"), "{st} {body}");
    let (st, _) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs", Some("site"), Some(r#"{"bookmark":"nope"}"#.into())).await;
    assert_eq!(st, StatusCode::CONFLICT);
    let (st, _) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs/99/rerun", Some("site"), None).await;
    assert_eq!(st, StatusCode::NOT_FOUND);
    // Anonymous callers cannot start runs.
    let (st, _) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs", None, Some("{}".into())).await;
    assert_ne!(st, StatusCode::CREATED);

    // A finished run re-runs as a new run of the same state.
    let row = db::repos::get(&h.app.db, "andy", "demo").await.unwrap().unwrap();
    for r in [&first[0], &second] {
        for j in db::flow::jobs(&h.app.db, r.id).await.unwrap() {
            sqlx::query("UPDATE flow_jobs SET status = 'passed' WHERE id = ?").bind(j.id).execute(&h.app.db).await.unwrap();
        }
        sqlx::query("UPDATE flow_runs SET status = 'passed', finished_ms = 1 WHERE id = ?").bind(r.id).execute(&h.app.db).await.unwrap();
    }
    let (st, body) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs/1/rerun", Some("site"), None).await;
    assert_eq!(st, StatusCode::CREATED, "{body}");
    let third = db::flow::run(&h.app.db, row.id, 4).await.unwrap().unwrap();
    assert_eq!((third.trigger.as_str(), third.status.as_str(), third.state.as_str()), ("rerun:1", "queued", first[0].state.as_str()));
    let checks = db::hooks::checks_for_state(&h.app.db, row.id, &third.state).await.unwrap();
    assert!(
        checks.iter().all(|c| c.status == "queued" && c.url.ends_with("/flow/4")),
        "the re-run's checks replace the old ones: {checks:?}"
    );
    let (_, page) = call(&h.app, "GET", "/andy/demo/flow/4", None, None).await;
    assert!(page.contains("re-run of #1 by andy"));
    assert!(!page.contains("RE-RUN"), "anonymous viewers get no button");

    // With Flow off, nothing starts.
    db::repos::set_flow(&h.app.db, row.id, "off").await.unwrap();
    let (st, body) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs", Some("site"), Some("{}".into())).await;
    assert!(st == StatusCode::CONFLICT && body.contains("not enabled"), "{body}");
    let (_, page) = call(&h.app, "GET", "/andy/demo/flow", None, None).await;
    assert!(page.contains("Flow is off for this repository"), "and the Flow tab says so");
}

/// The Flow API (hub spec §13.2): runs, a run with its jobs, a job's log,
/// checks by the heads a client has, and JSON for unknown API routes.
#[tokio::test(flavor = "multi_thread")]
async fn runs_jobs_and_logs_are_in_the_api() {
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "test.sh", "echo ok\n");
    let head = c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let run = runs_of(&h.app, "andy", "demo").await.remove(0);

    // A job runs and logs.
    let a = claim(&h.app).await.unwrap();
    let id = a["job_id"].as_i64().unwrap();
    let recs =
        serde_json::json!({ "records": [{ "t": 1, "stream": "stdout", "text": "one" }, { "t": 3, "stream": "stderr", "text": "three" }] });
    let (st, _) = call(&h.app, "POST", &format!("/api/v1/flow/jobs/{id}/log"), Some("runner-secret"), Some(recs.to_string())).await;
    assert!(st.is_success(), "{st}");

    let (st, body) = call(&h.app, "GET", "/api/v1/repos/andy/demo/flow/runs", None, None).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let runs: Vec<hub_api::FlowRun> = serde_json::from_str(&body).unwrap();
    assert_eq!(runs.len(), 1);
    assert_eq!((runs[0].number, runs[0].state.as_str(), runs[0].trigger.as_str()), (1, run.state.as_str(), "push"));
    assert!(runs[0].url.ends_with("/andy/demo/flow/1") && runs[0].jobs.is_empty());

    let (_, body) = call(&h.app, "GET", "/api/v1/repos/andy/demo/flow/runs/1", None, None).await;
    let one: hub_api::FlowRun = serde_json::from_str(&body).unwrap();
    let names: Vec<(&str, &str)> = one.jobs.iter().map(|j| (j.name.as_str(), j.status.as_str())).collect();
    assert_eq!(names.len(), 3);
    let running = one.jobs.iter().find(|j| j.status == "running").unwrap();
    assert_eq!(running.name, a["job"]["name"].as_str().unwrap());
    assert!(one.jobs.iter().any(|j| j.name == "test" && j.depends_on == ["build"]));

    let job = running.name.clone();
    let (st, body) = call(&h.app, "GET", &format!("/api/v1/repos/andy/demo/flow/runs/1/jobs/{job}/log"), None, None).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let lines: Vec<hub_api::FlowLogLine> = serde_json::from_str(&body).unwrap();
    assert!(lines.iter().any(|l| l.text == "three"), "{body}");
    let last = lines.last().unwrap().seq;
    let (_, body) = call(&h.app, "GET", &format!("/api/v1/repos/andy/demo/flow/runs/1/jobs/{job}/log?after={last}"), None, None).await;
    assert_eq!(body, "[]", "nothing after the last line");
    assert_eq!(call(&h.app, "GET", "/api/v1/repos/andy/demo/flow/runs/9", None, None).await.0, StatusCode::NOT_FOUND);
    assert_eq!(call(&h.app, "GET", "/api/v1/repos/andy/demo/flow/runs/1/jobs/nope/log", None, None).await.0, StatusCode::NOT_FOUND);

    // Checks by the state hash, by the head `oxy log` shows, and by bookmark.
    let by = |key: String| {
        let app = h.app.clone();
        async move {
            let (st, body) = call(&app, "GET", &format!("/api/v1/repos/andy/demo/checks/{key}"), None, None).await;
            assert_eq!(st, StatusCode::OK, "{key}: {body}");
            serde_json::from_str::<Vec<hub_api::Check>>(&body).unwrap().len()
        }
    };
    checks_when(&h.app, run.repo_id, &run.state, 3).await;
    assert_eq!(by(run.state.clone()).await, 3);
    assert_eq!(by(head.to_hex()).await, 3, "a single head names its state");
    assert_eq!(by("main".into()).await, 3);
    let (st, _) = call(&h.app, "GET", "/api/v1/repos/andy/demo/checks/no-such-bookmark", None, None).await;
    assert_eq!(st, StatusCode::NOT_FOUND);

    // An unknown API route answers with a problem document.
    let (st, body) = call(&h.app, "GET", "/api/v1/users/andy/repos", None, None).await;
    assert_eq!(st, StatusCode::NOT_FOUND);
    let v: serde_json::Value = serde_json::from_str(&body).unwrap();
    assert_eq!(v["status"], 404, "{body}");
}

async fn get_bytes(app: &Shared, path: &str) -> (StatusCode, Vec<u8>) {
    let res = oxygenhub::router(app.clone()).oneshot(Request::get(path).body(Body::empty()).unwrap()).await.unwrap();
    let status = res.status();
    (status, res.into_body().collect().await.unwrap().to_bytes().to_vec())
}

#[tokio::test(flavor = "multi_thread")]
async fn signed_receipts_are_checked_kept_and_shown() {
    use oxy_core::hash::{hash_object, ChangeHash, ObjectRef, ObjectType};
    use oxy_core::identity::SecretKey;
    use oxy_core::objects::StateObject;
    use oxy_core::receipt::SignedReceipt;
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    c.snapshot("test").unwrap();
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let run = runs_of(&h.app, "andy", "demo").await.remove(0);
    let heads: Vec<ChangeHash> = run.head_list().into_iter().filter_map(ChangeHash::from_hex).collect();
    let state = hash_object(ObjectType::State, &StateObject::new(heads.clone()).encode());
    let repo_id = h.app.repos.open(&RepoName::parse("andy/demo").unwrap()).unwrap().id().unwrap();
    let sk = SecretKey::generate();
    let runner = sk.identity("test-1 (OxygenFlow runner)", "", 1);
    let sign = |r: &serde_json::Value, state: ObjectRef, sk: &SecretKey, id: &oxy_core::identity::Identity| {
        let r: flow_core::Receipt = serde_json::from_value(r.clone()).unwrap();
        let run_url = format!("{url}/flow/{}", run.number);
        hex::encode(SignedReceipt::sign(r.to_oxy(), repo_id, state, &run_url, id, sk).encode())
    };

    let a = claim(&h.app).await.unwrap();
    let b = claim(&h.app).await.unwrap();
    let ph = a["pipeline_hash"].as_str().unwrap().to_string();
    let (a_id, b_id) = (a["job_id"].as_i64().unwrap(), b["job_id"].as_i64().unwrap());
    // The first signed receipt binds the runner token to its key, and is kept.
    let rb = receipt("build", &ph, 0);
    let body = serde_json::json!({ "exit_code": 0, "duration_ms": 700, "receipt": rb, "signed_receipt": sign(&rb, state, &sk, &runner) });
    assert_eq!(finish(&h.app, a_id, body).await, StatusCode::NO_CONTENT);
    // Another key under the same token: the result counts, the receipt is not kept.
    let other = SecretKey::generate();
    let rl = receipt("lint", &ph, 0);
    let body = serde_json::json!({ "exit_code": 0, "duration_ms": 300, "receipt": rl,
        "signed_receipt": sign(&rl, state, &other, &other.identity("mallory", "", 1)) });
    assert_eq!(finish(&h.app, b_id, body).await, StatusCode::NO_CONTENT);
    // The right key, but for another state.
    let t = claim(&h.app).await.unwrap();
    let rt = receipt("test", &ph, 0);
    let body = serde_json::json!({ "exit_code": 0, "duration_ms": 500, "receipt": rt,
        "signed_receipt": sign(&rt, ObjectRef([9; 32]), &sk, &runner) });
    assert_eq!(finish(&h.app, t["job_id"].as_i64().unwrap(), body).await, StatusCode::NO_CONTENT);

    let jobs = db::flow::jobs(&h.app.db, run.id).await.unwrap();
    let job = |n: &str| jobs.iter().find(|j| j.name == n).unwrap().clone();
    assert_eq!(job("build").status, "passed");
    assert_eq!((job("build").receipt_attest.len(), job("build").receipt_signer.as_str()), (64, "test-1 (OxygenFlow runner)"));
    assert_eq!(job("lint").status, "passed", "the job's result stands");
    assert!(job("lint").receipt_attest.is_empty());
    assert!(job("lint").receipt_note.contains("not the key this runner signs with"), "{}", job("lint").receipt_note);
    assert!(job("test").receipt_note.contains("another state"), "{}", job("test").receipt_note);

    // The repository keeps it, and anyone who can read the repository can
    // download it and check it on their own.
    let kept = h.app.repos.open(&RepoName::parse("andy/demo").unwrap()).unwrap().receipts(&heads).unwrap();
    assert_eq!(kept.len(), 1);
    let attest = job("build").receipt_attest;
    let (st, bytes) = get_bytes(&h.app, &format!("/andy/demo/receipts/{attest}")).await;
    assert_eq!(st, StatusCode::OK);
    let sr = SignedReceipt::decode(&bytes).unwrap();
    sr.verify().unwrap();
    assert_eq!((sr.state(), sr.receipt.job_name.as_str()), (state, "build"));
    assert_eq!(get_bytes(&h.app, &format!("/andy/demo/receipts/{}", "0".repeat(64))).await.0, StatusCode::NOT_FOUND);

    // Pages: the run names the signer or why a receipt was not kept; the
    // change whose state it is lists it.
    let (_, page) = call(&h.app, "GET", &format!("/andy/demo/flow/{}", run.number), None, None).await;
    assert!(page.contains("signed receipt</a> by test-1"), "run page");
    assert!(page.contains("receipt not kept"), "run page");
    let (_, page) = call(&h.app, "GET", &format!("/andy/demo/change/{}", heads[0].to_hex()), None, None).await;
    assert!(page.contains("<dt>Receipts</dt>") && page.contains(&format!("/andy/demo/receipts/{attest}")), "change page");
}

async fn upload(app: &Shared, job: i64, name: &str, bytes: &[u8]) -> (StatusCode, String) {
    let req = Request::post(format!("/api/v1/flow/jobs/{job}/artifacts/{name}"))
        .header(header::AUTHORIZATION, "Bearer runner-secret")
        .header(header::CONTENT_TYPE, "application/octet-stream")
        .body(Body::from(bytes.to_vec()))
        .unwrap();
    let res = oxygenhub::router(app.clone()).oneshot(req).await.unwrap();
    let status = res.status();
    (status, String::from_utf8_lossy(&res.into_body().collect().await.unwrap().to_bytes()).into_owned())
}

#[tokio::test(flavor = "multi_thread")]
async fn artifacts_are_kept_named_by_the_receipt_and_served() {
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(
        work.path(),
        ".flow/pipeline.kdl",
        &format!(
            "schema-version 1\nsandbox \"alpine\" {{\n    rootfs \"cas://blob/{ROOTFS}\"\n}}\njob \"build\" {{\n    sandbox \"alpine\"\n    exec \"make\"\n    output {{\n        artifact \"tool\" {{\n            path \"/src/out/tool\"\n        }}\n    }}\n}}\n"
        ),
    );
    c.snapshot("test").unwrap();
    c.commit("Build a tool", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let run = runs_of(&h.app, "andy", "demo").await.remove(0);
    let a = claim(&h.app).await.unwrap();
    assert_eq!(a["job"]["outputs"], serde_json::json!([{ "name": "tool", "path": "/src/out/tool" }]), "the runner is told");
    let id = a["job_id"].as_i64().unwrap();

    let body: Vec<u8> = (0..300_000u32).map(|i| (i % 253) as u8).collect();
    let (st, text) = upload(&h.app, id, "other", &body).await;
    assert_eq!(st, StatusCode::BAD_REQUEST, "an undeclared artifact: {text}");
    let (st, text) = upload(&h.app, id, "tool", &body).await;
    assert_eq!(st, StatusCode::OK, "{text}");
    let blob = serde_json::from_str::<serde_json::Value>(&text).unwrap()["blob"].as_str().unwrap().to_string();
    assert_eq!(blob, oxy_core::repo::Repo::blob_ref(&body).to_hex(), "the same blob the runner computes");

    // The receipt must name exactly the uploaded artifacts.
    let mut r = receipt("build", a["pipeline_hash"].as_str().unwrap(), 0);
    assert_eq!(
        finish(&h.app, id, serde_json::json!({ "exit_code": 0, "duration_ms": 700, "receipt": r.clone() })).await,
        StatusCode::BAD_REQUEST
    );
    r["outputs"] = serde_json::json!([blob]);
    assert_eq!(finish(&h.app, id, serde_json::json!({ "exit_code": 0, "duration_ms": 700, "receipt": r })).await, StatusCode::NO_CONTENT);

    // The run page lists it; anyone who can read the repository downloads it.
    let (_, page) = call(&h.app, "GET", &format!("/andy/demo/flow/{}", run.number), None, None).await;
    let link = format!("/andy/demo/flow/{}/artifacts/build/tool", run.number);
    assert!(page.contains(&link), "run page lists the artifact");
    let (st, got) = get_bytes(&h.app, &link).await;
    assert_eq!((st, got == body), (StatusCode::OK, true));
    assert_eq!(get_bytes(&h.app, &format!("/andy/demo/flow/{}/artifacts/build/none", run.number)).await.0, StatusCode::NOT_FOUND);

    // A writer attaches the artifact to a release from the run page; the
    // release then names the run and job that built it, as does the API.
    let form = |f: String| {
        Request::post("/register").header(header::CONTENT_TYPE, "application/x-www-form-urlencoded").body(Body::from(f)).unwrap()
    };
    let res = oxygenhub::router(h.app.clone())
        .oneshot(form("username=andy&email=a%40example.com&password=correct-horse-battery".into()))
        .await
        .unwrap();
    let cookie = res.headers().get(header::SET_COOKIE).unwrap().to_str().unwrap().split(';').next().unwrap().to_string();
    let with_cookie = |method: &str, path: &str, body: Option<String>| {
        let mut req = Request::builder().method(method).uri(path).header(header::COOKIE, cookie.clone());
        if body.is_some() {
            req = req.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded");
        }
        req.body(body.map(Body::from).unwrap_or_else(Body::empty)).unwrap()
    };
    let row = db::repos::get(&h.app.db, "andy", "demo").await.unwrap().unwrap();
    db::releases::create(&h.app.db, row.id, "v1", "One", "", false, None).await.unwrap();
    let res = oxygenhub::router(h.app.clone()).oneshot(with_cookie("GET", &format!("/andy/demo/flow/{}", run.number), None)).await.unwrap();
    let page = String::from_utf8_lossy(&res.into_body().collect().await.unwrap().to_bytes()).into_owned();
    let attach = format!("/andy/demo/flow/{}/artifacts/build/tool/attach", run.number);
    assert!(page.contains(&attach), "a writer is offered the release");
    let i = page.find("name=\"_csrf\" value=\"").unwrap() + "name=\"_csrf\" value=\"".len();
    let csrf = page[i..].split('"').next().unwrap().to_string();
    let release_id = db::releases::list(&h.app.db, row.id).await.unwrap()[0].id;
    let res = oxygenhub::router(h.app.clone())
        .oneshot(with_cookie("POST", &attach, Some(format!("_csrf={csrf}&release={release_id}"))))
        .await
        .unwrap();
    assert_eq!(res.status(), StatusCode::SEE_OTHER);
    let (_, page) = call(&h.app, "GET", "/andy/demo/releases/tag/v1", None, None).await;
    assert!(
        page.contains(">tool</a>") && page.contains(&format!("built by <a href=\"/andy/demo/flow/{0}\">run #{0}</a> build", run.number)),
        "{page}"
    );
    let (_, api) = call(&h.app, "GET", "/api/v1/repos/andy/demo/releases", None, None).await;
    let api: serde_json::Value = serde_json::from_str(&api).unwrap();
    let asset = &api[0]["assets"][0];
    assert_eq!(
        (asset["name"].as_str(), asset["built_by"]["run"].as_i64(), asset["built_by"]["job"].as_str()),
        (Some("tool"), Some(run.number), Some("build"))
    );
    assert!(asset["built_by"]["receipt"].is_null(), "that receipt was not signed");
}

/// A release's trust report (hub spec §9.1): a signed tag, an asset built
/// by a job with a signed receipt from the tagged state, and one uploaded
/// by hand; the page, its JSON export and the API.
#[tokio::test(flavor = "multi_thread")]
async fn a_release_is_audited_from_authors_to_receipts() {
    use oxy_core::hash::{hash_object, ChangeHash, ObjectType};
    use oxy_core::identity::SecretKey;
    use oxy_core::objects::StateObject;
    use oxy_core::receipt::SignedReceipt;
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let mut c = client(&url, work.path());
    c.create_identity("Andy", "[email protected]", oxy_core::repo::IdentityScope::Repo).unwrap();
    write(
        work.path(),
        ".flow/pipeline.kdl",
        &format!(
            "schema-version 1\nsandbox \"alpine\" {{\n    rootfs \"cas://blob/{ROOTFS}\"\n}}\njob \"build\" {{\n    sandbox \"alpine\"\n    exec \"make\"\n    output {{\n        artifact \"tool\" {{\n            path \"/src/out/tool\"\n        }}\n    }}\n}}\n"
        ),
    );
    c.snapshot("test").unwrap();
    c.commit("Build a tool", &[]).unwrap();
    let heads = c.bookmark_heads().unwrap();
    c.tag_create("v1", &heads, "first").unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));

    // The runner builds, uploads the artifact and signs its receipt.
    let run = runs_of(&h.app, "andy", "demo").await.remove(0);
    let a = claim(&h.app).await.unwrap();
    let id = a["job_id"].as_i64().unwrap();
    let body: Vec<u8> = b"#!/bin/sh\necho tool\n".to_vec();
    let (_, text) = upload(&h.app, id, "tool", &body).await;
    let blob = serde_json::from_str::<serde_json::Value>(&text).unwrap()["blob"].as_str().unwrap().to_string();
    let mut r = receipt("build", a["pipeline_hash"].as_str().unwrap(), 0);
    r["outputs"] = serde_json::json!([blob]);
    let heads_run: Vec<ChangeHash> = run.head_list().into_iter().filter_map(ChangeHash::from_hex).collect();
    let state = hash_object(ObjectType::State, &StateObject::new(heads_run).encode());
    let repo_id = h.app.repos.open(&RepoName::parse("andy/demo").unwrap()).unwrap().id().unwrap();
    let sk = SecretKey::generate();
    let runner = sk.identity("test-1 (OxygenFlow runner)", "", 1);
    let fr: flow_core::Receipt = serde_json::from_value(r.clone()).unwrap();
    let signed = hex::encode(SignedReceipt::sign(fr.to_oxy(), repo_id, state, &format!("{url}/flow/1"), &runner, &sk).encode());
    let done = serde_json::json!({ "exit_code": 0, "duration_ms": 900, "receipt": r, "signed_receipt": signed });
    assert_eq!(finish(&h.app, id, done).await, StatusCode::NO_CONTENT);

    // A release with the built asset and a file uploaded by hand.
    let row = db::repos::get(&h.app.db, "andy", "demo").await.unwrap().unwrap();
    let rel = db::releases::create(&h.app.db, row.id, "v1", "One", "", false, None).await.unwrap();
    oxygenhub::web::releases::store_asset(&h.app, rel.id, "tool", "", &body).await.unwrap();
    oxygenhub::web::releases::store_asset(&h.app, rel.id, "notes.txt", "", b"by hand\n").await.unwrap();

    let (st, page) = call(&h.app, "GET", "/andy/demo/releases/trust/v1", None, None).await;
    assert_eq!(st, StatusCode::OK);
    for want in [
        "Tag v1 is signed",
        "The one change in the release is authenticated",
        "tool: signed receipt from run #1 build",
        "a runner key of this hub",
        "tool: built from the tagged state",
        "tool: pinned environment",
        "notes.txt: no build found",
        "<svg class=\"trust-graph\"",
    ] {
        assert!(page.contains(want), "missing {want:?}");
    }
    let (_, releases) = call(&h.app, "GET", "/andy/demo/releases", None, None).await;
    assert!(releases.contains("/andy/demo/releases/trust/v1"), "the release links its audit");

    // The export carries the signed receipt, which checks on its own.
    let (st, json) = call(&h.app, "GET", "/andy/demo/releases/trust/v1?format=json", None, None).await;
    assert_eq!(st, StatusCode::OK);
    let t: serde_json::Value = serde_json::from_str(&json).unwrap();
    assert_eq!((t["tag"].as_str(), t["tag_signature"].as_str(), t["changes"].as_i64()), (Some("v1"), Some("valid"), Some(1)));
    assert_eq!(t["authors"][0]["who"]["name"], "Andy");
    let tool = t["assets"].as_array().unwrap().iter().find(|a| a["name"] == "tool").unwrap();
    let hex_receipt = tool["build"]["signed_receipt"].as_str().unwrap();
    let sr = SignedReceipt::decode(&hex::decode(hex_receipt).unwrap()).unwrap();
    sr.verify().unwrap();
    assert_eq!(sr.receipt.outputs[0].to_hex(), blob);
    let audit = t["audit"].as_array().unwrap();
    assert!(audit.iter().any(|c| c["verdict"] == "warn" && c["title"].as_str().unwrap().starts_with("notes.txt")));
    assert!(!audit.iter().any(|c| c["verdict"] == "fail"), "{audit:?}");
    // The API gives the same report; an unknown release is a 404.
    let (st, api) = call(&h.app, "GET", "/api/v1/repos/andy/demo/releases/trust/v1", None, None).await;
    assert_eq!(st, StatusCode::OK);
    assert_eq!(serde_json::from_str::<serde_json::Value>(&api).unwrap()["assets"].as_array().unwrap().len(), 2);
    assert_eq!(call(&h.app, "GET", "/api/v1/repos/andy/demo/releases/trust/v9", None, None).await.0, StatusCode::NOT_FOUND);
    assert_eq!(call(&h.app, "GET", "/andy/demo/releases/trust/v9", None, None).await.0, StatusCode::NOT_FOUND);
}

/// Runners (hub spec §13.2): an administrator makes a registration token,
/// a runner registers with it (binding its key), claims, shows on Admin →
/// Runners with its history, and is drained, disabled and removed.
#[tokio::test(flavor = "multi_thread")]
async fn runners_register_show_their_history_and_are_drained_and_removed() {
    use oxy_core::identity::SecretKey;
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    // A signed-in site administrator.
    let res = oxygenhub::router(h.app.clone())
        .oneshot(
            Request::post("/register")
                .header(header::CONTENT_TYPE, "application/x-www-form-urlencoded")
                .body(Body::from("username=carol&email=carol%40example.com&password=correct-horse-battery"))
                .unwrap(),
        )
        .await
        .unwrap();
    let cookie = res.headers().get(header::SET_COOKIE).unwrap().to_str().unwrap().split(';').next().unwrap().to_string();
    let admin = db::accounts::account_by_name(&h.app.db, "carol").await.unwrap().unwrap();
    db::accounts::set_admin(&h.app.db, admin.id, true).await.unwrap();
    let web = |method: &'static str, path: String, body: Option<String>| {
        let (app, cookie) = (h.app.clone(), cookie.clone());
        async move {
            let mut req = Request::builder().method(method).uri(path).header(header::COOKIE, cookie);
            if body.is_some() {
                req = req.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded");
            }
            let res = oxygenhub::router(app).oneshot(req.body(body.map(Body::from).unwrap_or_else(Body::empty)).unwrap()).await.unwrap();
            let st = res.status();
            (st, String::from_utf8_lossy(&res.into_body().collect().await.unwrap().to_bytes()).into_owned())
        }
    };
    let csrf_of = |page: &str| {
        let i = page.find("name=\"_csrf\" value=\"").unwrap() + "name=\"_csrf\" value=\"".len();
        page[i..].split('"').next().unwrap().to_string()
    };
    let (st, page) = web("GET", "/admin/runners".into(), None).await;
    assert_eq!(st, StatusCode::OK);
    assert!(page.contains("No runners yet"));
    assert_eq!(call(&h.app, "GET", "/admin/runners", None, None).await.0, StatusCode::NOT_FOUND, "administrators only");

    // A one-time registration token, shown once with the command.
    let (_, page) = web("POST", "/admin/runners/token".into(), Some(format!("_csrf={}", csrf_of(&page)))).await;
    let i = page.find("--token ").unwrap() + "--token ".len();
    let token: String = page[i..].chars().take_while(|c| !c.is_whitespace()).collect();
    assert!(token.starts_with("flowreg_"), "{token}");

    // The runner registers, binding its key; the token then is spent.
    let sk = SecretKey::generate();
    let identity = sk.identity("plex (OxygenFlow runner)", "", 1);
    let info = serde_json::json!({ "arch": "x86_64", "os": "linux", "version": "0.1.0", "labels": ["big-disk"], "max_jobs": 2, "memory_mib": 16384, "cpus": 4 });
    let reg = serde_json::json!({ "token": token, "name": "plex", "identity": hex::encode(identity.encode()), "info": info });
    let (st, body) = call(&h.app, "POST", "/api/v1/flow/runners/register", None, Some(reg.to_string())).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let credential = serde_json::from_str::<serde_json::Value>(&body).unwrap()["credential"].as_str().unwrap().to_string();
    assert!(credential.starts_with("flowrunner_"));
    assert_eq!(
        call(&h.app, "POST", "/api/v1/flow/runners/register", None, Some(reg.to_string())).await.0,
        StatusCode::FORBIDDEN,
        "used once"
    );

    // It claims with its credential; the hub.toml runner shows too.
    let claim_as = |who: String| {
        let app = h.app.clone();
        let info = info.clone();
        async move {
            let body = serde_json::json!({ "runner": "ignored", "runtimes": ["microvm"], "info": info }).to_string();
            call(&app, "POST", "/api/v1/flow/claim", Some(&who), Some(body)).await
        }
    };
    assert_eq!(claim_as(credential.clone()).await.0, StatusCode::NO_CONTENT, "nothing queued");
    assert_eq!(claim(&h.app).await, None);
    let (_, page) = web("GET", "/admin/runners".into(), None).await;
    assert!(
        page.contains(">plex</strong>") && page.contains("st-online") && page.contains("x86_64 · linux") && page.contains("big-disk"),
        "{page}"
    );
    assert!(page.contains(">test-1</strong>") && page.contains("hub.toml"), "the configured runner, by its claimed name");
    let plex = db::runners::list(&h.app.db).await.unwrap().into_iter().find(|r| r.name == "plex").unwrap();
    assert_eq!(plex.key, identity.hash().to_hex(), "bound at registration");

    // A run: plex takes a job, and its history shows it.
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    c.snapshot("test").unwrap();
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    runs_of(&h.app, "andy", "demo").await;
    let (st, body) = claim_as(credential.clone()).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let job: serde_json::Value = serde_json::from_str(&body).unwrap();
    let (_, page) = web("GET", format!("/admin/runners/{}", plex.id), None).await;
    assert!(
        page.contains("andy/demo") && page.contains(job["job"]["name"].as_str().unwrap()) && page.contains("Running now</dt><dd>1 job"),
        "{page}"
    );
    // The queue on Admin → Runners and through the API (#41, part 4).
    let (_, page) = web("GET", "/admin/runners".into(), None).await;
    assert!(page.contains("<h2>Queue</h2>") && page.contains("<td>plex</td>"), "{page}");
    let (_, tokens) = web("GET", "/settings/tokens".into(), None).await;
    let (_, made) = web("POST", "/settings/tokens".into(), Some(format!("_csrf={}&name=admin&scopes=user%3Aread", csrf_of(&tokens)))).await;
    let i = made.find("oxyhub_").unwrap();
    let admin_token: String = made[i..].chars().take_while(|c| c.is_ascii_alphanumeric() || *c == '_').collect();
    let (st, body) = call(&h.app, "GET", "/api/v1/runners", Some(&admin_token), None).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let listed: serde_json::Value = serde_json::from_str(&body).unwrap();
    let plex_row = listed.as_array().unwrap().iter().find(|r| r["name"] == "plex").unwrap();
    assert_eq!((plex_row["running"].as_i64(), plex_row["arch"].as_str()), (Some(1), Some("x86_64")));
    let (_, body) = call(&h.app, "GET", "/api/v1/runners/queue", Some(&admin_token), None).await;
    let queue: serde_json::Value = serde_json::from_str(&body).unwrap();
    assert!(queue.as_array().unwrap().iter().any(|q| q["status"] == "running" && q["runner"] == "plex"), "{queue}");
    let (_, body) = call(&h.app, "GET", &format!("/api/v1/runners/{}/jobs", plex.id), Some(&admin_token), None).await;
    let jobs: serde_json::Value = serde_json::from_str(&body).unwrap();
    assert_eq!(jobs[0]["repo"], "andy/demo", "{jobs}");
    assert_eq!(call(&h.app, "GET", "/api/v1/runners", None, None).await.0, StatusCode::NOT_FOUND, "administrators only");

    // Draining: no new work; removing while busy is refused.
    let (_, page) = web("GET", "/admin/runners".into(), None).await;
    let csrf = csrf_of(&page);
    web("POST", format!("/admin/runners/{}/state", plex.id), Some(format!("_csrf={csrf}&state=draining"))).await;
    assert_eq!(claim_as(credential.clone()).await.0, StatusCode::NO_CONTENT, "a draining runner takes nothing");
    let (_, page) = web("POST", format!("/admin/runners/{}/remove", plex.id), Some(format!("_csrf={csrf}"))).await;
    assert!(page.contains("is running jobs: drain it first"), "{page}");
    // Its job finishes; disabled, it is refused; removed, its credential is dead.
    let id = job["job_id"].as_i64().unwrap();
    let done = call(
        &h.app,
        "POST",
        &format!("/api/v1/flow/jobs/{id}/finish"),
        Some(&credential),
        Some(r#"{"exit_code":0,"duration_ms":5}"#.into()),
    )
    .await;
    assert_eq!(done.0, StatusCode::NO_CONTENT);
    web("POST", format!("/admin/runners/{}/state", plex.id), Some(format!("_csrf={csrf}&state=disabled"))).await;
    assert_eq!(claim_as(credential.clone()).await.0, StatusCode::FORBIDDEN);
    let (_, page) = web("POST", format!("/admin/runners/{}/remove", plex.id), Some(format!("_csrf={csrf}"))).await;
    assert!(page.contains("Removed plex."), "{page}");
    assert_eq!(claim_as(credential).await.0, StatusCode::UNAUTHORIZED);
}

/// Runner scopes (oxygenhub#41): a runner registered for one repository, or
/// one owner, takes only those jobs; a site runner takes any.
#[tokio::test(flavor = "multi_thread")]
async fn scoped_runners_take_only_their_scope() {
    use oxy_core::identity::SecretKey;
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    create(&h.app, "andy", "other").await;
    let register = |scope: db::runners::Scope, name: &'static str| {
        let app = h.app.clone();
        async move {
            let token = format!("flowreg_{name}");
            db::runners::add_registration(&app.db, &blake3::hash(token.as_bytes()).to_hex(), "admin", &scope).await.unwrap();
            let identity = SecretKey::generate().identity(name, "", 1);
            let info = serde_json::json!({ "arch": "x86_64", "os": "linux", "version": "0.1.0", "labels": [], "max_jobs": 1 });
            let reg = serde_json::json!({ "token": token, "name": name, "identity": hex::encode(identity.encode()), "info": info });
            let (st, body) = call(&app, "POST", "/api/v1/flow/runners/register", None, Some(reg.to_string())).await;
            assert_eq!(st, StatusCode::OK, "{body}");
            serde_json::from_str::<serde_json::Value>(&body).unwrap()["credential"].as_str().unwrap().to_string()
        }
    };
    let demo = db::repos::get(&h.app.db, "andy", "demo").await.unwrap().unwrap();
    let only_demo = register(db::runners::Scope { owner: "andy".into(), repo_id: Some(demo.id) }, "demo-runner").await;
    let carols = register(db::runners::Scope { owner: "carol".into(), repo_id: None }, "carol-runner").await;
    let andys = register(db::runners::Scope { owner: "andy".into(), repo_id: None }, "andy-runner").await;
    let claim_as = |who: String| {
        let app = h.app.clone();
        async move {
            let body = serde_json::json!({ "runner": "x", "runtimes": ["microvm"] }).to_string();
            let (st, body) = call(&app, "POST", "/api/v1/flow/claim", Some(&who), Some(body)).await;
            (st == StatusCode::OK).then(|| serde_json::from_str::<serde_json::Value>(&body).unwrap()["repo"].as_str().unwrap().to_string())
        }
    };
    // Jobs queued in andy/other only.
    let work = tempfile::tempdir().unwrap();
    let c = client(&format!("{}/andy/other", h.base), work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    c.snapshot("t").unwrap();
    c.commit("pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    assert!(!runs_of(&h.app, "andy", "other").await.is_empty());
    assert_eq!(claim_as(only_demo.clone()).await, None, "nothing of andy/demo is queued");
    assert_eq!(claim_as(carols).await, None, "nothing of carol's");
    assert_eq!(claim_as(andys).await.as_deref(), Some("andy/other"), "the owner's runner takes its repositories' jobs");
    // The site runner (hub.toml) takes any.
    assert_eq!(claim(&h.app).await.unwrap()["repo"], "andy/other");
    assert_eq!(claim_as(only_demo).await, None, "still nothing of its own");
}

/// Settings → Runners (oxygenhub#41, part 2b): a repository's admin adds a
/// runner for that repository alone, sees which runners serve it, and
/// removes only its own.
#[tokio::test(flavor = "multi_thread")]
async fn repository_admins_add_and_remove_their_own_runners() {
    use oxy_core::identity::SecretKey;
    let h = hub().await;
    let res = oxygenhub::router(h.app.clone())
        .oneshot(
            Request::post("/register")
                .header(header::CONTENT_TYPE, "application/x-www-form-urlencoded")
                .body(Body::from("username=andy&email=andy%40example.com&password=correct-horse-battery"))
                .unwrap(),
        )
        .await
        .unwrap();
    let cookie = res.headers().get(header::SET_COOKIE).unwrap().to_str().unwrap().split(';').next().unwrap().to_string();
    create(&h.app, "andy", "demo").await;
    let demo = db::repos::get(&h.app.db, "andy", "demo").await.unwrap().unwrap();
    let web = |method: &'static str, path: String, body: Option<String>| {
        let (app, cookie) = (h.app.clone(), cookie.clone());
        async move {
            let mut req = Request::builder().method(method).uri(path).header(header::COOKIE, cookie);
            if body.is_some() {
                req = req.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded");
            }
            let res = oxygenhub::router(app).oneshot(req.body(body.map(Body::from).unwrap_or_else(Body::empty)).unwrap()).await.unwrap();
            let st = res.status();
            (st, String::from_utf8_lossy(&res.into_body().collect().await.unwrap().to_bytes()).into_owned())
        }
    };
    let csrf_of = |page: &str| {
        let i = page.find("name=\"_csrf\" value=\"").unwrap() + "name=\"_csrf\" value=\"".len();
        page[i..].split('"').next().unwrap().to_string()
    };
    let (st, page) = web("GET", "/andy/demo/settings".into(), None).await;
    assert_eq!(st, StatusCode::OK);
    assert!(page.contains("<h2>Runners</h2>"), "the owner sees the section");
    let (_, page) = web("POST", "/andy/demo/settings/runners/token".into(), Some(format!("_csrf={}", csrf_of(&page)))).await;
    let i = page.find("--token ").unwrap() + "--token ".len();
    let token: String = page[i..].chars().take_while(|c| !c.is_whitespace()).collect();
    let identity = SecretKey::generate().identity("mine", "", 1);
    let info = serde_json::json!({ "arch": "x86_64", "os": "linux", "version": "0.1.0", "labels": [], "max_jobs": 1 });
    let reg = serde_json::json!({ "token": token, "name": "mine", "identity": hex::encode(identity.encode()), "info": info });
    let (st, body) = call(&h.app, "POST", "/api/v1/flow/runners/register", None, Some(reg.to_string())).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let mine = db::runners::list(&h.app.db).await.unwrap().into_iter().find(|r| r.name == "mine").unwrap();
    assert_eq!(mine.scope(), db::runners::Scope { owner: "andy".into(), repo_id: Some(demo.id) });
    // The site's runner (hub.toml, from its first claim) serves it too, but is not the owner's to remove.
    claim(&h.app).await;
    let site = db::runners::list(&h.app.db).await.unwrap().into_iter().find(|r| r.name == "test-1").unwrap();
    let (_, page) = web("GET", "/andy/demo/settings".into(), None).await;
    assert!(page.contains("<strong>mine</strong>") && page.contains("this repository"), "{page}");
    assert!(page.contains("<strong>test-1</strong>") && page.contains("the whole site"), "{page}");
    let csrf = csrf_of(&page);
    let (st, _) = web("POST", format!("/andy/demo/settings/runners/{}/remove", site.id), Some(format!("_csrf={csrf}"))).await;
    assert_eq!(st, StatusCode::NOT_FOUND, "not the repository's own");
    let (_, page) = web("POST", format!("/andy/demo/settings/runners/{}/remove", mine.id), Some(format!("_csrf={csrf}"))).await;
    assert!(page.contains("mine removed"), "{page}");
    assert!(db::runners::get(&h.app.db, mine.id).await.unwrap().is_none());

    // The same for an organisation's owner.
    let (_, page) = web("GET", "/orgs/new".into(), None).await;
    web("POST", "/orgs/new".into(), Some(format!("_csrf={}&name=acme&display_name=Acme", csrf_of(&page)))).await;
    let (st, page) = web("GET", "/orgs/acme/settings".into(), None).await;
    assert_eq!(st, StatusCode::OK);
    let (_, page) = web("POST", "/orgs/acme/settings/runners/token".into(), Some(format!("_csrf={}", csrf_of(&page)))).await;
    let i = page.find("--token ").unwrap() + "--token ".len();
    let token: String = page[i..].chars().take_while(|c| !c.is_whitespace()).collect();
    let identity = SecretKey::generate().identity("acme-runner", "", 1);
    let reg = serde_json::json!({ "token": token, "name": "acme-runner", "identity": hex::encode(identity.encode()), "info": info });
    assert_eq!(call(&h.app, "POST", "/api/v1/flow/runners/register", None, Some(reg.to_string())).await.0, StatusCode::OK);
    let acme = db::runners::list(&h.app.db).await.unwrap().into_iter().find(|r| r.name == "acme-runner").unwrap();
    assert_eq!(acme.scope(), db::runners::Scope { owner: "acme".into(), repo_id: None });
    let (_, page) = web("GET", "/orgs/acme/settings".into(), None).await;
    assert!(page.contains("<strong>acme-runner</strong>") && page.contains("the organisation's repositories"), "{page}");
    let (_, page) = web("POST", format!("/orgs/acme/settings/runners/{}/remove", acme.id), Some(format!("_csrf={}", csrf_of(&page)))).await;
    assert!(page.contains("acme-runner removed"), "{page}");
}

/// runs-on (oxygenhub#41, part 3): a job that asks for a label goes only to
/// a runner that has it, and waits for one otherwise.
#[tokio::test(flavor = "multi_thread")]
async fn jobs_that_ask_for_labels_wait_for_a_runner_with_them() {
    use oxy_core::identity::SecretKey;
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    let work = tempfile::tempdir().unwrap();
    let c = client(&format!("{}/andy/demo", h.base), work.path());
    let cross = "job \"cross\" {\n    sandbox \"alpine\"\n    runs-on \"riscv64\"\n    exec \"true\"\n}\n";
    write(work.path(), ".flow/pipeline.kdl", &pipeline(cross));
    c.snapshot("t").unwrap();
    c.commit("pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    assert!(!runs_of(&h.app, "andy", "demo").await.is_empty());
    // The site runner has no labels: it takes the others, never cross.
    let mut taken = Vec::new();
    while let Some(j) = claim(&h.app).await {
        taken.push(j["job"]["name"].as_str().unwrap().to_string());
    }
    assert!(!taken.contains(&"cross".to_string()) && taken.contains(&"lint".to_string()), "{taken:?}");
    let queue = db::flow::queue(&h.app.db).await.unwrap();
    let waiting = queue.iter().find(|q| q.job == "cross").unwrap();
    assert_eq!((waiting.status.as_str(), waiting.needs()), ("queued", vec!["riscv64".to_string()]));
    // A runner registered with the label takes it.
    let token = "flowreg_riscv";
    db::runners::add_registration(&h.app.db, &blake3::hash(token.as_bytes()).to_hex(), "admin", &Default::default()).await.unwrap();
    let identity = SecretKey::generate().identity("visionfive", "", 1);
    let info = serde_json::json!({ "arch": "riscv64", "os": "linux", "version": "0.1.0", "labels": ["riscv64"], "max_jobs": 1 });
    let reg = serde_json::json!({ "token": token, "name": "visionfive", "identity": hex::encode(identity.encode()), "info": info });
    let (st, body) = call(&h.app, "POST", "/api/v1/flow/runners/register", None, Some(reg.to_string())).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let credential = serde_json::from_str::<serde_json::Value>(&body).unwrap()["credential"].as_str().unwrap().to_string();
    let claim_body = serde_json::json!({ "runner": "x", "runtimes": ["microvm"], "info": info }).to_string();
    let (st, body) = call(&h.app, "POST", "/api/v1/flow/claim", Some(&credential), Some(claim_body)).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    assert_eq!(serde_json::from_str::<serde_json::Value>(&body).unwrap()["job"]["name"], "cross");
}

/// oxygenhub#71: a job its runner stopped reporting (restarted, or lost
/// the hub as the job ended) is ended once it has run past the time limit
/// and a grace: an error with the reason, its dependents skipped and its
/// run settled, as if the runner had reported the error.
#[tokio::test(flavor = "multi_thread")]
async fn a_job_no_runner_reports_is_ended_after_the_time_limit() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    create(&h.app, "andy", "lost").await;
    let url = format!("{}/andy/lost", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "test.sh", "echo ok\n");
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    assert_eq!(runs_of(&h.app, "andy", "lost").await.len(), 1);
    let job = claim(&h.app).await.expect("a job to claim");
    let id = job["job_id"].as_i64().unwrap();

    // Running, but well within the limit: left alone.
    assert_eq!(oxygenhub::flow::reap_lost_jobs(&h.app).await.unwrap(), 0);
    // Started long ago and never reported: ended.
    let old = db::now_ms() - 24 * 3600 * 1000;
    sqlx::query("UPDATE flow_jobs SET started_ms = ? WHERE id = ?").bind(old).bind(id).execute(&h.app.db).await.unwrap();
    assert_eq!(oxygenhub::flow::reap_lost_jobs(&h.app).await.unwrap(), 1);
    let j = db::flow::job(&h.app.db, id).await.unwrap().unwrap();
    assert_eq!(j.status, "error");
    assert!(j.error.starts_with("lost:"), "{}", j.error);
    let run = &runs_of(&h.app, "andy", "lost").await[0];
    let jobs = db::flow::jobs(&h.app.db, run.id).await.unwrap();
    assert!(jobs.iter().all(|j| j.status != "running"), "nothing left running");
    // A late report, should one come, does not overwrite the ending.
    assert_ne!(finish(&h.app, id, serde_json::json!({ "exit_code": 0 })).await, StatusCode::NO_CONTENT);
    assert_eq!(db::flow::job(&h.app.db, id).await.unwrap().unwrap().status, "error");
}

/// oxygenhub#73: a runner that lists the images it holds is offered only
/// the jobs whose rootfs is among them; one that does not say gets any.
#[tokio::test(flavor = "multi_thread")]
async fn a_runner_is_offered_only_jobs_whose_image_it_holds() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    create(&h.app, "andy", "images").await;
    let url = format!("{}/andy/images", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "test.sh", "echo ok\n");
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    assert_eq!(runs_of(&h.app, "andy", "images").await.len(), 1);
    let claim_with = |images: Vec<String>| {
        let body = serde_json::json!({ "runner": "test-1", "runtimes": ["microvm"], "images": images }).to_string();
        let app = h.app.clone();
        async move {
            let (st, body) = call(&app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(body)).await;
            (st, body)
        }
    };
    let (st, _) = claim_with(vec!["ab".repeat(32)]).await;
    assert_eq!(st, StatusCode::NO_CONTENT, "no job for a runner without the image");
    let jobs = db::flow::jobs(&h.app.db, runs_of(&h.app, "andy", "images").await[0].id).await.unwrap();
    let rootfs: serde_json::Value = serde_json::from_str(&jobs[0].sandbox).unwrap();
    let rootfs = rootfs["rootfs"].as_str().unwrap().to_string();
    let (st, body) = claim_with(vec!["ab".repeat(32), rootfs]).await;
    assert_eq!(st, StatusCode::OK, "{body}");
}

/// oxygenhub#71: a runner that lists the jobs it holds has the others it
/// was given ended at once (it restarted, or never received them), not at
/// the time limit; a held job is left alone.
#[tokio::test(flavor = "multi_thread")]
async fn a_job_its_runner_no_longer_holds_is_ended_at_its_next_claim() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();
    create(&h.app, "andy", "held").await;
    let url = format!("{}/andy/held", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "test.sh", "echo ok\n");
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    assert_eq!(runs_of(&h.app, "andy", "held").await.len(), 1);
    let id = claim(&h.app).await.expect("a job to claim")["job_id"].as_i64().unwrap();
    let claim_holding = |holding: Vec<i64>| {
        let body = serde_json::json!({ "runner": "test-1", "runtimes": ["microvm"], "holding": holding }).to_string();
        let app = h.app.clone();
        async move { call(&app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(body)).await.0 }
    };
    claim_holding(vec![id]).await;
    assert_eq!(db::flow::job(&h.app.db, id).await.unwrap().unwrap().status, "running", "a held job stays");
    claim_holding(vec![]).await;
    let j = db::flow::job(&h.app.db, id).await.unwrap().unwrap();
    assert_eq!(j.status, "error");
    assert!(j.error.starts_with("lost:") && j.error.contains("no longer holds"), "{}", j.error);
    let run = &runs_of(&h.app, "andy", "held").await[0];
    assert!(db::flow::jobs(&h.app.db, run.id).await.unwrap().iter().all(|j| j.status != "running"), "the run settles");
}
`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet
Keyboard
