use crate::*;

const ROOTFS: &str = "9b5c2a1e8f3d4e7a6b8c9d0e1f2a3b4c5d6e7f8a9b0c1d2e3f4a5b6c7d8e9f0a";

fn doc(body: &str) -> String {
    format!(
        "schema-version 1\n\nsandbox \"alpine\" {{\n    runtime \"microvm\"\n    rootfs \"cas://blob/{ROOTFS}\"\n    memory-limit \"512MiB\"\n    cpu-limit 2\n    network \"isolated\"\n}}\n\n{body}"
    )
}

fn problems(src: &str) -> Vec<(usize, String)> {
    Pipeline::parse(src).unwrap_err().0.into_iter().map(|p| (p.line, p.message)).collect()
}

#[test]
fn parses_the_m0_example() {
    let p = Pipeline::parse(&doc(r#"job "lint" {
    sandbox "alpine"
    scope "src/**" "README.md"
    env {
        LANG "C.UTF-8"
    }
    exec "grep" "-rn" "TODO" "src"
}

job "test" {
    sandbox "alpine"
    depends-on "lint"
    exec "sh" "-eu" "test.sh"
}
"#))
    .unwrap();
    let sb = &p.sandboxes["alpine"];
    assert_eq!((sb.runtime, sb.memory_mib, sb.cpus, sb.network), (Runtime::Microvm, 512, 2, Network::Isolated));
    assert_eq!(sb.rootfs, ROOTFS);
    let lint = &p.jobs["lint"];
    assert_eq!(lint.exec, ["grep", "-rn", "TODO", "src"]);
    assert_eq!(lint.scope, ["src/**", "README.md"]);
    assert_eq!(lint.env["LANG"], "C.UTF-8");
    assert_eq!(p.jobs["test"].depends_on, ["lint"]);
    assert_eq!(p.order().iter().map(|j| j.name.as_str()).collect::<Vec<_>>(), ["lint", "test"]);
    assert_eq!(p.depths()["test"], 1);
}

#[test]
fn defaults_and_gib() {
    let p = Pipeline::parse(&format!(
        "schema-version 1\nsandbox \"s\" {{\n  rootfs \"cas://blob/{ROOTFS}\"\n  memory-limit \"2GiB\"\n}}\njob \"a\" {{\n  sandbox \"s\"\n  exec \"true\"\n}}\n"
    ))
    .unwrap();
    let s = &p.sandboxes["s"];
    assert_eq!((s.runtime, s.memory_mib, s.cpus, s.network), (Runtime::Microvm, 2048, 1, Network::Isolated));
}

#[test]
fn shell_metacharacters_stay_literal_argv() {
    // Spec §10.1: every byte reaches argv as written; nothing adds a shell.
    let p = Pipeline::parse(&doc(r#"job "x" { sandbox "alpine"; exec "echo" "a; rm -rf / && $(id) | `x`" "b\nc"; }"#)).unwrap();
    assert_eq!(p.jobs["x"].exec, ["echo", "a; rm -rf / && $(id) | `x`", "b\nc"]);
}

#[test]
fn problems_carry_lines_and_are_all_reported() {
    let src = doc(r#"job "a" {
    sandbox "missing"
    exec "true"
    mount { source "oxy://workspace"; }
}
job "b" {
    sandbox "alpine"
    depends-on "nope"
    exec
}
flavour "x"
"#);
    let ps = problems(&src);
    let has = |line: usize, text: &str| ps.iter().any(|(l, m)| *l == line && m.contains(text));
    assert!(has(11, "which is not declared") && has(11, "\"missing\""), "{ps:?}");
    assert!(has(14, "mount source \"oxy://workspace\" is not in M0 yet"), "{ps:?}");
    assert!(has(16, "\"nope\""), "{ps:?}");
    assert!(has(19, "`exec` needs a program"), "{ps:?}");
    assert!(has(21, "unknown node `flavour`"), "{ps:?}");
}

#[test]
fn typed_values_are_not_coerced() {
    let src = format!(
        "schema-version 1\nsandbox \"s\" {{\n  rootfs \"cas://blob/{ROOTFS}\"\n  cpu-limit \"2\"\n  memory-limit \"1.5GiB\"\n}}\njob \"a\" {{\n  sandbox \"s\"\n  exec \"echo\" 1\n}}\n"
    );
    let ps = problems(&src);
    assert!(ps.iter().any(|(l, m)| *l == 4 && m.contains("integer, not a string")), "{ps:?}");
    assert!(ps.iter().any(|(l, m)| *l == 5 && m.contains("whole number")), "{ps:?}");
    assert!(ps.iter().any(|(l, m)| *l == 9 && m.contains("strings, not an integer")), "{ps:?}");
}

#[test]
fn m0_refuses_what_it_cannot_honour() {
    let src = format!("schema-version 1\nsandbox \"s\" {{\n  runtime \"wasm\"\n  rootfs \"cas://blob/{ROOTFS}\"\n  network \"wifi\"\n}}\njob \"a\" {{ sandbox \"s\"; exec \"true\"; }}\n");
    let ps = problems(&src);
    // A WASM sandbox has no image (oxygenflow#75).
    assert!(ps.iter().any(|(l, m)| *l == 2 && m.contains("takes no rootfs")), "{ps:?}");
    assert!(ps.iter().any(|(l, m)| *l == 5 && m.contains("isolated")), "{ps:?}");
    assert!(problems("schema-version 2\n").iter().any(|(_, m)| m.contains("schema-version 2")));
    assert!(problems(&format!(
        "schema-version 1\nsandbox \"s\" {{ rootfs \"cas://blob/{}\"; }}\njob \"a\" {{ sandbox \"s\"; exec \"true\"; }}",
        &ROOTFS[..60]
    ))
    .iter()
    .any(|(_, m)| m.contains("64 lower-case hex")));
}

#[test]
fn cycles_are_found() {
    let ps = problems(&doc(
        "job \"a\" { sandbox \"alpine\"; depends-on \"c\"; exec \"true\"; }\njob \"b\" { sandbox \"alpine\"; depends-on \"a\"; exec \"true\"; }\njob \"c\" { sandbox \"alpine\"; depends-on \"b\"; exec \"true\"; }\njob \"d\" { sandbox \"alpine\"; depends-on \"d\"; exec \"true\"; }\n",
    ));
    assert!(ps.iter().any(|(_, m)| m.contains("cycle: a → c → b → a")), "{ps:?}");
    assert!(ps.iter().any(|(_, m)| m.contains("\"d\" depends on itself")), "{ps:?}");
}

#[test]
fn invalid_kdl_has_a_line() {
    let ps = problems("schema-version 1\njob \"a\" {\n  exec \"unterminated\n}\n");
    assert!(!ps.is_empty() && ps[0].0 >= 2, "{ps:?}");
}

#[test]
fn canonical_form_ignores_layout_and_order() {
    let a = Pipeline::parse(&doc(
        "job \"b\" { sandbox \"alpine\"; exec \"true\"; }\njob \"a\" {\n  sandbox \"alpine\"\n  depends-on \"b\"\n  exec \"x\"\n}\n",
    ))
    .unwrap();
    let b = Pipeline::parse(&doc("// comment\njob \"a\" { exec \"x\"; depends-on \"b\"; sandbox \"alpine\"; }\n\n\njob \"b\" { exec \"true\"; sandbox \"alpine\"; }\n")).unwrap();
    assert_eq!(a.canonical(), b.canonical());
    assert_eq!(a.hash(), b.hash());
    let c = Pipeline::parse(&doc(
        "job \"b\" { sandbox \"alpine\"; exec \"true\"; }\njob \"a\" { sandbox \"alpine\"; depends-on \"b\"; exec \"y\"; }\n",
    ))
    .unwrap();
    assert_ne!(a.hash(), c.hash());
    let sb = &a.sandboxes["alpine"];
    assert_ne!(a.jobs["a"].hash(sb), c.jobs["a"].hash(sb));
    assert_eq!(a.jobs["b"].hash(sb), c.jobs["b"].hash(sb), "an unchanged job keeps its hash");
    // The canonical form round-trips through Oxygen's strict decoder.
    oxy_core::cbor::decode(&a.canonical()).unwrap();
}

#[test]
fn scope_tree_hash_sees_only_the_scope() {
    let f = |b: u8| [b; 32];
    let scope = vec!["src/**".to_string()];
    let base = scope_tree_hash(&scope, [("src/a.rs", f(1)), ("README.md", f(2))]);
    assert_eq!(base, scope_tree_hash(&scope, [("README.md", f(9)), ("src/a.rs", f(1))]), "outside the scope, and order, do not count");
    assert_ne!(base, scope_tree_hash(&scope, [("src/a.rs", f(3)), ("README.md", f(2))]));
    assert_ne!(scope_tree_hash(&[], [("README.md", f(2))]), scope_tree_hash(&[], [("README.md", f(9))]), "no scope is the whole tree");
}

#[test]
fn receipts_hash_as_oxygen_objects() {
    let r = Receipt {
        job_name: "lint".into(),
        pipeline_hash: [1; 32],
        scope_tree_hash: [2; 32],
        inputs: vec![],
        environment: [3; 32],
        outputs: vec![],
        exit_code: 0,
        duration_ms: 1234,
    };
    let body = r.encode();
    let mut want = format!("oxy1 receipt {}\0", body.len()).into_bytes();
    want.extend_from_slice(&body);
    assert_eq!(r.hash(), *blake3::hash(&want).as_bytes());
    let json = serde_json::to_string(&r).unwrap();
    assert_eq!(serde_json::from_str::<Receipt>(&json).unwrap(), r);
    // The same receipt as the Oxygen object a runner signs.
    assert_eq!(r.to_oxy().hash().0, r.hash());
    assert_eq!(r.to_oxy().encode(), body);
    let mut r2 = r.clone();
    r2.exit_code = 1;
    assert_ne!(r.hash(), r2.hash());
}

#[test]
fn ephemeral_mounts_parse_sort_and_hash() {
    let plain = Pipeline::parse(&doc("job \"b\" {\n    sandbox \"alpine\"\n    exec \"cargo\" \"build\"\n}\n")).unwrap();
    let p = Pipeline::parse(&doc(r#"job "b" {
    sandbox "alpine"
    mount {
        source "ephemeral://tmpfs"
        target "/tmp/cache"
        size "256MiB"
    }
    mount {
        source "ephemeral://disk"
        target "/src/target"
        size "8GiB"
    }
    exec "cargo" "build"
}
"#))
    .unwrap();
    let b = &p.jobs["b"];
    assert_eq!(
        b.mounts,
        vec![
            Mount { source: MountSource::Disk, target: "/src/target".into(), size_mib: 8192, cache: None, key: vec![] },
            Mount { source: MountSource::Tmpfs, target: "/tmp/cache".into(), size_mib: 256, cache: None, key: vec![] },
        ],
        "sorted by target"
    );
    assert_ne!(p.hash(), plain.hash(), "mounts are part of the job");
    // A job without mounts encodes as before mounts existed: no field 6.
    let v = oxy_core::cbor::decode(&plain.canonical()).unwrap();
    let text = format!("{v:?}");
    assert!(!text.contains("ephemeral"), "{text}");
    let json = serde_json::to_value(&plain.jobs["b"]).unwrap();
    assert!(json.get("mounts").is_none(), "absent from the API too");
    let back: Job = serde_json::from_value(json).unwrap();
    assert!(back.mounts.is_empty());
}

#[test]
fn a_disk_mount_can_be_a_cache_keyed_by_files() {
    let src = |extra: &str| {
        doc(&format!(
            "job \"b\" {{\n    sandbox \"alpine\"\n    mount {{\n        source \"ephemeral://disk\"\n        target \"/src/target\"\n        size \"8GiB\"\n{extra}    }}\n    exec \"cargo\" \"build\"\n}}\n"
        ))
    };
    let plain = Pipeline::parse(&src("")).unwrap();
    let cached = Pipeline::parse(&src("        cache \"target\"\n        key \"Cargo.lock\" \"rust-toolchain.toml\"\n")).unwrap();
    let m = &cached.jobs["b"].mounts[0];
    assert_eq!((m.cache.as_deref(), m.key.clone()), (Some("target"), vec!["Cargo.lock".to_string(), "rust-toolchain.toml".to_string()]));
    assert_ne!(cached.hash(), plain.hash(), "a cache is part of the job");
    let keyless = Pipeline::parse(&src("        cache \"target\"\n")).unwrap();
    assert_ne!(keyless.hash(), cached.hash());
    let ps = problems(&src("        cache \"no spaces\"\n"));
    assert!(ps.iter().any(|(_, m)| m.contains("cache name")), "{ps:?}");
    let ps = problems(&src("        key \"Cargo.lock\"\n"));
    assert!(ps.iter().any(|(_, m)| m.contains("goes with `cache`")), "{ps:?}");
    let ps = problems(&src("        cache \"t\"\n        key \"../x\"\n"));
    assert!(ps.iter().any(|(_, m)| m.contains("path in the repository")), "{ps:?}");
    let tmpfs = doc("job \"b\" {\n    sandbox \"alpine\"\n    mount {\n        source \"ephemeral://tmpfs\"\n        target \"/tmp/c\"\n        size \"64MiB\"\n        cache \"c\"\n    }\n    exec \"true\"\n}\n");
    assert!(problems(&tmpfs).iter().any(|(_, m)| m.contains("only an ephemeral://disk")));
}

#[test]
fn mounts_are_checked() {
    let job = |mount: &str| doc(&format!("job \"b\" {{\n    sandbox \"alpine\"\n    mount {{\n{mount}\n    }}\n    exec \"true\"\n}}\n"));
    let has = |src: String, want: &str| {
        let ps = problems(&src);
        assert!(ps.iter().any(|(_, m)| m.contains(want)), "{want:?} not in {ps:?}");
    };
    has(job("source \"oxy://workspace\"\ntarget \"/src/x\"\nsize \"1GiB\""), "not in M0 yet");
    has(job("source \"ephemeral://disk\"\ntarget \"/etc/x\"\nsize \"1GiB\""), "under /src or /tmp");
    has(job("source \"ephemeral://disk\"\ntarget \"/src\"\nsize \"1GiB\""), "under /src or /tmp");
    has(job("source \"ephemeral://disk\"\ntarget \"/src/../etc\"\nsize \"1GiB\""), "without . or ..");
    has(job("source \"ephemeral://disk\"\ntarget \"/src/x/\"\nsize \"1GiB\""), "trailing /");
    has(job("source \"ephemeral://disk\"\ntarget \"/src/x\"\nsize \"8MiB\""), "from 16MiB to 64GiB");
    has(job("source \"ephemeral://disk\"\ntarget \"/src/x\"\nsize \"65GiB\""), "from 16MiB to 64GiB");
    has(job("source \"ephemeral://disk\"\ntarget \"/src/x\"\nsize \"lots\""), "whole number");
    has(job("source \"ephemeral://disk\"\ntarget \"/src/x\""), "needs a size");
    has(job("source \"ephemeral://disk\"\ntarget \"/src/x\"\nsize \"1GiB\"\nreadonly #true"), "`readonly` is not in M0 yet");
    has(job("source \"ephemeral://disk\"\nsource \"ephemeral://tmpfs\"\ntarget \"/src/x\"\nsize \"1GiB\""), "given twice");
    let two = |a: &str, b: &str| {
        doc(&format!(
            "job \"b\" {{\n    sandbox \"alpine\"\n    mount {{ source \"ephemeral://disk\"; target \"{a}\"; size \"1GiB\"; }}\n    mount {{ source \"ephemeral://disk\"; target \"{b}\"; size \"1GiB\"; }}\n    exec \"true\"\n}}\n"
        ))
    };
    has(two("/src/t", "/src/t/deps"), "overlaps");
    has(two("/src/t", "/src/t"), "overlaps");
    assert!(Pipeline::parse(&two("/src/t", "/src/tt")).is_ok(), "a shared prefix is not nesting");
    let five: String =
        (0..5).map(|i| format!("    mount {{ source \"ephemeral://tmpfs\"; target \"/tmp/m{i}\"; size \"16MiB\"; }}\n")).collect();
    has(doc(&format!("job \"b\" {{\n    sandbox \"alpine\"\n{five}    exec \"true\"\n}}\n")), "at most 4 mounts");
}

#[test]
fn loopback_and_allow_listed_egress() {
    let src = |net: &str| {
        format!("schema-version 1\nsandbox \"s\" {{\n  rootfs \"cas://blob/{ROOTFS}\"\n  network {net}\n}}\njob \"a\" {{ sandbox \"s\"; exec \"true\"; }}\n")
    };
    let p = Pipeline::parse(&src("\"loopback\"")).unwrap();
    assert_eq!(p.sandboxes["s"].network, Network::Loopback);
    assert_ne!(p.hash(), Pipeline::parse(&src("\"isolated\"")).unwrap().hash(), "the network is part of the sandbox");

    // Egress names its hosts (oxygenflow#55), sorted; the sandbox then needs
    // a runner offering the egress label.
    let p =
        Pipeline::parse(&src("\"egress\" { allow \"registry.opentofu.org\" \"*.googleapis.com\"; allow \"*.googleapis.com\"; }")).unwrap();
    let s = &p.sandboxes["s"];
    assert_eq!(s.network, Network::Egress);
    assert_eq!(s.egress_allow, ["*.googleapis.com", "registry.opentofu.org"]);
    assert_eq!(s.runs_on, [crate::EGRESS_LABEL]);
    let other = Pipeline::parse(&src("\"egress\" { allow \"registry.opentofu.org\"; }")).unwrap();
    assert_ne!(p.hash(), other.hash(), "the allow-list is part of the sandbox");

    let has = |net: &str, want: &str| {
        let ps = problems(&src(net));
        assert!(ps.iter().any(|(_, m)| m.contains(want)), "{net}: {ps:?}");
    };
    has("\"egress\"", "needs a block");
    has("\"egress\" { }", "at least one `allow");
    has("\"egress\" { allow \"Example.COM\"; }", "lower-case host name");
    has("\"egress\" { allow \"localhost\"; }", "lower-case host name");
    has("\"egress\" { allow \"example.com:8443\"; }", "lower-case host name");
    has("\"egress\" { deny \"example.com\"; }", "unknown egress setting");
    has("\"loopback\" { allow \"example.com\"; }", "takes no block");
    let syscall = format!(
        "schema-version 1\nsandbox \"s\" {{\n  runtime \"syscall\"\n  rootfs \"cas://blob/{ROOTFS}\"\n  network \"egress\" {{ allow \"example.com\"; }}\n}}\njob \"a\" {{ sandbox \"s\"; exec \"true\"; }}\n"
    );
    assert!(problems(&syscall).iter().any(|(_, m)| m.contains("needs runtime \"microvm\"")));
}

#[test]
fn egress_patterns_match_names_and_names_under_a_domain() {
    let allow = vec!["*.googleapis.com".to_string(), "registry.opentofu.org".to_string()];
    for ok in ["storage.googleapis.com", "a.b.googleapis.com", "REGISTRY.opentofu.org.", "registry.opentofu.org"] {
        assert!(crate::egress_allows(&allow, ok), "{ok}");
    }
    for no in ["googleapis.com", "evilgoogleapis.com", "googleapis.com.evil.org", "opentofu.org", "x.registry.opentofu.org", ""] {
        assert!(!crate::egress_allows(&allow, no), "{no}");
    }
}

#[test]
fn secrets_mount_as_files_and_are_part_of_the_job() {
    let plain = Pipeline::parse(&doc("job \"d\" {\n    sandbox \"alpine\"\n    exec \"deploy\"\n}\n")).unwrap();
    let p = Pipeline::parse(&doc(r#"job "d" {
    sandbox "alpine"
    secret "token" {
        mount-path "/run/secrets/token"
    }
    secret "aws" {
        mount-path "/run/secrets/aws/credentials"
    }
    exec "deploy"
}"#))
    .unwrap();
    let d = &p.jobs["d"];
    assert_eq!(
        d.secrets,
        vec![
            Secret { name: "aws".into(), mount_path: "/run/secrets/aws/credentials".into() },
            Secret { name: "token".into(), mount_path: "/run/secrets/token".into() },
        ],
        "sorted by name"
    );
    assert_ne!(p.hash(), plain.hash(), "secrets are part of the job");
    for (bad, why) in [
        ("secret \"t\" {\n    mount-path \"/tmp/t\"\n}", "under /run/secrets/"),
        ("secret \"t\" {\n    mount-path \"/run/secrets/../x\"\n}", "under /run/secrets/"),
        ("secret \"t\" {\n}", "needs a mount-path"),
        ("secret \"t\" {\n    mount-path \"/run/secrets/a\"\n}\nsecret \"t\" {\n    mount-path \"/run/secrets/b\"\n}", "given twice"),
        (
            "secret \"t\" {\n    mount-path \"/run/secrets/a\"\n}\nsecret \"u\" {\n    mount-path \"/run/secrets/a\"\n}",
            "two files mount at",
        ),
        ("secret \"t\" {\n    vault-ref \"x\"\n    mount-path \"/run/secrets/a\"\n}", "vault-ref"),
    ] {
        let e = Pipeline::parse(&doc(&format!("job \"d\" {{\n    sandbox \"alpine\"\n    {bad}\n    exec \"deploy\"\n}}\n"))).unwrap_err();
        assert!(e.to_string().contains(why), "{bad}: {e}");
    }
}

#[test]
fn identities_are_asked_for_by_audience_and_mount_like_secrets() {
    let plain = Pipeline::parse(&doc("job \"d\" {\n    sandbox \"alpine\"\n    exec \"deploy\"\n}\n")).unwrap();
    let p = Pipeline::parse(&doc(r#"job "d" {
    sandbox "alpine"
    identity "sts.amazonaws.com" {
        mount-path "/run/secrets/aws-token"
    }
    identity "https://vault.example" {
        mount-path "/run/secrets/vault-token"
    }
    secret "key" {
        mount-path "/run/secrets/key"
    }
    exec "deploy"
}"#))
    .unwrap();
    let d = &p.jobs["d"];
    assert_eq!(
        d.identities.iter().map(|i| i.audience.as_str()).collect::<Vec<_>>(),
        ["https://vault.example", "sts.amazonaws.com"],
        "sorted"
    );
    assert_ne!(p.hash(), plain.hash(), "identities are part of the job");
    for (bad, why) in [
        ("identity \"a b\" {\n    mount-path \"/run/secrets/x\"\n}", "without spaces"),
        ("identity \"a\" {\n    mount-path \"/tmp/x\"\n}", "under /run/secrets/"),
        ("identity \"a\" {\n}", "needs a mount-path"),
        ("identity \"a\" {\n    mount-path \"/run/secrets/x\"\n}\nidentity \"a\" {\n    mount-path \"/run/secrets/y\"\n}", "given twice"),
        (
            "identity \"a\" {\n    mount-path \"/run/secrets/x\"\n}\nsecret \"s\" {\n    mount-path \"/run/secrets/x\"\n}",
            "two files mount at",
        ),
    ] {
        let e = Pipeline::parse(&doc(&format!("job \"d\" {{\n    sandbox \"alpine\"\n    {bad}\n    exec \"deploy\"\n}}\n"))).unwrap_err();
        assert!(e.to_string().contains(why), "{bad}: {e}");
    }
}

#[test]
fn a_sandbox_may_name_its_architecture() {
    let plain = Pipeline::parse(&doc("job \"b\" {\n    sandbox \"alpine\"\n    exec \"make\"\n}\n")).unwrap();
    assert_eq!(plain.sandboxes["alpine"].arch, None);
    let src =
        doc("job \"b\" {\n    sandbox \"alpine\"\n    exec \"make\"\n}\n").replace("memory-limit", "arch \"riscv64\"\n    memory-limit");
    let p = Pipeline::parse(&src).unwrap();
    assert_eq!(p.sandboxes["alpine"].arch.as_deref(), Some("riscv64"));
    assert_ne!(p.hash(), plain.hash(), "the architecture is part of the sandbox");
    let bad = src.replace("riscv64", "RISC V");
    assert!(Pipeline::parse(&bad).unwrap_err().to_string().contains("not an architecture name"));
    assert!(arch_ok("x86_64") && arch_ok("aarch64") && !arch_ok("x") && !arch_ok("A64"));
}

#[test]
fn jobs_are_told_what_they_build() {
    let p = Pipeline::parse(&doc(r#"job "build" {
    sandbox "alpine"
    env {
        FLOW_BOOKMARK "pinned-by-the-pipeline"
    }
    exec "make"
}"#))
    .unwrap();
    let job = p.jobs["build"].clone();
    let claim = crate::api::Claim {
        job_id: 7,
        run_id: 40,
        repo: "andy/demo".into(),
        clone_url: "https://hub/andy/demo".into(),
        token: "t".into(),
        heads: vec!["aa".into(), "bb".into()],
        pipeline_hash: "00".into(),
        job: job.clone(),
        sandbox: p.sandboxes["alpine"].clone(),
        timeout_s: 60,
        subrepos: vec![],
        state: "cc".into(),
        bookmark: "feature/x".into(),
        run_number: 3,
        env: [("CC".to_string(), "clang".to_string()), ("FLOW_REPO".to_string(), "spoofed".to_string())].into_iter().collect(),
        secrets: vec![],
        inputs: vec![],
        expected_ms: Some(17_700),
        checkpoint: None,
    };
    let env = job.with_flow_env(&claim.flow_env()).env;
    assert_eq!(env["CC"], "clang", "the hub's variables reach the job");
    assert_eq!(env["FLOW_REPO"], "andy/demo", "a hub variable cannot pose as a FLOW_ fact");
    assert_eq!(env["FLOW_HEADS"], "aa+bb");
    assert_eq!(env["FLOW_STATE"], "cc");
    assert_eq!(env["FLOW_RUN"], "3", "the run's number in its repository, not the hub's id");
    assert_eq!(env["FLOW_JOB"], "build");
    assert_eq!(env["FLOW_BOOKMARK"], "pinned-by-the-pipeline", "the pipeline's own value wins");
    // Unknown values (an older hub) are left out, not empty.
    let old = crate::api::Claim {
        state: String::new(),
        bookmark: String::new(),
        run_number: 0,
        env: Default::default(),
        secrets: vec![],
        ..claim
    };
    let names: Vec<String> = old.flow_env().into_iter().map(|(k, _)| k).collect();
    assert_eq!(names, ["FLOW_REPO", "FLOW_HEADS", "FLOW_JOB"]);
    // The expected time goes with the claim, and an older hub's claim
    // without it still reads (#76).
    let json = serde_json::to_value(&old).unwrap();
    assert_eq!(json["expected_ms"], 17_700);
    let mut without = json.clone();
    without.as_object_mut().unwrap().remove("expected_ms");
    assert_eq!(serde_json::from_value::<crate::api::Claim>(without).unwrap().expected_ms, None);
}

#[test]
fn a_sandbox_env_is_where_jobs_start() {
    let src = |sandbox_env: &str| {
        format!(
            "schema-version 1\nsandbox \"mos\" {{\n    rootfs \"cas://blob/{ROOTFS}\"\n{sandbox_env}}}\njob \"build\" {{\n    sandbox \"mos\"\n    env {{\n        X16EMU \"/opt/job\"\n    }}\n    exec \"make\"\n}}\n"
        )
    };
    let with =
        Pipeline::parse(&src("    env {\n        PATH \"/opt/llvm-mos/bin:/usr/bin:/bin\"\n        X16EMU \"/opt/x16emu\"\n    }\n"))
            .unwrap();
    let sb = &with.sandboxes["mos"];
    assert_eq!(sb.env["PATH"], "/opt/llvm-mos/bin:/usr/bin:/bin");
    let job = with.jobs["build"].in_sandbox(sb);
    assert_eq!(job.env["PATH"], "/opt/llvm-mos/bin:/usr/bin:/bin", "from the sandbox");
    assert_eq!(job.env["X16EMU"], "/opt/job", "the job's own value wins");
    // The sandbox's environment is part of the pipeline's hash; a sandbox
    // without one hashes as before.
    let without = Pipeline::parse(&src("")).unwrap();
    assert!(without.sandboxes["mos"].env.is_empty());
    assert_ne!(with.hash(), without.hash());
    let bad = problems(&src("    env {\n        BAD-NAME \"x\"\n    }\n"));
    assert!(bad.iter().any(|(_, m)| m.contains("letters, digits and _")), "{bad:?}");
}

#[test]
fn artifacts_are_declared_checked_and_hashed() {
    let src = |out: &str| {
        doc(&format!(
            "job \"build\" {{\n    sandbox \"alpine\"\n    mount {{\n        source \"ephemeral://disk\"\n        target \"/src/target\"\n        size \"1GiB\"\n    }}\n    mount {{\n        source \"ephemeral://tmpfs\"\n        target \"/src/tmp\"\n        size \"64MiB\"\n    }}\n    exec \"make\"\n{out}}}\n"
        ))
    };
    let p = Pipeline::parse(&src(
        "    output {\n        artifact \"tool\" {\n            path \"/src/target/release/tool\"\n            format \"blob\"\n        }\n        artifact \"docs\" {\n            path \"/src/docs/tool.1\"\n        }\n    }\n",
    ))
    .unwrap();
    let names: Vec<(&str, &str)> = p.jobs["build"].outputs.iter().map(|o| (o.name.as_str(), o.path.as_str())).collect();
    assert_eq!(names, [("docs", "/src/docs/tool.1"), ("tool", "/src/target/release/tool")], "sorted by name");
    // Artifacts are part of the job's hash; a job without them hashes as before.
    let plain = Pipeline::parse(&src("")).unwrap();
    let sb = &p.sandboxes["alpine"];
    assert_ne!(p.jobs["build"].hash(sb), plain.jobs["build"].hash(sb));
    assert!(plain.jobs["build"].outputs.is_empty());
    let json = serde_json::to_string(&plain.jobs["build"]).unwrap();
    assert!(!json.contains("outputs"), "no outputs field when there are none: {json}");

    let bad = |out: &str| problems(&src(out)).into_iter().map(|(_, m)| m).collect::<Vec<_>>().join("\n");
    assert!(bad("    output {\n        artifact \"x\" {\n            path \"/etc/passwd\"\n        }\n    }\n")
        .contains("must be an absolute path under /src"));
    assert!(bad("    output {\n        artifact \"x\" {\n            path \"/src/../etc\"\n        }\n    }\n")
        .contains("must be an absolute path under /src"));
    assert!(bad("    output {\n        artifact \"x\" {\n            path \"/src/tmp/a\"\n        }\n    }\n").contains("tmpfs"));
    assert!(bad("    output {\n        artifact \"x\" {\n        }\n    }\n").contains("needs a path"));
    assert!(bad("    output {\n        artifact \"x\" {\n            path \"/src/a\"\n        }\n        artifact \"x\" {\n            path \"/src/b\"\n        }\n    }\n")
        .contains("given twice"));
    assert!(bad("    output {\n        artifact \"x\" {\n            path \"/src/a\"\n            format \"tar\"\n        }\n    }\n")
        .contains("format"));

    // oxygenflow#37: `when "always"` keeps a failing job's report; the
    // default hashes and serialises as before.
    let p = Pipeline::parse(&src(
        "    output {\n        artifact \"junit\" {\n            path \"/src/target/junit.xml\"\n            when \"always\"\n        }\n        artifact \"tool\" {\n            path \"/src/target/tool\"\n            when \"passed\"\n        }\n    }\n",
    ))
    .unwrap();
    let outs = &p.jobs["build"].outputs;
    assert!(outs[0].always && outs[0].name == "junit");
    assert!(!outs[1].always);
    let json = serde_json::to_string(&outs[1]).unwrap();
    assert!(!json.contains("always"), "{json}");
    assert!(serde_json::to_string(&outs[0]).unwrap().contains("\"always\":true"));
    assert!(bad("    output {\n        artifact \"x\" {\n            path \"/src/a\"\n            when \"failed\"\n        }\n    }\n")
        .contains("\"passed\" (the default) or \"always\""));

    // oxygenhub#127: a report is a directory output; it shares names with
    // artifacts, and its kind is part of the job's hash.
    let as_kind = |kind: &str| {
        Pipeline::parse(&src(&format!("    output {{\n        {kind} \"perf\" {{\n            path \"/src/target/report\"\n            when \"always\"\n        }}\n    }}\n"))).unwrap()
    };
    let (report, artifact) = (as_kind("report"), as_kind("artifact"));
    let o = &report.jobs["build"].outputs[0];
    assert!(o.kind == OutputKind::Report && o.always && o.path == "/src/target/report");
    assert!(serde_json::to_string(o).unwrap().contains("\"kind\":\"report\""));
    assert!(!serde_json::to_string(&artifact.jobs["build"].outputs[0]).unwrap().contains("kind"));
    assert_ne!(report.jobs["build"].hash(sb), artifact.jobs["build"].hash(sb));
    assert!(bad("    output {\n        artifact \"x\" {\n            path \"/src/a\"\n        }\n        report \"x\" {\n            path \"/src/b\"\n        }\n    }\n")
        .contains("given twice"));
    assert!(bad("    output {\n        report \"x\" {\n        }\n    }\n").contains("needs a path"));
}

/// metalgear feedback 19 (oxygenflow#17): `\033` in a string is not KDL;
/// the problems say to write ESC as `\u{1b}`, which parses.
#[test]
fn ansi_escapes_are_written_as_unicode() {
    let bad = problems(&doc("job \"a\" {\n    sandbox \"alpine\"\n    exec \"printf\" \"\\033[32mok\\033[0m\"\n}\n"));
    assert!(bad.iter().any(|(line, m)| *line == 13 && m.contains("\\u{1b}")), "{bad:?}");
    let p = Pipeline::parse(&doc("job \"a\" {\n    sandbox \"alpine\"\n    exec \"printf\" \"\\u{1b}[32mok\\u{1b}[0m\"\n}\n")).unwrap();
    assert_eq!(p.jobs["a"].exec[1], "\u{1b}[32mok\u{1b}[0m");
}

#[test]
fn runs_on_asks_for_runner_labels() {
    let src = doc("job \"cross\" {\n    sandbox \"alpine\"\n    runs-on \"riscv64\" \"big-disk\"\n    exec \"make\"\n}\n")
        .replace("    network", "    runs-on \"gpu\"\n    network");
    let p = Pipeline::parse(&src).unwrap();
    let job = &p.jobs["cross"];
    assert_eq!(job.runs_on, vec!["big-disk", "riscv64"], "sorted");
    let sandbox = &p.sandboxes["alpine"];
    assert_eq!(sandbox.runs_on, vec!["gpu"]);
    assert_eq!(job.in_sandbox(sandbox).runs_on, vec!["big-disk", "gpu", "riscv64"], "with its sandbox's");
    // A pipeline without runs-on serialises as before, so its jobs' hashes stay.
    let plain = Pipeline::parse(&doc("job \"x\" {\n    sandbox \"alpine\"\n    exec \"true\"\n}\n")).unwrap();
    assert!(!serde_json::to_string(&plain.jobs["x"]).unwrap().contains("runs_on"));
    let bad = problems(&doc("job \"x\" {\n    sandbox \"alpine\"\n    runs-on \"RISCV\"\n    exec \"true\"\n}\n"));
    assert!(bad.iter().any(|(_, m)| m.contains("a label is lower-case")), "{bad:?}");
    let none = problems(&doc("job \"y\" {\n    sandbox \"alpine\"\n    runs-on\n    exec \"true\"\n}\n"));
    assert!(none.iter().any(|(_, m)| m.contains("needs one or more labels")), "{none:?}");
}

/// oxygenhub#244: a pipeline may lower its jobs' priority, for all of them
/// or one by one, never raise it.
#[test]
fn a_pipeline_lowers_its_jobs_priority_and_never_raises_it() {
    let jobs = "job \"nightly\" {\n    sandbox \"alpine\"\n    exec \"true\"\n}\njob \"check\" {\n    sandbox \"alpine\"\n    priority \"normal\"\n    exec \"true\"\n}\n";
    let p = Pipeline::parse(&format!("priority \"low\"\n{}", doc(jobs))).unwrap();
    assert_eq!(p.jobs["nightly"].priority.as_deref(), Some("low"), "the pipeline's");
    assert_eq!(p.jobs["check"].priority.as_deref(), Some("normal"), "its own");
    let one = Pipeline::parse(&doc("job \"a\" {\n    sandbox \"alpine\"\n    priority \"low\"\n    exec \"true\"\n}\n")).unwrap();
    assert_eq!(one.jobs["a"].priority.as_deref(), Some("low"));
    // None given: serialised as before, so its jobs' hashes stay.
    let plain = Pipeline::parse(&doc("job \"x\" {\n    sandbox \"alpine\"\n    exec \"true\"\n}\n")).unwrap();
    assert!(!serde_json::to_string(&plain.jobs["x"]).unwrap().contains("priority"));
    for high in ["high", "urgent"] {
        let bad = problems(&doc(&format!("job \"x\" {{\n    sandbox \"alpine\"\n    priority \"{high}\"\n    exec \"true\"\n}}\n")));
        assert!(bad.iter().any(|(_, m)| m.contains("not raise them")), "{high}: {bad:?}");
    }
    let odd = problems(&doc("job \"x\" {\n    sandbox \"alpine\"\n    priority \"soon\"\n    exec \"true\"\n}\n"));
    assert!(odd.iter().any(|(_, m)| m.contains("expected `low` or `normal`")), "{odd:?}");
}

/// oxygenflow#23: `timeout` on a job, or on its sandbox for every job in
/// it, as a duration from a minute to a day.
#[test]
fn a_job_says_how_long_it_may_run() {
    use crate::pipeline::{parse_duration, Pipeline};
    assert_eq!(parse_duration("90m"), Some(5400));
    assert_eq!(parse_duration("1h30m"), Some(5400));
    assert_eq!(parse_duration("2h"), Some(7200));
    assert_eq!(parse_duration("45s"), Some(45));
    for bad in ["", "90", "m", "30m1h", "1h1h", "2d", "-1h"] {
        assert_eq!(parse_duration(bad), None, "{bad:?}");
    }
    let root = "a".repeat(64);
    let text = format!(
        "schema-version 1\nsandbox \"s\" {{\n    runtime \"microvm\"\n    rootfs \"cas://blob/{root}\"\n    memory-limit \"1GiB\"\n    cpu-limit 1\n    network \"isolated\"\n    timeout \"2h\"\n}}\njob \"long\" {{\n    sandbox \"s\"\n    timeout \"3h30m\"\n    exec \"true\"\n}}\njob \"plain\" {{\n    sandbox \"s\"\n    exec \"true\"\n}}\n"
    );
    let p = Pipeline::parse(&text).unwrap();
    let job = |n: &str| p.jobs[n].in_sandbox(&p.sandboxes["s"]);
    assert_eq!(job("long").timeout_s, Some(12600), "the job's own");
    assert_eq!(job("plain").timeout_s, Some(7200), "the sandbox's");
    let bad = text.replace("timeout \"3h30m\"", "timeout \"2d\"");
    assert!(Pipeline::parse(&bad).unwrap_err().to_string().contains("timeout"));
    let long = text.replace("timeout \"3h30m\"", "timeout \"25h\"");
    assert!(Pipeline::parse(&long).unwrap_err().to_string().contains("from 1m to 24h"));
}

/// oxygenflow#24: `on` says what starts a job, at pipeline or job level;
/// a job without one runs on every push and by hand, as before.
#[test]
fn on_says_what_starts_a_job() {
    use crate::pipeline::{Event, Pipeline};
    let root = "a".repeat(64);
    let text = format!(
        "schema-version 1\non {{\n    push \"main\" \"feature/*\"\n    manual\n}}\nsandbox \"s\" {{\n    runtime \"microvm\"\n    rootfs \"cas://blob/{root}\"\n    memory-limit \"1GiB\"\n    cpu-limit 1\n    network \"isolated\"\n}}\njob \"test\" {{\n    sandbox \"s\"\n    exec \"true\"\n}}\njob \"dist\" {{\n    sandbox \"s\"\n    on {{\n        tag \"v*\"\n        manual\n    }}\n    exec \"true\"\n}}\njob \"nightly\" {{\n    sandbox \"s\"\n    on {{\n        schedule \"0 3 * * *\"\n    }}\n    exec \"true\"\n}}\n"
    );
    let p = Pipeline::parse(&text).unwrap();
    let runs = |job: &str, e: Event| p.on_of(&p.jobs[job]).matches(e);
    assert!(runs("test", Event::Push("main")) && runs("test", Event::Push("feature/9-x")));
    assert!(!runs("test", Event::Push("develop")), "not a bookmark the pipeline names");
    assert!(!runs("test", Event::Tag("v1.0.0")) && runs("test", Event::Manual));
    assert!(runs("dist", Event::Tag("v0.3.1")) && !runs("dist", Event::Tag("nightly")));
    assert!(!runs("dist", Event::Push("main")) && runs("dist", Event::Manual));
    assert!(runs("nightly", Event::Schedule) && !runs("nightly", Event::Manual));
    use crate::pipeline::When;
    let at = |hour, minute| Event::ScheduleAt(When { minute, hour, day: 2, month: 10, weekday: 5 });
    assert!(runs("nightly", at(3, 0)) && !runs("nightly", at(3, 1)) && !runs("test", at(3, 0)));
    // Without any `on`, as before.
    let plain = text.replace("on {\n    push \"main\" \"feature/*\"\n    manual\n}\n", "");
    let p = Pipeline::parse(&plain).unwrap();
    assert!(p.on_of(&p.jobs["test"]).matches(Event::Push("anything")) && p.on_of(&p.jobs["test"]).matches(Event::Manual));
    for (bad, why) in [
        ("on {\n}\n", "names no event"),
        ("on {\n    when \"x\"\n}\n", "unknown `on` event"),
        ("on {\n    schedule \"0 3 * *\"\n}\n", "five cron fields"),
    ] {
        let t = format!("{bad}{plain}");
        let e = Pipeline::parse(&t).unwrap_err().to_string();
        assert!(e.contains(why), "{bad:?}: {e}");
    }
}

#[test]
fn a_job_takes_an_artifact_of_a_job_it_depends_on() {
    let jobs = |apply: &str| {
        doc(&format!(
            "job \"plan\" {{\n    sandbox \"alpine\"\n    output {{ artifact \"tfplan\" {{ path \"/src/out/tfplan\"; }}; report \"r\" {{ path \"/src/out/r\"; }}; }}\n    exec \"plan\"\n}}\njob \"apply\" {{\n    sandbox \"alpine\"\n{apply}    exec \"apply\"\n}}\n"
        ))
    };
    let ok = jobs("    depends-on \"plan\"\n    input { artifact \"tfplan\" from \"plan\"; path \"/src/tfplan\"; }\n");
    let p = Pipeline::parse(&ok).unwrap();
    assert_eq!(p.jobs["apply"].inputs, [crate::Input { artifact: "tfplan".into(), from: "plan".into(), path: "/src/tfplan".into() }]);
    let without = Pipeline::parse(&jobs("    depends-on \"plan\"\n")).unwrap();
    assert_ne!(p.hash(), without.hash(), "inputs are part of the job");

    let has = |apply: &str, want: &str| {
        let ps = problems(&jobs(apply));
        assert!(ps.iter().any(|(_, m)| m.contains(want)), "{apply}: {ps:?}");
    };
    has("    input { artifact \"tfplan\" from \"plan\"; path \"/src/tfplan\"; }\n", "add depends-on \"plan\"");
    has("    depends-on \"plan\"\n    input { artifact \"nope\" from \"plan\"; path \"/src/x\"; }\n", "declares no artifact \"nope\"");
    has("    depends-on \"plan\"\n    input { artifact \"r\" from \"plan\"; path \"/src/x\"; }\n", "is a report");
    has("    input { artifact \"tfplan\" from \"ghost\"; path \"/src/x\"; }\n", "which is not declared");
    has("    depends-on \"plan\"\n    input { artifact \"tfplan\" from \"plan\"; path \"/etc/x\"; }\n", "absolute path under /src");
    has("    depends-on \"plan\"\n    input { artifact \"tfplan\" \"plan\"; path \"/src/x\"; }\n", "artifact \"tfplan\" from \"plan\"");
    has("    depends-on \"plan\"\n    input { artifact \"tfplan\" from \"plan\"; }\n", "needs a `path`");
    has(
        "    depends-on \"plan\"\n    input { artifact \"tfplan\" from \"plan\"; path \"/src/x\"; }\n    input { artifact \"tfplan\" from \"plan\"; path \"/src/x\"; }\n",
        "two inputs are put at",
    );
}

#[test]
fn a_job_can_wait_for_approval() {
    let job = |approval: &str| doc(&format!("job \"apply\" {{\n    sandbox \"alpine\"\n{approval}    exec \"apply\"\n}}\n"));
    let p = Pipeline::parse(&job("    approval { by \"maintain\"; others-only #true; expires \"2h\"; }\n")).unwrap();
    assert_eq!(
        p.jobs["apply"].approval,
        Some(crate::Approval { by: "maintain".into(), others_only: true, expires_s: 7200, sealed: false })
    );
    let plain = Pipeline::parse(&job("    approval\n")).unwrap();
    assert_eq!(
        plain.jobs["apply"].approval,
        Some(crate::Approval { by: "maintain".into(), others_only: false, expires_s: 86400, sealed: false }),
        "defaults"
    );
    // Sealed (oxygenhub#235): on by default for a job with an environment,
    // as it says otherwise.
    let sealed = |extra: &str| Pipeline::parse(&job(extra)).unwrap().jobs["apply"].approval.clone().unwrap().sealed;
    assert!(sealed("    environment \"prod\"\n    approval\n"));
    assert!(!sealed("    environment \"prod\"\n    approval { sealed #false; }\n"));
    assert!(sealed("    approval { sealed #true; }\n"));
    assert_ne!(Pipeline::parse(&job("    approval { sealed #true; }\n")).unwrap().hash(), plain.hash(), "part of the job");
    assert_eq!(Pipeline::parse(&job("")).unwrap().jobs["apply"].approval, None);
    assert_ne!(p.hash(), plain.hash(), "the approval is part of the job");
    let has = |a: &str, want: &str| {
        let ps = problems(&job(a));
        assert!(ps.iter().any(|(_, m)| m.contains(want)), "{a}: {ps:?}");
    };
    has("    approval { by \"read\"; }\n", "expected \"write\", \"maintain\" or \"admin\"");
    has("    approval { others-only \"yes\"; }\n", "#true or #false");
    has("    approval { expires \"800h\"; }\n", "from 1m to 720h");
    has("    approval { when \"later\"; }\n", "unknown approval setting");
    has("    approval { sealed \"yes\"; }\n", "#true or #false");
}

#[test]
fn a_job_names_the_environment_it_deploys_to() {
    let job = |e: &str| doc(&format!("job \"apply\" {{\n    sandbox \"alpine\"\n{e}    exec \"apply\"\n}}\n"));
    let p = Pipeline::parse(&job("    environment \"prod\"\n")).unwrap();
    assert_eq!(p.jobs["apply"].environment.as_deref(), Some("prod"));
    assert_ne!(p.hash(), Pipeline::parse(&job("")).unwrap().hash(), "part of the job");
    assert!(problems(&job("    environment \"Prod env\"\n")).iter().any(|(_, m)| m.contains("1 to 32 of a-z 0-9 -")));
}

/// oxygenflow#75: a WASM sandbox needs no rootfs and has no network; its
/// job's exec names the module.
#[test]
fn a_wasm_sandbox_runs_a_module() {
    let src = "schema-version 1\nsandbox \"w\" {\n  runtime \"wasm\"\n  memory-limit \"512MiB\"\n}\njob \"t\" {\n  sandbox \"w\"\n  exec \"target/wasm32-wasip1/debug/deps/*.wasm\" \"--quiet\"\n}\n";
    let p = Pipeline::parse(src).unwrap_or_else(|ps| panic!("{:?}", ps.0));
    let sb = &p.sandboxes["w"];
    assert_eq!((sb.runtime, sb.rootfs.as_str(), sb.network, sb.memory_mib), (Runtime::Wasm, "", Network::Isolated, 512));
    let networked =
        "schema-version 1\nsandbox \"w\" {\n  runtime \"wasm\"\n  network \"loopback\"\n}\njob \"t\" { sandbox \"w\"; exec \"a.wasm\"; }\n";
    assert!(problems(networked).iter().any(|(_, m)| m.contains("has no network")));
}