//! `/install.sh` and `/releases`: the script, run as `curl … | sh` runs it,
//! installing from a release laid out as `deploy/release.sh` lays one out,
//! or published as a repository's.

#[path = "../../knotless/tests/common/mod.rs"]
mod common;
use common::*;

use knotless_serve::releases::Source;
use sha2::{Digest, Sha256};
use std::path::Path;
use std::process::{Command, Output};
use std::sync::{Arc, Mutex};

const BINS: [&str; 7] = [
    "knotless",
    "git-remote-knotless",
    "knotless-mcp",
    "knotless-run",
    "knotless-triage",
    "knotless-agent",
    "knotless-claude-review",
];

/// What a release from before the runner, labeller and reviewer shipped
/// holds.
const OLD_BINS: usize = 3;

/// The binaries a release of `version` holds: 0.1.0 is one from before
/// the rest shipped.
fn bins(version: &str) -> &'static [&'static str] {
    if version == "0.1.0" {
        &BINS[..OLD_BINS]
    } else {
        &BINS
    }
}

/// The target this machine's script asks for.
fn target() -> String {
    let os = match std::env::consts::OS {
        "macos" => "apple-darwin",
        _ => "unknown-linux-musl",
    };
    format!("{}-{os}", std::env::consts::ARCH)
}

/// This machine's archive of `version` in `out`, its binaries scripts that
/// say who they are, staged in `dir`.
fn archive(dir: &Path, out: &Path, version: &str) -> std::path::PathBuf {
    let name = format!("knotless-{version}-{}", target());
    let staged = dir.join("staging").join(&name);
    std::fs::create_dir_all(&staged).unwrap();
    for bin in bins(version) {
        std::fs::write(staged.join(bin), format!("#!/bin/sh\necho {bin}\n")).unwrap();
    }
    std::fs::create_dir_all(out).unwrap();
    let archive = out.join(format!("{name}.tar.gz"));
    let tar = Command::new("tar")
        .arg("-czf")
        .arg(&archive)
        .arg("-C")
        .arg(dir.join("staging"))
        .arg(&name)
        .status()
        .unwrap();
    assert!(tar.success());
    archive
}

/// A release of `version` in `dir`.
fn release(dir: &Path, version: &str) {
    let name = format!("knotless-{version}-{}", target());
    let out = dir.join("releases").join(version);
    let archive = archive(dir, &out, version);
    let sum = format!("{:x}", Sha256::digest(std::fs::read(&archive).unwrap()));
    std::fs::write(
        out.join(format!("{name}.tar.gz.sha256")),
        format!("{sum}  {name}.tar.gz\n"),
    )
    .unwrap();
    std::fs::write(dir.join("releases/latest"), format!("{version}\n")).unwrap();
}

struct Server {
    base: String,
    _rt: tokio::runtime::Runtime,
}

fn serve(dir: Option<&Path>, public_url: Option<&str>) -> Server {
    let source = dir.map(|d| Source::Dir(d.join("releases")));
    let ctx = knotless::env::Ctx::new(knotless::env::Env::from_vars(|_| None));
    serve_from(source, public_url, ctx)
}

fn serve_from(source: Option<Source>, public_url: Option<&str>, ctx: knotless::env::Ctx) -> Server {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let listener = rt
        .block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
        .unwrap();
    let base = format!("http://{}", listener.local_addr().unwrap());
    let router = knotless_serve::releases::routes(source, public_url.map(String::from), ctx);
    rt.spawn(async move { axum::serve(listener, router).await.unwrap() });
    Server { base, _rt: rt }
}

fn get(url: &str, headers: &[(&str, &str)]) -> (u16, String) {
    let client = reqwest::blocking::Client::new();
    let mut req = client.get(url);
    for (k, v) in headers {
        req = req.header(*k, *v);
    }
    let res = req.send().unwrap();
    (res.status().as_u16(), res.text().unwrap())
}

/// The script from `server`, run with `env`, installing into `home`.
fn install(server: &Server, home: &Path, env: &[(&str, &str)]) -> Output {
    let (status, script) = get(&format!("{}/install.sh", server.base), &[]);
    assert_eq!(status, 200);
    Command::new("sh")
        .arg("-c")
        .arg(script)
        .env("HOME", home)
        .envs(env.iter().copied())
        .output()
        .unwrap()
}

#[test]
fn installing_a_release() {
    let dir = tempfile::tempdir().unwrap();
    release(dir.path(), "0.1.0");
    release(dir.path(), "0.2.0");
    let server = serve(Some(dir.path()), None);
    let home = dir.path().join("home");

    let out = install(&server, &home, &[]);
    let stdout = String::from_utf8_lossy(&out.stdout);
    assert!(
        out.status.success(),
        "{stdout}{}",
        String::from_utf8_lossy(&out.stderr)
    );
    assert!(stdout.contains(&format!("installing 0.2.0 for {}", target())));
    // Into ~/.local/bin, which is not on this PATH.
    assert!(stdout.contains("is not on your PATH"));
    assert!(stdout.contains("knotless login"));
    for bin in BINS {
        let ran = Command::new(home.join(".local/bin").join(bin))
            .output()
            .unwrap();
        assert_eq!(String::from_utf8_lossy(&ran.stdout).trim(), bin);
    }

    // A version asked for, into a directory on the PATH, over what is there.
    let bin = dir.path().join("bin");
    let path = format!("{}:{}", bin.display(), std::env::var("PATH").unwrap());
    let to = bin.to_str().unwrap();
    for _ in 0..2 {
        let out = install(
            &server,
            &home,
            &[
                ("KNOTLESS_VERSION", "0.1.0"),
                ("KNOTLESS_INSTALL_DIR", to),
                ("PATH", &path),
            ],
        );
        let stdout = String::from_utf8_lossy(&out.stdout);
        assert!(out.status.success());
        assert!(stdout.contains("installing 0.1.0"));
        assert!(!stdout.contains("is not on your PATH"));
    }
    assert!(bin.join("knotless-mcp").exists());
    // A release from before the runner shipped installs without it.
    assert!(!bin.join("knotless-run").exists());

    // No such version.
    let out = install(&server, &home, &[("KNOTLESS_VERSION", "9.9.9")]);
    assert!(!out.status.success());
    assert!(String::from_utf8_lossy(&out.stderr).contains("no knotless-9.9.9-"));

    // An archive that is not what its checksum says.
    let name = format!("knotless-0.2.0-{}.tar.gz", target());
    let sum = dir
        .path()
        .join("releases/0.2.0")
        .join(format!("{name}.sha256"));
    std::fs::write(&sum, format!("{}  {name}\n", "0".repeat(64))).unwrap();
    let out = install(&server, &home, &[]);
    assert!(!out.status.success());
    assert!(String::from_utf8_lossy(&out.stderr).contains("does not match its checksum"));
    std::fs::remove_file(&sum).unwrap();
    let out = install(&server, &home, &[]);
    assert!(String::from_utf8_lossy(&out.stderr).contains("no checksum"));

    // No release at all.
    std::fs::remove_file(dir.path().join("releases/latest")).unwrap();
    let out = install(&server, &home, &[]);
    assert!(!out.status.success());
    assert!(String::from_utf8_lossy(&out.stderr).contains("no release at"));
}

#[test]
fn where_the_script_fetches_from() {
    let dir = tempfile::tempdir().unwrap();
    let server = serve(Some(dir.path()), None);
    let url = format!("{}/install.sh", server.base);
    let (_, script) = get(&url, &[]);
    assert!(script.contains(&format!("releases=\"{}/releases\"", server.base)));
    // Behind a proxy that holds the certificate.
    let (_, script) = get(
        &url,
        &[("x-forwarded-proto", "https"), ("host", "git.example")],
    );
    assert!(script.contains("releases=\"https://git.example/releases\""));
    assert!(script.contains("curl -fsSL https://git.example/install.sh | sh"));

    // The URL this server is reached at, when it is told it.
    let server = serve(Some(dir.path()), Some("https://knot.example/"));
    let (_, script) = get(&format!("{}/install.sh", server.base), &[]);
    assert!(script.contains("releases=\"https://knot.example/releases\""));

    // A server with no releases hands out no script.
    let server = serve(None, None);
    assert_eq!(get(&format!("{}/install.sh", server.base), &[]).0, 404);
}

#[test]
fn installing_a_published_release() {
    let mut s = Sandbox::new();
    let handle = s.account.handle.clone();
    s.set("KNOTLESS_IDENTIFIER", &handle);
    let url = s.url("knotless");
    let work = s.work("w", 1);
    s.git(&work, &["push", "-q", &url, "main"]);
    let staged = s.path("built");
    for version in ["0.1.0", "0.2.0-rc.1"] {
        let archive = archive(&staged, &staged.join(version), version);
        let file = archive.to_string_lossy();
        let (code, out) = s.cli(
            &["release", "publish", &url, "knotless", version, &file],
            b"",
        );
        assert_eq!(code, 0, "{out}");
    }
    let source = Source::parse(&url).unwrap();
    let server = serve_from(Some(source), None, ctx(&s));
    let releases = format!("{}/releases", server.base);

    // The latest is the highest that is not a pre-release.
    assert_eq!(
        get(&format!("{releases}/latest"), &[]),
        (200, "0.1.0\n".into())
    );
    let home = s.path("home");
    let out = install(&server, &home, &[]);
    let stdout = String::from_utf8_lossy(&out.stdout);
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(out.status.success(), "{stdout}{stderr}");
    assert!(stdout.contains("installing 0.1.0"));
    let ran = Command::new(home.join(".local/bin/knotless"))
        .output()
        .unwrap();
    assert_eq!(String::from_utf8_lossy(&ran.stdout).trim(), "knotless");

    // Kept once read, and handed out from there again.
    let name = format!("knotless-0.2.0-rc.1-{}.tar.gz", target());
    let bytes = std::fs::read(staged.join("0.2.0-rc.1").join(&name)).unwrap();
    let sum = format!("{:x}", Sha256::digest(&bytes));
    for _ in 0..2 {
        let res = reqwest::blocking::get(format!("{releases}/0.2.0-rc.1/{name}")).unwrap();
        assert_eq!(res.bytes().unwrap().as_ref(), bytes.as_slice());
    }
    let kept = ctx(&s).env.cache_dir.join("serve/releases");
    assert!(kept.join(&sum).exists());
    let checksum = get(&format!("{releases}/0.2.0-rc.1/{name}.sha256"), &[]);
    assert_eq!(checksum, (200, format!("{sum}  {name}\n")));

    // What is not there.
    for missing in ["9.9.9/x", "0.1.0/x", "0.1.0/x.sha256"] {
        assert_eq!(
            get(&format!("{releases}/{missing}"), &[]).0,
            404,
            "{missing}"
        );
    }

    // A file that cannot be kept is not handed out.
    let blocked = s.path("blocked").to_string_lossy().into_owned();
    s.set("KNOTLESS_CACHE_DIR", &blocked);
    std::fs::create_dir_all(s.path("blocked/serve")).unwrap();
    std::fs::write(s.path("blocked/serve/releases"), b"").unwrap();
    let server = serve_from(Some(Source::parse(&url).unwrap()), None, ctx(&s));
    let name = format!("knotless-0.1.0-{}.tar.gz", target());
    let res = get(&format!("{}/releases/0.1.0/{name}", server.base), &[]);
    assert_eq!(res.0, 502);

    // A repository with no releases, and one that is not there.
    let empty = s.url("empty");
    s.git(&work, &["push", "-q", &empty, "main"]);
    let server = serve_from(Some(Source::parse(&empty).unwrap()), None, ctx(&s));
    let (status, body) = get(&format!("{}/releases/latest", server.base), &[]);
    assert_eq!((status, body.as_str()), (404, "no release\n"));
    let gone = Source::parse(&s.url("gone")).unwrap();
    let server = serve_from(Some(gone), None, ctx(&s));
    for path in ["latest", "0.1.0/x"] {
        let (status, body) = get(&format!("{}/releases/{path}", server.base), &[]);
        assert_eq!(status, 502, "{path}");
        assert!(body.contains("no such repository"), "{body}");
    }
}

/// An OCI image layout in `dir`: one layer holding `says`, its config and
/// manifest, and the index naming it. Returns the manifest's bytes and
/// digest, and the layer's digest.
fn image(dir: &Path, says: &str) -> (Vec<u8>, String, String) {
    let digest = |b: &[u8]| format!("sha256:{:x}", Sha256::digest(b));
    let blobs = dir.join("blobs/sha256");
    std::fs::create_dir_all(&blobs).unwrap();
    let put = |b: &[u8]| {
        let d = digest(b);
        std::fs::write(blobs.join(&d[7..]), b).unwrap();
        d
    };
    let layer = put(says.as_bytes());
    let config = br#"{"architecture":"amd64","os":"linux"}"#;
    let config_digest = put(config);
    let manifest = serde_json::json!({
        "schemaVersion": 2,
        "mediaType": "application/vnd.oci.image.manifest.v1+json",
        "config": {"mediaType": "application/vnd.oci.image.config.v1+json", "digest": config_digest, "size": config.len()},
        "layers": [{"mediaType": "application/vnd.oci.image.layer.v1.tar", "digest": layer, "size": says.len()}],
    })
    .to_string()
    .into_bytes();
    let manifest_digest = put(&manifest);
    let index = serde_json::json!({"schemaVersion": 2, "manifests": [
        {"mediaType": "application/vnd.oci.image.manifest.v1+json", "digest": manifest_digest, "size": manifest.len()}
    ]});
    std::fs::write(dir.join("index.json"), index.to_string()).unwrap();
    std::fs::write(dir.join("oci-layout"), r#"{"imageLayoutVersion":"1.0.0"}"#).unwrap();
    (manifest, manifest_digest, layer)
}

fn registry(ctx: knotless::env::Ctx) -> Server {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let listener = rt
        .block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
        .unwrap();
    let base = format!("http://{}", listener.local_addr().unwrap());
    let router = knotless_serve::oci::routes(ctx, None);
    rt.spawn(async move { axum::serve(listener, router).await.unwrap() });
    Server { base, _rt: rt }
}

#[test]
fn pulling_an_image() {
    let mut s = Sandbox::new();
    let handle = s.account.handle.clone();
    s.set("KNOTLESS_IDENTIFIER", &handle);
    let url = s.url("tools");
    let work = s.work("w", 1);
    s.git(&work, &["push", "-q", &url, "main"]);
    let (stable, rc) = (s.path("stable"), s.path("rc"));
    let (manifest, manifest_digest, layer) = image(&stable, "stable layer");
    let (_, rc_digest, _) = image(&rc, "rc layer");
    for (version, dir) in [("1.0.0", &stable), ("1.1.0-rc.1", &rc)] {
        let dir = dir.to_string_lossy();
        let (code, out) = s.cli(
            &["release", "publish", &url, "tools", version, "--oci", &dir],
            b"",
        );
        assert_eq!(code, 0, "{out}");
    }
    let server = registry(ctx(&s));
    let v2 = format!("{}/v2", server.base);
    let client = reqwest::blocking::Client::new();
    let fetch = |path: &str| client.get(format!("{v2}/{path}")).send().unwrap();

    // Asked first, it asks for a token; any, as a pull's is anyone's.
    let res = fetch("");
    assert_eq!(res.status(), 401);
    let asked = res.headers()["www-authenticate"]
        .to_str()
        .unwrap()
        .to_string();
    assert_eq!(
        asked,
        format!("Bearer realm=\"{v2}/token\",service=\"knotless\"")
    );
    let token: serde_json::Value = fetch("token").json().unwrap();
    assert_eq!(token["token"], "anonymous");
    let res = client
        .get(format!("{v2}/"))
        .bearer_auth("anonymous")
        .send()
        .unwrap();
    assert_eq!(
        res.headers()["docker-distribution-api-version"],
        "registry/2.0"
    );
    assert_eq!(res.text().unwrap(), "{}");
    // Its tags: the versions, and the latest.
    let tags: serde_json::Value = fetch(&format!("{handle}/tools/tags/list")).json().unwrap();
    let name = format!("{handle}/tools/tools");
    assert_eq!(
        tags,
        serde_json::json!({"name": name, "tags": ["1.1.0-rc.1", "1.0.0", "latest"]})
    );
    let none: serde_json::Value = fetch(&format!("{handle}/tools/other/tags/list"))
        .json()
        .unwrap();
    assert_eq!(none["tags"], serde_json::json!([]));

    // The latest is the highest that is not a pre-release, as its index names it.
    let res = fetch(&format!("{handle}/tools/manifests/latest"));
    assert_eq!(res.status(), 200);
    let headers = res.headers().clone();
    assert_eq!(
        headers["content-type"],
        "application/vnd.oci.image.manifest.v1+json"
    );
    assert_eq!(headers["docker-content-digest"], manifest_digest.as_str());
    assert_eq!(res.bytes().unwrap().as_ref(), manifest.as_slice());
    // By version, under the package's own name, and by digest.
    let res = fetch(&format!("{handle}/tools/tools/manifests/1.1.0-rc.1"));
    assert_eq!(res.headers()["docker-content-digest"], rc_digest.as_str());
    let res = fetch(&format!("{handle}/tools/manifests/{manifest_digest}"));
    assert_eq!(res.bytes().unwrap().as_ref(), manifest.as_slice());
    let head = client
        .head(format!("{v2}/{handle}/tools/manifests/latest"))
        .send()
        .unwrap();
    assert_eq!(head.status(), 200);
    assert_eq!(
        head.headers()["docker-content-digest"],
        manifest_digest.as_str()
    );
    // A layer, read once and kept.
    for _ in 0..2 {
        let res = fetch(&format!("{handle}/tools/blobs/{layer}"));
        assert_eq!(res.headers()["docker-content-digest"], layer.as_str());
        assert_eq!(res.text().unwrap(), "stable layer");
    }

    // What is not there, as the spec's errors.
    let code = |path: &str| {
        let res = fetch(path);
        let status = res.status().as_u16();
        let body: serde_json::Value = res.json().unwrap();
        (
            status,
            body["errors"][0]["code"].as_str().unwrap().to_string(),
        )
    };
    let none = format!("sha256:{}", "0".repeat(64));
    for (path, want) in [
        (
            format!("{handle}/tools/manifests/9.9.9"),
            "MANIFEST_UNKNOWN",
        ),
        (
            format!("{handle}/tools/manifests/{none}"),
            "MANIFEST_UNKNOWN",
        ),
        (format!("{handle}/tools/blobs/{none}"), "BLOB_UNKNOWN"),
        (
            format!("{handle}/tools/other/manifests/latest"),
            "MANIFEST_UNKNOWN",
        ),
        (format!("{handle}/manifests/latest"), "NAME_UNKNOWN"),
        (format!("{handle}/a%20b/manifests/latest"), "NAME_INVALID"),
        (format!("{handle}/gone/manifests/latest"), "NAME_UNKNOWN"),
        (
            format!("{handle}/tools/blobs/uploads/x"),
            "BLOB_UPLOAD_UNKNOWN",
        ),
    ] {
        assert_eq!(code(&path), (404, want.to_string()), "{path}");
    }

    // What cannot be kept is not handed out.
    let blocked = s.path("blocked").to_string_lossy().into_owned();
    s.set("KNOTLESS_CACHE_DIR", &blocked);
    std::fs::create_dir_all(s.path("blocked/serve")).unwrap();
    std::fs::write(s.path("blocked/serve/releases"), b"").unwrap();
    let server = registry(ctx(&s));
    let v2 = format!("{}/v2", server.base);
    for path in [
        format!("{handle}/tools/manifests/latest"),
        format!("{handle}/tools/manifests/{manifest_digest}"),
        format!("{handle}/tools/blobs/{layer}"),
    ] {
        let res = client.get(format!("{v2}/{path}")).send().unwrap();
        assert_eq!(res.status(), 502, "{path}");
    }
}

/// A push to `v2`, as a registry client makes one, with `token`.
#[derive(Clone)]
struct Pushing<'a> {
    v2: &'a str,
    name: String,
    token: &'a str,
    client: reqwest::blocking::Client,
}

impl Pushing<'_> {
    fn send(
        &self,
        method: reqwest::Method,
        path: &str,
        body: Vec<u8>,
    ) -> reqwest::blocking::Response {
        let url = format!("{}/{}{path}", self.v2, self.name);
        let req = self
            .client
            .request(method, url)
            .bearer_auth(self.token)
            .body(body);
        req.send().unwrap()
    }

    /// The error's status and code, of what was answered.
    fn code(res: reqwest::blocking::Response) -> (u16, String) {
        let status = res.status().as_u16();
        let body: serde_json::Value = res.json().unwrap();
        (
            status,
            body["errors"][0]["code"].as_str().unwrap().to_string(),
        )
    }

    /// Pushes `bytes` as one blob, in one request.
    fn blob(&self, bytes: &[u8]) -> String {
        let digest = format!("sha256:{:x}", Sha256::digest(bytes));
        let path = format!("/blobs/uploads/?digest={digest}");
        let res = self.send(reqwest::Method::POST, &path, bytes.to_vec());
        assert_eq!(res.status(), 201, "{:?}", res.text());
        digest
    }

    fn manifest(
        &self,
        reference: &str,
        manifest: &serde_json::Value,
    ) -> reqwest::blocking::Response {
        let url = format!("{}/{}/manifests/{reference}", self.v2, self.name);
        let req = self.client.put(url).bearer_auth(self.token);
        let req = req.header("content-type", "application/vnd.oci.image.manifest.v1+json");
        req.body(manifest.to_string()).send().unwrap()
    }
}

#[test]
fn pushing_an_image() {
    use reqwest::Method;
    let mut s = Sandbox::new();
    let handle = s.account.handle.clone();
    s.set("KNOTLESS_IDENTIFIER", &handle);
    let url = s.url("app");
    let work = s.work("w", 1);
    s.git(&work, &["push", "-q", &url, "main"]);
    let server = registry(ctx(&s));
    let v2 = format!("{}/v2", server.base);
    let client = reqwest::blocking::Client::new();

    // A token for the owner's app password, and none for a wrong one.
    let password = s.account.password.clone();
    let login = |v2: &str, password: &str| {
        let res = client
            .get(format!("{v2}/token"))
            .basic_auth(&handle, Some(password));
        res.send().unwrap()
    };
    assert_eq!(login(&v2, "wrong").status(), 401);
    let token: serde_json::Value = login(&v2, &password).json().unwrap();
    let token = token["token"].as_str().unwrap().to_string();
    assert_eq!(token.len(), 32);
    let res = client
        .get(format!("{v2}/"))
        .bearer_auth(&token)
        .send()
        .unwrap();
    assert_eq!(res.status(), 200);
    let p = Pushing {
        v2: &v2,
        name: format!("{handle}/app"),
        token: &token,
        client: client.clone(),
    };

    // A layer in parts: begun, gone on with, and ended.
    let layer: Vec<u8> = (0..3000u32).map(|i| (i % 251) as u8).collect();
    let layer_digest = format!("sha256:{:x}", Sha256::digest(&layer));
    let begun = p.send(Method::POST, "/blobs/uploads/", vec![]);
    assert_eq!(begun.status(), 202);
    let id = begun.headers()["docker-upload-uuid"]
        .to_str()
        .unwrap()
        .to_string();
    let at = begun.headers()["location"].to_str().unwrap();
    assert_eq!(at, format!("/v2/{handle}/app/app/blobs/uploads/{id}"));
    let going = p.send(
        Method::PATCH,
        &format!("/blobs/uploads/{id}"),
        layer[..1000].to_vec(),
    );
    assert_eq!(going.status(), 202);
    assert_eq!(going.headers()["range"], "0-999");
    // Ended with no digest, or the wrong one, it is refused, and gone.
    let end = format!("/blobs/uploads/{id}");
    let res = p.send(Method::PUT, &end, vec![]);
    assert_eq!(Pushing::code(res), (400, "DIGEST_INVALID".into()));
    let wrong = format!("{end}?digest=sha256:{}", "0".repeat(64));
    let res = p.send(Method::PUT, &wrong, layer[1000..].to_vec());
    assert_eq!(Pushing::code(res), (400, "DIGEST_INVALID".into()));
    let res = p.send(Method::PATCH, &end, vec![1]);
    assert_eq!(Pushing::code(res), (404, "BLOB_UPLOAD_UNKNOWN".into()));
    let res = p.send(Method::PUT, &wrong, vec![1]);
    assert_eq!(Pushing::code(res), (404, "BLOB_UPLOAD_UNKNOWN".into()));
    let begun = p.send(Method::POST, "/blobs/uploads/", vec![]);
    let id = begun.headers()["docker-upload-uuid"]
        .to_str()
        .unwrap()
        .to_string();
    p.send(
        Method::PATCH,
        &format!("/blobs/uploads/{id}"),
        layer[..1000].to_vec(),
    );
    let end = format!("/blobs/uploads/{id}?digest={layer_digest}");
    let res = p.send(Method::PUT, &end, layer[1000..].to_vec());
    assert_eq!(res.status(), 201);
    assert_eq!(
        res.headers()["docker-content-digest"],
        layer_digest.as_str()
    );
    // Pushed, it is there to read before any release holds it.
    let staged = format!("{v2}/{handle}/app/blobs/{layer_digest}");
    assert_eq!(client.head(&staged).send().unwrap().status(), 200);

    let config = p.blob(br#"{"architecture":"amd64","os":"linux"}"#);
    let image = serde_json::json!({
        "schemaVersion": 2,
        "mediaType": "application/vnd.oci.image.manifest.v1+json",
        "config": {"mediaType": "application/vnd.oci.image.config.v1+json", "digest": config, "size": 37},
        "layers": [{"mediaType": "application/vnd.oci.image.layer.v1.tar", "digest": layer_digest, "size": 3000}],
    });
    // The manifest by its digest, and an index naming it by a version.
    let image_digest = format!("sha256:{:x}", Sha256::digest(image.to_string()));
    let res = p.manifest(&format!("sha256:{}", "0".repeat(64)), &image);
    assert_eq!(Pushing::code(res), (400, "DIGEST_INVALID".into()));
    assert_eq!(p.manifest(&image_digest, &image).status(), 201);
    let index = serde_json::json!({
        "schemaVersion": 2,
        "mediaType": "application/vnd.oci.image.index.v1+json",
        "manifests": [{"mediaType": "application/vnd.oci.image.manifest.v1+json", "digest": image_digest, "size": 1}],
    });
    // Tags are versions; `latest` is computed.
    let res = p.manifest("latest", &index);
    assert_eq!(Pushing::code(res), (400, "TAG_INVALID".into()));
    let res = p.manifest("1.0.0", &index);
    assert_eq!(res.status(), 201, "{:?}", res.text());
    let res = p.manifest("1.0.0", &index);
    assert_eq!(Pushing::code(res), (403, "DENIED".into()));

    // Pulled as any image published as a release is.
    let fetch = |path: &str| {
        let res = client.get(format!("{v2}/{handle}/app/{path}"));
        res.send().unwrap()
    };
    let tags: serde_json::Value = fetch("tags/list").json().unwrap();
    assert_eq!(tags["tags"], serde_json::json!(["1.0.0", "latest"]));
    let top = fetch("manifests/latest");
    let index_type = "application/vnd.oci.image.index.v1+json";
    assert_eq!(top.headers()["content-type"], index_type);
    let child = fetch(&format!("manifests/{image_digest}"));
    assert_eq!(child.text().unwrap(), image.to_string());
    let pulled = fetch(&format!("blobs/{layer_digest}")).bytes().unwrap();
    assert_eq!(pulled.as_ref(), layer.as_slice());
    let (code, out) = s.cli(&["release", "list", &url], b"");
    assert_eq!(code, 0, "{out}");
    assert!(out.starts_with("app\t1.0.0\t"), "{out}");

    // What a manifest reaches must be pushed first.
    let config = format!("sha256:{}", "1".repeat(64));
    let missing = serde_json::json!({"config": {"digest": config, "size": 1}, "layers": []});
    let res = p.manifest("2.0.0", &missing);
    assert_eq!(Pushing::code(res), (400, "MANIFEST_BLOB_UNKNOWN".into()));
    let child = format!("sha256:{}", "2".repeat(64));
    let orphan = serde_json::json!({"manifests": [{"digest": child, "size": 1}]});
    let res = p.manifest("2.0.0", &orphan);
    assert_eq!(Pushing::code(res), (400, "MANIFEST_BLOB_UNKNOWN".into()));
    // Nor is what is not a manifest taken.
    let res = p.send(Method::PUT, "/manifests/2.0.0", b"not json".to_vec());
    assert_eq!(Pushing::code(res), (400, "MANIFEST_INVALID".into()));
    let res = p.send(Method::PUT, "/manifests/2.0.0", vec![b' '; (4 << 20) + 1]);
    assert_eq!(Pushing::code(res), (400, "MANIFEST_INVALID".into()));

    // Nothing else is posted, patched or put.
    let blob = format!("/blobs/{layer_digest}");
    for (method, path) in [
        (Method::POST, "/manifests/1.0.0"),
        (Method::PATCH, "/manifests/1.0.0"),
        (Method::PUT, blob.as_str()),
    ] {
        let res = p.send(method, path, vec![]);
        assert_eq!(Pushing::code(res), (405, "UNSUPPORTED".into()));
    }
    let res = client.post(format!("{v2}/nowhere")).send().unwrap();
    assert_eq!(Pushing::code(res), (405, "UNSUPPORTED".into()));

    // Only the owner pushes, with a token, to a repository there can be.
    let uploads = format!("{v2}/{handle}/app/blobs/uploads/");
    let res = client.post(uploads).send().unwrap();
    assert_eq!(res.status(), 401);
    let asked = res.headers()["www-authenticate"].to_str().unwrap();
    let scope = format!(",scope=\"repository:{handle}/app/app:push,pull\"");
    assert!(asked.ends_with(&scope), "{asked}");
    let other = Sandbox::new();
    for (name, status, code) in [
        (format!("{}/app", other.account.handle), 403, "DENIED"),
        ("nobody.invalid/app".to_string(), 404, "NAME_UNKNOWN"),
        (format!("{handle}/a%20b"), 404, "NAME_INVALID"),
    ] {
        let p = Pushing { name, ..p.clone() };
        let res = p.send(Method::POST, "/blobs/uploads/", vec![]);
        assert_eq!(Pushing::code(res), (status, code.to_string()));
    }

    // A push whose blobs cannot be kept fails.
    let blocked = s.path("blocked").to_string_lossy().into_owned();
    s.set("KNOTLESS_CACHE_DIR", &blocked);
    std::fs::create_dir_all(s.path("blocked/serve")).unwrap();
    std::fs::write(s.path("blocked/serve/uploads"), b"").unwrap();
    let server = registry(ctx(&s));
    let v2 = format!("{}/v2", server.base);
    let token: serde_json::Value = login(&v2, &password).json().unwrap();
    let p = Pushing {
        v2: &v2,
        name: format!("{handle}/app"),
        token: token["token"].as_str().unwrap(),
        client: client.clone(),
    };
    let res = p.send(Method::POST, "/blobs/uploads/", vec![]);
    assert_eq!(Pushing::code(res).0, 502);
    assert_eq!(Pushing::code(p.manifest(&image_digest, &image)).0, 502);
}

// A lost race is told apart from any other failed write, which says what
// went wrong rather than that the version was published.
#[test]
fn a_release_not_made_says_why() {
    use knotless::push;
    use knotless::records::{RELEASE_MIME, ReleaseFile, ReleasePart};
    use knotless::release::{Made, Target};
    use knotless::url::RepoUrl;
    use knotless::xrpc::Auth;

    let mut s = Sandbox::new();
    let handle = s.account.handle.clone();
    s.set("KNOTLESS_IDENTIFIER", &handle);
    let url = s.url("raced");
    let work = s.work("w", 1);
    s.git(&work, &["push", "-q", &url, "main"]);
    let ctx = ctx(&s);
    let url = RepoUrl::parse(&url).unwrap();
    let rt = tokio::runtime::Runtime::new().unwrap();
    rt.block_on(async {
        let session = knotless::session::create(&ctx, &handle, &s.account.password);
        let auth = session.await.unwrap().auth();
        let target = Target::open(&ctx, &url, "tool", "1.0.0").await.unwrap();
        let blobs = push::upload(&ctx, &target.remote, &auth, b"x", RELEASE_MIME);
        let file = ReleaseFile {
            name: "x".into(),
            size: 1,
            sha256: knotless::release::sha256(b"x"),
            parts: blobs
                .await
                .unwrap()
                .into_iter()
                .map(|blob| ReleasePart { blob })
                .collect(),
        };
        let stranger = Auth::Bearer("no such token".into());
        let create = |auth| target.create(auth, vec![file.clone()], Made::default());

        let refused = create(&stranger).await.unwrap_err();
        assert!(
            !format!("{refused:#}").contains("published already"),
            "{refused:#}"
        );

        create(&auth).await.unwrap();
        let lost = create(&auth).await.unwrap_err();
        assert!(
            format!("{lost:#}").contains("tool 1.0.0 is published already"),
            "{lost:#}"
        );
    });
}

/// An appview handing out the releases in `dir`, whose describeServer
/// answers what `described` holds.
fn appview(dir: &Path, described: Arc<Mutex<serde_json::Value>>) -> Server {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let listener = rt
        .block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
        .unwrap();
    let base = format!("http://{}", listener.local_addr().unwrap());
    let ctx = knotless::env::Ctx::new(knotless::env::Env::from_vars(|_| None));
    let source = Some(Source::Dir(dir.join("releases")));
    let describe = move || {
        let described = described.lock().unwrap().clone();
        async move { axum::Json(described) }
    };
    let router = knotless_serve::releases::routes(source, None, ctx).route(
        "/xrpc/dev.knotless.describeServer",
        axum::routing::get(describe),
    );
    rt.spawn(async move { axum::serve(listener, router).await.unwrap() });
    Server { base, _rt: rt }
}

#[test]
fn a_newer_version_is_told_of_and_an_older_one_refused() {
    let dir = tempfile::tempdir().unwrap();
    release(dir.path(), "999.0.0");
    let described = Arc::new(Mutex::new(serde_json::json!({"runners": []})));
    let server = appview(dir.path(), described.clone());
    let mut s = Sandbox::new();
    s.set("KNOTLESS_APPVIEW", &server.base);
    s.set("KNOTLESS_NO_UPDATE_CHECK", "");

    // Told on stderr that a newer one is out, and asked once a day.
    let (ok, out, err) = s.knotless(&["help"], b"");
    assert!(ok && out.contains("usage:"));
    assert!(err.contains("knotless 999.0.0 is out"), "{err}");
    let kept = s.path("cache/update.json");
    assert!(kept.exists());
    *described.lock().unwrap() = serde_json::json!({
        "runners": [],
        "client": {"minimum": "999.0.0", "reason": "Compaction loses objects."},
    });
    // A day is not up, so the floor is not known yet.
    let refused = |s: &Sandbox| s.knotless(&["nope"], b"").2.contains("asks for");
    assert!(!refused(&s));

    // A day on, the appview asks for 999.0.0, and nothing but what gets
    // there runs.
    let stale = || {
        let mut known: serde_json::Value =
            serde_json::from_slice(&std::fs::read(&kept).unwrap()).unwrap();
        known["checked"] = 0.into();
        std::fs::write(&kept, known.to_string()).unwrap();
    };
    stale();
    let (ok, _, err) = s.knotless(&["nope"], b"");
    assert!(!ok);
    assert!(
        err.contains("older than the 999.0.0 the appview asks for. Compaction loses objects."),
        "{err}"
    );
    assert!(!err.contains("is out"), "{err}");
    assert_eq!(s.cli(&["help"], b"").0, 0);
    // Unless told to go on anyway.
    s.set("KNOTLESS_IGNORE_MINIMUM", "1");
    let (ok, _, err) = s.knotless(&["nope"], b"");
    assert!(!ok && err.contains("unknown command nope"), "{err}");
    s.set("KNOTLESS_IGNORE_MINIMUM", "");

    // An appview that cannot be asked leaves what was known.
    stale();
    s.set("KNOTLESS_APPVIEW", "http://127.0.0.1:9");
    assert!(refused(&s));
    // Nor is there anything to know with checks off.
    s.set("KNOTLESS_NO_UPDATE_CHECK", "1");
    assert!(!refused(&s));
    s.set("KNOTLESS_NO_UPDATE_CHECK", "");
    s.set("KNOTLESS_APPVIEW", &server.base);

    // Upgraded, from the appview's script, into the directory asked for.
    let bin = dir.path().join("bin");
    s.set("KNOTLESS_INSTALL_DIR", &bin.to_string_lossy());
    let (ok, out, err) = s.knotless(&["upgrade"], b"");
    assert!(ok, "{err}");
    assert!(out.starts_with("knotless: installing 999.0.0"), "{out}");
    assert!(!err.contains("is out"), "{err}");
    let ran = Command::new(bin.join("knotless")).output().unwrap();
    assert_eq!(String::from_utf8_lossy(&ran.stdout).trim(), "knotless");
    // Or beside the binary that runs, when none is.
    s.set("KNOTLESS_INSTALL_DIR", "");
    let copied = dir.path().join("copied");
    std::fs::create_dir_all(&copied).unwrap();
    std::fs::copy(bin_dir().join("knotless"), copied.join("knotless")).unwrap();
    let out = s
        .command(&copied.join("knotless").to_string_lossy(), s.dir.path())
        .arg("upgrade")
        .output()
        .unwrap();
    assert!(
        out.status.success(),
        "{}",
        String::from_utf8_lossy(&out.stderr)
    );
    let ran = Command::new(copied.join("knotless")).output().unwrap();
    assert_eq!(String::from_utf8_lossy(&ran.stdout).trim(), "knotless");

    // What cannot be upgraded so.
    let upgrade = |s: &Sandbox| {
        let (ok, _, err) = s.knotless(&["upgrade"], b"");
        assert!(!ok);
        err
    };
    s.set("KNOTLESS_INSTALL_DIR", "/home/a/.cargo/bin");
    assert!(upgrade(&s).contains("upgrade it with cargo"));
    s.set("KNOTLESS_INSTALL_DIR", &bin.to_string_lossy());
    s.set("KNOTLESS_VERSION", "9.9.9");
    assert!(upgrade(&s).contains("no knotless-9.9.9-"));
    s.set("KNOTLESS_VERSION", "");
    let none = serve(None, None);
    s.set("KNOTLESS_APPVIEW", &none.base);
    assert!(upgrade(&s).contains("hands out no releases"));
    s.set("KNOTLESS_APPVIEW", "");
    assert!(upgrade(&s).contains("set KNOTLESS_APPVIEW"));
}