nixos/tests/deploy-guard.nix

# Aggregator test for modules/server-deploy-guard.nix.
#
# The jellyfin and minecraft check scripts are validated at build time by
# writeShellApplication's shellcheck / writePython3Bin's pyflakes, plus manual
# post-deploy verification on muffin. This test focuses on the aggregator and
# bypass contract with synthetic checks so failures in this file point at the
# aggregator itself rather than at Jellyfin/Minecraft availability.
{
  lib,
  pkgs,
  inputs,
  ...
}:
let
  baseServiceConfigs = import ../hosts/muffin/service-configs.nix;
  testServiceConfigs = lib.recursiveUpdate baseServiceConfigs {
    zpool_ssds = "";
    https.domain = "test.local";
  };

  alwaysOk = pkgs.writeShellApplication {
    name = "deploy-guard-check-synthetic-ok";
    text = ''echo "all clear"'';
  };

  alwaysFail = pkgs.writeShellApplication {
    name = "deploy-guard-check-synthetic-fail";
    text = ''
      echo "synthetic failure reason"
      exit 1
    '';
  };

  # Blocks only while /tmp/synth-fail exists. Lets the test script drive the
  # check's state without restarting the system.
  conditional = pkgs.writeShellApplication {
    name = "deploy-guard-check-conditional";
    text = ''
      if [[ -e /tmp/synth-fail ]]; then
        echo "conditional marker present"
        exit 1
      fi
      echo "condition clear"
    '';
  };

  # Hangs past the aggregator's timeout only while /tmp/synth-slow exists —
  # otherwise fast-path so the default state of other subtests is unaffected.
  slowIfMarker = pkgs.writeShellApplication {
    name = "deploy-guard-check-slow-if-marker";
    runtimeInputs = [ pkgs.coreutils ];
    text = ''
      if [[ -e /tmp/synth-slow ]]; then
        sleep 30
      fi
      echo "fast path"
    '';
  };
in
pkgs.testers.runNixOSTest {
  name = "deploy-guard";

  node.specialArgs = {
    inherit inputs lib;
    service_configs = testServiceConfigs;
    username = "testuser";
  };

  nodes.machine =
    { ... }:
    {
      imports = [
        ../modules/server-deploy-guard.nix
      ];

      environment.systemPackages = [ pkgs.jq ];

      services.deployGuard = {
        enable = true;
        timeout = 2;
        checks = {
          always-ok = {
            description = "synthetic always-pass";
            command = alwaysOk;
          };
          synthetic-fail = {
            description = "synthetic always-block";
            command = alwaysFail;
          };
          conditional = {
            description = "blocks while /tmp/synth-fail exists";
            command = conditional;
          };
          slow-if-marker = {
            description = "sleeps past timeout while /tmp/synth-slow exists";
            command = slowIfMarker;
          };
        };
      };
    };

  testScript = ''
    import json
    import time

    start_all()
    machine.wait_for_unit("multi-user.target")

    with subtest("baseline: mixed pass/block aggregates to blocked"):
        rc, out = machine.execute("deploy-guard-check 2>&1")
        assert rc == 1, f"expected blocked (rc=1), got rc={rc}\n{out}"
        assert "PASS: always-ok" in out, out
        assert "PASS: conditional" in out, out
        assert "PASS: slow-if-marker" in out, out
        assert "BLOCK: synthetic-fail" in out, out
        assert "synthetic failure reason" in out, out

    with subtest("bypass via DEPLOY_GUARD_BYPASS env"):
        rc, out = machine.execute("DEPLOY_GUARD_BYPASS=1 deploy-guard-check 2>&1")
        assert rc == 0, f"bypass should pass, got rc={rc}\n{out}"
        assert "BYPASS" in out, out

    with subtest("bypass via /run/deploy-guard-bypass is single-shot"):
        machine.succeed("touch /run/deploy-guard-bypass")
        rc, out = machine.execute("deploy-guard-check 2>&1")
        assert rc == 0, f"marker bypass should pass, got rc={rc}\n{out}"
        machine.fail("test -e /run/deploy-guard-bypass")
        rc, out = machine.execute("deploy-guard-check 2>&1")
        assert rc == 1, f"marker must be single-shot, got rc={rc}\n{out}"

    with subtest("conditional check toggles on marker file"):
        machine.succeed("touch /tmp/synth-fail")
        rc, out = machine.execute("deploy-guard-check 2>&1")
        assert rc == 1
        assert "BLOCK: conditional" in out, out
        assert "conditional marker present" in out, out
        machine.succeed("rm /tmp/synth-fail")
        rc, out = machine.execute("deploy-guard-check 2>&1")
        assert rc == 1
        assert "PASS: conditional" in out, out

    with subtest("per-check timeout kills runaway checks"):
        machine.succeed("touch /tmp/synth-slow")
        t0 = time.monotonic()
        rc, out = machine.execute("deploy-guard-check 2>&1")
        elapsed = time.monotonic() - t0
        assert rc == 1, f"expected block, got rc={rc}\n{out}"
        assert "BLOCK: slow-if-marker" in out, out
        assert "timed out" in out, out
        # timeout=2s per check; the whole run must finish well under 10s even
        # running every check serially.
        assert elapsed < 10, f"aggregator took {elapsed:.1f}s — did timeout misfire?"
        machine.succeed("rm /tmp/synth-slow")

    with subtest("--json output is well-formed"):
        rc, out = machine.execute("DEPLOY_GUARD_BYPASS=1 deploy-guard-check --json")
        data = json.loads(out.strip())
        assert data["bypassed"] is True, data
        assert data["ok"] is True, data
        assert data["checks"] == [], data

        rc, out = machine.execute("deploy-guard-check --json")
        data = json.loads(out.strip())
        assert data["bypassed"] is False, data
        assert data["ok"] is False, data
        names = {c["name"]: c for c in data["checks"]}
        assert set(names) == {
            "always-ok", "synthetic-fail", "conditional", "slow-if-marker"
        }, names
        assert names["always-ok"]["ok"] is True, names
        assert names["synthetic-fail"]["ok"] is False, names
        assert names["synthetic-fail"]["exit"] == 1, names
        assert "synthetic failure reason" in names["synthetic-fail"]["output"], names
  '';
}