197 lines
7.0 KiB
Python
197 lines
7.0 KiB
Python
"""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"
|