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.

scripts/althing-seat-tests/exit1-behavior-test.py

@@ -3,16 +3,14 @@
 Calls cron.monitor.check_monitor and cron.scheduler._apply_monitor_gate directly
 against fabricated job dicts and throwaway monitor scripts. No live job touched,
 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
-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")
-results = []
+
 
 def mkscript(name, body):
     p = os.path.join(SDIR, name)
@@ -21,74 +19,99 @@ def mkscript(name, body):
     os.chmod(p, 0o755)
     return os.path.relpath(p, SDIR)
 
-# Script S1: emits the sentinel then exit 1 (proposed repaired failure path)
-s1 = mkscript("_t_exit1.sh",
-    '#!/usr/bin/env bash\necho "PEEK-FAILED rc=2"\nexit 1\n')
-# Script S2: emits the sentinel then exit 0 (current production behaviour)
-s2 = mkscript("_t_exit0.sh",
-    '#!/usr/bin/env bash\necho "PEEK-FAILED rc=2"\nexit 0\n')
 
 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}}
 
-# --- T1: exit 1 -> check_monitor ok=False, no state persisted
+
-job = fake_job("_t_exit1_job", s1)
+def main():
-out = check_monitor(job)
+    sys.path.insert(0, os.path.expanduser("~/.hermes/hermes-agent"))
-results.append(("T1 exit1 => ok=False", out.ok is False))
+    os.environ.setdefault("HERMES_HOME", os.path.expanduser("~/.hermes"))
-results.append(("T1 error carries sentinel", "PEEK-FAILED" in (out.error or "")))
+
-from cron.jobs import get_job
+    from cron.monitor import check_monitor, hash_monitor_output
-persisted = get_job("_t_exit1_job")
+    from cron.scheduler import _apply_monitor_gate
-results.append(("T1 nothing persisted (no monitor_state on a real store)",
+    from cron.jobs import get_job
+
+    results = []
+
+    # Script S1: emits the sentinel then exit 1 (proposed repaired failure path)
+    s1 = mkscript("_t_exit1.sh",
+        '#!/usr/bin/env bash\necho "PEEK-FAILED rc=2"\nexit 1\n')
+    # Script S2: emits the sentinel then exit 0 (current production behaviour)
+    s2 = mkscript("_t_exit0.sh",
+        '#!/usr/bin/env bash\necho "PEEK-FAILED rc=2"\nexit 0\n')
+
+    try:
+        # --- T1: exit 1 -> check_monitor ok=False, no state persisted
+        job = fake_job("_t_exit1_job", s1)
+        out = check_monitor(job)
+        results.append(("T1 exit1 => ok=False", out.ok is False))
+        results.append(("T1 error carries sentinel", "PEEK-FAILED" in (out.error or "")))
+        persisted = get_job("_t_exit1_job")
+        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")))
 
-# --- T2: exit 0 -> ok=True changed=True (sentinel becomes a persisted hash)
+        # --- T2: exit 0 -> ok=True changed=True (sentinel becomes a persisted hash)
-job0 = fake_job("_t_exit0_job", s2)
+        job0 = fake_job("_t_exit0_job", s2)
-out0 = check_monitor(job0)
+        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)",
                         suppressed.ok is True and suppressed.changed is False))
 
-# --- T3: _apply_monitor_gate on the exit-1 job returns the error early-result
+        # --- T3: _apply_monitor_gate on the exit-1 job returns the error early-result
-early, prompt, ctx = _apply_monitor_gate(fake_job("_t_exit1_job", s1), "_t_exit1_job", "t", None)
+        early, prompt, ctx = _apply_monitor_gate(fake_job("_t_exit1_job", s1), "_t_exit1_job", "t", None)
-ok_gate = (early is not None and early[0] is False and "source failed" in early[1])
+        ok_gate = (early is not None and early[0] is False and "source failed" in early[1])
-results.append(("T3 gate early-returns failure alert, no agent run", ok_gate))
+        results.append(("T3 gate early-returns failure alert, no agent run", ok_gate))
-# repeated failing tick: gate must alert AGAIN (nothing persisted to suppress against)
+        # repeated failing tick: gate must alert AGAIN (nothing persisted to suppress against)
-early2, _, _ = _apply_monitor_gate(fake_job("_t_exit1_job", s1), "_t_exit1_job", "t", None)
+        early2, _, _ = _apply_monitor_gate(fake_job("_t_exit1_job", s1), "_t_exit1_job", "t", None)
-results.append(("T3 second failing tick alerts again (no dedup state)",
+        results.append(("T3 second failing tick alerts again (no dedup state)",
                         early2 is not None and early2[0] is False))
 
-# --- T4: sustained-failure then recovery: no backoff, no auto-pause, job keeps firing.
+        # --- T4: sustained-failure then recovery: no backoff, no auto-pause, job keeps firing.
-# Source claim (svos-dev): the monitor-failure branch early-returns only;
+        # Source claim (svos-dev): the monitor-failure branch early-returns only;
-# _block_and_pause_job is unreachable from it. Behavioral close: N consecutive
+        # _block_and_pause_job is unreachable from it. Behavioral close: N consecutive
-# failing ticks each alert (no dedup/pause), a recovery tick passes the gate
+        # failing ticks each alert (no dedup/pause), a recovery tick passes the gate
-# open (early=None), and the job is never left paused/absent-of-schedule.
+        # open (early=None), and the job is never left paused/absent-of-schedule.
-fail_alerts = 0
+        fail_alerts = 0
-for _ in range(5):
+        for _ in range(5):
             early_t, _, ctx_t = _apply_monitor_gate(fake_job("_t_exit1_job", s1), "_t_exit1_job", "t", None)
             if early_t is not None and early_t[0] is False:
                 fail_alerts += 1
-results.append(("T4 5 consecutive failing ticks each alert (no backoff/disable)", fail_alerts == 5))
+        results.append(("T4 5 consecutive failing ticks each alert (no backoff/disable)", fail_alerts == 5))
-after = get_job("_t_exit1_job")
+        after = get_job("_t_exit1_job")
-results.append(("T4 job not paused/disabled by failures (absent or state!=paused)",
+        results.append(("T4 job not paused/disabled by failures (absent or state!=paused)",
                         after is None or after.get("state") not in ("paused", "blocked")))
-# recovery: a script that succeeds with a NEW output passes the gate open
+        # recovery: a script that succeeds with a NEW output passes the gate open
-s3 = mkscript("_t_recover.sh", '#!/usr/bin/env bash\necho "[9001]"\nexit 0\n')
+        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))
-os.remove(os.path.join(SDIR, s3))
+    finally:
-
+        for f in (s1, s2):
-# cleanup
+            try:
-for f in (s1, s2):
                 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)
-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()

scripts/althing-seat-tests/hash_gate_repro.py

@@ -9,28 +9,34 @@ Controls per forseti (2740):
   - crash case: persist detection, simulate death before 'read', reload state
     as after restart, rerun with SAME id set -> suppressed (the finding)
 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
-sys.path.insert(0, "/home/lkraven/.hermes/hermes-agent")
-from cron import monitor
 
-STATE = {}  # in-memory stand-in for jobs.json monitor_state
 
-def fake_persist(job_id, new_hash, output):
+def main():
+    sys.path.insert(0, "/home/lkraven/.hermes/hermes-agent")
+    from cron import monitor
+
+    STATE = {}  # in-memory stand-in for jobs.json monitor_state
+
+    def fake_persist(job_id, new_hash, output):
         STATE[job_id] = {"last_output_hash": new_hash, "snapshot": output}
 
-monitor._persist_monitor_state = fake_persist
+    monitor._persist_monitor_state = fake_persist
-monitor._read_last_output = lambda job_id: STATE.get(job_id, {}).get("snapshot", "")
+    monitor._read_last_output = lambda job_id: STATE.get(job_id, {}).get("snapshot", "")
-monitor._write_last_output = lambda job_id, output: STATE[job_id].__setitem__("snapshot", output)
+    monitor._write_last_output = lambda job_id, output: STATE[job_id].__setitem__("snapshot", output)
 
-OUTPUT = [""]  # current observed monitor output, set per tick
+    OUTPUT = [""]  # current observed monitor output, set per tick
 
-def fake_source(job):
+    def fake_source(job):
         return True, OUTPUT[0]
 
-monitor._run_monitor_source = fake_source
+    monitor._run_monitor_source = fake_source
 
-def tick(ids):
+    def tick(ids):
         """One monitor tick with the given unread id set. Returns (ok, changed)."""
         OUTPUT[0] = "\n".join(f"[{i}]" for i in ids) + "\n"
         job = {"id": "disposable-test-job", "monitor_script": "stub"}
@@ -39,8 +45,8 @@ def tick(ids):
         out = monitor.check_monitor(job)
         return out.ok, getattr(out, "changed", None)
 
-results = []
+    results = []
-for rep in range(1, 4):
+    for rep in range(1, 4):
         # --- consumption control: a SUCCESSFUL run consumes -> output genuinely changes.
         # (Was: same ids after success, which only proved unchanged-input suppression.
         #  Fix: after the run consumes, the mailbox is empty, so the next observation
@@ -79,9 +85,13 @@ for rep in range(1, 4):
         r4 = tick(["200", "201", "202"])        # a genuinely new message re-wakes,
         results.append((f"rep{rep} recovery-on-new-mail: {r4}", r4 == (True, True)))
 
-ok = True
+    ok = True
-for line, passed in results:
+    for line, passed in results:
         print(("PASS " if passed else "FAIL ") + line)
         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()