227 lines
9.4 KiB
Python
227 lines
9.4 KiB
Python
|
"""Tests for linux-schedule-parity (v1.1 task 6)."""
|
||
|
|
|
||
|
|
import importlib.util
|
||
|
|
import json
|
||
|
|
import platform
|
||
|
|
import sys
|
||
|
|
from pathlib import Path
|
||
|
|
from unittest.mock import patch, MagicMock
|
||
|
|
|
||
|
|
spec = importlib.util.spec_from_file_location(
|
||
|
|
"st", str(Path(__file__).resolve().parent.parent / "scripts" / "status.py")
|
||
|
|
)
|
||
|
|
st = importlib.util.module_from_spec(spec)
|
||
|
|
spec.loader.exec_module(st)
|
||
|
|
|
||
|
|
LOOP_TICK_SCRIPT_SH = st.LOOP_TICK_SCRIPT_SH
|
||
|
|
|
||
|
|
|
||
|
|
class TestInstallCronBlock:
|
||
|
|
def test_writes_fresh_block(self, tmp_path):
|
||
|
|
with patch("subprocess.run") as mock_run:
|
||
|
|
mock_run.return_value.returncode = 0
|
||
|
|
mock_run.return_value.stdout = ""
|
||
|
|
rc = st._install_cron_block("my-loop", tmp_path, 3600)
|
||
|
|
assert rc == 0
|
||
|
|
# Second call to crontab - contains the block
|
||
|
|
input_call = mock_run.call_args_list[-1]
|
||
|
|
stdin = input_call.kwargs["input"]
|
||
|
|
assert "# automaton-loop:my-loop" in stdin
|
||
|
|
assert "# end automaton-loop:my-loop" in stdin
|
||
|
|
assert "*/60 * * * *" in stdin
|
||
|
|
|
||
|
|
def test_strips_prior_block(self, tmp_path):
|
||
|
|
existing = (
|
||
|
|
"# automaton-loop:my-loop\n"
|
||
|
|
"*/60 * * * * /old/stub\n"
|
||
|
|
"# end automaton-loop:my-loop\n"
|
||
|
|
)
|
||
|
|
with patch("subprocess.run") as mock_run:
|
||
|
|
mock_run.return_value.returncode = 0
|
||
|
|
mock_run.return_value.stdout = existing
|
||
|
|
rc = st._install_cron_block("my-loop", tmp_path, 3600)
|
||
|
|
assert rc == 0
|
||
|
|
input_call = mock_run.call_args_list[-1]
|
||
|
|
stdin = input_call.kwargs["input"]
|
||
|
|
blocks = [l for l in stdin.splitlines()
|
||
|
|
if l.strip().startswith("# automaton-loop:my-loop")]
|
||
|
|
assert len(blocks) == 1, "should have exactly one block"
|
||
|
|
|
||
|
|
def test_returns_2_on_write_error(self, tmp_path):
|
||
|
|
from subprocess import SubprocessError
|
||
|
|
with patch("subprocess.run") as mock_run:
|
||
|
|
mock_run.side_effect = [
|
||
|
|
MagicMock(returncode=0, stdout=""),
|
||
|
|
SubprocessError("crontab write failed"),
|
||
|
|
]
|
||
|
|
rc = st._install_cron_block("my-loop", tmp_path, 3600)
|
||
|
|
assert rc == 2
|
||
|
|
|
||
|
|
def test_rounds_interval_to_minutes(self, tmp_path):
|
||
|
|
with patch("subprocess.run") as mock_run:
|
||
|
|
mock_run.return_value.returncode = 0
|
||
|
|
mock_run.return_value.stdout = ""
|
||
|
|
st._install_cron_block("my-loop", tmp_path, 90)
|
||
|
|
input_call = mock_run.call_args_list[-1]
|
||
|
|
stdin = input_call.kwargs["input"]
|
||
|
|
assert "*/1 * * * *" in stdin
|
||
|
|
|
||
|
|
def test_minimum_interval(self, tmp_path):
|
||
|
|
with patch("subprocess.run") as mock_run:
|
||
|
|
mock_run.return_value.returncode = 0
|
||
|
|
mock_run.return_value.stdout = ""
|
||
|
|
st._install_cron_block("my-loop", tmp_path, 30)
|
||
|
|
input_call = mock_run.call_args_list[-1]
|
||
|
|
stdin = input_call.kwargs["input"]
|
||
|
|
assert "*/1 * * * *" in stdin
|
||
|
|
|
||
|
|
|
||
|
|
class TestEnableScheduleLinux:
|
||
|
|
def _loop_path(self, tmp_path, name="my-loop"):
|
||
|
|
"""Build a fake loop dir structure that _enable_schedule expects."""
|
||
|
|
loop_dir = tmp_path / ".automaton" / "loops" / name
|
||
|
|
loop_dir.mkdir(parents=True, exist_ok=True)
|
||
|
|
return loop_dir
|
||
|
|
|
||
|
|
def test_reinstalls_cron_when_stub_exists(self, tmp_path):
|
||
|
|
loop_dir = self._loop_path(tmp_path)
|
||
|
|
stub = loop_dir / LOOP_TICK_SCRIPT_SH
|
||
|
|
stub.write_text("#!/bin/bash\necho tick\n")
|
||
|
|
cfg = {"schedule": {"interval_seconds": 7200}}
|
||
|
|
(loop_dir / "loop.json").write_text(json.dumps(cfg))
|
||
|
|
with patch("platform.system", return_value="Linux"):
|
||
|
|
with patch("subprocess.run") as mock_run:
|
||
|
|
mock_run.return_value.returncode = 0
|
||
|
|
mock_run.return_value.stdout = ""
|
||
|
|
st._enable_schedule("my-loop", tmp_path)
|
||
|
|
has_crontab_write = False
|
||
|
|
for c in mock_run.call_args_list:
|
||
|
|
_, kwargs = c
|
||
|
|
if kwargs.get("input") and "*/120 * * * *" in kwargs["input"]:
|
||
|
|
has_crontab_write = True
|
||
|
|
break
|
||
|
|
assert has_crontab_write, "expected crontab write with */120"
|
||
|
|
|
||
|
|
def test_warns_when_stub_missing(self, tmp_path, capsys):
|
||
|
|
self._loop_path(tmp_path)
|
||
|
|
with patch("platform.system", return_value="Linux"):
|
||
|
|
with patch("subprocess.run") as mock_run:
|
||
|
|
st._enable_schedule("my-loop", tmp_path)
|
||
|
|
mock_run.assert_not_called()
|
||
|
|
captured = capsys.readouterr()
|
||
|
|
assert "WARNING" in captured.err
|
||
|
|
assert "no tick stub" in captured.err
|
||
|
|
|
||
|
|
def test_reads_interval_from_cfg(self, tmp_path):
|
||
|
|
loop_dir = self._loop_path(tmp_path)
|
||
|
|
stub = loop_dir / LOOP_TICK_SCRIPT_SH
|
||
|
|
stub.write_text("#!/bin/bash\n")
|
||
|
|
(loop_dir / "loop.json").write_text(
|
||
|
|
json.dumps({"schedule": {"interval_seconds": 180}}))
|
||
|
|
with patch("platform.system", return_value="Linux"):
|
||
|
|
with patch("subprocess.run") as mock_run:
|
||
|
|
mock_run.return_value.returncode = 0
|
||
|
|
mock_run.return_value.stdout = ""
|
||
|
|
st._enable_schedule("my-loop", tmp_path)
|
||
|
|
found = False
|
||
|
|
for c in mock_run.call_args_list:
|
||
|
|
_, kwargs = c
|
||
|
|
inp = kwargs.get("input", "")
|
||
|
|
if "*/3 * * * *" in inp:
|
||
|
|
found = True
|
||
|
|
break
|
||
|
|
assert found, "expected */3 interval"
|
||
|
|
|
||
|
|
def test_fallback_interval_when_garbage(self, tmp_path):
|
||
|
|
loop_dir = self._loop_path(tmp_path)
|
||
|
|
stub = loop_dir / LOOP_TICK_SCRIPT_SH
|
||
|
|
stub.write_text("#!/bin/bash\n")
|
||
|
|
(loop_dir / "loop.json").write_text(
|
||
|
|
json.dumps({"schedule": {"interval_seconds": "twenty"}}))
|
||
|
|
with patch("platform.system", return_value="Linux"):
|
||
|
|
with patch("subprocess.run") as mock_run:
|
||
|
|
mock_run.return_value.returncode = 0
|
||
|
|
mock_run.return_value.stdout = ""
|
||
|
|
st._enable_schedule("my-loop", tmp_path)
|
||
|
|
found = False
|
||
|
|
for c in mock_run.call_args_list:
|
||
|
|
_, kwargs = c
|
||
|
|
inp = kwargs.get("input", "")
|
||
|
|
if "*/60 * * * *" in inp:
|
||
|
|
found = True
|
||
|
|
break
|
||
|
|
assert found, "expected */60 fallback interval"
|
||
|
|
|
||
|
|
def test_idempotent_two_calls(self, tmp_path):
|
||
|
|
loop_dir = self._loop_path(tmp_path)
|
||
|
|
stub = loop_dir / LOOP_TICK_SCRIPT_SH
|
||
|
|
stub.write_text("#!/bin/bash\n")
|
||
|
|
(loop_dir / "loop.json").write_text(
|
||
|
|
json.dumps({"schedule": {"interval_seconds": 3600}}))
|
||
|
|
existing = "# automaton-loop:my-loop\n*/60 * * * * /stub\n# end automaton-loop:my-loop\n"
|
||
|
|
with patch("platform.system", return_value="Linux"):
|
||
|
|
with patch("subprocess.run") as mock_run:
|
||
|
|
mock_run.return_value.returncode = 0
|
||
|
|
mock_run.return_value.stdout = existing
|
||
|
|
st._enable_schedule("my-loop", tmp_path)
|
||
|
|
st._enable_schedule("my-loop", tmp_path)
|
||
|
|
last_input = ""
|
||
|
|
for c in mock_run.call_args_list:
|
||
|
|
_, kwargs = c
|
||
|
|
if kwargs.get("input"):
|
||
|
|
last_input = kwargs["input"]
|
||
|
|
blocks = [l for l in last_input.splitlines()
|
||
|
|
if l.strip().startswith("# automaton-loop:")]
|
||
|
|
assert len(blocks) == 1, "should have exactly one block after two calls"
|
||
|
|
|
||
|
|
def test_darwin_branch_unchanged(self, tmp_path):
|
||
|
|
self._loop_path(tmp_path)
|
||
|
|
plist = Path.home() / "Library" / "LaunchAgents" / f"com.automaton.loop.my-loop.plist"
|
||
|
|
disabled = plist.with_suffix(".plist.disabled")
|
||
|
|
try:
|
||
|
|
disabled.parent.mkdir(parents=True, exist_ok=True)
|
||
|
|
disabled.write_text("fake")
|
||
|
|
with patch("platform.system", return_value="Darwin"):
|
||
|
|
st._enable_schedule("my-loop", tmp_path)
|
||
|
|
assert plist.exists(), "Darwin should rename .disabled back"
|
||
|
|
assert not disabled.exists()
|
||
|
|
finally:
|
||
|
|
plist.unlink(missing_ok=True)
|
||
|
|
disabled.unlink(missing_ok=True)
|
||
|
|
|
||
|
|
def test_windows_branch_unchanged(self, tmp_path):
|
||
|
|
self._loop_path(tmp_path)
|
||
|
|
with patch("platform.system", return_value="Windows"):
|
||
|
|
with patch("subprocess.run") as mock_run:
|
||
|
|
mock_run.return_value.returncode = 0
|
||
|
|
st._enable_schedule("my-loop", tmp_path)
|
||
|
|
mock_run.assert_called_once()
|
||
|
|
|
||
|
|
|
||
|
|
class TestDisableScheduleLinux:
|
||
|
|
def test_still_strips_after_extraction(self, tmp_path):
|
||
|
|
loop_dir = tmp_path / ".automaton" / "loops" / "my-loop"
|
||
|
|
loop_dir.mkdir(parents=True, exist_ok=True)
|
||
|
|
stub = loop_dir / LOOP_TICK_SCRIPT_SH
|
||
|
|
stub.write_text("#!/bin/bash\n")
|
||
|
|
existing = (
|
||
|
|
"# automaton-loop:my-loop\n"
|
||
|
|
"*/60 * * * * /stub\n"
|
||
|
|
"# end automaton-loop:my-loop\n"
|
||
|
|
"# unrelated\n"
|
||
|
|
)
|
||
|
|
with patch("platform.system", return_value="Linux"):
|
||
|
|
with patch("subprocess.run") as mock_run:
|
||
|
|
mock_run.return_value.returncode = 0
|
||
|
|
mock_run.return_value.stdout = existing
|
||
|
|
st._disable_schedule("my-loop", tmp_path)
|
||
|
|
inputs = []
|
||
|
|
for c in mock_run.call_args_list:
|
||
|
|
_, kwargs = c
|
||
|
|
inp = kwargs.get("input")
|
||
|
|
if inp:
|
||
|
|
inputs.append(inp)
|
||
|
|
assert len(inputs) == 1, "expected one crontab - write"
|
||
|
|
assert "# automaton-loop:my-loop" not in inputs[0], "block should be stripped"
|
||
|
|
assert "# unrelated" in inputs[0], "unrelated lines preserved"
|