operator
113 lines
4.1 KiB
Python
113 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
|
"""test_tmux_server_watchdog.py — Death-capture transition logic.
|
|
|
|
Covers: steady state is quiet, pid change yields restart, alive->dead
|
|
yields a death bundle, dead->alive yields started, corrupt state file
|
|
is tolerated.
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
REPO_ROOT = Path("/home/super/Projects/NetVM")
|
|
BIN_DIR = REPO_ROOT / "bin"
|
|
sys.path.insert(0, str(BIN_DIR))
|
|
|
|
import tmux_server_watchdog as w
|
|
|
|
|
|
class TestEvaluate(unittest.TestCase):
|
|
def test_steady_alive_is_quiet(self):
|
|
prev = {"/s": {"pid": 100, "since": "t"}}
|
|
new, events = w.evaluate(prev, {"/s": 100})
|
|
self.assertEqual(events, [])
|
|
self.assertEqual(new["/s"]["pid"], 100)
|
|
|
|
def test_pid_change_is_restart(self):
|
|
prev = {"/s": {"pid": 100, "since": "t"}}
|
|
new, events = w.evaluate(prev, {"/s": 200})
|
|
self.assertEqual(len(events), 1)
|
|
self.assertEqual(events[0]["type"], "restart")
|
|
self.assertEqual(events[0]["old_pid"], 100)
|
|
|
|
def test_alive_to_dead_is_death(self):
|
|
prev = {"/s": {"pid": 100, "since": "t"}}
|
|
new, events = w.evaluate(prev, {"/s": None})
|
|
self.assertEqual(len(events), 1)
|
|
self.assertEqual(events[0]["type"], "death")
|
|
self.assertIsNone(new["/s"]["pid"])
|
|
|
|
def test_dead_stays_dead_is_quiet(self):
|
|
prev = {"/s": {"pid": None, "died": "t", "last_pid": 100}}
|
|
_, events = w.evaluate(prev, {"/s": None})
|
|
self.assertEqual(events, [])
|
|
|
|
def test_dead_to_alive_is_started(self):
|
|
prev = {"/s": {"pid": None, "died": "t", "last_pid": 100}}
|
|
_, events = w.evaluate(prev, {"/s": 300})
|
|
self.assertEqual(len(events), 1)
|
|
self.assertEqual(events[0]["type"], "started")
|
|
|
|
def test_unknown_socket_first_seen_is_started(self):
|
|
_, events = w.evaluate({}, {"/s": 300})
|
|
self.assertEqual(events[0]["type"], "started")
|
|
|
|
|
|
class TestCheck(unittest.TestCase):
|
|
def _iso(self, td):
|
|
state = str(td / "servers.json")
|
|
log = str(td / "deaths.jsonl")
|
|
p1 = mock.patch.object(w, "STATE_FILE", state)
|
|
p2 = mock.patch.object(w, "DEATH_LOG", log)
|
|
return p1, p2, state, log
|
|
|
|
def test_death_writes_bundle(self):
|
|
import tempfile
|
|
with tempfile.TemporaryDirectory() as td:
|
|
td = Path(td)
|
|
p1, p2, state, log = self._iso(td)
|
|
with open(state, "w") as f:
|
|
json.dump({"/s": {"pid": 100, "since": "t"}}, f)
|
|
with p1, p2, \
|
|
mock.patch.object(w, "probe", return_value=None), \
|
|
mock.patch.object(w, "collect_forensics",
|
|
return_value={"ts": "t", "socket": "/s",
|
|
"last_pid": 100}):
|
|
res = w.check(sockets=["/s"])
|
|
self.assertEqual(res["events"][0]["type"], "death")
|
|
bundle = json.loads(open(log).read().strip())
|
|
self.assertEqual(bundle["event"], "death")
|
|
self.assertEqual(bundle["last_pid"], 100)
|
|
self.assertIsNone(json.load(open(state))["/s"]["pid"])
|
|
|
|
def test_dry_run_writes_nothing(self):
|
|
import tempfile
|
|
with tempfile.TemporaryDirectory() as td:
|
|
td = Path(td)
|
|
p1, p2, state, log = self._iso(td)
|
|
with p1, p2, \
|
|
mock.patch.object(w, "probe", return_value=100):
|
|
res = w.check(sockets=["/s"], dry_run=True)
|
|
self.assertEqual(res["events"][0]["type"], "started")
|
|
self.assertFalse(Path(state).exists())
|
|
self.assertFalse(Path(log).exists())
|
|
|
|
def test_corrupt_state_tolerated(self):
|
|
import tempfile
|
|
with tempfile.TemporaryDirectory() as td:
|
|
td = Path(td)
|
|
p1, p2, state, _ = self._iso(td)
|
|
with open(state, "w") as f:
|
|
f.write("{{{nope")
|
|
with p1, p2, \
|
|
mock.patch.object(w, "probe", return_value=100):
|
|
res = w.check(sockets=["/s"])
|
|
self.assertEqual(res["events"][0]["type"], "started")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|