//! Tests for [`super`].

use super::*;

/// Nothing deployed, so nothing pinned: the tests that exercise the count
/// alone pass this, and the ones that exercise pinning build their own set.
fn no_pins() -> PinnedReleases {
    PinnedReleases::none()
}

use crate::topology::NodeCompanion;
use async_trait::async_trait;
use ops_exec::{CapabilitySet, LocalExec, LogSink, SshExec};
use std::os::unix::process::ExitStatusExt;
use std::sync::{Arc, Mutex as StdMutex};
use std::time::SystemTime;

// ---- placement ----
//
// The whole table, because the interesting cases are the two where one side
// said nothing. Treating silence as agreement is how a wrong-architecture
// deploy would get through, and it is the shape a "check it before you call"
// guard tends to end up with.

fn node_on(platform: Option<&str>) -> Node {
    Node {
        name: crate::domain::NodeId::new("n1"),
        ssh_target: "deploy@n1".into(),
        release_root: "/opt/x".into(),
        platform: platform.map(|p| Platform::parse(p).unwrap()),
        base_image: None,
        libc: None,
        service_name: "x.service".into(),
        config_check_env_file: None,
        actuate: crate::topology::default_actuate(),
        observe: crate::topology::default_observe(),
        health_url: None,
        companions: Vec::new(),
    }
}

/// A node that declares a glibc older than the bundle needs is refused
/// before the rsync, and the message names both numbers so the operator
/// knows which side to fix.
#[tokio::test]
async fn a_bundle_above_the_node_s_declared_glibc_is_refused_before_the_rsync() {
    let dir = tempfile::tempdir().unwrap();
    let exe = std::fs::read(std::env::current_exe().unwrap()).unwrap();
    std::fs::write(dir.path().join("bin"), &exe).unwrap();
    let Some(floor) = crate::elf::glibc_floor(&exe) else {
        return; // a static test binary states no floor; nothing to compare
    };

    let mut node = node_on(None);
    node.libc = Some("2.0".into()); // older than anything real
    let err = check_bundle_fits_node(&node, dir.path())
        .await
        .expect_err("a bundle above the node's glibc must be refused");
    // `{:#}` walks the context chain: the outermost context is the
    // `FailureStage`, whose Display is the operator-facing "nothing moved"
    // line, and the cause below it is the reason.
    let msg = format!("{err:#}");
    assert!(
        msg.contains(&floor.to_string()) && msg.contains("2.0"),
        "the refusal must name both numbers: {msg}"
    );
    assert_eq!(
        stage_of(&err),
        Some(FailureStage::BeforeSwap),
        "refusing here must be recoverable: nothing has moved yet"
    );
}

#[tokio::test]
async fn a_bundle_within_the_node_s_declared_glibc_passes() {
    let dir = tempfile::tempdir().unwrap();
    let exe = std::fs::read(std::env::current_exe().unwrap()).unwrap();
    std::fs::write(dir.path().join("bin"), &exe).unwrap();

    let mut node = node_on(None);
    node.libc = Some("99.0".into()); // newer than anything real
    check_bundle_fits_node(&node, dir.path())
        .await
        .expect("a bundle the node can load must pass");
}

/// The three ways there is nothing to compare. All three pass, because
/// "cannot verify" is not "known bad" — the same call `arch_guard_script`
/// makes for an unmapped architecture.
#[tokio::test]
async fn nothing_to_compare_is_a_pass_not_a_refusal() {
    let dir = tempfile::tempdir().unwrap();
    let exe = std::fs::read(std::env::current_exe().unwrap()).unwrap();
    std::fs::write(dir.path().join("bin"), &exe).unwrap();

    // 1. The node declares no libc.
    let node = node_on(None);
    check_bundle_fits_node(&node, dir.path()).await.unwrap();

    // 2. The node's declared libc is not a version (a config typo).
    let mut typo = node_on(None);
    typo.libc = Some("noble".into());
    check_bundle_fits_node(&typo, dir.path()).await.unwrap();

    // 3. The bundle holds no ELF, so it states no floor.
    let empty = tempfile::tempdir().unwrap();
    std::fs::write(empty.path().join("style.css"), b"body{}").unwrap();
    let mut strict = node_on(None);
    strict.libc = Some("2.0".into());
    check_bundle_fits_node(&strict, empty.path())
        .await
        .expect("a bundle with no binaries has no floor to exceed");
}

#[test]
fn matching_platforms_are_placeable() {
    let node = node_on(Some("linux/aarch64"));
    let art = Platform::parse("linux/aarch64").unwrap();
    let p = Placement::check(&node, Path::new("/r/abc"), Some(&art)).expect("a match places");
    assert_eq!(p.bundle(), Path::new("/r/abc"));
    assert_eq!(p.node().name.as_str(), "n1");
}

#[test]
fn a_different_architecture_is_refused() {
    // The failure this type exists for: pom's aarch64 bundle reaching the
    // x86_64 box, which execs nothing and takes the watcher down.
    let node = node_on(Some("linux/x86_64"));
    let art = Platform::parse("linux/aarch64").unwrap();
    let err = Placement::check(&node, Path::new("/r/abc"), Some(&art)).unwrap_err();
    assert!(
        matches!(err, PlacementError::Mismatch { .. }),
        "expected a mismatch, got {err}"
    );
    // The message has to name both, or an operator cannot tell which half
    // is wrong.
    let msg = err.to_string();
    assert!(
        msg.contains("linux/x86_64") && msg.contains("linux/aarch64"),
        "{msg}"
    );
}

#[test]
fn a_silent_node_refuses_a_stated_artifact() {
    // Not "the node probably runs it". A node that never said what it is
    // cannot vouch for a bundle that did, and the pairing that looks
    // harmless here is exactly the one that ships the wrong half of a
    // two-architecture release.
    let node = node_on(None);
    let art = Platform::parse("linux/aarch64").unwrap();
    assert!(matches!(
        Placement::check(&node, Path::new("/r/abc"), Some(&art)),
        Err(PlacementError::NodeSilent { .. })
    ));
}

#[test]
fn a_stated_node_refuses_a_silent_artifact() {
    let node = node_on(Some("linux/aarch64"));
    assert!(matches!(
        Placement::check(&node, Path::new("/r/abc"), None),
        Err(PlacementError::ArtifactSilent { .. })
    ));
}

#[test]
fn both_silent_is_the_single_platform_world_and_still_places() {
    // MNW is here and stays here. Its nodes declare nothing and its builds
    // record nothing, which is the truth about a product with one build host
    // and one architecture. The moment either side starts stating, the other
    // has to as well — that is the forcing function, and it is why this cell
    // is the only admissible non-match.
    let node = node_on(None);
    Placement::check(&node, Path::new("/r/abc"), None).expect("the pre-pom world still ships");
}

#[test]
fn platform_parsing_is_a_shape_not_a_spelling() {
    assert_eq!(
        Platform::parse("Linux/AArch64").unwrap(),
        Platform::parse("linux/aarch64").unwrap(),
        "case is not a distinction between two machines"
    );
    for bad in ["linux", "linux/", "/aarch64", "linux/aarch64/gnu", ""] {
        assert!(Platform::parse(bad).is_err(), "{bad:?} should not parse");
    }
}

// ---- failure stage ----
//
// The 2026-08-01 prod deploy failed its pre-swap config check, and the
// rollback then failed the same way — which left the node safely on the old
// version, and was reported as "it remains on the new version, manual
// intervention needed". These pin the distinction the reporting layer now
// depends on.

#[test]
fn a_pre_swap_failure_is_recoverable_as_such() {
    let e = anyhow::anyhow!("Permission denied")
        .context("pre-swap config check failed")
        .context(FailureStage::BeforeSwap);
    assert_eq!(stage_of(&e), Some(FailureStage::BeforeSwap));
    // The reason survives alongside the stage; the stage does not replace it.
    let rendered = format!("{e:#}");
    assert!(
        rendered.contains("pre-swap config check failed"),
        "{rendered}"
    );
    assert!(rendered.contains("Permission denied"), "{rendered}");
}

#[test]
fn a_post_swap_failure_is_recoverable_as_such() {
    let e = anyhow::anyhow!("unit failed to start")
        .context("companion x deploy failed (server already swapped)")
        .context(FailureStage::AtOrAfterSwap);
    assert_eq!(stage_of(&e), Some(FailureStage::AtOrAfterSwap));
}

#[test]
fn an_unannotated_failure_has_no_stage() {
    // Must be None, not a default. A caller seeing None has to treat the
    // node as indeterminate; inferring "before the swap" would reintroduce
    // the original bug pointing the other way, which is the dangerous way.
    let e = anyhow::anyhow!("something older, from before stages existed");
    assert_eq!(stage_of(&e), None);
}

// ---- env file readability probe ----

#[tokio::test]
async fn readability_probe_passes_on_a_readable_file() {
    let tmp = tempfile::tempdir().unwrap();
    let f = tmp.path().join("ok.env");
    tokio::fs::write(&f, "A=1\n").await.unwrap();
    let script = readability_probe_script(&f.to_string_lossy());
    let out = run_checked(&local_executor(), &script, "probe").await;
    assert!(out.is_ok(), "{:?}", out.err().map(|e| format!("{e:#}")));
}

#[tokio::test]
async fn readability_probe_names_the_user_and_mode_when_unreadable() {
    // Root can read anything, so a mode-based test would pass spuriously
    // there. Skip rather than assert something false. No libc dependency
    // for one probe: a 0-mode temp file is readable iff we are root.
    let probe_dir = tempfile::tempdir().unwrap();
    let probe_file = probe_dir.path().join("root-check");
    tokio::fs::write(&probe_file, "x").await.unwrap();
    tokio::fs::set_permissions(
        &probe_file,
        std::os::unix::fs::PermissionsExt::from_mode(0o000),
    )
    .await
    .unwrap();
    if tokio::fs::read(&probe_file).await.is_ok() {
        return; // running as root
    }
    let tmp = tempfile::tempdir().unwrap();
    let f = tmp.path().join("locked.env");
    tokio::fs::write(&f, "A=1\n").await.unwrap();
    tokio::fs::set_permissions(&f, std::os::unix::fs::PermissionsExt::from_mode(0o000))
        .await
        .unwrap();

    let script = readability_probe_script(&f.to_string_lossy());
    let err = run_checked(&local_executor(), &script, "probe")
        .await
        .expect_err("an unreadable file must fail the probe");
    let msg = format!("{err:#}");
    // The two things the raw bash error does not tell you.
    assert!(msg.contains("cannot read"), "{msg}");
    assert!(msg.contains("mode 0") || msg.contains("mode "), "{msg}");
}

#[tokio::test]
async fn readability_probe_distinguishes_missing_from_unreadable() {
    let tmp = tempfile::tempdir().unwrap();
    let missing = tmp.path().join("nope.env");
    let script = readability_probe_script(&missing.to_string_lossy());
    let err = run_checked(&local_executor(), &script, "probe")
        .await
        .expect_err("a missing file must fail the probe");
    let msg = format!("{err:#}");
    assert!(msg.contains("does not exist"), "{msg}");
}

#[test]
fn the_two_stages_read_differently() {
    // These strings end up in an operator's terminal during an incident.
    let before = FailureStage::BeforeSwap.to_string();
    let after = FailureStage::AtOrAfterSwap.to_string();
    assert!(before.contains("previous version"), "{before}");
    assert!(after.contains("indeterminate"), "{after}");
    assert_ne!(before, after);
}

/// A LocalExec granted the default node capabilities (deploy + restart).
fn local_executor() -> LocalExec {
    LocalExec::new(CapabilitySet::from_tokens(
        ["deploy", "restart"],
        ["health"],
    ))
}

#[tokio::test]
async fn deploy_local_copies_multiple_binaries_and_swaps_symlink() {
    let tmp = tempfile::tempdir().unwrap();
    let root = tmp.path();

    let src_dir = root.join("src");
    tokio::fs::create_dir_all(&src_dir).await.unwrap();
    let primary = src_dir.join("makenotwork");
    let admin = src_dir.join("mnw-admin");
    tokio::fs::write(&primary, b"PRIMARY").await.unwrap();
    tokio::fs::write(&admin, b"ADMIN").await.unwrap();

    let release_root = root.join("releases-root");
    tokio::fs::create_dir_all(&release_root).await.unwrap();

    // Stage into staging/<build_id> (no publish yet).
    let staging = stage_local_bundle(&release_root, 42, &[primary.clone(), admin.clone()])
        .await
        .expect("stage_local_bundle should succeed");
    assert_eq!(staging, release_root.join("staging").join("42"));
    assert!(
        !release_root.join("current").exists(),
        "staging must not publish or flip current"
    );

    // Publish content-addressed at releases/<digest16>.
    let released = finalize_local_release(&release_root, &staging, "deadbeefcafe0000", &no_pins())
        .await
        .expect("finalize_local_release should succeed");
    assert_eq!(
        released,
        release_root.join("releases").join("deadbeefcafe0000")
    );
    assert!(
        !staging.exists(),
        "staging dir is consumed by the publish rename"
    );
    assert_eq!(
        tokio::fs::read(released.join("makenotwork")).await.unwrap(),
        b"PRIMARY"
    );
    assert_eq!(
        tokio::fs::read(released.join("mnw-admin")).await.unwrap(),
        b"ADMIN"
    );

    let current = release_root.join("current");
    let target = tokio::fs::read_link(&current).await.unwrap();
    assert_eq!(target.to_string_lossy(), "releases/deadbeefcafe0000");
    let via_current = tokio::fs::read(current.join("makenotwork")).await.unwrap();
    assert_eq!(via_current, b"PRIMARY");
}

#[tokio::test]
async fn finalize_second_release_swaps_symlink_and_keeps_old_dir() {
    let tmp = tempfile::tempdir().unwrap();
    let root = tmp.path();
    let src_dir = root.join("src");
    tokio::fs::create_dir_all(&src_dir).await.unwrap();
    let bin = src_dir.join("server");
    tokio::fs::write(&bin, b"V1").await.unwrap();

    let release_root = root.join("rr");
    tokio::fs::create_dir_all(&release_root).await.unwrap();

    // Two builds, distinct digests (distinct content) -> two release dirs.
    let s1 = stage_local_bundle(&release_root, 1, std::slice::from_ref(&bin))
        .await
        .unwrap();
    finalize_local_release(&release_root, &s1, "1111111111111111", &no_pins())
        .await
        .unwrap();
    tokio::fs::write(&bin, b"V2").await.unwrap();
    let s2 = stage_local_bundle(&release_root, 2, std::slice::from_ref(&bin))
        .await
        .unwrap();
    finalize_local_release(&release_root, &s2, "2222222222222222", &no_pins())
        .await
        .unwrap();

    assert!(
        release_root
            .join("releases/1111111111111111/server")
            .exists()
    );
    assert!(
        release_root
            .join("releases/2222222222222222/server")
            .exists()
    );
    let target = tokio::fs::read_link(release_root.join("current"))
        .await
        .unwrap();
    assert_eq!(target.to_string_lossy(), "releases/2222222222222222");
    let via_current = tokio::fs::read(release_root.join("current/server"))
        .await
        .unwrap();
    assert_eq!(via_current, b"V2");
}

#[tokio::test]
async fn finalize_reuses_an_existing_release_of_the_same_digest() {
    let tmp = tempfile::tempdir().unwrap();
    let root = tmp.path();
    let bin = root.join("server");
    tokio::fs::write(&bin, b"BYTES").await.unwrap();
    let release_root = root.join("rr");
    tokio::fs::create_dir_all(&release_root).await.unwrap();

    let s1 = stage_local_bundle(&release_root, 1, std::slice::from_ref(&bin))
        .await
        .unwrap();
    finalize_local_release(&release_root, &s1, "abc123abc123abc1", &no_pins())
        .await
        .unwrap();
    // Same digest rebuilt (e.g. a re-run at the same content): finalize must
    // reuse the existing release and drop the redundant staging dir, not error.
    let s2 = stage_local_bundle(&release_root, 2, std::slice::from_ref(&bin))
        .await
        .unwrap();
    let released = finalize_local_release(&release_root, &s2, "abc123abc123abc1", &no_pins())
        .await
        .expect("finalize is idempotent on a repeated digest");
    assert_eq!(released, release_root.join("releases/abc123abc123abc1"));
    assert!(!s2.exists(), "redundant staging dropped");
}

#[tokio::test]
async fn manifest_verify_script_passes_on_match_fails_on_drift_and_skips_when_absent() {
    // The node-side verification is a shell running `sha256sum -c MANIFEST`;
    // drive the real script through bash to prove it accepts a good bundle,
    // rejects a tampered one, and no-ops on a legacy (MANIFEST-less) bundle.
    let dir = tempfile::tempdir().unwrap();
    tokio::fs::write(dir.path().join("server"), b"BINARY")
        .await
        .unwrap();
    tokio::fs::create_dir(dir.path().join("static"))
        .await
        .unwrap();
    tokio::fs::write(dir.path().join("static/app.css"), b"body{}")
        .await
        .unwrap();
    let digest = crate::bundle::digest_dir(dir.path()).await.unwrap();
    tokio::fs::write(dir.path().join("MANIFEST"), digest.manifest.as_bytes())
        .await
        .unwrap();

    let run = |d: &std::path::Path| {
        let script = manifest_verify_script(d.to_str().unwrap());
        async move {
            Command::new("bash")
                .arg("-c")
                .arg(&script)
                .output()
                .await
                .unwrap()
        }
    };

    let ok = run(dir.path()).await;
    assert!(
        ok.status.success(),
        "matching bundle verifies: {}",
        String::from_utf8_lossy(&ok.stderr)
    );

    // Drift one file: sha256sum -c must fail (current symlink left intact).
    tokio::fs::write(dir.path().join("static/app.css"), b"TAMPERED")
        .await
        .unwrap();
    let bad = run(dir.path()).await;
    assert!(!bad.status.success(), "a drifted file fails verification");

    // Legacy bundle with no MANIFEST: skip, not fail.
    let legacy = tempfile::tempdir().unwrap();
    tokio::fs::write(legacy.path().join("server"), b"x")
        .await
        .unwrap();
    let skip = run(legacy.path()).await;
    assert!(
        skip.status.success(),
        "a bundle without a MANIFEST skips verification rather than failing"
    );
}

#[tokio::test]
async fn gc_local_releases_keeps_last_n_by_mtime() {
    let tmp = tempfile::tempdir().unwrap();
    let root = tmp.path();
    let releases = root.join("releases");
    tokio::fs::create_dir_all(&releases).await.unwrap();

    let total = RELEASES_TO_KEEP + 3;
    let mut names = Vec::new();
    for i in 0..total {
        let name = format!("v{i:02}");
        let dir = releases.join(&name);
        tokio::fs::create_dir(&dir).await.unwrap();
        let f = std::fs::File::open(&dir).unwrap();
        let when =
            SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(1_700_000_000 + i as u64);
        let times = std::fs::FileTimes::new().set_modified(when);
        f.set_times(times).unwrap();
        names.push(name);
    }

    gc_local_releases(root, &no_pins()).await.unwrap();

    let surviving_expected: Vec<_> = names
        .iter()
        .skip(total - RELEASES_TO_KEEP)
        .cloned()
        .collect();
    for name in &surviving_expected {
        assert!(releases.join(name).exists(), "expected to survive: {name}");
    }
    for name in names.iter().take(total - RELEASES_TO_KEEP) {
        assert!(
            !releases.join(name).exists(),
            "expected to be pruned: {name}"
        );
    }
}

#[tokio::test]
async fn gc_local_releases_never_evicts_a_pinned_dir() {
    // The 2026-08-25 shape exactly: the oldest dir is the one production is
    // running, and enough newer rebuilds exist to push it past the count.
    // Under the count alone it was the first thing deleted.
    let tmp = tempfile::tempdir().unwrap();
    let root = tmp.path();
    let releases = root.join("releases");
    tokio::fs::create_dir_all(&releases).await.unwrap();

    let total = RELEASES_TO_KEEP + 3;
    let mut names = Vec::new();
    for i in 0..total {
        let name = format!("v{i:02}");
        let dir = releases.join(&name);
        tokio::fs::create_dir(&dir).await.unwrap();
        let f = std::fs::File::open(&dir).unwrap();
        let when =
            SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(1_700_000_000 + i as u64);
        f.set_times(std::fs::FileTimes::new().set_modified(when))
            .unwrap();
        names.push(name);
    }

    // The two oldest: what a tier is running and what it would roll back to.
    let pinned: PinnedReleases = [names[0].clone(), names[1].clone()].into_iter().collect();
    gc_local_releases(root, &pinned).await.unwrap();

    for name in [&names[0], &names[1]] {
        assert!(
            releases.join(name).exists(),
            "a referenced artifact was evicted: {name}"
        );
    }
    // And the count still applies to everything else, from a floor that the
    // pinned pair did not eat into: the newest RELEASES_TO_KEEP unpinned
    // dirs survive, so pinning two costs two extra slots rather than two of
    // the five.
    let unpinned: Vec<_> = names.iter().filter(|n| !pinned.contains(n)).collect();
    let cut = unpinned.len() - RELEASES_TO_KEEP;
    for name in unpinned.iter().take(cut) {
        assert!(
            !releases.join(name).exists(),
            "expected to be pruned: {name}"
        );
    }
    for name in unpinned.iter().skip(cut) {
        assert!(releases.join(name).exists(), "expected to survive: {name}");
    }
}

#[tokio::test]
async fn gc_local_releases_keeps_a_pinned_dir_that_is_not_even_present() {
    // A pinned name with nothing on disk must not disturb the count. This is
    // the state the bug leaves behind, and gc runs again while it holds.
    let tmp = tempfile::tempdir().unwrap();
    let root = tmp.path();
    let releases = root.join("releases");
    tokio::fs::create_dir_all(&releases).await.unwrap();
    for i in 0..=RELEASES_TO_KEEP {
        tokio::fs::create_dir(releases.join(format!("v{i}")))
            .await
            .unwrap();
    }
    let pinned: PinnedReleases = ["gone-already".to_string()].into_iter().collect();
    gc_local_releases(root, &pinned).await.unwrap();

    let left = std::fs::read_dir(&releases).unwrap().count();
    assert_eq!(left, RELEASES_TO_KEEP);
}

#[tokio::test]
async fn gc_local_releases_noop_when_below_threshold() {
    let tmp = tempfile::tempdir().unwrap();
    let root = tmp.path();
    let releases = root.join("releases");
    tokio::fs::create_dir_all(&releases).await.unwrap();
    for i in 0..3 {
        tokio::fs::create_dir(releases.join(format!("v{i}")))
            .await
            .unwrap();
    }
    gc_local_releases(root, &no_pins()).await.unwrap();
    for i in 0..3 {
        assert!(releases.join(format!("v{i}")).exists());
    }
}

// ---- remote gc ----
//
// Driven through `LocalExec`, so these run the real shell the node runs
// rather than asserting on the script's text. The script is the half of the
// remote gc that can be wrong, and it is wrong with `rm -rf`.

/// `releases/` with `total` dirs named `v00..`, oldest first by mtime.
async fn releases_by_age(root: &Path, total: usize) -> Vec<String> {
    let releases = root.join("releases");
    tokio::fs::create_dir_all(&releases).await.unwrap();
    let mut names = Vec::new();
    for i in 0..total {
        let name = format!("v{i:02}");
        let dir = releases.join(&name);
        tokio::fs::create_dir(&dir).await.unwrap();
        // A file inside, so a deletion is visible as more than an empty dir.
        tokio::fs::write(dir.join("makenotwork"), b"x")
            .await
            .unwrap();
        let f = std::fs::File::open(&dir).unwrap();
        let when =
            SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(1_700_000_000 + i as u64);
        f.set_times(std::fs::FileTimes::new().set_modified(when))
            .unwrap();
        names.push(name);
    }
    names
}

/// Nothing pinned: the newest `RELEASES_TO_KEEP` survive.
#[tokio::test]
async fn gc_remote_releases_keeps_last_n_by_mtime_when_nothing_is_pinned() {
    let tmp = tempfile::tempdir().unwrap();
    let root = tmp.path();
    let total = RELEASES_TO_KEEP + 3;
    let names = releases_by_age(root, total).await;

    gc_remote_releases(&local_executor(), root.to_str().unwrap(), &no_pins())
        .await
        .unwrap();

    let releases = root.join("releases");
    for name in names.iter().take(total - RELEASES_TO_KEEP) {
        assert!(!releases.join(name).exists(), "expected pruned: {name}");
    }
    for name in names.iter().skip(total - RELEASES_TO_KEEP) {
        assert!(releases.join(name).exists(), "expected to survive: {name}");
    }
}

/// The done condition, on the node: the dirs a tier's current and previous
/// artifacts name survive even when they are the oldest on disk and well
/// past the count. Same shape as the host-store test above, which is the
/// point — the two stores now answer the same question the same way.
#[tokio::test]
async fn gc_remote_releases_never_evicts_a_pinned_dir() {
    let tmp = tempfile::tempdir().unwrap();
    let root = tmp.path();
    let total = RELEASES_TO_KEEP + 3;
    let names = releases_by_age(root, total).await;

    let pinned: PinnedReleases = [names[0].clone(), names[1].clone()].into_iter().collect();
    gc_remote_releases(&local_executor(), root.to_str().unwrap(), &pinned)
        .await
        .unwrap();

    let releases = root.join("releases");
    for name in [&names[0], &names[1]] {
        assert!(
            releases.join(name).exists(),
            "a referenced artifact was evicted from the node: {name}"
        );
    }
    // And pinning does not spend the count's slots, again matching the host.
    let unpinned: Vec<_> = names.iter().filter(|n| !pinned.contains(n)).collect();
    let cut = unpinned.len() - RELEASES_TO_KEEP;
    for name in unpinned.iter().take(cut) {
        assert!(!releases.join(name).exists(), "expected pruned: {name}");
    }
    for name in unpinned.iter().skip(cut) {
        assert!(releases.join(name).exists(), "expected to survive: {name}");
    }
}

/// Every dir pinned means the loop deletes nothing and the script still
/// exits 0. Worth its own test because the obvious implementation of this
/// filter is `grep -v`, which exits 1 when it selects no lines and would
/// have failed the deploy here under `set -e`.
#[tokio::test]
async fn gc_remote_releases_succeeds_when_everything_is_pinned() {
    let tmp = tempfile::tempdir().unwrap();
    let root = tmp.path();
    let names = releases_by_age(root, RELEASES_TO_KEEP + 3).await;
    let pinned: PinnedReleases = names.iter().cloned().collect();

    gc_remote_releases(&local_executor(), root.to_str().unwrap(), &pinned)
        .await
        .unwrap();

    let releases = root.join("releases");
    for name in &names {
        assert!(releases.join(name).exists(), "expected to survive: {name}");
    }
}

/// A `releases/` that does not exist is not an error: a node's first deploy
/// creates the dir, and gc runs on the same path.
#[tokio::test]
async fn gc_remote_releases_is_a_noop_when_the_store_is_missing() {
    let tmp = tempfile::tempdir().unwrap();
    gc_remote_releases(&local_executor(), tmp.path().to_str().unwrap(), &no_pins())
        .await
        .unwrap();
}

/// Names reach the script as positional parameters, so a name that looks
/// like shell must be compared whole rather than expanded or split. None of
/// these can be a digest16, but the pre-identity names are version strings
/// and the pinned set is data read out of a database.
#[tokio::test]
async fn gc_remote_releases_quotes_pinned_names() {
    let tmp = tempfile::tempdir().unwrap();
    let root = tmp.path();
    let releases = root.join("releases");
    tokio::fs::create_dir_all(&releases).await.unwrap();
    let awkward = ["a b", "x'y", "*"];
    for name in awkward {
        tokio::fs::create_dir(releases.join(name)).await.unwrap();
    }
    // Enough newer dirs that the count alone would evict all three.
    let filler: Vec<String> = (0..=RELEASES_TO_KEEP).map(|i| format!("f{i}")).collect();
    for name in &filler {
        tokio::fs::create_dir(releases.join(name)).await.unwrap();
    }

    let pinned: PinnedReleases = awkward.iter().map(|s| (*s).to_string()).collect();
    gc_remote_releases(&local_executor(), root.to_str().unwrap(), &pinned)
        .await
        .unwrap();

    for name in awkward {
        assert!(releases.join(name).exists(), "expected to survive: {name}");
    }
}

/// A pinned name matches a whole directory name, never a prefix of one.
/// `case`-with-globbing or a `grep -F` without `-x` would keep `v0` and
/// `v01` both because one contains the other, quietly widening the pinned
/// set past what the database said.
#[tokio::test]
async fn gc_remote_releases_matches_whole_names_not_prefixes() {
    let tmp = tempfile::tempdir().unwrap();
    let root = tmp.path();
    let names = releases_by_age(root, RELEASES_TO_KEEP + 3).await;

    // Pin the oldest by an exact name; its neighbours share the prefix.
    let pinned: PinnedReleases = [names[0].clone()].into_iter().collect();
    gc_remote_releases(&local_executor(), root.to_str().unwrap(), &pinned)
        .await
        .unwrap();

    let releases = root.join("releases");
    assert!(
        releases.join(&names[0]).exists(),
        "the pinned dir was evicted"
    );
    assert!(
        !releases.join(&names[1]).exists(),
        "a dir sharing the pinned name's prefix was treated as pinned"
    );
}

#[tokio::test]
async fn gc_local_releases_noop_when_releases_dir_missing() {
    let tmp = tempfile::tempdir().unwrap();
    gc_local_releases(tmp.path(), &no_pins()).await.unwrap();
}

#[tokio::test]
async fn deploy_remote_fails_cleanly_when_host_unreachable() {
    // 192.0.2.0/24 is reserved for documentation and routes nowhere.
    // ConnectTimeout=10 limits the test wallclock to ~10s worst case.
    let tmp = tempfile::tempdir().unwrap();
    let staged = tmp.path().join("releases").join("0.0.1");
    tokio::fs::create_dir_all(&staged).await.unwrap();
    tokio::fs::write(staged.join("server"), b"x").await.unwrap();

    let node = crate::topology::Node {
        platform: None,
        base_image: None,
        libc: None,
        name: "unreachable".into(),
        ssh_target: "deploy@192.0.2.1".into(),
        release_root: "/opt/never".into(),
        service_name: "makenotwork.service".into(),
        health_url: None,
        config_check_env_file: None,
        actuate: crate::topology::default_actuate(),
        observe: crate::topology::default_observe(),
        companions: Vec::new(),
    };
    let executor = SshExec::new(
        node.ssh_target.clone(),
        CapabilitySet::from_tokens(["deploy", "restart"], ["health"]),
    );

    let placement = Placement::check(&node, &staged, None).expect("both sides silent");
    let result = deploy_node(&executor, placement, "0.0.1", "server", Some(&no_pins())).await;
    let err = result.expect_err("deploy to unreachable host should fail");
    let msg = format!("{err:#}");
    // Don't pin exact wording, just that the failure is attributed (ssh /
    // rsync / connection) and that no panic / hang happened.
    assert!(
        msg.contains("ssh")
            || msg.contains("rsync")
            || msg.contains("connection")
            || msg.contains("Connection"),
        "unexpected error: {msg}"
    );
}

#[tokio::test]
async fn deploy_node_with_local_ssh_target_swaps_symlink() {
    // ssh_target="local" routes to the local fast-path: just a symlink
    // swap, no remote calls.
    let tmp = tempfile::tempdir().unwrap();
    let release_root = tmp.path().to_path_buf();
    let staged = release_root.join("releases").join("0.0.1");
    tokio::fs::create_dir_all(&staged).await.unwrap();
    tokio::fs::write(staged.join("server"), b"x").await.unwrap();

    let node = crate::topology::Node {
        platform: None,
        base_image: None,
        libc: None,
        name: "local-dev".into(),
        ssh_target: "local".into(),
        release_root: release_root.to_string_lossy().into_owned(),
        service_name: "makenotwork.service".into(),
        health_url: None,
        config_check_env_file: None,
        actuate: crate::topology::default_actuate(),
        observe: crate::topology::default_observe(),
        companions: Vec::new(),
    };
    let executor = local_executor();

    let out = deploy_node(
        &executor,
        Placement::check(&node, &staged, None).unwrap(),
        "0.0.1",
        "server",
        Some(&no_pins()),
    )
    .await
    .unwrap();
    assert_eq!(out, staged);
    let target = tokio::fs::read_link(release_root.join("current"))
        .await
        .unwrap();
    assert_eq!(target.to_string_lossy(), "releases/0.0.1");
}

// ---- swap_and_restart_script: symlink/restart consistency ----

async fn run_script(script: &str) -> std::process::Output {
    Command::new("sh")
        .arg("-c")
        .arg(script)
        .output()
        .await
        .unwrap()
}

async fn setup_release_root(with_current: bool) -> tempfile::TempDir {
    let tmp = tempfile::tempdir().unwrap();
    let root = tmp.path();
    tokio::fs::create_dir_all(root.join("releases/old"))
        .await
        .unwrap();
    tokio::fs::create_dir_all(root.join("releases/new"))
        .await
        .unwrap();
    if with_current {
        std::os::unix::fs::symlink("releases/old", root.join("current")).unwrap();
    }
    tmp
}

#[tokio::test]
async fn swap_and_restart_keeps_new_symlink_when_restart_succeeds() {
    let tmp = setup_release_root(true).await;
    let root = tmp.path().to_string_lossy().into_owned();
    let out = run_script(&swap_and_restart_script(&root, "new", "true")).await;
    assert!(
        out.status.success(),
        "script should succeed when restart succeeds"
    );
    let target = tokio::fs::read_link(tmp.path().join("current"))
        .await
        .unwrap();
    assert_eq!(
        target.to_string_lossy(),
        "releases/new",
        "symlink advanced to new"
    );
}

#[tokio::test]
async fn swap_and_restart_rolls_symlink_back_when_restart_fails() {
    // The bug: a restart failure after the flip must NOT leave `current`
    // pointing at the new (un-activated) release.
    let tmp = setup_release_root(true).await;
    let root = tmp.path().to_string_lossy().into_owned();
    let out = run_script(&swap_and_restart_script(&root, "new", "false")).await;
    assert!(!out.status.success(), "script must fail when restart fails");
    let target = tokio::fs::read_link(tmp.path().join("current"))
        .await
        .unwrap();
    assert_eq!(
        target.to_string_lossy(),
        "releases/old",
        "symlink rolled back to prev so a later restart can't silently activate new",
    );
}

// ---- arch_guard_script: wrong-arch artifacts fail closed ----

/// A 20-byte stub whose ELF e_machine field (offset 18, 2 bytes LE) is set.
fn elf_stub_with_machine(b18: u8, b19: u8) -> tempfile::NamedTempFile {
    let mut data = vec![0u8; 20];
    data[18] = b18;
    data[19] = b19;
    let f = tempfile::NamedTempFile::new().unwrap();
    std::fs::write(f.path(), &data).unwrap();
    f
}

/// e_machine low byte for the host running the test, if mapped.
fn host_machine_lo() -> Option<u8> {
    match std::env::consts::ARCH {
        "x86_64" => Some(0x3e),
        "aarch64" => Some(0xb7),
        _ => None,
    }
}

#[tokio::test]
async fn arch_guard_passes_for_matching_binary() {
    let Some(lo) = host_machine_lo() else { return };
    let f = elf_stub_with_machine(lo, 0x00);
    let out = run_script(&arch_guard_script(&f.path().to_string_lossy())).await;
    assert!(
        out.status.success(),
        "matching arch must pass: {}",
        String::from_utf8_lossy(&out.stderr),
    );
}

#[tokio::test]
async fn arch_guard_fails_closed_for_wrong_binary() {
    // Use the other arch's e_machine so it can't match the host.
    let wrong = match std::env::consts::ARCH {
        "x86_64" => 0xb7,  // aarch64 binary on an x86_64 node
        "aarch64" => 0x3e, // x86_64 binary on an aarch64 node
        _ => return,
    };
    let f = elf_stub_with_machine(wrong, 0x00);
    let out = run_script(&arch_guard_script(&f.path().to_string_lossy())).await;
    assert!(
        !out.status.success(),
        "wrong-arch binary must fail closed before the symlink swap"
    );
}

// ---- ldd_guard_script: a binary this node cannot resolve fails closed ----

/// A fake `ldd` on PATH that prints `body` and exits `code`, so the guard's
/// three outcomes can be exercised without a binary that genuinely fails to
/// link. The real `ldd` cannot be made to produce a `not found` on demand.
async fn run_ldd_guard_with_fake(body: &str, code: i32) -> std::process::Output {
    let dir = tempfile::tempdir().unwrap();
    let fake = dir.path().join("ldd");
    std::fs::write(
        &fake,
        format!("#!/bin/sh\ncat <<'EOF'\n{body}\nEOF\nexit {code}\n"),
    )
    .unwrap();
    let mut perms = std::fs::metadata(&fake).unwrap().permissions();
    std::os::unix::fs::PermissionsExt::set_mode(&mut perms, 0o755);
    std::fs::set_permissions(&fake, perms).unwrap();
    let bin = dir.path().join("subject");
    std::fs::write(&bin, b"x").unwrap();
    Command::new("sh")
        .arg("-c")
        .arg(ldd_guard_script(&bin.to_string_lossy()))
        .env("PATH", format!("{}:/usr/bin:/bin", dir.path().display()))
        .output()
        .await
        .unwrap()
}

#[tokio::test]
async fn ldd_guard_fails_closed_on_an_unsatisfiable_symbol_version() {
    // The exact failure Bento's glibc_check used to catch at build time, and
    // the reason this guard exists: right arch, resolves every library, and
    // still cannot exec because the node's glibc is older than the build
    // host's.
    let out = run_ldd_guard_with_fake(
        "\tlinux-vdso.so.1 (0x00007fff)\n\
         \t/lib/x86_64-linux-gnu/libc.so.6: version `GLIBC_2.40' not found (required by ./pom)\n\
         \tlibc.so.6 => /lib/x86_64-linux-gnu/libc.so.6 (0x00007f00)",
        0,
    )
    .await;
    assert!(
        !out.status.success(),
        "an unsatisfiable symbol version must fail before the symlink swap"
    );
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(
        stderr.contains("GLIBC_2.40"),
        "the offending line must reach the operator, not just a verdict: {stderr}"
    );
}

#[tokio::test]
async fn ldd_guard_fails_closed_on_a_missing_library() {
    let out = run_ldd_guard_with_fake("\tlibfoo.so.1 => not found", 0).await;
    assert!(!out.status.success(), "a missing library must fail closed");
}

#[tokio::test]
async fn ldd_guard_passes_a_resolvable_binary() {
    let out = run_ldd_guard_with_fake(
        "\tlibc.so.6 => /lib/x86_64-linux-gnu/libc.so.6 (0x00007f00)",
        0,
    )
    .await;
    assert!(
        out.status.success(),
        "a fully resolved binary must pass: {}",
        String::from_utf8_lossy(&out.stderr),
    );
}

#[tokio::test]
async fn ldd_guard_passes_a_static_binary() {
    // ldd exits non-zero for these. Nothing to resolve is not a failure.
    let out = run_ldd_guard_with_fake("\tnot a dynamic executable", 1).await;
    assert!(
        out.status.success(),
        "a static binary has no dependencies to satisfy: {}",
        String::from_utf8_lossy(&out.stderr),
    );
}

#[tokio::test]
async fn ldd_guard_fails_when_ldd_errors_for_another_reason() {
    // Not the static case: ldd said something else and exited non-zero. We
    // do not know the binary is fine, so we do not say it is.
    let out = run_ldd_guard_with_fake("ldd: cannot read file", 1).await;
    assert!(
        !out.status.success(),
        "an unexplained ldd failure must not read as a pass"
    );
}

#[tokio::test]
async fn ldd_guard_skips_when_the_node_has_no_ldd() {
    // Cannot verify is not known bad, matching arch_guard's unmapped-arch
    // call. PATH holds nothing, so `command -v ldd` finds none.
    let dir = tempfile::tempdir().unwrap();
    let bin = dir.path().join("subject");
    std::fs::write(&bin, b"x").unwrap();
    // Absolute path to the shell: PATH is what this test empties, so
    // resolving `sh` through it would fail before the script ever ran.
    let out = Command::new("/bin/sh")
        .arg("-c")
        .arg(ldd_guard_script(&bin.to_string_lossy()))
        .env("PATH", dir.path().display().to_string())
        .output()
        .await
        .unwrap();
    assert!(
        out.status.success(),
        "a node with no ldd must not fail the deploy: {}",
        String::from_utf8_lossy(&out.stderr),
    );
}

#[tokio::test]
async fn swap_and_restart_first_deploy_failure_has_no_prev_to_restore() {
    // No prior `current`. A restart failure leaves `current` at new (the only
    // version) and still reports failure — documented degenerate case.
    let tmp = setup_release_root(false).await;
    let root = tmp.path().to_string_lossy().into_owned();
    let out = run_script(&swap_and_restart_script(&root, "new", "false")).await;
    assert!(!out.status.success(), "script must fail when restart fails");
    let target = tokio::fs::read_link(tmp.path().join("current"))
        .await
        .unwrap();
    assert_eq!(
        target.to_string_lossy(),
        "releases/new",
        "no prev existed to roll back to"
    );
}

// ---- config_check_script: systemd-faithful env loading ----

#[tokio::test]
async fn config_check_script_loads_values_with_shell_metachars() {
    // The bug: `. env_file` expands/word-splits values, so a URL or a
    // password containing a shell metacharacter is mangled — it dropped
    // DATABASE_URL to empty on a real node, which would fail every deploy.
    // The export-loop must load such a value intact. The "binary" is a
    // checker script (a real path, like a deployed binary) that exits 0 only
    // if the var arrived byte-for-byte — it compares against the expected
    // value read from a file, so nothing re-interprets the metacharacters.
    let tricky = "postgres://u:p$ss;w&rd@h/db `x` $(y)";
    // Plain files in a tempdir: no lingering write fd, so the checker can be
    // exec'd (a NamedTempFile stays open and would ETXTBSY).
    let dir = tempfile::tempdir().unwrap();
    let expected_path = dir.path().join("expected");
    std::fs::write(&expected_path, tricky).unwrap(); // no trailing newline

    let env_path = dir.path().join("node.env");
    std::fs::write(
        &env_path,
        format!(
            "# a comment\n\nDATABASE_URL={tricky}\nOTHER=plain\nEXPECTED_FILE={ef}\n",
            ef = expected_path.display(),
        ),
    )
    .unwrap();

    let checker_path = dir.path().join("checker.sh");
    std::fs::write(
        &checker_path,
        "#!/bin/sh\nwant=$(cat \"$EXPECTED_FILE\")\n\
         [ \"$DATABASE_URL\" = \"$want\" ] || { echo \"DB [$DATABASE_URL] != [$want]\" >&2; exit 1; }\n\
         [ \"$OTHER\" = plain ] || { echo \"OTHER [$OTHER]\" >&2; exit 1; }\n",
    )
    .unwrap();
    std::fs::set_permissions(
        &checker_path,
        std::os::unix::fs::PermissionsExt::from_mode(0o755),
    )
    .unwrap();

    let script = config_check_script(&env_path.to_string_lossy(), &checker_path.to_string_lossy());
    let out = run_script(&script).await;
    assert!(
        out.status.success(),
        "value with shell metachars must load intact; stderr: {}",
        String::from_utf8_lossy(&out.stderr),
    );
}

// ---- install-companion.sh: the node-side guard rails ----

/// Run the shipped installer script with three args; returns its exit code.
/// Exercises the real file rather than a copy of its logic, because the
/// script is the ONLY control on a NOPASSWD sudo grant.
fn run_installer(src: &str, dst: &str, service: &str) -> i32 {
    let script =
        std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("../deploy/install-companion.sh");
    std::process::Command::new("bash")
        .arg(&script)
        .args([src, dst, service])
        .output()
        .expect("running install-companion.sh")
        .status
        .code()
        .expect("script exited via signal")
}

// Guards run before any filesystem write, so these never install anything.
// Exit 3 = refused by a guard; exit 4 = guards passed, src simply absent.
const REFUSED: i32 = 3;
const PASSED_GUARDS: i32 = 4;

#[test]
fn installer_refuses_a_dst_that_escapes_opt_via_dotdot() {
    // `/opt/../etc/...` matches a bare `/opt/*` glob. With the sudoers
    // wildcard that meant `install -m 0755` as root to anywhere, plus a
    // restart of any unit — so the path must be normalised before the test.
    assert_eq!(
        run_installer(
            "/opt/mnw/releases/1.0.0/companions/mnw-cli",
            "/opt/../etc/systemd/system/evil.service",
            "mnw-cli.service",
        ),
        REFUSED,
    );
}

#[test]
fn installer_refuses_a_src_that_escapes_the_bundle_via_dotdot() {
    assert_eq!(
        run_installer(
            "/opt/mnw/releases/1.0.0/companions/../../../../../etc/shadow",
            "/opt/mnw-cli/mnw-cli",
            "mnw-cli.service",
        ),
        REFUSED,
    );
}

#[test]
fn installer_accepts_the_real_companion_paths() {
    // The guards must not have been tightened into uselessness: the shape
    // Sando actually sends has to get past them. It stops at the missing
    // src (exit 4), which is proof the guards accepted it.
    assert_eq!(
        run_installer(
            "/opt/mnw/releases/1.0.0/companions/mnw-cli",
            "/opt/mnw-cli/mnw-cli",
            "mnw-cli.service",
        ),
        PASSED_GUARDS,
    );
}

#[test]
fn installer_refuses_a_service_name_with_a_path_separator() {
    assert_eq!(
        run_installer(
            "/opt/mnw/releases/1.0.0/companions/mnw-cli",
            "/opt/mnw-cli/mnw-cli",
            "../../etc/evil.service",
        ),
        REFUSED,
    );
}

// ---- install_companion_cmd: shape + quoting ----

#[test]
fn install_companion_cmd_shape_and_quoting() {
    let cmd = install_companion_cmd(
        "/opt/mnw/releases/0.10.14/companions/mnw-cli",
        "/opt/mnw-cli/mnw-cli",
        "mnw-cli.service",
    );
    // Routes through the wrapper (single sudoers grant), sudo-invoked, with
    // src, dst, service in that order.
    assert!(cmd.starts_with("sudo "), "must be sudo-invoked: {cmd}");
    assert!(
        cmd.contains("/usr/local/lib/mnw/install-companion.sh"),
        "{cmd}"
    );
    let installer_pos = cmd.find("install-companion.sh").unwrap();
    let src_pos = cmd.find("companions/mnw-cli").unwrap();
    let dst_pos = cmd.find("/opt/mnw-cli/mnw-cli").unwrap();
    let svc_pos = cmd.find("mnw-cli.service").unwrap();
    assert!(
        installer_pos < src_pos && src_pos < dst_pos && dst_pos < svc_pos,
        "arg order: {cmd}"
    );
}

#[test]
fn install_companion_cmd_quotes_metachars() {
    // A path with a space/quote must be shell-safe (defense in depth even
    // though these come from operator config).
    let cmd = install_companion_cmd("/a b/src", "/dst'x", "u.service");
    let out = std::process::Command::new("sh")
        .arg("-c")
        .arg(format!(
            "set -- {}; echo \"$#\"",
            cmd.strip_prefix("sudo ").unwrap()
        ))
        .output()
        .unwrap();
    // installer + 3 args = 4 positional words after quoting.
    assert_eq!(
        String::from_utf8_lossy(&out.stdout).trim(),
        "4",
        "quoting split wrong: {cmd}"
    );
}

#[tokio::test]
async fn config_check_script_propagates_binary_failure() {
    // A required var missing (the binary exits non-zero) must fail the check.
    let env = tempfile::NamedTempFile::new().unwrap();
    std::fs::write(env.path(), "FOO=bar\n").unwrap();
    let script = config_check_script(&env.path().to_string_lossy(), "false");
    let out = run_script(&script).await;
    assert!(
        !out.status.success(),
        "a non-zero MNW_CHECK_CONFIG exit must fail the check"
    );
}

#[tokio::test]
async fn deploy_node_denied_when_executor_lacks_deploy_grant() {
    // Defense in depth: an executor without the deploy grant refuses the
    // step before any filesystem / ssh action.
    let tmp = tempfile::tempdir().unwrap();
    let release_root = tmp.path().to_path_buf();
    let staged = release_root.join("releases").join("0.0.1");
    tokio::fs::create_dir_all(&staged).await.unwrap();

    let node = crate::topology::Node {
        platform: None,
        base_image: None,
        libc: None,
        name: "local-dev".into(),
        ssh_target: "local".into(),
        release_root: release_root.to_string_lossy().into_owned(),
        service_name: "makenotwork.service".into(),
        health_url: None,
        config_check_env_file: None,
        actuate: vec!["restart".into()], // no deploy
        observe: vec![],
        companions: Vec::new(),
    };
    let executor = LocalExec::new(CapabilitySet::from_tokens(["restart"], Vec::<&str>::new()));
    let err = deploy_node(
        &executor,
        Placement::check(&node, &staged, None).unwrap(),
        "0.0.1",
        "server",
        Some(&no_pins()),
    )
    .await
    .unwrap_err();
    assert!(
        format!("{err:#}").contains("capability denied"),
        "expected capability denial"
    );
}

// ---- FakeExec: the deploy_remote choreography without a real host ----
//
// deploy_node's local fast-path is covered above with a real LocalExec, but
// the remote path (rsync + arch guard + config-drift + swap + companions +
// gc) short-circuits on `ssh_target != "local"` and so never ran under test
// without a reachable node. FakeExec records every executor call in order
// and can be told to fail one shell step (matched by substring) or the rsync
// push, so the ordering and the fail-closed-before-swap contract are
// assertable in-process.

struct FakeExec {
    caps: CapabilitySet,
    calls: Arc<StdMutex<Vec<String>>>,
    /// The first `run_streaming` whose script contains this substring exits
    /// non-zero (a failed shell step), e.g. the arch guard.
    fail_run_matching: Option<String>,
    /// `push_dir` (the rsync) returns an error.
    fail_push_dir: bool,
}

impl FakeExec {
    fn new() -> Self {
        Self {
            caps: CapabilitySet::from_tokens(["deploy", "restart"], ["health"]),
            calls: Arc::new(StdMutex::new(Vec::new())),
            fail_run_matching: None,
            fail_push_dir: false,
        }
    }
    fn log(&self) -> Vec<String> {
        self.calls.lock().unwrap().clone()
    }
}

#[async_trait]
impl Executor for FakeExec {
    async fn run_streaming(&self, step: &Step, _sink: &mut dyn LogSink) -> Result<RunOutput> {
        // Every deploy step is a `Step::shell`, so the script is argv's tail.
        let script = step.argv.last().cloned().unwrap_or_default();
        self.calls.lock().unwrap().push(format!("run:{script}"));
        let fail = self
            .fail_run_matching
            .as_deref()
            .is_some_and(|m| script.contains(m));
        Ok(RunOutput {
            status: std::process::ExitStatus::from_raw(if fail { 1 << 8 } else { 0 }),
            stdout: Vec::new(),
            stderr: if fail {
                b"fake step failure".to_vec()
            } else {
                Vec::new()
            },
        })
    }
    async fn pull_file(&self, _remote: &Path, _local: &Path, _opts: &SyncOpts) -> Result<()> {
        self.calls.lock().unwrap().push("pull_file".into());
        Ok(())
    }
    async fn pull_dir(&self, _remote: &Path, _local: &Path, _opts: &SyncOpts) -> Result<()> {
        self.calls.lock().unwrap().push("pull_dir".into());
        Ok(())
    }
    async fn pull_glob(&self, _glob: &str, _local: &Path, _opts: &SyncOpts) -> Result<()> {
        self.calls.lock().unwrap().push("pull_glob".into());
        Ok(())
    }
    async fn push_dir(&self, _local: &Path, remote: &Path, _opts: &SyncOpts) -> Result<()> {
        self.calls
            .lock()
            .unwrap()
            .push(format!("push_dir:{}", remote.display()));
        if self.fail_push_dir {
            anyhow::bail!("fake rsync failure");
        }
        Ok(())
    }
    fn capabilities(&self) -> &CapabilitySet {
        &self.caps
    }
}

fn remote_node(config_check: bool, companions: Vec<NodeCompanion>) -> Node {
    Node {
        platform: None,
        base_image: None,
        libc: None,
        name: "web-a".into(),
        ssh_target: "deploy@web-a".into(),
        release_root: "/opt/mnw".into(),
        service_name: "makenotwork.service".into(),
        health_url: None,
        config_check_env_file: config_check.then(|| "/etc/mnw/node.env".to_string()),
        actuate: crate::topology::default_actuate(),
        observe: crate::topology::default_observe(),
        companions,
    }
}

fn companion() -> NodeCompanion {
    NodeCompanion {
        name: "mnw-cli".into(),
        install_path: "/opt/mnw-cli/mnw-cli".into(),
        service_name: "mnw-cli.service".into(),
    }
}

/// Index of the first recorded call whose text contains `needle` (panics if
/// absent — the assertion message names what was missing).
fn pos(log: &[String], needle: &str) -> usize {
    log.iter()
        .position(|c| c.contains(needle))
        .unwrap_or_else(|| panic!("no call matched {needle:?} in {log:#?}"))
}

#[tokio::test]
async fn deploy_remote_runs_the_full_choreography_in_order() {
    // A node opted into the config-drift check and carrying one companion:
    // mkdir -> rsync -> arch guard -> config check -> swap+restart ->
    // companion install -> gc, in that order.
    let tmp = tempfile::tempdir().unwrap();
    let staged = tmp.path().join("releases").join("0.9.0");
    tokio::fs::create_dir_all(&staged).await.unwrap();

    let node = remote_node(true, vec![companion()]);
    let exec = FakeExec::new();
    let out = deploy_node(
        &exec,
        Placement::check(&node, &staged, None).unwrap(),
        "0.9.0",
        "makenotwork",
        Some(&no_pins()),
    )
    .await
    .expect("deploy_remote should succeed against the fake");
    assert_eq!(out, PathBuf::from("/opt/mnw/releases/0.9.0"));

    let log = exec.log();
    let mkdir = pos(&log, "mkdir -p");
    let rsync = pos(&log, "push_dir:/opt/mnw/releases/0.9.0");
    let arch = pos(&log, "e_machine");
    let cfg = pos(&log, "MNW_CHECK_CONFIG=1");
    let swap = pos(&log, "reload-or-restart");
    let comp = pos(&log, "install-companion.sh");
    let gc = pos(&log, "ls -1t");
    assert!(
        mkdir < rsync && rsync < arch && arch < cfg && cfg < swap && swap < comp && comp < gc,
        "deploy steps out of order: {log:#?}"
    );
}

#[tokio::test]
async fn deploy_remote_aborts_before_swap_when_rsync_fails() {
    // The rsync failing must fail the deploy BEFORE the symlink swap — the
    // "current symlink left intact" contract. Assert the swap never ran.
    let tmp = tempfile::tempdir().unwrap();
    let staged = tmp.path().join("releases").join("0.9.0");
    tokio::fs::create_dir_all(&staged).await.unwrap();

    let node = remote_node(false, Vec::new());
    let mut exec = FakeExec::new();
    exec.fail_push_dir = true;
    let err = deploy_node(
        &exec,
        Placement::check(&node, &staged, None).unwrap(),
        "0.9.0",
        "makenotwork",
        Some(&no_pins()),
    )
    .await
    .expect_err("rsync failure must fail the deploy");
    assert!(
        format!("{err:#}").contains("rsync"),
        "error should attribute the rsync: {err:#}"
    );
    let log = exec.log();
    assert!(
        !log.iter().any(|c| c.contains("reload-or-restart")),
        "swap must not run after a failed rsync: {log:#?}"
    );
}

#[tokio::test]
async fn deploy_remote_aborts_before_swap_when_arch_guard_fails() {
    // A wrong-arch binary must fail closed before the swap. The fake fails
    // the arch-guard shell step; the swap must not follow.
    let tmp = tempfile::tempdir().unwrap();
    let staged = tmp.path().join("releases").join("0.9.0");
    tokio::fs::create_dir_all(&staged).await.unwrap();

    let node = remote_node(false, Vec::new());
    let mut exec = FakeExec::new();
    exec.fail_run_matching = Some("e_machine".into());
    let err = deploy_node(
        &exec,
        Placement::check(&node, &staged, None).unwrap(),
        "0.9.0",
        "makenotwork",
        Some(&no_pins()),
    )
    .await
    .expect_err("arch mismatch must fail the deploy");
    assert!(
        format!("{err:#}").contains("architecture"),
        "error should mention the arch check: {err:#}"
    );
    let log = exec.log();
    assert!(
        !log.iter().any(|c| c.contains("reload-or-restart")),
        "swap must not run after a failed arch guard: {log:#?}"
    );
}

#[tokio::test]
async fn deploy_remote_skips_config_check_when_node_opts_out() {
    // No config_check_env_file => the pre-swap config check is skipped, but
    // the rest of the choreography (including the swap) still runs.
    let tmp = tempfile::tempdir().unwrap();
    let staged = tmp.path().join("releases").join("0.9.0");
    tokio::fs::create_dir_all(&staged).await.unwrap();

    let node = remote_node(false, Vec::new());
    let exec = FakeExec::new();
    deploy_node(
        &exec,
        Placement::check(&node, &staged, None).unwrap(),
        "0.9.0",
        "makenotwork",
        Some(&no_pins()),
    )
    .await
    .unwrap();
    let log = exec.log();
    assert!(
        !log.iter().any(|c| c.contains("MNW_CHECK_CONFIG=1")),
        "config check must be skipped when the node opts out: {log:#?}"
    );
    assert!(
        log.iter().any(|c| c.contains("reload-or-restart")),
        "the swap must still run: {log:#?}"
    );
}

/// A companion is guarded on the same terms as the primary, and BEFORE the
/// swap. Unguarded, the first thing to notice a bad companion is its unit
/// failing to start during the install loop, which runs after the server has
/// already been restarted.
#[tokio::test]
async fn companions_are_guarded_before_the_swap() {
    let tmp = tempfile::tempdir().unwrap();
    let staged = tmp.path().join("releases").join("0.9.0");
    tokio::fs::create_dir_all(&staged).await.unwrap();

    let node = remote_node(false, vec![companion()]);
    let exec = FakeExec::new();
    deploy_node(
        &exec,
        Placement::check(&node, &staged, None).unwrap(),
        "0.9.0",
        "makenotwork",
        Some(&no_pins()),
    )
    .await
    .unwrap();

    let log = exec.log();
    // The companion's own arch and loader checks, named by its path so they
    // cannot be confused with the primary's.
    let guard = pos(&log, "companions/mnw-cli");
    let swap = pos(&log, "reload-or-restart");
    let install = pos(&log, "install-companion.sh");
    assert!(
        guard < swap && swap < install,
        "a companion must be guarded before the swap and installed after it: {log:#?}"
    );
    let companion_guards = log
        .iter()
        .filter(|c| c.contains("companions/mnw-cli") && !c.contains("install-companion.sh"))
        .count();
    assert_eq!(
        companion_guards, 2,
        "both guards must run against the companion, not just one: {log:#?}"
    );
}

/// And failing one of them fails the promote with the service intact, which
/// is the whole point of moving the check ahead of the swap.
#[tokio::test]
async fn a_companion_failing_its_guard_aborts_before_the_swap() {
    let tmp = tempfile::tempdir().unwrap();
    let staged = tmp.path().join("releases").join("0.9.0");
    tokio::fs::create_dir_all(&staged).await.unwrap();

    let node = remote_node(false, vec![companion()]);
    let mut exec = FakeExec::new();
    // Fails the first script naming the companion, which is its arch guard.
    // The primary's guards name the primary and are unaffected.
    exec.fail_run_matching = Some("companions/mnw-cli".into());
    let err = deploy_node(
        &exec,
        Placement::check(&node, &staged, None).unwrap(),
        "0.9.0",
        "makenotwork",
        Some(&no_pins()),
    )
    .await
    .expect_err("a bad companion must fail the deploy");

    let msg = format!("{err:#}");
    assert!(
        msg.contains("mnw-cli"),
        "the refusal must name which companion: {msg}"
    );
    assert_eq!(
        stage_of(&err),
        Some(FailureStage::BeforeSwap),
        "a companion guard failing must leave the service intact: {msg}"
    );
    let log = exec.log();
    assert!(
        !log.iter().any(|c| c.contains("reload-or-restart")),
        "swap must not run after a failed companion guard: {log:#?}"
    );
    assert!(
        !log.iter().any(|c| c.contains("install-companion.sh")),
        "nothing should be installed after a failed companion guard: {log:#?}"
    );
}

/// The guards and the installer must read the same path. A guard checking a
/// path the installer does not use is a check of nothing, and passes.
#[test]
fn the_guarded_companion_path_is_the_one_installed() {
    let release_dir = "/opt/mnw/releases/0.9.0";
    let src = companion_src(release_dir, "mnw-cli");
    assert_eq!(src, "/opt/mnw/releases/0.9.0/companions/mnw-cli");
    let cmd = install_companion_cmd(&src, "/opt/mnw-cli/mnw-cli", "mnw-cli.service");
    assert!(
        cmd.contains(&src),
        "the installer must read the path the guards checked: {cmd}"
    );
}

#[tokio::test]
async fn deploy_remote_installs_companion_after_the_swap() {
    // Companions are After= the server: their install must land after the
    // symlink swap + service restart, never before.
    let tmp = tempfile::tempdir().unwrap();
    let staged = tmp.path().join("releases").join("0.9.0");
    tokio::fs::create_dir_all(&staged).await.unwrap();

    let node = remote_node(false, vec![companion()]);
    let exec = FakeExec::new();
    deploy_node(
        &exec,
        Placement::check(&node, &staged, None).unwrap(),
        "0.9.0",
        "makenotwork",
        Some(&no_pins()),
    )
    .await
    .unwrap();
    let log = exec.log();
    assert!(
        pos(&log, "reload-or-restart") < pos(&log, "install-companion.sh"),
        "companion install must follow the swap: {log:#?}"
    );
}