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 \"egress\"\n}}\njob \"a\" {{ sandbox \"s\"; exec \"true\"; }}\n");
    let ps = problems(&src);
    assert!(ps.iter().any(|(l, m)| *l == 3 && m.contains("wasm")), "{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 },
            Mount { source: MountSource::Tmpfs, target: "/tmp/cache".into(), size_mib: 256 },
        ],
        "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 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_is_in_m0_and_egress_is_not() {
    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");
    assert!(problems(&src("egress")).iter().any(|(l, m)| *l == 4 && m.contains("not in M0")));
}

#[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,
    };
    let env = job.with_flow_env(&claim.flow_env()).env;
    assert_eq!(env["FLOW_REPO"], "andy/demo");
    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, ..claim };
    let names: Vec<String> = old.flow_env().into_iter().map(|(k, _)| k).collect();
    assert_eq!(names, ["FLOW_REPO", "FLOW_HEADS", "FLOW_JOB"]);
}

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