"""Tests for host-side fleet evidence (bin/host_evidence.py). Covers log-line classification, verdict assembly (silent-when-healthy runs vs. fresh failures vs. stale/missing runs), and the effective status mapping used by `box fleet status` in PID/net-blind shells. """ import importlib.util import unittest from datetime import datetime, timedelta, timezone from pathlib import Path REPO_ROOT = Path(__file__).resolve().parent.parent def _load(mod_name, rel_path): spec = importlib.util.spec_from_file_location(mod_name, REPO_ROOT / rel_path) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return mod he = _load("host_evidence_test", "bin/host_evidence.py") NOW = datetime(2026, 10, 6, 8, 40, 0, tzinfo=timezone.utc) def _ts(dt): return dt.strftime("[%Y-%m-%dT%H:%M:%SZ]") def _run(minutes_ago): return NOW - timedelta(minutes=minutes_ago) class ParseLogTs(unittest.TestCase): def test_valid(self): self.assertEqual( he.parse_log_ts("[2026-10-06T05:55:25Z] [opm] relay restart FAILED"), datetime(2026, 10, 6, 5, 55, 25, tzinfo=timezone.utc)) def test_garbage(self): self.assertIsNone(he.parse_log_ts("cdp relay already running")) self.assertIsNone(he.parse_log_ts("")) self.assertIsNone(he.parse_log_ts("[2026-13-99T99:99:99Z] [muse] x")) class ClassifyChromebox(unittest.TestCase): def test_down(self): self.assertEqual( he.classify_chromebox_line("[muse] relaunch FAILED (x) — needs operator attention"), "down") def test_healthy(self): self.assertEqual( he.classify_chromebox_line("[muse] relaunch OK (attempt 2)"), "healthy") self.assertEqual( he.classify_chromebox_line( "[muse] tunnel restart recovered warp egress, chrome relaunch not needed"), "healthy") def test_degraded(self): for line in ( "[muse] relaunch attempt 3 not healthy yet (CDP up but no page)", "[muse] tunnel restart did not recover egress (x), proceeding with chrome relaunch", "[muse] unhealthy (x), relaunching chromebox", "[muse] warp partition detected (x), restarting tunnel via netvm-node-up.sh", "[muse] browser launched recently (pid 1), skipping relaunch (probably still starting)", ): self.assertEqual(he.classify_chromebox_line(line), "degraded", line) def test_no_verdict(self): for line in ( "[2026-10-06T08:00:00Z] [muse] log rotated", "cdp relay already running", "tunnel already up (egress=1.2.3.4), skipping handshake wait", "node=muse netns=warp-muse egress=1.2.3.4", "Traceback (most recent call last):", ): self.assertIsNone(he.classify_chromebox_line(line), line) class ClassifyRelay(unittest.TestCase): def test_down(self): self.assertEqual( he.classify_relay_line("[opm] relay restart FAILED on 10.201.157.2:9440 — needs operator"), "down") self.assertEqual( he.classify_relay_line("[pip] FAIL_LOUD: host veth ve-x missing IP 10.201.87.1"), "down") # Trailing "restarting" with no OK/FAILED after it: crashed mid-restart. self.assertEqual( he.classify_relay_line("[muse] relay unhealthy on 10.201.35.2:9410, restarting"), "down") def test_healthy(self): self.assertEqual( he.classify_relay_line("[646] relay restarted OK on 10.201.202.2:9430"), "healthy") def test_no_verdict(self): self.assertIsNone(he.classify_relay_line(" File \"/x/netvm-cdp-relay.py\", line 37, in main")) self.assertIsNone(he.classify_relay_line("")) class LastVerdict(unittest.TestCase): def test_newest_wins_per_node(self): lines = [ "%s [muse] relaunch OK (attempt 1)" % _ts(_run(30)), "%s [pip] relaunch FAILED (x)" % _ts(_run(20)), "%s [muse] unhealthy (x), relaunching chromebox" % _ts(_run(5)), "%s [muse] log rotated" % _ts(_run(1)), ] v, ts, _ = he.last_verdict(lines, "muse", he.classify_chromebox_line) self.assertEqual(v, "degraded") self.assertEqual(ts, _run(5)) v, _, _ = he.last_verdict(lines, "pip", he.classify_chromebox_line) self.assertEqual(v, "down") def test_none_when_no_verdict(self): self.assertIsNone(he.last_verdict( ["[2026-10-06T08:00:00Z] [muse] log rotated", "noise"], "muse", he.classify_chromebox_line)) self.assertIsNone(he.last_verdict([], "muse", he.classify_chromebox_line)) class BrowserVerdict(unittest.TestCase): def test_silent_run_is_healthy(self): v, d = he.browser_verdict("muse", [], _run(2), now=NOW) self.assertEqual(v, "healthy") v, _ = he.browser_verdict( "muse", ["%s [muse] relaunch FAILED (x)" % _ts(_run(60))], _run(2), now=NOW) self.assertEqual(v, "healthy") # old failure, silent since def test_fresh_failure_wins(self): v, _ = he.browser_verdict( "muse", ["%s [muse] relaunch FAILED (x)" % _ts(_run(1))], _run(2), now=NOW) self.assertEqual(v, "down") v, _ = he.browser_verdict( "muse", ["%s [muse] unhealthy (x), relaunching chromebox" % _ts(_run(1))], _run(2), now=NOW) self.assertEqual(v, "degraded") def test_fresh_recovery_is_healthy(self): v, _ = he.browser_verdict( "muse", ["%s [muse] tunnel restart recovered warp egress, chrome relaunch not needed" % _ts(_run(1))], _run(2), now=NOW) self.assertEqual(v, "healthy") def test_stale_or_missing_run_is_unknown(self): v, _ = he.browser_verdict("muse", [], _run(30), now=NOW) self.assertEqual(v, "unknown") v, _ = he.browser_verdict("muse", [], None, now=NOW) self.assertEqual(v, "unknown") def test_def_uses_watchdog_like_core_nodes(self): v, _ = he.browser_verdict("def", [], _run(2), now=NOW) self.assertEqual(v, "healthy") # silent run v, _ = he.browser_verdict("def", [], None, now=NOW) self.assertEqual(v, "unknown") # no timer run (yet) v, _ = he.browser_verdict( "def", ["%s [def] relaunch FAILED (x)" % _ts(_run(1))], _run(2), now=NOW) self.assertEqual(v, "down") def test_uncovered_node_uses_chrome_log_freshness(self): v, _ = he.browser_verdict("ghost", [], None, chrome_log_mtime=_run(5), now=NOW) self.assertEqual(v, "healthy") v, d = he.browser_verdict("ghost", [], None, chrome_log_mtime=_run(60), now=NOW) self.assertEqual(v, "unknown") v, _ = he.browser_verdict("ghost", [], None, chrome_log_mtime=None, now=NOW) self.assertEqual(v, "unknown") class CdpVerdict(unittest.TestCase): def test_silent_run_is_healthy(self): v, _ = he.cdp_verdict("opm", [], _run(4), now=NOW) self.assertEqual(v, "healthy") def test_fresh_failure_is_down(self): v, _ = he.cdp_verdict( "opm", ["%s [opm] relay restart FAILED on 10.201.157.2:9440" % _ts(_run(1))], _run(4), now=NOW) self.assertEqual(v, "down") def test_stale_or_missing_run_is_unknown(self): v, _ = he.cdp_verdict("opm", [], _run(60), now=NOW) self.assertEqual(v, "unknown") v, _ = he.cdp_verdict("opm", [], None, now=NOW) self.assertEqual(v, "unknown") def test_dev_def_monitored_like_core_nodes(self): for node in ("def", "dev"): v, _ = he.cdp_verdict(node, [], _run(4), now=NOW) self.assertEqual(v, "healthy", node) # silent run v, _ = he.cdp_verdict( node, ["%s [%s] relay restart FAILED on 10.0.0.1:1" % (_ts(_run(1)), node)], _run(4), now=NOW) self.assertEqual(v, "down", node) def test_unmonitored_node_is_unknown(self): v, d = he.cdp_verdict("ghost", [], _run(1), now=NOW) self.assertEqual(v, "unknown") self.assertIn("no relay-monitor coverage", d) class EffectiveStatus(unittest.TestCase): def test_live_local_signals_win(self): self.assertEqual(he.effective_status(True, True, "down", "down"), ("ACTIVE", "local")) self.assertEqual(he.effective_status(True, False, "healthy", "healthy"), ("CDP_DOWN", "local")) def test_cdp_without_proc_is_active(self): self.assertEqual(he.effective_status(False, True, "unknown", "unknown"), ("ACTIVE", "local")) def test_blind_with_evidence(self): self.assertEqual( he.effective_status(False, False, "healthy", "healthy"), ("ACTIVE", "host-evidence")) self.assertEqual( he.effective_status(False, False, "degraded", "healthy"), ("ACTIVE", "host-evidence")) self.assertEqual( he.effective_status(False, False, "down", "healthy"), ("STOPPED", "host-evidence")) self.assertEqual( he.effective_status(False, False, "healthy", "down"), ("CDP_DOWN", "host-evidence")) def test_blind_without_evidence_is_unknown(self): self.assertEqual( he.effective_status(False, False, "unknown", "unknown"), ("UNKNOWN", "host-evidence")) self.assertEqual( he.effective_status(False, False, "healthy", "unknown"), ("UNKNOWN", "host-evidence")) self.assertEqual( he.effective_status(False, False, "unknown", "healthy"), ("UNKNOWN", "host-evidence")) class CollectSeams(unittest.TestCase): def test_assembles_per_node_verdicts(self): starts = {"cdp-relay-watchdog.service": _run(4), "chromebox-watchdog@muse.service": _run(1), "chromebox-watchdog@def.service": _run(1)} relay_lines = ["%s [muse] relay restarted OK on 10.201.35.2:9410" % _ts(_run(3))] chrome_lines = ["%s [muse] relaunch OK (attempt 1)" % _ts(_run(1))] out = he.collect(["muse", "def"], _journal_starts=starts, _relay_lines=relay_lines, _chromebox_lines=chrome_lines, _chrome_mtimes={"def": _run(5)}, _now=NOW) self.assertEqual(out["muse"]["browser"], "healthy") self.assertEqual(out["muse"]["cdp"], "healthy") # def is watchdog-covered: silent runs read healthy (chrome-log # freshness is no longer consulted for covered nodes). self.assertEqual(out["def"]["browser"], "healthy") self.assertEqual(out["def"]["cdp"], "healthy") class CoverageTuples(unittest.TestCase): def test_watchdog_tuples_cover_registry(self): reg = _load("netvm_registry_cov", "bin/netvm-registry.py") for node in reg.active_nodes(): self.assertIn(node, he.RELAY_NODES) self.assertIn(node, he.CHROMEBOX_NODES) if __name__ == "__main__": unittest.main()