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 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![],

    };

    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();