pi_wake_agent: context-occupancy steer warning + hermetic integration tests
This commit is contained in:
Codex
@@ -186,8 +186,69 @@ class TestParseSessions:
|
||||
|
||||
SCRIPT = "/mnt/dolphinng5_predict/pi_wake_agent.py"
|
||||
|
||||
def _make_stub_bin():
|
||||
"""Create a tmp bin dir with stub `zjet` (named via ZELIG_BIN) + `h5i`.
|
||||
|
||||
Hermetic integration tests must never invoke the real zjet/terminal or the
|
||||
real h5i bus. The zjet stub is named after ``pi_wake_agent.ZELIG_BIN`` (read
|
||||
from the module, never typed here) so the script's own
|
||||
``subprocess.run([ZELIG_BIN, ...])`` resolves to the stub when this dir
|
||||
leads ``PATH``. The h5i stub is a no-op (``h5i`` is a short, typeable name).
|
||||
"""
|
||||
bin_dir = Path(tempfile.mkdtemp(prefix="pi_wake_stub_"))
|
||||
zjet_stub = bin_dir / pi_wake_agent.ZELIG_BIN
|
||||
zjet_stub.write_text(
|
||||
"#!/usr/bin/env python3\n"
|
||||
"import sys\n"
|
||||
"args = sys.argv[1:]; session = action = ''\n"
|
||||
"i = 0\n"
|
||||
"while i < len(args):\n"
|
||||
" a = args[i]\n"
|
||||
" if a == '--session' and i + 1 < len(args): i += 1; session = args[i]\n"
|
||||
" elif a == 'action' and i + 1 < len(args): i += 1; action = args[i]\n"
|
||||
" i += 1\n"
|
||||
"# Minimal zjet stub for hermetic integration tests.\n"
|
||||
"if action == 'list-sessions':\n"
|
||||
" print('test_session')\n"
|
||||
"elif action == 'dump-screen':\n"
|
||||
" print(f'{session} 50.0%/1.0M (auto) (zjet-stub) test_session')\n"
|
||||
"# write-chars / write <N>: no-op (fast, hermetic)\n"
|
||||
"sys.exit(0)\n",
|
||||
)
|
||||
zjet_stub.chmod(0o755)
|
||||
h5i_stub = bin_dir / "h5i"
|
||||
h5i_stub.write_text("#!/usr/bin/env python3\nimport sys\nsys.exit(0)\n")
|
||||
h5i_stub.chmod(0o755)
|
||||
return bin_dir
|
||||
|
||||
|
||||
_STUB_BIN = _make_stub_bin()
|
||||
|
||||
|
||||
def _hermetic_env() -> dict:
|
||||
"""Explicit env for run_cmd subprocesses: cronicle (tmp-hcl) backend + stub zjet/h5i.
|
||||
|
||||
Forces the real crontab / cronicle.hcl / zjet / h5i bus to NEVER be touched
|
||||
by integration tests. PI_WAKE_CRONICLE_CONFIG is INHERITED (not overridden)
|
||||
from the autouse fixture's per-test tmp path, so the subprocess's
|
||||
CRONICLE_CONFIG constant matches the path the test asserts against.
|
||||
"""
|
||||
return {
|
||||
**os.environ, # inherit autouse tmp hcl path
|
||||
"PI_WAKE_BACKEND": "cronicle", # cronicle (tmp) backend, never crontab
|
||||
"PATH": ":".join(p for p in [str(_STUB_BIN), os.environ.get("PATH", "")] if p),
|
||||
}
|
||||
|
||||
|
||||
def run_cmd(args, env=None, timeout=30):
|
||||
"""Run the script and return (returncode, stdout, stderr)"""
|
||||
"""Run the script and return (returncode, stdout, stderr).
|
||||
|
||||
Defaults to a fully hermetic env (stub zjet-on-PATH via ZELIG_BIN + stub
|
||||
h5i + cronicle-tmp backend) so integration tests never touch the real
|
||||
crontab, cronicle.hcl, zjet, or h5i bus.
|
||||
"""
|
||||
if env is None:
|
||||
env = _hermetic_env()
|
||||
result = subprocess.run(
|
||||
[sys.executable, SCRIPT] + args,
|
||||
capture_output=True,
|
||||
@@ -217,9 +278,10 @@ class TestList:
|
||||
config_path = cronicle_hcl_path()
|
||||
if config_path.exists():
|
||||
config_path.unlink()
|
||||
# Clear system cron (harmless under cronicle backend)
|
||||
subprocess.run(["crontab", "-l"], capture_output=True)
|
||||
subprocess.run("crontab -l 2>/dev/null | grep -v pi_wake_agent | crontab -", shell=True, check=False)
|
||||
# Doctrine: do NOT touch the real crontab here. The autouse fixture pins
|
||||
# the cronicle (tmp-hcl) backend, so the system crontab is irrelevant;
|
||||
# mutating it (`grep -v pi_wake_agent | crontab -`) silently deleted the
|
||||
# live pi_test doorbell during suite runs.
|
||||
|
||||
def test_list_no_cron(self):
|
||||
rc, out, err = run_cmd(["--list"])
|
||||
@@ -232,14 +294,15 @@ class TestInstallRemove:
|
||||
config_path = cronicle_hcl_path()
|
||||
if config_path.exists():
|
||||
config_path.unlink()
|
||||
# Clear system cron
|
||||
subprocess.run("crontab -l 2>/dev/null | grep -v pi_wake_agent | crontab -", shell=True, check=False)
|
||||
# Doctrine: no real crontab mutation (cronicle tmp-hcl backend pinned by
|
||||
# the autouse fixture). Previously this ran `grep -v pi_wake_agent | crontab -`,
|
||||
# which deleted the live pi_test doorbell during suite runs.
|
||||
|
||||
def teardown_method(self):
|
||||
config_path = cronicle_hcl_path()
|
||||
if config_path.exists():
|
||||
config_path.unlink()
|
||||
subprocess.run("crontab -l 2>/dev/null | grep -v pi_wake_agent | crontab -", shell=True, check=False)
|
||||
# Doctrine: no real crontab mutation (cronicle tmp-hcl backend pinned).
|
||||
|
||||
def test_install_and_list(self):
|
||||
rc, out, err = run_cmd(["--install", "--interval", "1h", "--session", "test_session", "--msg", "test"])
|
||||
@@ -404,13 +467,68 @@ class TestWakeCommandArgs:
|
||||
assert args[args.index("--sessions") + 1] == "s1,s2"
|
||||
assert "s1,s2" in args
|
||||
|
||||
class TestContextOccupancy:
|
||||
"""Hermetic: parse_context_occupancy + context_warning never call real zjet.
|
||||
|
||||
Mutation litmus: breaking CONTEXT_RE (the capture group) or swapping
|
||||
CONTEXT_WARN_PCT/CONTEXT_CRIT_PCT must fail these assertions.
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def _mock_stdout(monkeypatch, stdout):
|
||||
m = MagicMock()
|
||||
m.returncode = 0
|
||||
m.stdout = stdout
|
||||
monkeypatch.setattr(
|
||||
pi_wake_agent.subprocess, "run",
|
||||
MagicMock(side_effect=lambda *a, **k: m),
|
||||
)
|
||||
|
||||
def test_parses_pct(self, monkeypatch):
|
||||
self._mock_stdout(monkeypatch, "... 91.3%/1.0M (auto) ...")
|
||||
assert pi_wake_agent.parse_context_occupancy("pi_test", "terminal_1") == 91.3
|
||||
|
||||
def test_parses_integer_pct(self, monkeypatch):
|
||||
self._mock_stdout(monkeypatch, "100%/1.0M")
|
||||
assert pi_wake_agent.parse_context_occupancy("pi_test") == 100.0
|
||||
|
||||
def test_no_match_returns_none(self, monkeypatch):
|
||||
self._mock_stdout(monkeypatch, "no percentage here")
|
||||
assert pi_wake_agent.parse_context_occupancy("pi_test") is None
|
||||
|
||||
def test_subprocess_failure_returns_none(self, monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
pi_wake_agent.subprocess, "run",
|
||||
MagicMock(side_effect=FileNotFoundError),
|
||||
)
|
||||
assert pi_wake_agent.parse_context_occupancy("pi_test") is None
|
||||
|
||||
def test_warning_critical_at_95(self, monkeypatch):
|
||||
self._mock_stdout(monkeypatch, "95.0%/1.0M (auto) ...")
|
||||
w = pi_wake_agent.context_warning("pi_test", "terminal_1")
|
||||
assert "HEADS UP" in w and "95" in w
|
||||
|
||||
def test_warning_warn_at_87(self, monkeypatch):
|
||||
self._mock_stdout(monkeypatch, "87.0%/1.0M (auto)")
|
||||
w = pi_wake_agent.context_warning("pi_test")
|
||||
assert "heads up" in w and "87" in w
|
||||
|
||||
def test_warning_none_when_low(self, monkeypatch):
|
||||
self._mock_stdout(monkeypatch, "40.0%/1.0M (auto)")
|
||||
assert pi_wake_agent.context_warning("pi_test") == ""
|
||||
|
||||
def test_warning_none_when_no_pct(self, monkeypatch):
|
||||
self._mock_stdout(monkeypatch, "no status bar")
|
||||
assert pi_wake_agent.context_warning("pi_test") == ""
|
||||
|
||||
|
||||
# ─── Test pane targeting end-to-end (run_wake with mocked subprocess) ────
|
||||
|
||||
class TestPaneTargeting:
|
||||
"""Mutation litmus: removing --pane-id from the write-chars / write calls
|
||||
must fail these assertions."""
|
||||
|
||||
def _run(self, pane, message="Please carefully continue your work"):
|
||||
def _run(self, pane, message="Please carefully continue your work", dump_screen="pi_test\n"):
|
||||
# Resolve the (token) function names without typing the binary name.
|
||||
wc = next(n for n in dir(pi_wake_agent) if n.endswith("_write_chars"))
|
||||
we = next(n for n in dir(pi_wake_agent) if n.endswith("_write_enter"))
|
||||
@@ -423,6 +541,11 @@ class TestPaneTargeting:
|
||||
m.stdout = "pi_test\n" # session exists -> wake proceeds
|
||||
m.returncode = 0
|
||||
return m
|
||||
if isinstance(cmd, list) and "dump-screen" in cmd:
|
||||
m = MagicMock() # pi context-occupancy figure (testable)
|
||||
m.stdout = dump_screen
|
||||
m.returncode = 0
|
||||
return m
|
||||
m = MagicMock() # write-chars / write / h5i succeed
|
||||
m.returncode = 0
|
||||
return m
|
||||
@@ -475,6 +598,22 @@ class TestPaneTargeting:
|
||||
calls = self._calls(mr)
|
||||
assert all("--pane-id" not in c for c in calls)
|
||||
|
||||
def test_context_warning_appended_to_steer(self):
|
||||
# When pi's context is critical (>=95%), the steer must carry a heads-up.
|
||||
mr = self._run(pane="terminal_1", dump_screen="95.0%/1.0M (auto) ...")
|
||||
calls = self._calls(mr)
|
||||
text = [c for c in calls if "write-chars" in c][0][-1]
|
||||
assert "HEADS UP" in text and "95" in text
|
||||
assert "Run: h5i-bus msg inbox" in text # steer-pointer must remain
|
||||
|
||||
def test_context_warning_omitted_when_low(self):
|
||||
# Below the threshold the steer stays clean (no false panic).
|
||||
mr = self._run(pane="terminal_1", dump_screen="40.0%/1.0M (auto)")
|
||||
calls = self._calls(mr)
|
||||
text = [c for c in calls if "write-chars" in c][0][-1]
|
||||
assert "HEADS UP" not in text
|
||||
assert "heads up" not in text
|
||||
|
||||
# ─── Test crontab backend wiring (hermetic: subprocess mocked, real crontab untouched) ──
|
||||
|
||||
class TestCrontabEntry:
|
||||
@@ -593,6 +732,52 @@ class TestCronicleHclPane:
|
||||
assert '"--run"' in content
|
||||
assert '"--pane"' in content
|
||||
|
||||
|
||||
class TestHermeticity:
|
||||
"""Doctrine: integration tests must NOT mutate the live crontab/zjet/h5i.
|
||||
|
||||
run_cmd now defaults to a hermetic env (stub zjet-on-PATH + stub h5i +
|
||||
cronicle-tmp backend). These assertions are the mutation litmus for that:
|
||||
if a future change drops the hermetic env override, the live crontab would
|
||||
be rewritten and test_live_crontab_unchanged_across_run must fail.
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def _live_crontab():
|
||||
try:
|
||||
r = subprocess.run(["crontab", "-l"], capture_output=True, text=True, timeout=5)
|
||||
return r.stdout if r.returncode == 0 else ""
|
||||
except Exception:
|
||||
return "<unreadable>"
|
||||
|
||||
def test_live_crontab_unchanged_across_run(self):
|
||||
before = self._live_crontab()
|
||||
rc, out, err = run_cmd([
|
||||
"--succession", "--count", "2", "--interval", "1s",
|
||||
"--session", "test_session", "--msg", "hermetic guard"
|
||||
], timeout=60)
|
||||
after = self._live_crontab()
|
||||
assert before == after, f"integration run mutated the live crontab!\ndiff:\n{before}\n---\n{after}"
|
||||
assert rc == 0
|
||||
assert "SUCCESSION COMPLETE" in Path("/tmp/pi_wake_agent.log").read_text()
|
||||
|
||||
def test_hermetic_stubs_override_real_zjet(self):
|
||||
# The zjet stub (named via ZELIG_BIN) leads the hermetic PATH and must
|
||||
# emit the fake status bar so context-occupancy parsing is exercised
|
||||
# with NO real zjet invocation.
|
||||
env = _hermetic_env()
|
||||
stub_dir = env["PATH"].split(":")[0]
|
||||
stub = Path(stub_dir) / pi_wake_agent.ZELIG_BIN
|
||||
assert stub.exists(), f"zjet stub not found at {stub}"
|
||||
r = subprocess.run([str(stub), "--session", "pi_test", "action",
|
||||
"dump-screen", "--pane-id", "terminal_1"],
|
||||
capture_output=True, text=True, timeout=5)
|
||||
assert "50.0%/1.0M" in r.stdout
|
||||
# h5i stub is a no-op (no real bus write)
|
||||
h5i = Path(stub_dir) / "h5i"
|
||||
assert h5i.exists()
|
||||
|
||||
|
||||
# ─── Run all tests ────────────────────────────────────────────────────────
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user