operator
0065d11e97
- bin/muse_choice_watcher.py + systemd/muse-choices-reconcile.*: automatic choice answering and timer reconciliation - bin/digest.py: fleet log and health summarization - docs/BOX-*-HTTPS.md: comprehensive HTTPS execution contracts and API documentation - docs/MUSE-CHOICES-POLICY.md & docs/SUPERVISION-SPEC.md: autonomous execution specs - tests/test_*.py: unit test suites for HTTPS API, choice watcher, fleet heal, and swarm pruning
80 lines
3.0 KiB
Python
80 lines
3.0 KiB
Python
"""The bl-side #lobby relay must stay opt-in.
|
|
|
|
fleet-alert-check.sh invokes bin/fleet-alert-relay.sh only when
|
|
FLEET_BL_RELAY=1: the container-side hook is the live pager, and running
|
|
both double-posts every alert (2026-10-06). These tests run a copy of the
|
|
checker with stubbed-out fleet commands and assert the relay stub is (not)
|
|
invoked.
|
|
"""
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
CHECKER = REPO_ROOT / "bin" / "fleet-alert-check.sh"
|
|
|
|
|
|
class BlRelayGate(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
root = Path(self.tmp.name)
|
|
self.bindir = root / "bin"
|
|
self.bindir.mkdir()
|
|
# Copy the real checker so BIN-relative lookups hit our stubs.
|
|
shutil.copy(CHECKER, self.bindir / "fleet-alert-check.sh")
|
|
(self.bindir / "fleet-alert-relay.sh").write_text(
|
|
"#!/bin/bash\necho RELAY-RAN >> \"$CALLS\"\n")
|
|
(self.bindir / "fleet-alert-relay.sh").chmod(0o755)
|
|
(self.bindir / "netvm-registry.py").write_text(
|
|
"#!/usr/bin/env python3\n") # no nodes: all fleet loops skip
|
|
(self.bindir / "netvm-registry.py").chmod(0o755)
|
|
(self.bindir / "box-ctl.py").write_text("#!/usr/bin/env python3\n")
|
|
(self.bindir / "box-ctl.py").chmod(0o755)
|
|
fakebin = root / "fakebin"
|
|
fakebin.mkdir()
|
|
(fakebin / "sudo").write_text("#!/bin/bash\necho sudo-stub >&2\nexit 1\n")
|
|
(fakebin / "sudo").chmod(0o755)
|
|
self.state = root / "state"
|
|
self.state.mkdir()
|
|
self.calls = root / "calls.log"
|
|
self.env = dict(os.environ)
|
|
self.env["PATH"] = str(fakebin) + ":/usr/bin:/bin"
|
|
self.env["FLEET_ALERT_STATE_DIR"] = str(self.state)
|
|
self.env["CALLS"] = str(self.calls)
|
|
self.env.pop("FLEET_BL_RELAY", None)
|
|
|
|
def run_checker(self, **extra):
|
|
env = dict(self.env)
|
|
env.update(extra)
|
|
# NOTE: appends 1-2 lines to the shared /tmp/fleet-alert-check.log
|
|
# (LOG path is hardcoded); same as any production timer run.
|
|
return subprocess.run(
|
|
["bash", str(self.bindir / "fleet-alert-check.sh")],
|
|
capture_output=True, text=True, env=env, timeout=120)
|
|
|
|
def relay_ran(self):
|
|
return self.calls.exists() and "RELAY-RAN" in self.calls.read_text()
|
|
|
|
def test_relay_not_invoked_by_default(self):
|
|
r = self.run_checker()
|
|
self.assertEqual(r.returncode, 0, r.stderr[-2000:])
|
|
self.assertFalse(self.relay_ran())
|
|
|
|
def test_relay_invoked_when_opted_in(self):
|
|
r = self.run_checker(FLEET_BL_RELAY="1")
|
|
self.assertEqual(r.returncode, 0, r.stderr[-2000:])
|
|
self.assertTrue(self.relay_ran())
|
|
|
|
def test_dry_run_never_invokes_relay(self):
|
|
r = self.run_checker(FLEET_BL_RELAY="1", FLEET_ALERT_DRY_RUN="1")
|
|
self.assertEqual(r.returncode, 0, r.stderr[-2000:])
|
|
self.assertFalse(self.relay_ran())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|