102 lines
3.8 KiB
Python
102 lines
3.8 KiB
Python
|
"""Tests for setup-fed watchdog supervision.
|
||
|
|
|
||
|
|
Covers bin/ensure-node-supervision.sh (registry row + timer unit,
|
||
|
|
idempotent), its hook in netvm-node-up.sh, and the registry-driven
|
||
|
|
node lists in relay-health-check.sh / cdp-latency-check.sh.
|
||
|
|
|
||
|
|
Shell is driven for real (LIB_ONLY sourcing where the script runs on
|
||
|
|
import); UNIT_DIR/NODES_MD overrides keep everything in scratch dirs.
|
||
|
|
"""
|
||
|
|
import subprocess
|
||
|
|
import tempfile
|
||
|
|
import unittest
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||
|
|
BIN = REPO_ROOT / "bin"
|
||
|
|
|
||
|
|
|
||
|
|
def _bash(prog, extra_env=None):
|
||
|
|
env = {"PATH": "/usr/bin:/bin"}
|
||
|
|
env.update(extra_env or {})
|
||
|
|
return subprocess.run(["bash", "-c", prog], capture_output=True,
|
||
|
|
text=True, env=env, timeout=30)
|
||
|
|
|
||
|
|
|
||
|
|
class EnsureSupervision(unittest.TestCase):
|
||
|
|
def setUp(self):
|
||
|
|
self.tmp = tempfile.TemporaryDirectory()
|
||
|
|
self.unit_dir = str(Path(self.tmp.name) / "units")
|
||
|
|
Path(self.unit_dir).mkdir()
|
||
|
|
self.nodes_md = str(Path(self.tmp.name) / "NODES.md")
|
||
|
|
Path(self.nodes_md).write_text(
|
||
|
|
"# NetVM Nodes (bl)\n\n"
|
||
|
|
"| node | netns | egress_ip | cdp_port | status | agent |\n"
|
||
|
|
"|------|-------|-----------|----------|--------|-------|\n"
|
||
|
|
"| muse | warp-muse | 1.2.3.4 | 9410 | active | muse |\n")
|
||
|
|
self.env = {"NODES_MD": self.nodes_md, "UNIT_DIR": self.unit_dir}
|
||
|
|
|
||
|
|
def tearDown(self):
|
||
|
|
self.tmp.cleanup()
|
||
|
|
|
||
|
|
def ensure(self, *args):
|
||
|
|
return _bash("'%s' %s" % (
|
||
|
|
BIN / "ensure-node-supervision.sh",
|
||
|
|
" ".join("'%s'" % a for a in args)), self.env)
|
||
|
|
|
||
|
|
def test_adds_registry_row_and_timer(self):
|
||
|
|
r = self.ensure("pip")
|
||
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
||
|
|
body = Path(self.nodes_md).read_text()
|
||
|
|
self.assertRegex(body, r"\| pip \| warp-pip \| \S+ \| 9420 \| active \|")
|
||
|
|
unit = Path(self.unit_dir) / "chromebox-watchdog-pip.timer"
|
||
|
|
self.assertTrue(unit.exists())
|
||
|
|
text = unit.read_text()
|
||
|
|
self.assertIn("Unit=chromebox-watchdog@pip.service", text)
|
||
|
|
self.assertIn("OnUnitActiveSec=2min", text)
|
||
|
|
|
||
|
|
def test_idempotent(self):
|
||
|
|
self.assertEqual(self.ensure("pip").returncode, 0)
|
||
|
|
before = Path(self.nodes_md).read_text()
|
||
|
|
r = self.ensure("pip")
|
||
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
||
|
|
self.assertEqual(Path(self.nodes_md).read_text(), before)
|
||
|
|
self.assertIn("already", r.stdout)
|
||
|
|
|
||
|
|
def test_honors_port_override(self):
|
||
|
|
env = dict(self.env, CDP_PORT_OVERRIDE="9470")
|
||
|
|
r = _bash("'%s' newnode" % (BIN / "ensure-node-supervision.sh"), env)
|
||
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
||
|
|
self.assertRegex(Path(self.nodes_md).read_text(),
|
||
|
|
r"\| newnode \| warp-newnode \| \S+ \| 9470 \| active \|")
|
||
|
|
|
||
|
|
def test_node_up_hooks_ensure(self):
|
||
|
|
text = (BIN / "netvm-node-up.sh").read_text()
|
||
|
|
self.assertIn("ensure-node-supervision.sh", text)
|
||
|
|
|
||
|
|
|
||
|
|
class CheckScriptCoverage(unittest.TestCase):
|
||
|
|
def watched(self, script, guard):
|
||
|
|
prog = "source '%s'\nwatched_nodes\n" % (BIN / script)
|
||
|
|
return _bash(prog, {guard: "1"})
|
||
|
|
|
||
|
|
def test_relay_health_covers_registry(self):
|
||
|
|
r = self.watched("relay-health-check.sh",
|
||
|
|
"RELAY_HEALTH_CHECK_LIB_ONLY")
|
||
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
||
|
|
nodes = set(r.stdout.split())
|
||
|
|
for n in ("muse", "pip", "646", "opm", "def", "dev"):
|
||
|
|
self.assertIn(n, nodes)
|
||
|
|
|
||
|
|
def test_cdp_latency_covers_registry(self):
|
||
|
|
r = self.watched("cdp-latency-check.sh",
|
||
|
|
"CDP_LATENCY_CHECK_LIB_ONLY")
|
||
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
||
|
|
nodes = set(r.stdout.split())
|
||
|
|
for n in ("muse", "pip", "646", "opm", "def", "dev"):
|
||
|
|
self.assertIn(n, nodes)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main()
|