automaton/tests/test_cleanup_done.py

"""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 "<integer>120</integer>" 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()