"""Tests for the task cleanup process (--cleanup-done) and its scheduled install. Covers: - cmd_cleanup_done: age threshold (by .state mtime), move, dry-run, skip (dest exists, untracked, non-complete, below threshold), --json summary. - --install-cleanup-schedule: stub + launchd plist written to AUTOMATON_LAUNCHAGENTS_DIR override (no host bleed). - _all_task_dirs excludes the 'complete' container (so --audit/--list stay clean once tasks are archived). """ import importlib.util import json import os import subprocess import sys import time from pathlib import Path import pytest _STATUS_PATH = Path.home() / ".automaton" / "scripts" / "status.py" _spec = importlib.util.spec_from_file_location("st_cleanup", _STATUS_PATH) st = importlib.util.module_from_spec(_spec) _spec.loader.exec_module(st) def _task_ns(days=None, dry_run=False, json_out=False, project=None): class A: pass a = A() a.days = days a.dry_run = dry_run a.json_output = json_out a.project = project return a def _make_complete_task(project_root: Path, name: str, age_days: float = 0.0) -> Path: task_dir = project_root / ".automaton" / "tasks" / name task_dir.mkdir(parents=True) (task_dir / ".state").write_text("complete\n") (task_dir / "VERDICT.md").write_text("## Status: PASS\n") if age_days: when = time.time() - age_days * 86400.0 os.utime(task_dir / ".state", (when, when)) return task_dir @pytest.fixture def tmp_project(tmp_path): (tmp_path / ".automaton" / "tasks").mkdir(parents=True) return tmp_path class TestCleanupDone: def test_moves_old_complete_task(self, tmp_project): _make_complete_task(tmp_project, "old-task", age_days=8.0) rc = st.cmd_cleanup_done(_task_ns(days=7, project=str(tmp_project))) assert rc == 0 dest = tmp_project / ".automaton" / "tasks" / "complete" / "old-task" assert dest.exists() assert (dest / ".state").exists() def test_keeps_recent_complete_task(self, tmp_project): td = _make_complete_task(tmp_project, "fresh-task", age_days=1.0) st.cmd_cleanup_done(_task_ns(days=7, project=str(tmp_project))) assert td.exists() # not moved assert not (tmp_project / ".automaton" / "tasks" / "complete" / "fresh-task").exists() def test_dry_run_does_not_move(self, tmp_project): td = _make_complete_task(tmp_project, "dry-task", age_days=10.0) rc = st.cmd_cleanup_done(_task_ns(days=7, dry_run=True, project=str(tmp_project))) assert rc == 0 assert td.exists() # still in place assert not (tmp_project / ".automaton" / "tasks" / "complete").exists() def test_skip_untracked(self, tmp_project): # dir exists but no .state -> untracked, ignored untracked = tmp_project / ".automaton" / "tasks" / "no-state" untracked.mkdir(parents=True) rc = st.cmd_cleanup_done(_task_ns(days=0, project=str(tmp_project))) assert rc == 0 assert untracked.exists() # untouched def test_skip_non_complete(self, tmp_project): td = tmp_project / ".automaton" / "tasks" / "wip" td.mkdir(parents=True) (td / ".state").write_text("implement\n") rc = st.cmd_cleanup_done(_task_ns(days=0, project=str(tmp_project))) assert rc == 0 assert td.exists() assert (td / ".state").read_text().strip() == "implement" def test_skip_when_dest_exists(self, tmp_project, capsys): _make_complete_task(tmp_project, "dup", age_days=9.0) dest = tmp_project / ".automaton" / "tasks" / "complete" / "dup" dest.mkdir(parents=True) (dest / ".state").write_text("complete\n") rc = st.cmd_cleanup_done(_task_ns(days=7, project=str(tmp_project))) assert rc == 0 out = capsys.readouterr().out assert "skipped" in out and "dest_exists" in out def test_json_summary(self, tmp_project, capsys): _make_complete_task(tmp_project, "json-old", age_days=9.0) _make_complete_task(tmp_project, "json-new", age_days=1.0) st.cmd_cleanup_done(_task_ns(days=7, json_out=True, project=str(tmp_project))) out = capsys.readouterr().out last = out.strip().splitlines()[-1] summary = json.loads(last) assert summary["days"] == 7 assert summary["dry_run"] is False assert "json-old" in summary["moved"] assert "json-new" not in summary["moved"] def test_all_task_dirs_excludes_complete_container(self, tmp_project): _make_complete_task(tmp_project, "active-done", age_days=0.0) archive = tmp_project / ".automaton" / "tasks" / "complete" / "archived" archive.mkdir(parents=True) names = {n for n, _ in st._all_task_dirs(str(tmp_project))} assert "active-done" in names assert "complete" not in names assert "archived" not in names class TestInstallCleanupScheduleIsolation: def _run_cli(self, args, env_dir): env = os.environ.copy() env["AUTOMATON_LAUNCHAGENTS_DIR"] = str(env_dir) cmd = [sys.executable, str(_STATUS_PATH)] + args return subprocess.run(cmd, capture_output=True, text=True, env=env) def test_plist_written_to_override_dir_not_host(self, tmp_path): agents = tmp_path / "agents" agents.mkdir() host = Path.home() / "Library" / "LaunchAgents" / "com.automaton.cleanup.plist" # Snapshot host plist state before the run. A pre-existing real install # is OK — only a write during this test would be a bleed. host_before = host.stat().st_mtime_ns if host.exists() else None res = self._run_cli( ["--install-cleanup-schedule", "--days", "5", "--interval", "120", "--project", str(tmp_path)], agents, ) assert res.returncode == 0, res.stdout + res.stderr plist = agents / "com.automaton.cleanup.plist" assert plist.exists() host_after = host.stat().st_mtime_ns if host.exists() else None assert host_before == host_after, "host plist was modified by the test run" content = plist.read_text() assert "com.automaton.cleanup" in content assert "120" in content stub = Path(st.AUTOMATON_DIR) / "scripts" / "automaton-cleanup.sh" assert stub.exists() assert "--days 5" in stub.read_text() @pytest.mark.skipif(sys.platform != "darwin", reason="Darwin plist branch") def test_uninstall_via_disabled_rename(self, tmp_path): agents = tmp_path / "agents" agents.mkdir() res = self._run_cli( ["--install-cleanup-schedule", "--project", str(tmp_path)], agents) assert res.returncode == 0 plist = agents / "com.automaton.cleanup.plist" assert plist.exists()