Complete tasks 3-7: harden verdict parsing, outputs retention, base branch, linux schedule parity, claim loop task

tests/test_claim_loop_task.py

@@ -0,0 +1,196 @@
+"""Tests for --claim-loop-task (cross-loop ownership).
+
+Covers R1-R12 from tasks/add-claim-loop-task/SPEC.md.
+"""
+
+import argparse
+import json
+import os
+import sys
+import subprocess
+import time
+from pathlib import Path
+from unittest.mock import MagicMock, patch, call
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent.parent / "scripts"))
+
+import status as st
+
+STATUS = Path.home() / ".automaton" / "scripts" / "status.py"
+
+
+def _run(args, project=None, expect_failure=False):
+    cmd = [sys.executable, str(STATUS)]
+    if project:
+        cmd.extend(["--project", str(project)])
+    cmd.extend(args)
+    res = subprocess.run(cmd, capture_output=True, text=True)
+    if not expect_failure:
+        assert res.returncode == 0, f"cmd {cmd!r} exited {res.returncode}:\n{res.stdout}\n{res.stderr}"
+    return res.stdout.strip(), res.stderr.strip(), res.returncode
+
+
+def _loop_state(loop_path):
+    return json.loads((loop_path / ".state.loop").read_text())
+
+
+def _create_loop(project, name="ci-loop", template="ci-triage"):
+    out, err, code = _run(["--create-loop", name, "--from-template", template], project)
+    assert code == 0, out + err
+    return project / ".automaton" / "loops" / name
+
+
+@pytest.fixture
+def tmp_project(tmp_path):
+    (tmp_path / ".automaton" / "tasks").mkdir(parents=True)
+    return tmp_path
+
+
+@pytest.fixture
+def two_loops(tmp_project):
+    lp1 = _create_loop(tmp_project, "loop-alpha")
+    lp2 = _create_loop(tmp_project, "loop-beta")
+    return tmp_project, lp1, lp2
+
+
+class TestClaimSucceedsNoOneOwns:
+    def test_claim_succeeds_no_one_owns(self, two_loops):
+        project, lp1, lp2 = two_loops
+        out, err, code = _run(["--claim-loop-task", "loop-alpha", "--task", "fix-X"], project)
+        assert code == 0, err
+        assert "OK" in out
+        s = _loop_state(lp1)
+        assert s["current_task"] == "fix-X"
+
+
+class TestClaimRefusesOtherLoopOwns:
+    def test_claim_refuses_other_loop_owns(self, two_loops):
+        project, lp1, lp2 = two_loops
+        s2 = _loop_state(lp2)
+        s2["current_task"] = "fix-X"
+        st._write_state_loop(lp2, s2)
+        out, err, code = _run(["--claim-loop-task", "loop-alpha", "--task", "fix-X"], project,
+                              expect_failure=True)
+        assert code == 2, f"expected exit 2, got {code}: {out} {err}"
+        assert "task_already_claimed:loop-beta" in err
+        s1 = _loop_state(lp1)
+        assert s1["current_task"] is None
+
+
+class TestClaimIdempotentSelfOwns:
+    def test_claim_idempotent_self_owns(self, two_loops):
+        project, lp1, lp2 = two_loops
+        s1 = _loop_state(lp1)
+        s1["current_task"] = "fix-X"
+        st._write_state_loop(lp1, s1)
+        mtime_before = (lp1 / ".state.loop").stat().st_mtime_ns
+        time.sleep(0.01)
+        out, err, code = _run(["--claim-loop-task", "loop-alpha", "--task", "fix-X"], project)
+        assert code == 0, err
+        assert "already_self_claimed" in out
+        mtime_after = (lp1 / ".state.loop").stat().st_mtime_ns
+        assert mtime_after == mtime_before, "state should not be re-written on idempotent claim"
+
+
+class TestClaimUntrackedLoop:
+    def test_claim_untracked_loop(self, tmp_project):
+        project = tmp_project
+        out, err, code = _run(["--claim-loop-task", "ghost-loop", "--task", "fix-X"], project,
+                              expect_failure=True)
+        assert code == 2
+        assert "UNTRACKED" in err
+
+
+class TestClaimMissingTask:
+    def test_claim_missing_task(self, tmp_project):
+        _create_loop(tmp_project, "loop-alpha")
+        out, err, code = _run(["--claim-loop-task", "loop-alpha"],
+                              tmp_project, expect_failure=True)
+        assert code == 2
+        assert "--task" in err
+
+
+class TestClaimPausedLoop:
+    def test_claim_paused_loop_ownership_check(self, two_loops):
+        project, lp1, lp2 = two_loops
+        s2 = _loop_state(lp2)
+        s2["current_task"] = "fix-X"
+        s2["status"] = "paused"
+        st._write_state_loop(lp2, s2)
+        out, err, code = _run(["--claim-loop-task", "loop-alpha", "--task", "fix-X"], project,
+                              expect_failure=True)
+        assert code == 2, "paused loop's claim should still block"
+        assert "task_already_claimed:loop-beta" in err
+
+
+class TestCrossLoopSelfHealingRace:
+    def test_self_healing_race(self, two_loops):
+        project, lp1, lp2 = two_loops
+        s2 = _loop_state(lp2)
+        s2["current_task"] = "fix-X"
+        st._write_state_loop(lp2, s2)
+        out, err, code = _run(["--claim-loop-task", "loop-alpha", "--task", "fix-X"], project,
+                              expect_failure=True)
+        assert code == 2
+        assert "task_already_claimed" in err
+        s2["current_task"] = None
+        st._write_state_loop(lp2, s2)
+        out, err, code = _run(["--claim-loop-task", "loop-alpha", "--task", "fix-X"], project)
+        assert code == 0
+        s1 = _loop_state(lp1)
+        assert s1["current_task"] == "fix-X"
+
+
+class TestReleaseLogic:
+    def _release(self, loop_path, current_task, project):
+        state = _loop_state(loop_path)
+        task_dir = project / ".automaton" / "tasks" / current_task
+        task_state = task_dir / ".state"
+        if task_state.exists():
+            phase = task_state.read_text().strip()
+            if phase in ("complete", "human_intervention"):
+                state["current_task"] = None
+                st._write_state_loop(loop_path, state)
+                return True
+        return False
+
+    def test_release_on_complete(self, two_loops):
+        project, lp1, lp2 = two_loops
+        s1 = _loop_state(lp1)
+        s1["current_task"] = "some-task"
+        st._write_state_loop(lp1, s1)
+        task_dir = project / ".automaton" / "tasks" / "some-task"
+        task_dir.mkdir(parents=True)
+        (task_dir / ".state").write_text("complete")
+        released = self._release(lp1, "some-task", project)
+        assert released
+        s_after = _loop_state(lp1)
+        assert s_after["current_task"] is None
+
+    def test_release_on_human_intervention(self, two_loops):
+        project, lp1, lp2 = two_loops
+        s1 = _loop_state(lp1)
+        s1["current_task"] = "some-task"
+        st._write_state_loop(lp1, s1)
+        task_dir = project / ".automaton" / "tasks" / "some-task"
+        task_dir.mkdir(parents=True)
+        (task_dir / ".state").write_text("human_intervention")
+        released = self._release(lp1, "some-task", project)
+        assert released
+        s_after = _loop_state(lp1)
+        assert s_after["current_task"] is None
+
+    def test_no_release_on_implement(self, two_loops):
+        project, lp1, lp2 = two_loops
+        s1 = _loop_state(lp1)
+        s1["current_task"] = "some-task"
+        st._write_state_loop(lp1, s1)
+        task_dir = project / ".automaton" / "tasks" / "some-task"
+        task_dir.mkdir(parents=True)
+        (task_dir / ".state").write_text("implement")
+        released = self._release(lp1, "some-task", project)
+        assert not released
+        s_after = _loop_state(lp1)
+        assert s_after["current_task"] == "some-task"