Guard seat test scripts behind __main__; clean _t_* cron output residue

Both harnesses executed top-to-bottom on import, so anything collecting
them (pytest, a glob import) ran live-store side effects. main() guards
plus finally-block cleanup of the _t_* cron output dirs the fabricated
jobs leave behind.
This commit is contained in:
vh
2026-09-19 07:04:15 -07:00
parent 0dc8e9096e
commit 0fe4da64c4
2 changed files with 160 additions and 127 deletions
@@ -3,16 +3,14 @@
Calls cron.monitor.check_monitor and cron.scheduler._apply_monitor_gate directly Calls cron.monitor.check_monitor and cron.scheduler._apply_monitor_gate directly
against fabricated job dicts and throwaway monitor scripts. No live job touched, against fabricated job dicts and throwaway monitor scripts. No live job touched,
no agent woken. Exit codes: 0 = all assertions held, 1 = mismatch. no agent woken. Exit codes: 0 = all assertions held, 1 = mismatch.
Guarded behind __main__: importing this module (e.g. by pytest collection)
must not execute it or touch the live cron store.
""" """
import os, sys, json, tempfile import os, sys, json, tempfile
sys.path.insert(0, os.path.expanduser("~/.hermes/hermes-agent"))
os.environ.setdefault("HERMES_HOME", os.path.expanduser("~/.hermes"))
from cron.monitor import check_monitor
from cron.scheduler import _apply_monitor_gate
SDIR = os.path.expanduser("~/.hermes/scripts") SDIR = os.path.expanduser("~/.hermes/scripts")
results = []
def mkscript(name, body): def mkscript(name, body):
p = os.path.join(SDIR, name) p = os.path.join(SDIR, name)
@@ -21,6 +19,23 @@ def mkscript(name, body):
os.chmod(p, 0o755) os.chmod(p, 0o755)
return os.path.relpath(p, SDIR) return os.path.relpath(p, SDIR)
def fake_job(jid, script, prior_hash=None):
st = {"last_output_hash": prior_hash} if prior_hash else None
return {"id": jid, "name": jid, "monitor_script": script,
"monitor_state": st, "schedule": {"kind": "interval", "minutes": 5}}
def main():
sys.path.insert(0, os.path.expanduser("~/.hermes/hermes-agent"))
os.environ.setdefault("HERMES_HOME", os.path.expanduser("~/.hermes"))
from cron.monitor import check_monitor, hash_monitor_output
from cron.scheduler import _apply_monitor_gate
from cron.jobs import get_job
results = []
# Script S1: emits the sentinel then exit 1 (proposed repaired failure path) # Script S1: emits the sentinel then exit 1 (proposed repaired failure path)
s1 = mkscript("_t_exit1.sh", s1 = mkscript("_t_exit1.sh",
'#!/usr/bin/env bash\necho "PEEK-FAILED rc=2"\nexit 1\n') '#!/usr/bin/env bash\necho "PEEK-FAILED rc=2"\nexit 1\n')
@@ -28,17 +43,12 @@ s1 = mkscript("_t_exit1.sh",
s2 = mkscript("_t_exit0.sh", s2 = mkscript("_t_exit0.sh",
'#!/usr/bin/env bash\necho "PEEK-FAILED rc=2"\nexit 0\n') '#!/usr/bin/env bash\necho "PEEK-FAILED rc=2"\nexit 0\n')
def fake_job(jid, script, prior_hash=None): try:
st = {"last_output_hash": prior_hash} if prior_hash else None
return {"id": jid, "name": jid, "monitor_script": script,
"monitor_state": st, "schedule": {"kind": "interval", "minutes": 5}}
# --- T1: exit 1 -> check_monitor ok=False, no state persisted # --- T1: exit 1 -> check_monitor ok=False, no state persisted
job = fake_job("_t_exit1_job", s1) job = fake_job("_t_exit1_job", s1)
out = check_monitor(job) out = check_monitor(job)
results.append(("T1 exit1 => ok=False", out.ok is False)) results.append(("T1 exit1 => ok=False", out.ok is False))
results.append(("T1 error carries sentinel", "PEEK-FAILED" in (out.error or ""))) results.append(("T1 error carries sentinel", "PEEK-FAILED" in (out.error or "")))
from cron.jobs import get_job
persisted = get_job("_t_exit1_job") persisted = get_job("_t_exit1_job")
results.append(("T1 nothing persisted (no monitor_state on a real store)", results.append(("T1 nothing persisted (no monitor_state on a real store)",
persisted is None or not (persisted.get("monitor_state") or {}).get("last_output_hash"))) persisted is None or not (persisted.get("monitor_state") or {}).get("last_output_hash")))
@@ -49,7 +59,6 @@ out0 = check_monitor(job0)
results.append(("T2 exit0 => ok=True changed=True", out0.ok is True and out0.changed is True)) results.append(("T2 exit0 => ok=True changed=True", out0.ok is True and out0.changed is True))
# simulate what a real store would now hold (job not in the store, so update_job # simulate what a real store would now hold (job not in the store, so update_job
# has nothing to write; fabricate the persisted hash the same way the monitor did) # has nothing to write; fabricate the persisted hash the same way the monitor did)
from cron.monitor import hash_monitor_output
persisted_hash = hash_monitor_output("PEEK-FAILED rc=2") persisted_hash = hash_monitor_output("PEEK-FAILED rc=2")
suppressed = check_monitor(fake_job("_t_exit0_job", s2, prior_hash=persisted_hash)) suppressed = check_monitor(fake_job("_t_exit0_job", s2, prior_hash=persisted_hash))
results.append(("T2 re-run with persisted hash suppressed (outage wakes once per transition)", results.append(("T2 re-run with persisted hash suppressed (outage wakes once per transition)",
@@ -83,12 +92,26 @@ s3 = mkscript("_t_recover.sh", '#!/usr/bin/env bash\necho "[9001]"\nexit 0\n')
early_r, _, ctx_r = _apply_monitor_gate(fake_job("_t_rec_job", s3), "_t_rec_job", "t", None) early_r, _, ctx_r = _apply_monitor_gate(fake_job("_t_rec_job", s3), "_t_rec_job", "t", None)
results.append(("T4 recovery tick passes gate open (early=None, monitor ctx present)", results.append(("T4 recovery tick passes gate open (early=None, monitor ctx present)",
early_r is None and ctx_r is not None)) early_r is None and ctx_r is not None))
os.remove(os.path.join(SDIR, s3)) finally:
# cleanup
for f in (s1, s2): for f in (s1, s2):
try:
os.remove(os.path.join(SDIR, f)) os.remove(os.path.join(SDIR, f))
except FileNotFoundError:
pass
try:
os.remove(os.path.join(SDIR, "_t_recover.sh"))
except FileNotFoundError:
pass
# cron output dirs the fabricated job ids produced (check_monitor persists
# last-output snapshots under ~/.hermes/cron/output/<job id>/)
import shutil
for d in ("_t_exit1_job", "_t_exit0_job", "_t_rec_job"):
shutil.rmtree(os.path.expanduser(f"~/.hermes/cron/output/{d}"), ignore_errors=True)
for name, ok in results: for name, ok in results:
print(("PASS" if ok else "FAIL"), "-", name) print(("PASS" if ok else "FAIL"), "-", name)
sys.exit(0 if all(ok for _, ok in results) else 1) sys.exit(0 if all(ok for _, ok in results) else 1)
if __name__ == "__main__":
main()
@@ -9,8 +9,14 @@ Controls per forseti (2740):
- crash case: persist detection, simulate death before 'read', reload state - crash case: persist detection, simulate death before 'read', reload state
as after restart, rerun with SAME id set -> suppressed (the finding) as after restart, rerun with SAME id set -> suppressed (the finding)
Each condition repeated 3x. Each condition repeated 3x.
Guarded behind __main__: importing this module (e.g. by pytest collection)
must not monkeypatch the live cron.monitor module or execute it.
""" """
import sys, copy import sys, copy
def main():
sys.path.insert(0, "/home/lkraven/.hermes/hermes-agent") sys.path.insert(0, "/home/lkraven/.hermes/hermes-agent")
from cron import monitor from cron import monitor
@@ -85,3 +91,7 @@ for line, passed in results:
ok = ok and passed ok = ok and passed
print("ALL PASS" if ok else "SOME FAILED") print("ALL PASS" if ok else "SOME FAILED")
sys.exit(0 if ok else 1) sys.exit(0 if ok else 1)
if __name__ == "__main__":
main()