88 lines
3.3 KiB
Python
88 lines
3.3 KiB
Python
|
"""Tests for the agent-health.sh restart circuit breaker.
|
||
|
|
|
||
|
|
Drives the real shell functions (sourced with AGENT_HEALTH_LIB_ONLY=1)
|
||
|
|
against a scratch STATE_DIR/LOG: futile-restart counting, circuit open,
|
||
|
|
half-open probe after cooldown, and reset on success.
|
||
|
|
"""
|
||
|
|
import subprocess
|
||
|
|
import tempfile
|
||
|
|
import unittest
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||
|
|
SCRIPT = REPO_ROOT / "bin" / "agent-health.sh"
|
||
|
|
|
||
|
|
|
||
|
|
def _run(state_dir, log, snippet):
|
||
|
|
prog = (
|
||
|
|
"source '%s'\n"
|
||
|
|
"STATE_DIR='%s'; LOG='%s'\n"
|
||
|
|
"%s\n" % (SCRIPT, state_dir, log, snippet)
|
||
|
|
)
|
||
|
|
env = {"AGENT_HEALTH_LIB_ONLY": "1", "PATH": "/usr/bin:/bin"}
|
||
|
|
return subprocess.run(["bash", "-c", prog], capture_output=True,
|
||
|
|
text=True, env=env, timeout=30)
|
||
|
|
|
||
|
|
|
||
|
|
class CircuitBreaker(unittest.TestCase):
|
||
|
|
def setUp(self):
|
||
|
|
self.tmp = tempfile.TemporaryDirectory()
|
||
|
|
self.state = str(Path(self.tmp.name) / "state")
|
||
|
|
Path(self.state).mkdir()
|
||
|
|
self.log = str(Path(self.tmp.name) / "log")
|
||
|
|
|
||
|
|
def tearDown(self):
|
||
|
|
self.tmp.cleanup()
|
||
|
|
|
||
|
|
def bash(self, snippet):
|
||
|
|
return _run(self.state, self.log, snippet)
|
||
|
|
|
||
|
|
def test_allows_when_closed(self):
|
||
|
|
r = self.bash("circuit_allows def")
|
||
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
||
|
|
|
||
|
|
def test_allows_below_threshold(self):
|
||
|
|
r = self.bash("circuit_note_restart def fail\n"
|
||
|
|
"circuit_note_restart def fail\n"
|
||
|
|
"circuit_allows def")
|
||
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
||
|
|
self.assertEqual((Path(self.state) / "futile-def").read_text().strip(), "2")
|
||
|
|
|
||
|
|
def test_opens_after_threshold_and_alerts(self):
|
||
|
|
r = self.bash("circuit_note_restart def fail\n"
|
||
|
|
"circuit_note_restart def fail\n"
|
||
|
|
"circuit_note_restart def fail")
|
||
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
||
|
|
self.assertTrue((Path(self.state) / "circuit-def").exists())
|
||
|
|
self.assertIn("ALERT", r.stdout)
|
||
|
|
self.assertIn("ALERT", Path(self.log).read_text())
|
||
|
|
r2 = self.bash("circuit_allows def")
|
||
|
|
self.assertNotEqual(r2.returncode, 0)
|
||
|
|
self.assertIn("CIRCUIT OPEN", Path(self.log).read_text())
|
||
|
|
|
||
|
|
def test_half_open_after_cooldown(self):
|
||
|
|
old = "echo $(( $(date +%%s) - 1900 )) > '%s/circuit-def'" % self.state
|
||
|
|
r = self.bash(old + "\ncircuit_allows def")
|
||
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
||
|
|
self.assertIn("half-open", Path(self.log).read_text())
|
||
|
|
|
||
|
|
def test_ok_resets(self):
|
||
|
|
r = self.bash("circuit_note_restart def fail\n"
|
||
|
|
"circuit_note_restart def fail\n"
|
||
|
|
"circuit_note_restart def fail\n"
|
||
|
|
"circuit_note_restart def ok")
|
||
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
||
|
|
self.assertFalse((Path(self.state) / "futile-def").exists())
|
||
|
|
self.assertFalse((Path(self.state) / "circuit-def").exists())
|
||
|
|
|
||
|
|
def test_per_agent_isolation(self):
|
||
|
|
r = self.bash("circuit_note_restart def fail\n"
|
||
|
|
"circuit_note_restart def fail\n"
|
||
|
|
"circuit_note_restart def fail\n"
|
||
|
|
"circuit_allows pip")
|
||
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main()
|