draupnir 0.1.9

//! **Test-matrix verdict rows for draupnir's NoCloud seed builder** — draupnir's
//! entry into nornir's constellation-wide introspection-coverage gate. Every check
//! asserts on the *actual* return value AND records the verdict as one
//! functional-status row.
//!
//! Follows the skidbladnir / ordning-core reference pattern. With
//! `--features testmatrix` each emit becomes a real
//! `nornir_testmatrix::functional_status` row (nornir's matrix reads them back);
//! without it `draupnir::functional_status` is a compiled-out `#[inline]` no-op
//! (no nornir dep), so this file is a plain assert-only test in the default build.
//! The `assert!` is the gate; the emit is the observation (nornir's `assert_emit!`
//! doctrine).

use draupnir::seed;
use draupnir::CloudInit;
use draupnir::{
    await_power_state, boot_and_await, boot_fleet_and_await, boot_fleet_and_await_parallel,
    BmcEndpoint, Boot, BootSpec, Error, Lifecycle, Machine, MemberOutcome, NetMode, PowerState,
    Result, WaitOptions,
};
use std::cell::{Cell, RefCell};
use std::collections::{HashSet, VecDeque};
use std::time::Duration;

/// A backend that is both [`Boot`] and [`Lifecycle`], reporting a scripted sequence
/// of power states — lets the cross-backend power readback seam be asserted with no
/// live instance (mirrors the lib unit-test mock).
struct ScriptedNode {
    states: RefCell<VecDeque<PowerState>>,
    fallback: PowerState,
    status_calls: Cell<usize>,
}

impl ScriptedNode {
    fn new(seq: impl IntoIterator<Item = PowerState>, fallback: PowerState) -> Self {
        Self {
            states: RefCell::new(seq.into_iter().collect()),
            fallback,
            status_calls: Cell::new(0),
        }
    }
}

impl Boot for ScriptedNode {
    fn boot(&self, spec: &BootSpec) -> Result<Machine> {
        Ok(Machine::started(format!("id-{}", spec.name), spec))
    }
}

impl Lifecycle for ScriptedNode {
    fn power_on(&self, _m: &Machine) -> Result<()> {
        Ok(())
    }
    fn power_off(&self, _m: &Machine) -> Result<()> {
        Ok(())
    }
    fn status(&self, _m: &Machine) -> Result<PowerState> {
        self.status_calls.set(self.status_calls.get() + 1);
        Ok(self
            .states
            .borrow_mut()
            .pop_front()
            .unwrap_or(self.fallback))
    }
}

/// A fleet backend where members named in `dead` never power on (their `status`
/// stays Off) while the rest boot and report On — lets the fleet-level partial
/// readback rollup be asserted with no live instances. Keys on `Machine::spec_name`.
struct FleetNode {
    dead: HashSet<String>,
}

impl Boot for FleetNode {
    fn boot(&self, spec: &BootSpec) -> Result<Machine> {
        Ok(Machine::started(format!("id-{}", spec.name), spec))
    }
}

impl Lifecycle for FleetNode {
    fn power_on(&self, _m: &Machine) -> Result<()> {
        Ok(())
    }
    fn power_off(&self, _m: &Machine) -> Result<()> {
        Ok(())
    }
    fn status(&self, m: &Machine) -> Result<PowerState> {
        if self.dead.contains(&m.spec_name) {
            Ok(PowerState::Off)
        } else {
            Ok(PowerState::On)
        }
    }
}

/// A `Sync` fleet backend that yields a *mixed* rollup with no live instances:
/// `unbootable` members fail at `boot()` (`Error`), `dead` members boot but never
/// power on (`Timeout`), the rest come `Up`. `Sync`, so it drives both the serial and
/// the parallel fleet-boot path — letting the two be asserted equal.
struct MixedNode {
    dead: HashSet<String>,
    unbootable: HashSet<String>,
}

impl Boot for MixedNode {
    fn boot(&self, spec: &BootSpec) -> Result<Machine> {
        if self.unbootable.contains(&spec.name) {
            return Err(Error::Unsupported(format!("no slot for {}", spec.name)));
        }
        Ok(Machine::started(format!("id-{}", spec.name), spec))
    }
}

impl Lifecycle for MixedNode {
    fn power_on(&self, _m: &Machine) -> Result<()> {
        Ok(())
    }
    fn power_off(&self, _m: &Machine) -> Result<()> {
        Ok(())
    }
    fn status(&self, m: &Machine) -> Result<PowerState> {
        if self.dead.contains(&m.spec_name) {
            Ok(PowerState::Off)
        } else {
            Ok(PowerState::On)
        }
    }
}

/// Assert on the real value AND record a GREEN matrix row under `draupnir/<surface>`.
macro_rules! assert_emit {
    ($surface:expr, $check:expr, $ok:expr, $($detail:tt)+) => {{
        let ok: bool = $ok;
        let detail = format!($($detail)+);
        draupnir::functional_status(concat!("draupnir/", $surface), $check, ok, &detail);
        assert!(ok, "{}::{} — {}", $surface, $check, detail);
    }};
}

#[test]
fn validate_rejects_empty_required_payload_paths() {
    // The pre-flight gate rejects an empty required path/ref on every image source
    // (parity with the Disk kernel/disk checks) so a mistyped spec fails at
    // validate(), not deep in a backend call. Records a `draupnir/spec` matrix row.
    let bmc = BmcEndpoint {
        host: "https://bmc-42.dc.example".into(),
        username: "admin".into(),
        system_id: "System.Embedded.1".into(),
    };
    let empty_kernel = matches!(
        BootSpec::kvm_kernel_rootfs("kr", "", "/rootfs.cpio.gz").validate(),
        Err(Error::Spec(_))
    );
    let empty_oci = matches!(
        BootSpec::container("cache", "").validate(),
        Err(Error::Spec(_))
    );
    let empty_iso = matches!(
        BootSpec::redfish_iso("node", "  ", bmc).validate(),
        Err(Error::Spec(_))
    );
    // A well-formed spec still validates (no regression).
    let good_ok = BootSpec::container("cache", "redis:7").validate().is_ok();
    assert_emit!(
        "spec",
        "validate_rejects_empty_required_payload_paths",
        empty_kernel && empty_oci && empty_iso && good_ok,
        "empty kernel/OCI-ref/ISO rejected (kernel:{empty_kernel} oci:{empty_oci} iso:{empty_iso}), good spec ok:{good_ok}"
    );
}

#[test]
fn validate_rejects_zero_or_duplicate_container_ports() {
    // Published container ports are exposed + host-bound verbatim, so a `0` port
    // (never a real published port) or a doubled port (a self-colliding host
    // binding) is rejected at validate(), not deep in the container backend.
    // Records a `draupnir/spec` matrix row.
    let zero_port = matches!(
        BootSpec::container("cache", "redis:7")
            .with_port(0)
            .validate(),
        Err(Error::Spec(_))
    );
    let dup_port = matches!(
        BootSpec::container("cache", "redis:7")
            .with_port(8080)
            .with_port(8080)
            .validate(),
        Err(Error::Spec(_))
    );
    // A distinct, non-zero port set still validates (no regression).
    let good_ok = BootSpec::container("cache", "redis:7")
        .with_port(8080)
        .with_port(8443)
        .validate()
        .is_ok();
    assert_emit!(
        "spec",
        "validate_rejects_zero_or_duplicate_container_ports",
        zero_port && dup_port && good_ok,
        "0 port rejected:{zero_port}, duplicate port rejected:{dup_port}, distinct port set ok:{good_ok}"
    );
}

#[test]
fn validate_rejects_container_only_cmd_or_ports_on_other_backends() {
    // `cmd` and `ports` are container-only; KVM/Redfish silently ignore them, so
    // setting either there is a misconfiguration that would vanish without a trace.
    // validate() rejects it (parity with a bmc on a non-Redfish backend). Records a
    // `draupnir/spec` matrix row.
    let bmc = BmcEndpoint {
        host: "https://bmc-42.dc.example".into(),
        username: "admin".into(),
        system_id: "System.Embedded.1".into(),
    };
    let kvm_cmd = matches!(
        BootSpec::kvm_kernel_rootfs("appliance", "/bzImage", "/rootfs.cpio.gz")
            .with_cmd(["/bin/init"])
            .validate(),
        Err(Error::Spec(_))
    );
    let kvm_port = matches!(
        BootSpec::kvm_disk("disky", "/bzImage", "/disk.qcow2")
            .with_port(8080)
            .validate(),
        Err(Error::Spec(_))
    );
    let redfish_cmd = matches!(
        BootSpec::redfish_iso("node", "/boot.iso", bmc.clone())
            .with_cmd(["/bin/init"])
            .validate(),
        Err(Error::Spec(_))
    );
    let redfish_port = matches!(
        BootSpec::redfish_iso("node", "/boot.iso", bmc)
            .with_port(443)
            .validate(),
        Err(Error::Spec(_))
    );
    // A container still carries both (no regression).
    let container_ok = BootSpec::container("web", "nginx:latest")
        .with_cmd(["nginx", "-g", "daemon off;"])
        .with_port(8080)
        .validate()
        .is_ok();
    assert_emit!(
        "spec",
        "validate_rejects_container_only_cmd_or_ports_on_other_backends",
        kvm_cmd && kvm_port && redfish_cmd && redfish_port && container_ok,
        "kvm cmd:{kvm_cmd} kvm port:{kvm_port} redfish cmd:{redfish_cmd} redfish port:{redfish_port}, container ok:{container_ok}"
    );
}

#[test]
fn net_mode_renders_oci_value_and_is_container_only() {
    // The container network mode (airgap `--network none`) is expressed on the spec
    // and rendered to its OCI `HostConfig.network_mode` value. `NetMode::None` =>
    // "none" (the airgap wire jera threads through); `Default` => no value (unchanged
    // create body). It is container vocabulary EXCEPT the airgap: the kvm backend
    // honours `NetMode::None` as tunnr's `-nic none`, so validate() accepts that one
    // pair; Host/Bridge on KVM and anything non-default on Redfish stay rejected
    // (parity with cmd/ports). Records a `draupnir/spec` matrix row.

    // Rendering: the OCI value + the cross-repo round-trip (jera's oci_value ->
    // NetMode::from_oci_value -> oci_value) is stable.
    let none_val = NetMode::None.oci_value() == Some("none");
    let default_none = NetMode::Default.oci_value().is_none() && NetMode::Default.is_default();
    let host_val = NetMode::Host.oci_value() == Some("host");
    let roundtrip = NetMode::from_oci_value(Some("none")) == NetMode::None
        && NetMode::from_oci_value(None) == NetMode::Default
        && NetMode::from_oci_value(Some("bogus")) == NetMode::Default;

    // The builder sets it on the spec.
    let set_on_spec = BootSpec::container("job", "busybox:latest")
        .with_net(NetMode::None)
        .net
        == NetMode::None;

    // Container-only save the airgap: Host on KVM and Host on Redfish are rejected;
    // a default net is fine everywhere; a container carrying net=none still validates
    // (no regression); and KVM + net=none — the airgap the backend implements as
    // `-nic none` — validates too.
    let bmc = BmcEndpoint {
        host: "https://bmc-42.dc.example".into(),
        username: "admin".into(),
        system_id: "System.Embedded.1".into(),
    };
    let kvm_net_rejected = matches!(
        BootSpec::kvm_kernel_rootfs("appliance", "/bzImage", "/rootfs.cpio.gz")
            .with_net(NetMode::Host)
            .validate(),
        Err(Error::Spec(_))
    );
    let kvm_airgap_ok = BootSpec::kvm_kernel_rootfs("appliance", "/bzImage", "/rootfs.cpio.gz")
        .with_net(NetMode::None)
        .validate()
        .is_ok();
    let redfish_net_rejected = matches!(
        BootSpec::redfish_iso("node", "/boot.iso", bmc)
            .with_net(NetMode::Host)
            .validate(),
        Err(Error::Spec(_))
    );
    let kvm_default_ok = BootSpec::kvm_kernel_rootfs("appliance", "/bzImage", "/rootfs.cpio.gz")
        .with_net(NetMode::Default)
        .validate()
        .is_ok();
    let container_airgap_ok = BootSpec::container("job", "busybox:latest")
        .with_net(NetMode::None)
        .validate()
        .is_ok();

    assert_emit!(
        "spec",
        "net_mode_renders_oci_value_and_is_container_only",
        none_val
            && default_none
            && host_val
            && roundtrip
            && set_on_spec
            && kvm_net_rejected
            && kvm_airgap_ok
            && redfish_net_rejected
            && kvm_default_ok
            && container_airgap_ok,
        "none:{none_val} default:{default_none} host:{host_val} roundtrip:{roundtrip} set:{set_on_spec} \
         kvm-reject:{kvm_net_rejected} kvm-airgap-ok:{kvm_airgap_ok} redfish-reject:{redfish_net_rejected} \
         kvm-default-ok:{kvm_default_ok} container-airgap-ok:{container_airgap_ok}"
    );
}

#[test]
fn container_exec_builds_argv() {
    // The container control/exec surface Skidbladnir's live start/stop/exec swap
    // needs: a detached container hands back a Machine, and a surviving handle can
    // `exec` INTO it. Asserts (a) the pure `exec_argv` builds the canonical
    // `["exec", id, argv…]` podman-exec command, (b) `ContainerControl::exec`
    // assembles that exact command + returns the engine's ExecOutcome (proven by a
    // recorder mock, no daemon), and (c) the container-only guard rejects a
    // non-container Machine and an empty argv (parity with the net/cmd/ports checks).
    // Records a `draupnir/container` matrix row. The live bollard `/exec` drive is a
    // Loki integration/smoke concern, deferred.
    use draupnir::container::{exec_argv, ContainerControl, ContainerState, ExecOutcome};
    use std::cell::RefCell;

    #[derive(Default)]
    struct ExecRecorder {
        seen: RefCell<Vec<Vec<String>>>,
    }
    impl ContainerControl for ExecRecorder {
        fn container_state(&self, _m: &Machine) -> ContainerState {
            ContainerState::Running
        }
        fn drain_logs(&self, _m: &Machine) -> Vec<String> {
            Vec::new()
        }
        fn stop(&self, _m: &Machine) {}
        fn exec_command(&self, command: &[String]) -> Result<ExecOutcome> {
            self.seen.borrow_mut().push(command.to_vec());
            Ok(ExecOutcome {
                exit_code: Some(0),
                stdout: vec!["PONG".into()],
                stderr: vec![],
            })
        }
    }

    // (a) pure command builder.
    let argv_built = exec_argv("draupnir-cache", &["redis-cli", "ping"])
        == vec!["exec", "draupnir-cache", "redis-cli", "ping"];

    // (b) exec assembles the command + returns the outcome (recorder mock, no daemon).
    let recorder = ExecRecorder::default();
    let m = Machine::started("draupnir-cache", &BootSpec::container("cache", "redis:7"));
    let outcome = recorder.exec(&m, &["redis-cli", "ping"]).unwrap();
    let exec_records_command = recorder.seen.borrow().as_slice()
        == [vec![
            "exec".to_string(),
            "draupnir-cache".to_string(),
            "redis-cli".to_string(),
            "ping".to_string(),
        ]];
    let exec_returns_outcome = outcome.exit_code == Some(0) && outcome.stdout == ["PONG"];

    // (c) container-only guard: a KVM Machine + an empty argv are both rejected, and
    // the guard runs before the engine (no command recorded).
    let guard = ExecRecorder::default();
    let kvm = Machine::started(
        "vm-1",
        &BootSpec::kvm_kernel_rootfs("appliance", "/bzImage", "/rootfs.cpio.gz"),
    );
    let kvm_rejected = matches!(guard.exec(&kvm, &["ls"]), Err(Error::Spec(_)));
    let empty_argv_rejected = matches!(guard.exec(&m, &[]), Err(Error::Spec(_)));
    let guard_before_engine = guard.seen.borrow().is_empty();

    assert_emit!(
        "container",
        "container_exec_builds_argv",
        argv_built
            && exec_records_command
            && exec_returns_outcome
            && kvm_rejected
            && empty_argv_rejected
            && guard_before_engine,
        "argv:{argv_built} records-cmd:{exec_records_command} returns-outcome:{exec_returns_outcome} \
         kvm-reject:{kvm_rejected} empty-reject:{empty_argv_rejected} guard-first:{guard_before_engine}"
    );
}

#[test]
fn await_power_state_confirms_a_boot_readback_across_backends() {
    // The cross-backend boot-status readback: boot() returns before an instance is
    // actually up, so a consumer polls await_power_state / boot_and_await until it
    // reports On. Asserts the reached-Ok, never-reached-timeout, and reject-Unknown
    // arms plus the boot_and_await one-call seam. Records a `draupnir/lifecycle` row.
    let bounded = WaitOptions::bounded(Duration::from_millis(200), Duration::from_millis(1));

    // Reaches On after two Unknowns → Ok.
    let up = ScriptedNode::new(
        [PowerState::Unknown, PowerState::Unknown, PowerState::On],
        PowerState::On,
    );
    let m = Machine::started("node", &BootSpec::container("c", "redis:7"));
    let reached_ok = await_power_state(&up, &m, PowerState::On, &bounded).is_ok();

    // Stuck Off forever → a bounded wait times out (Backend), never a hang.
    let dead = ScriptedNode::new(std::iter::empty(), PowerState::Off);
    let timed_out = matches!(
        await_power_state(
            &dead,
            &m,
            PowerState::On,
            &WaitOptions::bounded(Duration::from_millis(20), Duration::from_millis(1))
        ),
        Err(Error::Backend(_))
    );

    // Awaiting Unknown is nonsensical → rejected as Spec before any poll.
    let idle = ScriptedNode::new([PowerState::On], PowerState::On);
    let unknown_rejected = matches!(
        await_power_state(&idle, &m, PowerState::Unknown, &bounded),
        Err(Error::Spec(_))
    ) && idle.status_calls.get() == 0;

    // boot_and_await: boot then confirm On, returning the live Machine.
    let node = ScriptedNode::new([PowerState::Unknown, PowerState::On], PowerState::On);
    let confirmed = boot_and_await(&node, &BootSpec::container("cache", "redis:7"), &bounded)
        .map(|m| m.id == "id-cache" && m.power == PowerState::On)
        .unwrap_or(false);

    assert_emit!(
        "lifecycle",
        "await_power_state_confirms_a_boot_readback",
        reached_ok && timed_out && unknown_rejected && confirmed,
        "reached On:{reached_ok} never-up times out:{timed_out} await-Unknown rejected:{unknown_rejected} boot_and_await confirms On:{confirmed}"
    );
}

#[test]
fn boot_fleet_and_await_rolls_up_who_is_ready() {
    // The fleet-level provision-readback seam jera's dispatcher consumes: boot N
    // members, await each to On under a bounded per-member budget, and roll up a
    // per-member verdict (up|timeout|error) plus aggregate ready/failed counts. A
    // dead member is a per-member Timeout, never a hang — one bad node can't stall
    // the rollup. Records a `draupnir/lifecycle` matrix row.
    let bounded = WaitOptions::bounded(Duration::from_millis(200), Duration::from_millis(1));

    // A fleet of 3 where the middle member never powers on → node-2 = Timeout,
    // node-1 / node-3 = Up, ready 2 / failed 1.
    let backend = FleetNode {
        dead: HashSet::from(["node-2".to_string()]),
    };
    let one = BootSpec::redfish_iso(
        "node",
        "/images/installer.iso",
        BmcEndpoint {
            host: "https://bmc-42.dc.example".into(),
            username: "admin".into(),
            system_id: "System.Embedded.1".into(),
        },
    );
    let rollup = boot_fleet_and_await(&one, 3, &backend, &bounded);
    let names_in_order = rollup
        .members
        .iter()
        .map(|m| m.name.as_str())
        .eq(["node-1", "node-2", "node-3"]);
    let node2_timeout = matches!(rollup.members[1].outcome, MemberOutcome::Timeout(_));
    let others_up = matches!(rollup.members[0].outcome, MemberOutcome::Up(_))
        && matches!(rollup.members[2].outcome, MemberOutcome::Up(_));
    let partial_counts = rollup.ready() == 2 && rollup.failed() == 1 && !rollup.all_ready();

    // A fully-healthy fleet is all_ready (ready == n, no failures).
    let healthy = FleetNode {
        dead: HashSet::new(),
    };
    let up = boot_fleet_and_await(
        &BootSpec::container("cache", "redis:7"),
        4,
        &healthy,
        &bounded,
    );
    let all_up = up.ready() == 4 && up.failed() == 0 && up.all_ready();

    assert_emit!(
        "lifecycle",
        "boot_fleet_and_await_rolls_up_who_is_ready",
        names_in_order && node2_timeout && others_up && partial_counts && all_up,
        "order:{names_in_order} node-2 timeout:{node2_timeout} others up:{others_up} partial(2/1):{partial_counts} full-fleet all-ready:{all_up}"
    );
}

#[test]
fn boot_fleet_and_await_parallel_matches_the_serial_rollup() {
    // The parallel fleet-boot fans each member onto its own scoped thread (pure std)
    // so N nodes are awaited concurrently — but it MUST produce the identical
    // FleetReadback the serial path does: same per-member verdicts, same fleet order,
    // same aggregate tallies. Mixed fleet of 5 (node-3 dead → Timeout, node-5
    // unbootable → Error, node-1/2/4 Up) asserted equal both ways. Records a
    // `draupnir/lifecycle` matrix row.
    let bounded = WaitOptions::bounded(Duration::from_millis(200), Duration::from_millis(1));
    let backend = MixedNode {
        dead: HashSet::from(["node-3".to_string()]),
        unbootable: HashSet::from(["node-5".to_string()]),
    };
    let spec = BootSpec::container("node", "redis:7");

    let serial = boot_fleet_and_await(&spec, 5, &backend, &bounded);
    let parallel = boot_fleet_and_await_parallel(&spec, 5, &backend, &bounded);

    let identical = parallel == serial;
    let mixed_shape = parallel.members.len() == 5
        && matches!(parallel.members[0].outcome, MemberOutcome::Up(_))
        && matches!(parallel.members[2].outcome, MemberOutcome::Timeout(_))
        && matches!(parallel.members[4].outcome, MemberOutcome::Error(_))
        && parallel.ready() == 3
        && parallel.failed() == 2;

    assert_emit!(
        "lifecycle",
        "boot_fleet_and_await_parallel_matches_serial",
        identical && mixed_shape,
        "parallel==serial:{identical} mixed(3 up/1 timeout/1 error):{mixed_shape}"
    );
}

#[test]
fn seed_files_render_exact_user_data_and_meta_data() {
    let user_data = "#cloud-config\nhostname: web-01\nruncmd:\n  - [systemctl, start, holger]\n";
    let ci = CloudInit {
        user_data: user_data.into(),
        meta_data: Some("instance-id: iid-web-01\nlocal-hostname: web-01\n".into()),
        network_config: None,
    };
    let files = seed::seed_files(&ci);
    assert_emit!(
        "seed",
        "seed_files_render_exact_user_data_and_meta_data",
        files.len() == 2
            && files[0] == ("user-data", user_data.to_string())
            && files[1]
                == (
                    "meta-data",
                    "instance-id: iid-web-01\nlocal-hostname: web-01\n".to_string()
                ),
        "{} files: user-data verbatim + meta-data instance-id/hostname exact",
        files.len()
    );
}

#[test]
fn seed_files_default_meta_data_carries_instance_id_and_hostname() {
    let files = seed::seed_files(&CloudInit::user_data("#cloud-config\n"));
    assert_emit!(
        "seed",
        "omitted_meta_data_gets_default_instance_id_and_hostname",
        files[1].0 == "meta-data"
            && files[1].1 == seed::DEFAULT_META_DATA
            && files[1].1.contains("instance-id:")
            && files[1].1.contains("local-hostname:"),
        "default meta-data = {:?}",
        files[1].1
    );
}

#[test]
fn seed_files_emit_optional_network_config() {
    let net = CloudInit::user_data("#cloud-config\n")
        .with_network_config("version: 2\nethernets:\n  eth0:\n    dhcp4: true\n");
    let files = seed::seed_files(&net);
    assert_emit!(
        "seed",
        "network_config_is_the_optional_third_file",
        files.len() == 3
            && files[2]
                == (
                    "network-config",
                    "version: 2\nethernets:\n  eth0:\n    dhcp4: true\n".to_string()
                ),
        "{} files with a network-config",
        files.len()
    );
}

#[test]
fn write_seed_dir_lands_files_verbatim_on_disk() {
    let dir = std::env::temp_dir().join(format!("draupnir-tm-seeddir-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&dir);
    let ci = CloudInit {
        user_data: "#cloud-config\npackages: [curl]\n".into(),
        meta_data: Some("instance-id: iid-42\nlocal-hostname: node-42\n".into()),
        network_config: None,
    };
    seed::write_seed_dir(&dir, &ci).unwrap();
    let ud = std::fs::read_to_string(dir.join("user-data")).unwrap_or_default();
    let md = std::fs::read_to_string(dir.join("meta-data")).unwrap_or_default();
    assert_emit!(
        "seed",
        "write_seed_dir_lands_files_verbatim",
        ud == "#cloud-config\npackages: [curl]\n"
            && md == "instance-id: iid-42\nlocal-hostname: node-42\n",
        "seed dir at {} carries user-data + meta-data verbatim",
        dir.display()
    );
    let _ = std::fs::remove_dir_all(&dir);
}

/// RED-WHEN-BROKEN: a **failed boot** and a **failed rollback** each land as a RED
/// (FAIL) functional-status row — the failure surfaces jera's matrix reads back. A
/// fake/failed instance (a backend whose `boot` and `power_off` both error) drives
/// the real [`draupnir::boot`] and [`draupnir::rollback`] surfaces; each call must
/// (a) propagate the `Err` and (b) record its verdict as a **RED** row. If the emit
/// were dropped or the verdict inverted (a broken boot reported GREEN), this fails.
/// The GREEN control (a healthy instance) proves the row is not stuck-red. The
/// drain-and-assert half needs the `testmatrix` feature (only then are rows
/// buffered); the default build still asserts the propagated `Err`, so it is
/// red-when-broken either way.
#[test]
fn failed_boot_and_rollback_emit_red_rows() {
    // A "fake/failed instance": every boot and every power-off fails.
    struct DeadBackend;
    impl Boot for DeadBackend {
        fn boot(&self, spec: &BootSpec) -> Result<Machine> {
            Err(Error::Backend(format!("no host slot for {}", spec.name)))
        }
    }
    impl Lifecycle for DeadBackend {
        fn power_on(&self, _m: &Machine) -> Result<()> {
            Ok(())
        }
        fn power_off(&self, _m: &Machine) -> Result<()> {
            Err(Error::Backend("power-off refused".into()))
        }
        fn status(&self, _m: &Machine) -> Result<PowerState> {
            Ok(PowerState::Off)
        }
    }

    let dead = DeadBackend;
    let spec = BootSpec::container("doomed", "redis:7");

    // (a) A failed backend boot propagates as Err(Backend) — and emits a RED row.
    let boot_err = matches!(draupnir::boot(&spec, &dead), Err(Error::Backend(_)));
    // A failed power-off (rollback) propagates as Err(Backend) — and emits a RED row.
    let m = Machine::started("doomed", &spec);
    let rollback_err = matches!(draupnir::rollback(&dead, &m), Err(Error::Backend(_)));

    // A GREEN control: a healthy backend boots + rolls back cleanly (row not stuck-red).
    let healthy = ScriptedNode::new([PowerState::On], PowerState::On);
    let boot_ok = draupnir::boot(&spec, &healthy).is_ok();
    let rollback_ok = draupnir::rollback(&healthy, &m).is_ok();

    assert!(
        boot_err,
        "a failed backend boot must propagate as Err(Backend)"
    );
    assert!(
        rollback_err,
        "a failed power-off must propagate as Err(Backend)"
    );
    assert!(
        boot_ok && rollback_ok,
        "a healthy backend boots and rolls back cleanly"
    );

    // (b) With the matrix feature ON, prove the two failures are RED rows and the two
    // healthy calls are GREEN rows in the drained buffer (the emit wiring itself).
    #[cfg(feature = "testmatrix")]
    {
        let rows = draupnir::drain_status_rows();
        let has = |component: &str, check: &str, ok: bool| {
            rows.iter()
                .any(|(c, k, o)| c == component && k == check && *o == ok)
        };
        assert!(
            has("draupnir/boot", "boot", false),
            "failed boot => RED row"
        );
        assert!(
            has("draupnir/lifecycle", "rollback", false),
            "failed rollback => RED row"
        );
        assert!(
            has("draupnir/boot", "boot", true),
            "healthy boot => GREEN row"
        );
        assert!(
            has("draupnir/lifecycle", "rollback", true),
            "healthy rollback => GREEN row"
        );
    }
}

/// The vfat seed image round-trips label + contents (feature `seed`); when the
/// feature is off the authoring is compiled out — a PARTIAL row would be emitted by
/// a matrix run, but here it is simply not asserted.
#[cfg(feature = "seed")]
#[test]
fn build_seed_image_roundtrips_cidata_label_and_contents() {
    use std::io::Read;
    let out = std::env::temp_dir().join(format!("draupnir-tm-seed-{}.img", std::process::id()));
    let _ = std::fs::remove_file(&out);
    let ci = CloudInit {
        user_data: "#cloud-config\nruncmd:\n  - [echo, hi]\n".into(),
        meta_data: Some("instance-id: iid-img\nlocal-hostname: img-host\n".into()),
        network_config: None,
    };
    seed::build_seed_image(&out, &ci).unwrap();
    let img = std::fs::File::options()
        .read(true)
        .write(true)
        .open(&out)
        .unwrap();
    let fs = fatfs::FileSystem::new(img, fatfs::FsOptions::new()).unwrap();
    let label_ok = fs.volume_label().to_ascii_lowercase() == seed::NOCLOUD_LABEL;
    let mut ud = String::new();
    fs.root_dir()
        .open_file("user-data")
        .unwrap()
        .read_to_string(&mut ud)
        .unwrap();
    assert_emit!(
        "seed",
        "build_seed_image_roundtrips_cidata_label_and_contents",
        label_ok && ud == "#cloud-config\nruncmd:\n  - [echo, hi]\n",
        "vfat label=cidata:{label_ok}, user-data survives the round-trip"
    );
    drop(fs);
    let _ = std::fs::remove_file(&out);
}