From 616a23c0b382f77d5bf8ada29f5730ce44d598d1 Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 15 Sep 2026 11:17:36 +0200 Subject: [PATCH] feat(pi_wake_agent): --at HHMM one-off self-cleaning wake mode --- pi_wake_agent.py | 248 +++++++++++++++++++++++++++++- test_pi_wake_agent.py | 350 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 590 insertions(+), 8 deletions(-) diff --git a/pi_wake_agent.py b/pi_wake_agent.py index f2d2d06..1a74b36 100644 --- a/pi_wake_agent.py +++ b/pi_wake_agent.py @@ -25,7 +25,7 @@ import sys import threading import time from pathlib import Path -from typing import List, Optional +from typing import List, Optional, Tuple # ─── Constants ──────────────────────────────────────────────────────────── SCRIPT_PATH = Path(__file__).resolve() @@ -187,13 +187,18 @@ def cronicle_schedule_name(sessions: List[str], interval: str) -> str: return f"pi_wake_{sessions_str}_{interval}" -def wake_command_args(sessions: List[str], interval: str, message: str, pane: Optional[str] = None) -> List[str]: +def wake_command_args(sessions: List[str], interval: Optional[str], message: str, pane: Optional[str] = None) -> List[str]: """Build the argv list for the --run wake command (shared by both backends). - Always passes --interval so the cronicle/crnd entry is self-describing. - --pane targets a specific zjet pane (e.g. the bottom Soundcheck pane). + ``interval`` is the recurring cadence used by cron/crontab entries; pass + ``None`` for a one-off --run invocation (e.g. --at, which is time-of-day + triggered, not interval driven) so --interval is omitted. --pane targets a + specific zjet pane (e.g. the bottom Soundcheck pane). """ - args = [str(SCRIPT_PATH), "--run", "--interval", interval, "--sessions", ",".join(sessions)] + args = [str(SCRIPT_PATH), "--run"] + if interval is not None: + args += ["--interval", interval] + args += ["--sessions", ",".join(sessions)] if pane: args += ["--pane", pane] args += ["--msg", message] @@ -347,9 +352,199 @@ def remove_cron(sessions: List[str], interval: str, logger: logging.Logger) -> N logger.info(f"Removed cron for {sessions} ({interval})") +# ─── One-off Time-of-Day Scheduling (--at) ─────────────────────────────── +AT_TIME_RE = re.compile(r"^(\d{1,2}):(\d{2})$|^(\d{4})$") + + +def parse_at_time(spec: str) -> Tuple[str, str, str]: + """Parse a one-off wake time-of-day. + + Accepts 'HH:MM' (e.g. '01:50', '1:05') or 'HHMM' (e.g. '0150'). Returns + (hhmm, minute, hour), each zero-padded, e.g. ('0150', '50', '01'). Raises + ValueError on an unparseable or out-of-range time. + """ + m = re.match(r"^(\d{1,2}):(\d{2})$", spec) + if m: + hour, minute = int(m.group(1)), int(m.group(2)) + else: + m = re.match(r"^(\d{4})$", spec) + if not m: + raise ValueError(f"--at expects HH:MM or HHMM (e.g. 01:50 or 0150), got {spec!r}") + hour, minute = int(spec[:2]), int(spec[2:]) + if not (0 <= hour <= 23) or not (0 <= minute <= 59): + raise ValueError(f"--at time out of range: {spec!r} (hour 0-23, minute 0-59)") + return f"{hour:02d}{minute:02d}", f"{minute:02d}", f"{hour:02d}" + + +def cron_at_comment(sessions: List[str], hhmm: str) -> str: + """Crontab/crnd tag for a one-off --at entry (distinct from recurring).""" + sessions_str = ",".join(sessions) + return f"{CRON_COMMENT_PREFIX}:{sessions_str}:at:{hhmm}" + + +def cron_at_schedule_name(sessions: List[str], hhmm: str) -> str: + """Cronicle schedule name for a one-off --at entry (starts with 'pi_wake_').""" + return f"pi_wake_{'_'.join(sessions)}_at_{hhmm}" + + +def _at_wake_command(sessions: List[str], message: str, pane: Optional[str] = None) -> str: + """Shell command string for the --run wake invoked by a one-off --at entry. + + Mirrors install_cron's command construction but omits --interval: a one-off + is time-of-day triggered, not interval driven (and --run ignores --interval + anyway, so omitting it is both cleaner and accurate). + """ + sessions_arg = f"'{','.join(sessions)}'" + cmd = f"{PYTHON_BIN} {SCRIPT_PATH} --run --sessions {sessions_arg} --msg {shlex.quote(message)}" + if pane: + cmd += f" --pane {shlex.quote(pane)}" + return cmd + + +def _strip_lines_by_tag(lines: List[str], tag: str) -> List[str]: + """Drop crontab lines whose comment contains `tag` (idempotent install/remove).""" + return [line for line in lines if tag not in line] + + +def build_at_cron_entry(sessions: List[str], hhmm: str, message: str, pane: Optional[str] = None) -> str: + """Build the full crontab line for a one-off, self-cleaning --at wake. + + Layout: + 'min hour * * * cd && export PATH=... && export H5I_AGENT=... && ; (crontab -l | grep -v '' ; echo) | crontab - # ' + The self-strip suffix removes THIS line (matched by its unique 'at:' + tag) after it fires, so the entry is strictly one-off and never clobbers + unrelated entries (the tag is distinct from the recurring + 'pi_wake_agent::' tag). + """ + _, minute, hour = parse_at_time(hhmm) + comment = cron_at_comment(sessions, hhmm) + wake_cmd = _at_wake_command(sessions, message, pane) + self_strip = f"crontab -l 2>/dev/null | grep -v '{comment}'" + cmd = (f"cd {H5I_BUS_ROOT} && export PATH={CRON_PATH} && export H5I_AGENT={H5I_AGENT} && " + f"{wake_cmd} ; ({self_strip} ; echo) | crontab - # {comment}") + return f"{minute} {hour} * * * {cmd}" + + +def crontab_install_at(sessions: List[str], hhmm: str, message: str, logger: logging.Logger, pane: Optional[str] = None) -> None: + """Install a one-off, self-cleaning crontab entry at the given time-of-day (crnd backend). + + Idempotent: a pre-existing entry with the same 'at:' tag is replaced + rather than duplicated. Unrelated crontab entries are preserved verbatim. + """ + comment = cron_at_comment(sessions, hhmm) + entry = build_at_cron_entry(sessions, hhmm, message, pane) + result = subprocess.run(["crontab", "-l"], capture_output=True, text=True) + existing = result.stdout if result.returncode == 0 else "" + lines = _strip_lines_by_tag(existing.splitlines(), comment) + lines.append(entry) + subprocess.run(["crontab", "-"], input="\n".join(lines) + "\n", text=True, check=True) + logger.info(f"Installed one-off at {hhmm} -> {sessions} (self-cleaning)") + + +def crontab_remove_at(sessions: List[str], hhmm: str, logger: logging.Logger) -> None: + """Remove a pending one-off --at entry from the crontab.""" + comment = cron_at_comment(sessions, hhmm) + result = subprocess.run(["crontab", "-l"], capture_output=True, text=True) + existing = result.stdout if result.returncode == 0 else "" + lines = _strip_lines_by_tag(existing.splitlines(), comment) + new_cron = "\n".join(lines) + ("\n" if lines else "") + subprocess.run(["crontab", "-"], input=new_cron, text=True, check=True) + logger.info(f"Removed one-off at {hhmm} for {sessions}") + + +def cronicle_install_at(sessions: List[str], hhmm: str, message: str, logger: logging.Logger, pane: Optional[str] = None) -> None: + """Install a one-off wake schedule into the cronicle HCL config (tmp-hcl in tests). + + Cronicle's `cron` field accepts the standard 5-field expression, so a + time-of-day one-off is written as 'min hour * * *'. On this host the cronicle + daemon is NOT running (see AGENTS.md §38.10), so these schedules are INERT + and this path is only exercised by hermetic tests asserting the HCL string; + the LIVE one-off scheduler is crontab (crontab_install_at). + """ + _, minute, hour = parse_at_time(hhmm) + schedule_name = cron_at_schedule_name(sessions, hhmm) + config_path = cronicle_config_path() + existing = config_path.read_text() if config_path.exists() else "" + lines = existing.splitlines() + new_lines: List[str] = [] + i = 0 + while i < len(lines): + if f'schedule "{schedule_name}"' in lines[i]: + brace_count = 0 + while i < len(lines): + brace_count += lines[i].count('{') + brace_count -= lines[i].count('}') + i += 1 + if brace_count == 0: + break + continue + new_lines.append(lines[i]) + i += 1 + new_schedule = f'''\nschedule "{schedule_name}" {{ + cron = "{minute} {hour} * * *" + + task "wake" {{ + command = [{", ".join(chr(34) + a + chr(34) for a in wake_command_args(sessions, None, message, pane))}] + }} +}}''' + config_path.write_text("\n".join(new_lines) + "\n\n" + new_schedule + "\n") + logger.info(f"Installed cronicle one-off: {schedule_name} -> {sessions} at {hhmm}") + + +def cronicle_remove_at(sessions: List[str], hhmm: str, logger: logging.Logger) -> None: + """Remove a one-off wake schedule from the cronicle HCL config.""" + schedule_name = cron_at_schedule_name(sessions, hhmm) + config_path = cronicle_config_path() + if not config_path.exists(): + return + existing = config_path.read_text() + lines = existing.splitlines() + new_lines = [] + i = 0 + while i < len(lines): + if f'schedule "{schedule_name}"' in lines[i]: + brace_count = 0 + while i < len(lines): + brace_count += lines[i].count('{') + brace_count -= lines[i].count('}') + i += 1 + if brace_count == 0: + break + continue + new_lines.append(lines[i]) + i += 1 + config_path.write_text("\n".join(new_lines) + "\n") + logger.info(f"Removed cronicle one-off: {schedule_name}") + + +def install_at(sessions: List[str], hhmm: str, message: str, logger: logging.Logger, pane: Optional[str] = None) -> None: + """Install a one-off wake at HH:MM, routing through the configured backend. + + Cronicle (tmp-hcl) when selected (keeps tests hermetic) and the system + crontab (crnd) otherwise. The crontab path is strictly one-off: the entry + self-strips its tag after firing. ``hhmm`` is re-parsed here to validate. + """ + parse_at_time(hhmm) + if use_cronicle(): + cronicle_install_at(sessions, hhmm, message, logger, pane) + else: + crontab_install_at(sessions, hhmm, message, logger, pane) + logger.info(f"Scheduled one-off wake at {hhmm} for {sessions}") + + +def remove_at(sessions: List[str], hhmm: str, logger: logging.Logger) -> None: + """Remove a pending one-off --at wake (routes through the configured backend).""" + parse_at_time(hhmm) + if use_cronicle(): + cronicle_remove_at(sessions, hhmm, logger) + else: + crontab_remove_at(sessions, hhmm, logger) + + def list_cron(logger: logging.Logger) -> None: if use_cronicle(): cronicle_list(logger) + cronicle_list(logger) else: print("=== pi_wake_agent cron entries ===") result = subprocess.run(["crontab", "-l"], capture_output=True, text=True) @@ -609,13 +804,16 @@ def parse_sessions(raw) -> List[str]: # ─── Usage ────────────────────────────────────────────────────────────── def create_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( - description="pi_wake_agent.py — Reusable multi-agent wake-up timer with self-cron/daemon/succession", + description="pi_wake_agent.py — Reusable multi-agent wake-up timer with self-cron/daemon/succession/one-off --at", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" EXAMPLES: # Install recurring 1-hour timer for one session pi_wake_agent.py --install --interval 1h --session cc_UV_dev0_Fb + # One-off wake at 01:50 AM — fires once, then self-removes from crontab + pi_wake_agent.py --at 0150 --session cc_UV_dev0_Fb --msg "continue" + # Install 30-minute timer for multiple sessions pi_wake_agent.py --install --interval 30m --sessions "cc_UV_dev0_Fb,cc_UV_dev1_48" --msg "Wake up!" @@ -639,6 +837,12 @@ EXAMPLES: # Validate sessions pi_wake_agent.py --validate --session cc_UV_dev0_Fb + + # One-off wake at 01:50 AM (fires once, then self-removes from crontab) + pi_wake_agent.py --at 0150 --session cc_UV_dev0_Fb --msg "continue" + + # Cancel a pending one-off + pi_wake_agent.py --remove --at 0150 --session cc_UV_dev0_Fb """ ) parser.add_argument("--install", action="store_const", const="install", dest="mode", help="Install recurring cron timer") @@ -658,6 +862,13 @@ EXAMPLES: parser.add_argument("--msg", default=f"Operator says CONTINUE. {AGENT_NICK} here, saying hi!", help="Wake message") parser.add_argument("--debug", action="store_true", help="Enable debug logging") parser.add_argument("--dry-run", action="store_true", help="Show what would be done without executing") + parser.add_argument("--at", dest="at_time", default=None, metavar="HHMM", + help=("Schedule a ONE-OFF wake at HH:MM or HHMM (e.g. " + "0150 or 01:50). Fires once at that time-of-day then " + "self-removes from the crontab (crnd backend). Combine " + "with --remove to cancel, or --dry-run to preview. " + "Cannot combine with --once/--daemon/--succession/" + "--run/--list/--status/--validate.")) parser.set_defaults(mode="install") return parser @@ -677,6 +888,22 @@ def main() -> int: # Setup logging logger = setup_logging(args.debug) + # --at HHMM: one-off time-of-day wake (self-cleaning). Mutually exclusive + # with --once/--daemon/--succession/--run which own their own timing. + if args.at_time is not None: + if args.mode not in ("install", "remove"): + logger.error("--at HHMM may only be used alone (one-off install) or with " + "--remove; it cannot combine with --once/--daemon/" + "--succession/--run/--list/--status/--validate") + return 1 + try: + parse_at_time(args.at_time) # fail-fast validation before install/remove + except ValueError as e: + logger.error(f"Invalid --at time {args.at_time!r}: {e}") + return 1 + if args.mode == "install": + args.mode = "at" + # Validate if len(sessions) == 0 and args.mode not in ("list", "status"): logger.error("--session or --sessions required") @@ -703,7 +930,9 @@ def main() -> int: return 0 try: - if args.mode == "install": + if args.mode == "at": + install_at(sessions, args.at_time, args.msg, logger, args.pane) + elif args.mode == "install": install_cron(sessions, args.interval, args.msg, logger, args.pane) elif args.mode == "once": run_once(sessions, args.interval, args.msg, logger, args.pane) @@ -714,7 +943,10 @@ def main() -> int: elif args.mode == "run": run_wake(sessions, args.msg, logger, args.pane) elif args.mode == "remove": - remove_cron(sessions, args.interval, logger) + if args.at_time is not None: + remove_at(sessions, args.at_time, logger) + else: + remove_cron(sessions, args.interval, logger) elif args.mode == "list": list_cron(logger) elif args.mode == "status": diff --git a/test_pi_wake_agent.py b/test_pi_wake_agent.py index 79a13e4..0970507 100644 --- a/test_pi_wake_agent.py +++ b/test_pi_wake_agent.py @@ -32,6 +32,16 @@ from pi_wake_agent import ( cron_comment, parse_sessions, wake_command_args, + parse_at_time, + cron_at_comment, + cron_at_schedule_name, + build_at_cron_entry, + crontab_install_at, + crontab_remove_at, + cronicle_install_at, + cronicle_remove_at, + install_at, + remove_at, use_cronicle, cronicle_available, cronicle_daemon_running, @@ -778,6 +788,346 @@ class TestHermeticity: assert h5i.exists() +# ─── Test --at one-off time-of-day scheduling ───────────────────────────── + +class TestParseAtTime: + """Pure: HH:MM / HHMM parsing + range validation (mutation litmus).""" + + def test_hhmm_four_digit(self): + assert parse_at_time("0150") == ("0150", "50", "01") + + def test_hhmm_rejects_three_digits(self): + # HHMM is exactly 4 digits (0150); a bare 930 is ambiguous -> rejected. + with pytest.raises(ValueError): + parse_at_time("930") + + def test_hh_mm_colon(self): + assert parse_at_time("01:50") == ("0150", "50", "01") + assert parse_at_time("9:05") == ("0905", "05", "09") + + def test_midnight_and_last_minute(self): + assert parse_at_time("0000") == ("0000", "00", "00") + assert parse_at_time("2359") == ("2359", "59", "23") + + def test_bad_format(self): + with pytest.raises(ValueError): + parse_at_time("999") # 3 digits, not HHMM + with pytest.raises(ValueError): + parse_at_time("abc") # non-numeric + with pytest.raises(ValueError): + parse_at_time("12:5") # minute not 2 digits (regex is strict) + with pytest.raises(ValueError): + parse_at_time("") # empty + with pytest.raises(ValueError): + parse_at_time("1:50:00") # HH:MM:SS rejected + + def test_out_of_range(self): + with pytest.raises(ValueError): + parse_at_time("2400") # hour 24 + with pytest.raises(ValueError): + parse_at_time("2500") # hour 25 + with pytest.raises(ValueError): + parse_at_time("1260") # minute 60 + with pytest.raises(ValueError): + parse_at_time("12:99") # minute 99 + with pytest.raises(ValueError): + parse_at_time("23:60") + + def test_rejects_extra_digits(self): + with pytest.raises(ValueError): + parse_at_time("01500") # 5 digits + + +class TestCronAtTag: + """Tag + schedule-name generation (distinct from recurring entries).""" + + def test_comment_format(self): + assert cron_at_comment(["cdx_FLIGHT13"], "0150") == "pi_wake_agent:cdx_FLIGHT13:at:0150" + + def test_comment_multiple_sessions(self): + assert cron_at_comment(["a", "b"], "0150") == "pi_wake_agent:a,b:at:0150" + + def test_schedule_name_format(self): + assert cron_at_schedule_name(["cdx_FLIGHT13"], "0150") == "pi_wake_cdx_FLIGHT13_at_0150" + + def test_tag_distinct_from_recurring(self): + recurring = cron_comment(["cdx_FLIGHT13"], "20m") # pi_wake_agent:cdx_FLIGHT13:20m + at_tag = cron_at_comment(["cdx_FLIGHT13"], "0150") # ...:at:0150 + assert "at:" in at_tag + assert at_tag != recurring + assert "at:" not in recurring # recurring tag never looks like an at-tag + + +class TestBuildAtCronEntry: + """Pure: the crontab line for a one-off, self-cleaning --at wake.""" + + def _entry(self, **over): + kw = dict(sessions=["cdx_FLIGHT13"], hhmm="0150", message="continue", pane=None) + kw.update(over) + return build_at_cron_entry(**kw) + + def test_schedule_is_time_of_day(self): + e = self._entry() + assert e.startswith("50 01 * * * "), e # 01:50 -> min=50 hour=01 + + def test_wake_omits_interval(self): + e = self._entry() + assert "--interval" not in e # one-off is not interval driven + + def test_wake_has_run_sessions_and_message(self): + e = self._entry() + assert "--run" in e + assert "cdx_FLIGHT13" in e + assert "continue" in e + + def test_self_strip_targets_the_at_tag_not_recurring(self): + e = self._entry() + at_tag = cron_at_comment(["cdx_FLIGHT13"], "0150") + recurring = cron_comment(["cdx_FLIGHT13"], "20m") + assert f"grep -v '{at_tag}'" in e # strips THIS line + assert f"grep -v '{recurring}'" not in e # never strips recurring + + def test_self_strip_sits_after_wake_cmd(self): + e = self._entry() + wake = pi_wake_agent._at_wake_command(["cdx_FLIGHT13"], "continue") + assert wake in e # wake runs before self-strip + assert e.rstrip().endswith(f"# {cron_at_comment(['cdx_FLIGHT13'], '0150')}") + + def test_pane_threaded_into_command(self): + e = self._entry(pane="terminal_1") + cmd_part = e.split(" # ")[0] + assert "--pane" in cmd_part and "terminal_1" in cmd_part + + def test_pane_default_omitted(self): + e = self._entry(pane=None) + assert "--pane" not in e + + +class TestCrontabAtBackend: + """Hermetic crontab-backend tests for --at (subprocess.run mocked). + + Mirrors TestCrontabEntry: real crontab is never touched; we capture the + `crontab -` stdin to assert the time-of-day schedule, the self-strip suffix, + the at-tag comment, and that the pre-existing doorbell survives. Mutation + litmus: dropping the self-strip suffix must fail test_self_strip_present. + """ + + _DOORBELL = ( + "*/20 * * * * cd /mnt/dolphinng5_predict && export PATH=/usr/local/sbin && " + "/usr/bin/python3 /mnt/dolphinng5_predict/pi_wake_agent.py --run --interval 20m " + "--sessions 'pi_test' --msg 'Please carefully continue your work' --pane terminal_1 " + "# pi_wake_agent:pi_test:20m\n" + ) + + @staticmethod + def _mock_crontab_at(monkeypatch, crontab_before, captured): + def fake_run(cmd, *args, **kwargs): + c = list(cmd) + if c == ["crontab", "-l"]: + m = MagicMock(); m.stdout = crontab_before; m.returncode = 0; return m + if c == ["crontab", "-"]: + captured.append(kwargs.get("input", "")) + m = MagicMock(); m.returncode = 0; return m + m = MagicMock(); m.returncode = 0; return m + monkeypatch.setenv("PI_WAKE_BACKEND", "crontab") + monkeypatch.setattr(pi_wake_agent.subprocess, "run", MagicMock(side_effect=fake_run)) + + def test_install_sets_time_of_day_schedule(self, monkeypatch): + captured = [] + self._mock_crontab_at(monkeypatch, self._DOORBELL, captured) + pi_wake_agent.crontab_install_at(["cdx_FLIGHT13"], "0150", "continue", + logging.getLogger("t"), pane=None) + assert len(captured) == 1, "crontab - not invoked by crontab_install_at" + new = captured[0] + assert "50 01 * * *" in new, new # at-entry is mid-crontab + assert "pi_wake_agent:cdx_FLIGHT13:at:0150" in new + at_line = [ln for ln in new.splitlines() if "pi_wake_agent:cdx_FLIGHT13:at:0150" in ln] + assert at_line and "--interval" not in at_line[0] # at-ENTRY omits --interval (doorbell keeps its 20m) + + def test_install_self_strip_present(self, monkeypatch): + captured = [] + self._mock_crontab_at(monkeypatch, self._DOORBELL, captured) + pi_wake_agent.crontab_install_at(["cdx_FLIGHT13"], "0150", "continue", logging.getLogger("t")) + new = captured[0] + assert "grep -v" in new + assert "crontab -" in new # self-strip rewrites crontab + + def test_install_idempotent_replaces_same_tag(self, monkeypatch): + """Installing twice must NOT duplicate (replace, not append).""" + pre = self._DOORBELL + ( + "50 1 * * * cd /mnt/dolphinng5_predict && ... # pi_wake_agent:cdx_FLIGHT13:at:0150\n" + ) + captured = [] + self._mock_crontab_at(monkeypatch, pre, captured) + pi_wake_agent.crontab_install_at(["cdx_FLIGHT13"], "0150", "continue", logging.getLogger("t")) + new = captured[0] + assert sum(1 for ln in new.splitlines() if "pi_wake_agent:cdx_FLIGHT13:at:0150" in ln) == 1 # one line (old replaced) + assert "pi_wake_agent:pi_test:20m" in new # doorbell intact + + def test_install_preserves_doorbell_and_unrelated(self, monkeypatch): + captured = [] + self._mock_crontab_at(monkeypatch, self._DOORBELL, captured) + pi_wake_agent.crontab_install_at(["cdx_FLIGHT13"], "0150", "continue", logging.getLogger("t")) + new = captured[0] + assert "pi_wake_agent:pi_test:20m" in new # doorbell intact + assert "pi_wake_agent:cdx_FLIGHT13:at:0150" in new # new at-entry present + assert self._DOORBELL.strip() in new # pre-existing line verbatim + + def test_remove_clears_at_entry(self, monkeypatch): + installed = self._DOORBELL + ( + "50 1 * * * cd /mnt/dolphinng5_predict && ... # pi_wake_agent:cdx_FLIGHT13:at:0150\n" + ) + captured = [] + self._mock_crontab_at(monkeypatch, installed, captured) + pi_wake_agent.crontab_remove_at(["cdx_FLIGHT13"], "0150", logging.getLogger("t")) + new = captured[0] + assert "pi_wake_agent:cdx_FLIGHT13:at:0150" not in new + assert "pi_wake_agent:pi_test:20m" in new # doorbell preserved + + def test_remove_missing_tag_noop(self, monkeypatch): + """Removing an absent --at entry leaves the crontab content unchanged.""" + captured = [] + self._mock_crontab_at(monkeypatch, self._DOORBELL, captured) + pi_wake_agent.crontab_remove_at(["nope"], "0150", logging.getLogger("t")) + new = captured[0] + assert "at:0150" not in new + assert new.strip() == self._DOORBELL.strip() + + def test_install_at_routes_to_crontab_when_selected(self, monkeypatch): + captured = [] + self._mock_crontab_at(monkeypatch, self._DOORBELL, captured) # forces crontab backend + pi_wake_agent.install_at(["cdx_FLIGHT13"], "0150", "continue", logging.getLogger("t")) + assert captured and "50 01 * * *" in captured[0] + + +class TestCronicleAtHcl: + """cronicle backend (tmp-hcl): --at writes a one-off schedule block. + + Asserts the generated HCL string (hermetic: tmp config, never the real + cronicle.hcl). Mirrors TestCronicleHslPane's shape + mutation litmus. + """ + + def setup_method(self): + p = cronicle_hcl_path() + if p.exists(): + p.unlink() + + def teardown_method(self): + p = cronicle_hcl_path() + if p.exists(): + p.unlink() + + def test_install_hcl_block(self): + rc, out, err = run_cmd(["--at", "0150", "--session", "cdx_FLIGHT13", + "--msg", "continue"]) + assert rc == 0, f"install --at failed: {err}" + content = cronicle_hcl_path().read_text() + assert 'schedule "pi_wake_cdx_FLIGHT13_at_0150"' in content + assert 'cron = "50 01 * * *"' in content + assert '"--run"' in content + assert '"cdx_FLIGHT13"' in content + assert '"continue"' in content + assert '"--interval"' not in content # one-off omits --interval + assert 'command = [' in content # valid HCL array (mutation litmus) + + def test_install_hcl_has_pane(self): + rc, out, err = run_cmd(["--at", "0150", "--session", "cdx_FLIGHT13", + "--pane", "terminal_1", "--msg", "continue"]) + assert rc == 0, f"install --at failed: {err}" + content = cronicle_hcl_path().read_text() + assert '"--pane"' in content + assert '"terminal_1"' in content + + def test_install_is_idempotent_replaces_block(self): + """Reinstalling the same --at schedule replaces, not duplicates.""" + run_cmd(["--at", "0150", "--session", "cdx_FLIGHT13", "--msg", "continue"]) + run_cmd(["--at", "0150", "--session", "cdx_FLIGHT13", "--msg", "again"]) + content = cronicle_hcl_path().read_text() + assert content.count('schedule "pi_wake_cdx_FLIGHT13_at_0150"') == 1 + assert '"again"' in content + + def test_remove_clears_hcl_block(self): + cronicle_hcl_path().write_text( + 'schedule "pi_wake_cdx_FLIGHT13_at_0150" {\n' + ' cron = "50 1 * * *"\n' + ' task "wake" { command = ["x"] }\n' + '}\n' + 'schedule "pi_wake_keep_20m" {\n' + ' cron = "@every 20m"\n' + ' task "wake" { command = ["y"] }\n' + '}\n' + ) + rc, out, err = run_cmd(["--remove", "--at", "0150", "--session", "cdx_FLIGHT13"]) + assert rc == 0, f"remove --at failed: {err}" + content = cronicle_hcl_path().read_text() + assert 'schedule "pi_wake_cdx_FLIGHT13_at_0150"' not in content + assert 'schedule "pi_wake_keep_20m"' in content # unrelated schedule preserved + + +class TestAtDryRunHermeticity: + """--at --dry-run must not touch crontab, cronicle.hcl, or zjet/h5i.""" + + def setup_method(self): + p = cronicle_hcl_path() + if p.exists(): + p.unlink() + + def teardown_method(self): + p = cronicle_hcl_path() + if p.exists(): + p.unlink() + + def test_dry_run_does_not_install(self): + before_log = Path(LOG_FILE).read_text() if LOG_FILE.exists() else "" + rc, out, err = run_cmd(["--at", "0150", "--session", "cdx_FLIGHT13", + "--msg", "continue", "--dry-run"]) + assert rc == 0 + assert not cronicle_hcl_path().exists() # no HCL written (dry-run => no side effects) + # dry-run branch logs via the FileHandler (DEBUG-level); stderr's StreamHandler + # is WARNING-only, so assert on the log file, not on captured err/out. + after_log = Path(LOG_FILE).read_text() + assert "DRY RUN" in after_log + assert after_log.count("DRY RUN") > before_log.count("DRY RUN") # a new entry was emitted + + def test_dry_run_does_not_touch_live_crontab(self): + before = TestHermeticity._live_crontab() + rc, out, err = run_cmd(["--at", "0150", "--session", "cdx_FLIGHT13", + "--msg", "continue", "--dry-run"]) + after = TestHermeticity._live_crontab() + assert rc == 0 + assert before == after, "dry-run mutated live crontab" + assert "at:0150" not in (after or "") + + +class TestAtLiveHermeticity: + """Mutation litmus: integration runs must NOT mutate the live crontab or + the real cronicle.hcl (the autouse hermetic env routes --at to the cronicle + tmp-hcl backend, so the live crontab is never rewritten).""" + + @staticmethod + def _live_cronicle_hcl(): + p = Path("/mnt/dolphinng5_predict/cronicle.hcl") + return p.read_text() if p.exists() else "" + + def test_live_crontab_unchanged_across_at(self): + before = TestHermeticity._live_crontab() + rc, out, err = run_cmd(["--at", "0150", "--session", "cdx_FLIGHT13", + "--msg", "hermetic guard"]) + after = TestHermeticity._live_crontab() + assert rc == 0 + assert before == after, "integration run mutated the live crontab" + assert "pi_wake_cdx_FLIGHT13_at_0150" not in before # not pre-installed + + def test_live_cronicle_hcl_unchanged_across_at(self): + before = self._live_cronicle_hcl() + rc, out, err = run_cmd(["--at", "0150", "--session", "cdx_FLIGHT13", + "--msg", "hermetic guard"]) + after = self._live_cronicle_hcl() + assert rc == 0 + assert before == after, "integration run mutated real cronicle.hcl" + assert "pi_wake_cdx_FLIGHT13_at_0150" not in before + + # ─── Run all tests ──────────────────────────────────────────────────────── if __name__ == "__main__":