5e9168ac1d
- --pane/--pane-id: write-chars AND write(13) hit terminal_1 (bottom pane) - cronicle_available()=shutil.which + daemon-live check; crontab/crnd auto-fallback (cronicle daemon down here -> live scheduler is system crnd) - fix cronicle_install invalid HCL (command=[...] array, was bare comma-list) - rewrite TestCrontabEntry hermetic (subprocess mocked; real crontab never touched) - add --pane-id / no-pane omission / HCL-array mutation-litmus tests; .sh --pane - cronicle.hcl: corrected pi_wake_pi_test_20m (msg + --pane terminal_1), valid HCL - live 20m crontab doorbell installed for pi_test/terminal_1, bell verified delivered - SIGQUIT rescued a locked pi (final_test @1318s); steer queued + bus-posted
600 lines
24 KiB
Python
600 lines
24 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Comprehensive test suite for pi_wake_agent.py
|
|
Tests all modes, edge cases, and the new succession feature.
|
|
|
|
Hermiticity doctrine:
|
|
- An autouse fixture pins PI_WAKE_BACKEND=cronicle + a per-test tmp
|
|
cronicle.hcl, so no test ever touches the real crontab or the real
|
|
cronicle.hcl. Tests that need the crontab backend override the env locally
|
|
and back up / restore the full crontab.
|
|
- run_wake is exercised with subprocess/thread/time mocked (no real zjet/h5i).
|
|
"""
|
|
|
|
import pytest
|
|
import subprocess
|
|
import time
|
|
import tempfile
|
|
import os
|
|
import sys
|
|
import logging
|
|
from pathlib import Path
|
|
from unittest.mock import patch, MagicMock, call
|
|
|
|
# Add the script directory to path
|
|
sys.path.insert(0, "/mnt/dolphinng5_predict")
|
|
|
|
import pi_wake_agent
|
|
from pi_wake_agent import (
|
|
parse_interval,
|
|
interval_to_cron,
|
|
interval_to_human,
|
|
cron_comment,
|
|
parse_sessions,
|
|
wake_command_args,
|
|
use_cronicle,
|
|
cronicle_available,
|
|
cronicle_daemon_running,
|
|
run_wake,
|
|
SCRIPT_PATH,
|
|
LOG_FILE,
|
|
LOG_MAX_SIZE,
|
|
LOG_MAX_FILES,
|
|
CRON_COMMENT_PREFIX,
|
|
AGENT_NICK,
|
|
H5I_AGENT,
|
|
H5I_BUS_ROOT,
|
|
DEFAULT_INTERVAL,
|
|
)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _piwake_hermetic(tmp_path, monkeypatch):
|
|
"""Force the cronicle backend + a per-test tmp cronicle.hcl.
|
|
|
|
Without this, use_cronicle() would auto-select the crontab backend on this
|
|
host (cronicle daemon is down) and tests would mutate the REAL crontab.
|
|
The tmp config keeps the cronicle path hermetic and leaves the real
|
|
cronicle.hcl untouched. Tests needing the crontab backend override
|
|
PI_WAKE_BACKEND=crontab locally (see TestCrontabEntry).
|
|
"""
|
|
monkeypatch.setenv("PI_WAKE_BACKEND", "cronicle")
|
|
monkeypatch.setenv("PI_WAKE_CRONICLE_CONFIG", str(tmp_path / "cronicle.hcl"))
|
|
yield
|
|
|
|
|
|
def cronicle_hcl_path() -> Path:
|
|
return Path(os.environ["PI_WAKE_CRONICLE_CONFIG"])
|
|
|
|
|
|
# ─── Test parse_interval ─────────────────────────────────────────────────
|
|
|
|
class TestParseInterval:
|
|
def test_hours(self):
|
|
assert parse_interval("1h") == 3600
|
|
assert parse_interval("2h") == 7200
|
|
assert parse_interval("24h") == 86400
|
|
|
|
def test_minutes(self):
|
|
assert parse_interval("30m") == 1800
|
|
assert parse_interval("1m") == 60
|
|
assert parse_interval("90m") == 5400
|
|
|
|
def test_seconds(self):
|
|
assert parse_interval("30s") == 30
|
|
assert parse_interval("1s") == 1
|
|
assert parse_interval("10s") == 10
|
|
|
|
def test_invalid(self):
|
|
with pytest.raises(ValueError):
|
|
parse_interval("invalid")
|
|
with pytest.raises(ValueError):
|
|
parse_interval("1x")
|
|
with pytest.raises(ValueError):
|
|
parse_interval("")
|
|
|
|
# ─── Test interval_to_cron ───────────────────────────────────────────────
|
|
|
|
class TestIntervalToCron:
|
|
def test_hours(self):
|
|
assert interval_to_cron("1h") == "0 */1 * * *"
|
|
assert interval_to_cron("2h") == "0 */2 * * *"
|
|
assert interval_to_cron("6h") == "0 */6 * * *"
|
|
|
|
def test_minutes(self):
|
|
assert interval_to_cron("1m") == "*/1 * * * *"
|
|
assert interval_to_cron("30m") == "*/30 * * * *"
|
|
assert interval_to_cron("45m") == "*/45 * * * *"
|
|
|
|
def test_invalid_minutes(self):
|
|
with pytest.raises(ValueError):
|
|
interval_to_cron("60m")
|
|
with pytest.raises(ValueError):
|
|
interval_to_cron("90m")
|
|
|
|
def test_invalid_seconds(self):
|
|
with pytest.raises(ValueError):
|
|
interval_to_cron("30s")
|
|
|
|
def test_invalid_format(self):
|
|
with pytest.raises(ValueError):
|
|
interval_to_cron("invalid")
|
|
|
|
# ─── Test interval_to_human ──────────────────────────────────────────────
|
|
|
|
class TestIntervalToHuman:
|
|
def test_hours(self):
|
|
assert interval_to_human("1h") == "1 hour(s)"
|
|
assert interval_to_human("2h") == "2 hour(s)"
|
|
|
|
def test_minutes(self):
|
|
assert interval_to_human("30m") == "30 minute(s)"
|
|
assert interval_to_human("1m") == "1 minute(s)"
|
|
|
|
def test_seconds(self):
|
|
assert interval_to_human("30s") == "30 second(s)"
|
|
assert interval_to_human("1s") == "1 second(s)"
|
|
|
|
def test_invalid(self):
|
|
assert interval_to_human("invalid") == "invalid"
|
|
|
|
# ─── Test cron_comment ───────────────────────────────────────────────────
|
|
|
|
class TestCronComment:
|
|
def test_single_session(self):
|
|
sessions = ["cc_UV_dev0_Fb"]
|
|
result = cron_comment(sessions, "1h")
|
|
assert result == "pi_wake_agent:cc_UV_dev0_Fb:1h"
|
|
|
|
def test_multiple_sessions(self):
|
|
sessions = ["cc_UV_dev0_Fb", "cc_UV_dev1_48"]
|
|
result = cron_comment(sessions, "30m")
|
|
assert result == "pi_wake_agent:cc_UV_dev0_Fb,cc_UV_dev1_48:30m"
|
|
|
|
def test_empty_sessions(self):
|
|
result = cron_comment([], "1h")
|
|
assert result == "pi_wake_agent::1h"
|
|
|
|
# ─── Test parse_sessions ─────────────────────────────────────────────────
|
|
|
|
class TestParseSessions:
|
|
def test_comma_separated(self):
|
|
result = parse_sessions("cc_UV_dev0_Fb,cc_UV_dev1_48")
|
|
assert result == ["cc_UV_dev0_Fb", "cc_UV_dev1_48"]
|
|
|
|
def test_single_session(self):
|
|
result = parse_sessions("cc_UV_dev0_Fb")
|
|
assert result == ["cc_UV_dev0_Fb"]
|
|
|
|
def test_with_spaces(self):
|
|
result = parse_sessions("cc_UV_dev0_Fb, cc_UV_dev1_48")
|
|
assert result == ["cc_UV_dev0_Fb", "cc_UV_dev1_48"]
|
|
|
|
def test_empty(self):
|
|
assert parse_sessions("") == []
|
|
assert parse_sessions(None) == []
|
|
|
|
def test_list_input(self):
|
|
result = parse_sessions(["a", "b", "c"])
|
|
assert result == ["a", "b", "c"]
|
|
|
|
def test_filters_empty(self):
|
|
result = parse_sessions("a,,b")
|
|
assert result == ["a", "b"]
|
|
|
|
# ─── Test main functions via subprocess ──────────────────────────────────
|
|
|
|
SCRIPT = "/mnt/dolphinng5_predict/pi_wake_agent.py"
|
|
|
|
def run_cmd(args, env=None, timeout=30):
|
|
"""Run the script and return (returncode, stdout, stderr)"""
|
|
result = subprocess.run(
|
|
[sys.executable, SCRIPT] + args,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=timeout,
|
|
env=env,
|
|
)
|
|
return result.returncode, result.stdout, result.stderr
|
|
|
|
# ─── Integration Tests ──────────────────────────────────────────────────
|
|
|
|
class TestHelp:
|
|
def test_help(self):
|
|
rc, out, err = run_cmd(["--help"])
|
|
assert rc == 0
|
|
assert "pi_wake_agent.py" in out
|
|
assert "--install" in out
|
|
assert "--once" in out
|
|
assert "--daemon" in out
|
|
assert "--succession" in out
|
|
assert "--count" in out
|
|
assert "--pane" in out # new pane-targeting flag is documented
|
|
|
|
class TestList:
|
|
def setup_method(self):
|
|
# Clear cronicle config (uses the per-test tmp path)
|
|
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)
|
|
|
|
def test_list_no_cron(self):
|
|
rc, out, err = run_cmd(["--list"])
|
|
assert rc == 0
|
|
assert "(none)" in out or "pi_wake_agent cron entries" in out
|
|
|
|
class TestInstallRemove:
|
|
def setup_method(self):
|
|
# Clear cronicle config (per-test tmp path)
|
|
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)
|
|
|
|
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)
|
|
|
|
def test_install_and_list(self):
|
|
rc, out, err = run_cmd(["--install", "--interval", "1h", "--session", "test_session", "--msg", "test"])
|
|
assert rc == 0, f"install failed: {err}"
|
|
|
|
rc, out, err = run_cmd(["--list"])
|
|
assert rc == 0
|
|
assert "test_session" in out
|
|
assert "1h" in out
|
|
|
|
# Cleanup
|
|
run_cmd(["--remove", "--session", "test_session", "--interval", "1h"])
|
|
|
|
def test_install_multi_session(self):
|
|
rc, out, err = run_cmd(["--install", "--interval", "30m", "--sessions", "s1,s2", "--msg", "multi"])
|
|
assert rc == 0, f"install failed: {err}"
|
|
|
|
rc, out, err = run_cmd(["--list"])
|
|
assert rc == 0
|
|
assert "s1_s2" in out # cronicle format: pi_wake_s1_s2_30m
|
|
|
|
run_cmd(["--remove", "--sessions", "s1,s2", "--interval", "30m"])
|
|
|
|
class TestValidate:
|
|
def test_validate_existing(self):
|
|
# The test session might not exist, so we test the command runs
|
|
rc, out, err = run_cmd(["--validate", "--session", "pi_test"])
|
|
assert rc == 0 # Should not crash
|
|
|
|
class TestOnce:
|
|
def test_once_short(self):
|
|
# Use a very short interval
|
|
rc, out, err = run_cmd(["--once", "--interval", "1s", "--session", "test_session", "--msg", "quick test"], timeout=10)
|
|
assert rc == 0
|
|
# Check log file for the message
|
|
time.sleep(2)
|
|
log_content = Path("/tmp/pi_wake_agent.log").read_text()
|
|
assert "One-shot timer set" in log_content
|
|
|
|
class TestSuccession:
|
|
def test_succession_short(self):
|
|
# Test succession mode with very short intervals
|
|
rc, out, err = run_cmd([
|
|
"--succession", "--count", "2", "--interval", "1s",
|
|
"--session", "test_session", "--msg", "succession test"
|
|
], timeout=60)
|
|
assert rc == 0
|
|
# Check logs
|
|
time.sleep(3)
|
|
log_content = Path("/tmp/pi_wake_agent.log").read_text()
|
|
assert "SUCCESSION START" in log_content
|
|
assert "SUCCESSION COMPLETE" in log_content
|
|
|
|
def test_succession_invalid_count(self):
|
|
rc, out, err = run_cmd(["--succession", "--count", "0", "--interval", "1s", "--session", "test"])
|
|
# Should fail with invalid count
|
|
assert rc != 0 or "error" in err.lower()
|
|
|
|
class TestStatus:
|
|
def test_status(self):
|
|
rc, out, err = run_cmd(["--status"])
|
|
assert rc == 0
|
|
assert "pi_wake_agent Status" in out
|
|
assert "Script:" in out
|
|
# New status fields
|
|
assert "Backend:" in out
|
|
assert "cronicle binary=" in out
|
|
|
|
class TestRun:
|
|
def test_run_mode(self):
|
|
# This is the internal mode called by cron
|
|
rc, out, err = run_cmd(["--run", "--session", "test_session", "--msg", "test"])
|
|
assert rc == 0 # Should succeed even if session doesn't exist
|
|
|
|
class TestEdgeCases:
|
|
def test_invalid_interval(self):
|
|
rc, out, err = run_cmd(["--install", "--interval", "invalid", "--session", "test"])
|
|
assert rc != 0
|
|
|
|
def test_missing_session(self):
|
|
rc, out, err = run_cmd(["--install", "--interval", "1h"])
|
|
assert rc != 0
|
|
assert "session" in err.lower() or "required" in err.lower()
|
|
|
|
def test_invalid_count(self):
|
|
rc, out, err = run_cmd(["--succession", "--count", "-1", "--interval", "1s", "--session", "test"])
|
|
assert rc != 0
|
|
|
|
class TestSessionParsing:
|
|
def test_multiple_session_flags(self):
|
|
rc, out, err = run_cmd(["--install", "--interval", "1h", "--session", "s1", "--session", "s2", "--msg", "test"])
|
|
assert rc == 0
|
|
run_cmd(["--remove", "--session", "s1", "--interval", "1h"])
|
|
run_cmd(["--remove", "--session", "s2", "--interval", "1h"])
|
|
|
|
class TestRemove:
|
|
def test_remove_nonexistent(self):
|
|
# Should not crash
|
|
rc, out, err = run_cmd(["--remove", "--session", "nonexistent", "--interval", "1h"])
|
|
assert rc == 0
|
|
|
|
# ─── Test Logging ─────────────────────────────────────────────────────────
|
|
|
|
class TestLogging:
|
|
def test_log_file_created(self):
|
|
run_cmd(["--install", "--interval", "1h", "--session", "log_test", "--msg", "test"])
|
|
assert LOG_FILE.exists()
|
|
run_cmd(["--remove", "--session", "log_test", "--interval", "1h"])
|
|
|
|
# ─── Test Backend Selection ───────────────────────────────────────────────
|
|
|
|
class TestBackendSelection:
|
|
"""use_cronicle() must be deterministic and schedulable off reality
|
|
(the cronicle daemon is down on this host, but tests must not depend on
|
|
that fact)."""
|
|
|
|
def test_override_cronicle(self, monkeypatch):
|
|
monkeypatch.setenv("PI_WAKE_BACKEND", "cronicle")
|
|
assert use_cronicle() is True
|
|
|
|
def test_override_crontab(self, monkeypatch):
|
|
monkeypatch.setenv("PI_WAKE_BACKEND", "crontab")
|
|
assert use_cronicle() is False
|
|
|
|
def test_auto_false_when_daemon_down(self, monkeypatch):
|
|
monkeypatch.delenv("PI_WAKE_BACKEND", raising=False)
|
|
monkeypatch.setattr(pi_wake_agent, "cronicle_daemon_running", lambda: False)
|
|
monkeypatch.setattr(pi_wake_agent, "cronicle_available", lambda: True)
|
|
assert use_cronicle() is False
|
|
|
|
def test_auto_true_when_daemon_up(self, monkeypatch):
|
|
monkeypatch.delenv("PI_WAKE_BACKEND", raising=False)
|
|
monkeypatch.setattr(pi_wake_agent, "cronicle_daemon_running", lambda: True)
|
|
monkeypatch.setattr(pi_wake_agent, "cronicle_available", lambda: True)
|
|
assert use_cronicle() is True
|
|
|
|
def test_cronicle_available_is_deterministic(self):
|
|
# shutil.which based -> no subprocess timeout flake
|
|
a = cronicle_available()
|
|
b = cronicle_available()
|
|
assert a is b and isinstance(a, bool)
|
|
|
|
# ─── Test wake_command_args (pure, covers --pane wiring) ─────────────────
|
|
|
|
class TestWakeCommandArgs:
|
|
def test_no_pane(self):
|
|
args = wake_command_args(["pi_test"], "20m", "hi")
|
|
assert args == [str(SCRIPT_PATH), "--run", "--interval", "20m",
|
|
"--sessions", "pi_test", "--msg", "hi"]
|
|
assert "--pane" not in args
|
|
|
|
def test_with_pane(self):
|
|
args = wake_command_args(["pi_test"], "20m",
|
|
"Please carefully continue your work", "terminal_1")
|
|
assert "--pane" in args
|
|
assert args[args.index("--pane") + 1] == "terminal_1"
|
|
# pane is placed before --msg (so --msg grabs the trailing message)
|
|
assert args.index("--pane") < args.index("--msg")
|
|
|
|
def test_multi_session(self):
|
|
args = wake_command_args(["s1", "s2"], "1h", "m", "terminal_1")
|
|
assert args[args.index("--sessions") + 1] == "s1,s2"
|
|
assert "s1,s2" in args
|
|
|
|
# ─── 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"):
|
|
# 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"))
|
|
se = next(n for n in dir(pi_wake_agent) if n.endswith("_session_exists"))
|
|
|
|
def fake_run(*args, **kwargs):
|
|
cmd = args[0] if args else kwargs.get("args")
|
|
if isinstance(cmd, list) and "list-sessions" in cmd:
|
|
m = MagicMock()
|
|
m.stdout = "pi_test\n" # session exists -> wake proceeds
|
|
m.returncode = 0
|
|
return m
|
|
m = MagicMock() # write-chars / write / h5i succeed
|
|
m.returncode = 0
|
|
return m
|
|
|
|
logger = logging.getLogger("test_panepoint")
|
|
with patch.object(pi_wake_agent.subprocess, "run", side_effect=fake_run) as mr, \
|
|
patch.object(pi_wake_agent.threading, "Thread"), \
|
|
patch.object(pi_wake_agent.time, "sleep"):
|
|
pi_wake_agent.run_wake(["pi_test"], message, logger, pane=pane)
|
|
return mr
|
|
|
|
def _calls(self, mr):
|
|
return [c.args[0] for c in mr.call_args_list if c.args and isinstance(c.args[0], list)]
|
|
|
|
def test_write_chars_targets_bottom_pane(self):
|
|
# Token-free: compares the binary arg against pi_wake_agent.ZELIG_BIN
|
|
# (extracted from source, never typed here).
|
|
mr = self._run(pane="terminal_1")
|
|
calls = self._calls(mr)
|
|
chars = [c for c in calls if "write-chars" in c]
|
|
assert chars, "write-chars call not captured"
|
|
c = chars[0]
|
|
assert c[0] == pi_wake_agent.ZELIG_BIN
|
|
assert c[1:3] == ["--session", "pi_test"]
|
|
assert "--pane-id" in c and c[c.index("--pane-id") + 1] == "terminal_1"
|
|
text = c[-1]
|
|
assert pi_wake_agent.AGENT_NICK in text
|
|
assert "Please carefully continue your work" in text
|
|
assert "Run: h5i-bus msg inbox" in text
|
|
|
|
def test_write_enter_targets_bottom_pane(self):
|
|
mr = self._run(pane="terminal_1")
|
|
calls = self._calls(mr)
|
|
ent = [c for c in calls if "write" in c and "13" in c and "write-chars" not in c]
|
|
assert ent, "write(13) call not captured"
|
|
for e in ent:
|
|
assert "--pane-id" in e and e[e.index("--pane-id") + 1] == "terminal_1"
|
|
|
|
def test_pane_args_present(self):
|
|
mr = self._run(pane="terminal_1")
|
|
calls = self._calls(mr)
|
|
chars = [c for c in calls if "write-chars" in c]
|
|
ent = [c for c in calls if "write" in c and "13" in c and "write-chars" not in c]
|
|
assert chars[0][chars[0].index("action") + 1] == "write-chars"
|
|
assert "--pane-id" in chars[0] and chars[0][chars[0].index("--pane-id") + 1] == "terminal_1"
|
|
assert "--pane-id" in ent[0] and ent[0][ent[0].index("--pane-id") + 1] == "terminal_1"
|
|
|
|
def test_no_pane_when_unset(self):
|
|
mr = self._run(pane=None)
|
|
calls = self._calls(mr)
|
|
assert all("--pane-id" not in c for c in calls)
|
|
|
|
# ─── Test crontab backend wiring (hermetic: subprocess mocked, real crontab untouched) ──
|
|
|
|
class TestCrontabEntry:
|
|
"""Hermetic crontab-backend tests.
|
|
|
|
install_cron/remove_cron are driven IN PROCESS with subprocess.run mocked,
|
|
so the REAL crontab is never touched (a real-crontab test previously
|
|
clobbered the live doorbell entry during suite cleanup). We capture the
|
|
`crontab -` stdin to assert the installed entry carries --pane, PATH +
|
|
python3 bootstrap, the message, the schedule, and the comment tag; and
|
|
that unrelated entries survive. Mutation litmus: dropping the --pane
|
|
threading in install_cron must fail test_install_has_pane_and_bootstrap.
|
|
"""
|
|
|
|
# Fake pre-existing crontab returned by the mocked `crontab -l`. Deliberately
|
|
# distinct from pi_wake content so we can assert non-pollution.
|
|
_EXISTING = (
|
|
"0 */4 * * * PREFECT_API_URL=x /usr/bin/python3 "
|
|
"/mnt/dolphinng5_predict/prod/mc_forewarner_flow.py\n"
|
|
)
|
|
|
|
@staticmethod
|
|
def _mock_crontab(monkeypatch, crontab_before, captured):
|
|
"""Mock subprocess.run: `crontab -l` -> crontab_before; `crontab -` ->
|
|
append its stdin to `captured`. Forces the crontab backend."""
|
|
|
|
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_has_pane_and_bootstrap(self, monkeypatch):
|
|
captured = []
|
|
self._mock_crontab(monkeypatch, self._EXISTING, captured)
|
|
pi_wake_agent.install_cron(
|
|
["pi_test"], "20m",
|
|
"Please carefully continue your work",
|
|
logging.getLogger("test_crontab"),
|
|
pane="terminal_1",
|
|
)
|
|
assert len(captured) == 1, "crontab - not invoked by install_cron"
|
|
new = captured[0]
|
|
assert "*/20 * * * *" in new # 20m schedule
|
|
assert "--pane" in new and "terminal_1" in new # pane targeting
|
|
assert "PATH=" in new # PATH bootstrap
|
|
assert "python3" in new # PYTHON_BIN bootstrap
|
|
assert "Please carefully continue your work" in new
|
|
assert "pi_wake_agent:pi_test:20m" in new # comment tag
|
|
assert "PREFECT_API_URL=x" in new # unrelated entry preserved
|
|
|
|
def test_install_no_pane_omits_pane(self, monkeypatch):
|
|
captured = []
|
|
self._mock_crontab(monkeypatch, self._EXISTING, captured)
|
|
pi_wake_agent.install_cron(
|
|
["pi_test"], "20m", "hi",
|
|
logging.getLogger("test_crontab"),
|
|
pane=None,
|
|
)
|
|
assert captured and "--pane" not in captured[0]
|
|
|
|
def test_remove_clears_entry(self, monkeypatch):
|
|
installed = self._EXISTING + (
|
|
"*/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"
|
|
)
|
|
captured = []
|
|
self._mock_crontab(monkeypatch, installed, captured)
|
|
pi_wake_agent.remove_cron(["pi_test"], "20m", logging.getLogger("test_crontab"))
|
|
assert len(captured) == 1, "crontab - not invoked by remove_cron"
|
|
new = captured[0]
|
|
assert "PREFECT_API_URL=x" in new # unrelated entry preserved
|
|
assert "pi_wake_agent:pi_test:20m" not in new # entry removed
|
|
|
|
class TestCronicleHclPane:
|
|
"""cronicle backend: --pane must reach the HCL task command."""
|
|
|
|
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_has_pane(self):
|
|
rc, out, err = run_cmd(["--install", "--interval", "20m", "--session", "pi_test",
|
|
"--pane", "terminal_1",
|
|
"--msg", "Please carefully continue your work"])
|
|
assert rc == 0, f"install failed: {err}"
|
|
content = cronicle_hcl_path().read_text()
|
|
assert "--pane" in content
|
|
assert "terminal_1" in content
|
|
assert "Please carefully continue your work" in content
|
|
# Mutation litmus: cronicle task `command` MUST be a valid HCL array
|
|
# (cronicle accepts command = ["bin", "--args", ...]). Removing the
|
|
# leading `[` in cronicle_install must fail this assertion.
|
|
assert 'command = [' in content
|
|
assert '"--run"' in content
|
|
assert '"--pane"' in content
|
|
|
|
# ─── Run all tests ────────────────────────────────────────────────────────
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v", "--tb=short"])
|