317 lines
13 KiB
Python
317 lines
13 KiB
Python
|
"""Tests for scripts/autopilot.py — functional correctness review.
|
||
|
|
|
||
|
|
Covers bugs found during 2026-06-25 review:
|
||
|
|
1. PHASE_PRIORITY missing code_review
|
||
|
|
2. cmd_drive suggested implement→bug_find (illegal; should be implement→code_review)
|
||
|
|
3. cmd_drive had no handler for code_review / code_review:approved
|
||
|
|
4. _all_tasks didn't skip tasks/complete/ archive dir
|
||
|
|
5. _all_tasks didn't recurse into subtasks/
|
||
|
|
6. _all_tasks used wrong path for non-framework projects
|
||
|
|
7. is_terminal treated human_intervention as terminal (dead-code handler)
|
||
|
|
8. needs_user_input didn't flag human_intervention as blocked
|
||
|
|
"""
|
||
|
|
|
||
|
|
import importlib.util
|
||
|
|
import sys
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Optional
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
_AP_PATH = Path.home() / ".automaton" / "scripts" / "autopilot.py"
|
||
|
|
_spec = importlib.util.spec_from_file_location("autopilot_mod", _AP_PATH)
|
||
|
|
ap = importlib.util.module_from_spec(_spec)
|
||
|
|
_spec.loader.exec_module(ap)
|
||
|
|
|
||
|
|
|
||
|
|
def _make_task(project: Path, name: str, phase: str = "new") -> Path:
|
||
|
|
"""Create a task dir + .state (no subprocess)."""
|
||
|
|
if project == ap.AUTOMATON_DIR:
|
||
|
|
tp = project / "tasks" / name
|
||
|
|
else:
|
||
|
|
tp = project / ".automaton" / "tasks" / name
|
||
|
|
tp.mkdir(parents=True, exist_ok=True)
|
||
|
|
(tp / ".state").write_text(phase + "\n")
|
||
|
|
return tp
|
||
|
|
|
||
|
|
|
||
|
|
def _make_subtask(project: Path, parent: str, name: str, phase: str = "new") -> Path:
|
||
|
|
if project == ap.AUTOMATON_DIR:
|
||
|
|
sp = project / "tasks" / parent / "subtasks" / name
|
||
|
|
else:
|
||
|
|
sp = project / ".automaton" / "tasks" / parent / "subtasks" / name
|
||
|
|
sp.mkdir(parents=True, exist_ok=True)
|
||
|
|
(sp / ".state").write_text(phase + "\n")
|
||
|
|
return sp
|
||
|
|
|
||
|
|
|
||
|
|
def _write_verdict(task_path: Path, first_line: str) -> None:
|
||
|
|
(task_path / "VERDICT.md").write_text(first_line + "\nrest of verdict\n")
|
||
|
|
|
||
|
|
|
||
|
|
class _Args:
|
||
|
|
"""Minimal args object for cmd_* functions."""
|
||
|
|
def __init__(self, **kwargs):
|
||
|
|
self.project = kwargs.get("project")
|
||
|
|
self.max_iterations = kwargs.get("max_iterations")
|
||
|
|
self.delay = kwargs.get("delay")
|
||
|
|
self.threshold = kwargs.get("threshold")
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def tmp_project(tmp_path):
|
||
|
|
(tmp_path / ".automaton" / "tasks").mkdir(parents=True)
|
||
|
|
return tmp_path
|
||
|
|
|
||
|
|
|
||
|
|
class TestPhasePriority:
|
||
|
|
"""Bug 1: code_review was missing from PHASE_PRIORITY."""
|
||
|
|
|
||
|
|
def test_code_review_in_priority(self):
|
||
|
|
assert "code_review" in ap.PHASE_PRIORITY
|
||
|
|
|
||
|
|
def test_code_review_priority_between_implement_and_bug_find(self):
|
||
|
|
assert ap.PHASE_PRIORITY["code_review"] > ap.PHASE_PRIORITY["implement"]
|
||
|
|
assert ap.PHASE_PRIORITY["code_review"] < ap.PHASE_PRIORITY["bug_find"]
|
||
|
|
|
||
|
|
def test_human_intervention_in_priority(self):
|
||
|
|
assert "human_intervention" in ap.PHASE_PRIORITY
|
||
|
|
|
||
|
|
def test_priorities_monotonically_increase_along_pipeline(self):
|
||
|
|
pipeline = [
|
||
|
|
"new", "research", "decomposition", "design", "test_design",
|
||
|
|
"implement", "code_review", "bug_find", "adversarial_bug_find",
|
||
|
|
"doc_review", "referee",
|
||
|
|
]
|
||
|
|
for i in range(len(pipeline) - 1):
|
||
|
|
assert ap.PHASE_PRIORITY[pipeline[i]] < ap.PHASE_PRIORITY[pipeline[i + 1]], \
|
||
|
|
f"{pipeline[i]} ({ap.PHASE_PRIORITY[pipeline[i]]}) should be < {pipeline[i+1]} ({ap.PHASE_PRIORITY[pipeline[i+1]]})"
|
||
|
|
|
||
|
|
|
||
|
|
class TestCmdDriveCodeReview:
|
||
|
|
"""Bugs 2+3: implement→bug_find was illegal; code_review handlers missing."""
|
||
|
|
|
||
|
|
def test_implement_suggests_code_review_not_bug_find(self, tmp_project, capsys):
|
||
|
|
_make_task(tmp_project, "task-a", "implement")
|
||
|
|
args = _Args(project=str(tmp_project))
|
||
|
|
ap.cmd_drive(args)
|
||
|
|
out = capsys.readouterr().out
|
||
|
|
assert "--transition code_review" in out
|
||
|
|
assert "--transition bug_find" not in out
|
||
|
|
|
||
|
|
def test_code_review_suggests_awaiting_approval(self, tmp_project, capsys):
|
||
|
|
_make_task(tmp_project, "task-cr", "code_review")
|
||
|
|
args = _Args(project=str(tmp_project))
|
||
|
|
ap.cmd_drive(args)
|
||
|
|
out = capsys.readouterr().out
|
||
|
|
assert "--transition code_review:awaiting_approval" in out
|
||
|
|
|
||
|
|
def test_code_review_approved_suggests_bug_find(self, tmp_project, capsys):
|
||
|
|
_make_task(tmp_project, "task-cr2", "code_review:approved")
|
||
|
|
args = _Args(project=str(tmp_project))
|
||
|
|
ap.cmd_drive(args)
|
||
|
|
out = capsys.readouterr().out
|
||
|
|
assert "--transition bug_find" in out
|
||
|
|
|
||
|
|
def test_implement_task_not_stalled(self, tmp_project, capsys):
|
||
|
|
"""Ensure cmd_drive produces a transition command for implement phase."""
|
||
|
|
_make_task(tmp_project, "task-impl", "implement")
|
||
|
|
args = _Args(project=str(tmp_project))
|
||
|
|
rc = ap.cmd_drive(args)
|
||
|
|
out = capsys.readouterr().out
|
||
|
|
assert rc == 0
|
||
|
|
assert "status.py --transition" in out
|
||
|
|
|
||
|
|
|
||
|
|
class TestAllTasksSkipComplete:
|
||
|
|
"""Bug 4: _all_tasks included tasks/complete/ as a task."""
|
||
|
|
|
||
|
|
def test_complete_dir_excluded(self, tmp_project):
|
||
|
|
_make_task(tmp_project, "active-task", "implement")
|
||
|
|
complete_dir = tmp_project / ".automaton" / "tasks" / "complete"
|
||
|
|
complete_dir.mkdir(parents=True)
|
||
|
|
(complete_dir / "old-task").mkdir()
|
||
|
|
(complete_dir / "old-task" / ".state").write_text("complete\n")
|
||
|
|
|
||
|
|
tasks = ap._all_tasks(tmp_project)
|
||
|
|
names = [t[0] for t in tasks]
|
||
|
|
assert "active-task" in names
|
||
|
|
assert "complete" not in names
|
||
|
|
|
||
|
|
|
||
|
|
class TestAllTasksSubtasks:
|
||
|
|
"""Bug 5: _all_tasks didn't recurse into subtasks/."""
|
||
|
|
|
||
|
|
def test_subtask_found(self, tmp_project):
|
||
|
|
_make_task(tmp_project, "parent-task", "decomposition:approved")
|
||
|
|
_make_subtask(tmp_project, "parent-task", "sub-a", "implement")
|
||
|
|
_make_subtask(tmp_project, "parent-task", "sub-b", "new")
|
||
|
|
|
||
|
|
tasks = ap._all_tasks(tmp_project)
|
||
|
|
names = [t[0] for t in tasks]
|
||
|
|
assert "parent-task" in names
|
||
|
|
assert "parent-task/sub-a" in names
|
||
|
|
assert "parent-task/sub-b" in names
|
||
|
|
|
||
|
|
def test_subtask_scan_all_tasks(self, tmp_project):
|
||
|
|
_make_task(tmp_project, "parent-task", "decomposition:approved")
|
||
|
|
_make_subtask(tmp_project, "parent-task", "sub-a", "implement")
|
||
|
|
|
||
|
|
scanned = ap.scan_all_tasks(tmp_project)
|
||
|
|
names = [t["name"] for t in scanned]
|
||
|
|
assert "parent-task" in names
|
||
|
|
assert "parent-task/sub-a" in names
|
||
|
|
|
||
|
|
|
||
|
|
class TestAllTasksProjectPath:
|
||
|
|
"""Bug 6: _all_tasks used project_dir/tasks instead of project_dir/.automaton/tasks."""
|
||
|
|
|
||
|
|
def test_non_framework_project_path(self, tmp_project):
|
||
|
|
_make_task(tmp_project, "proj-task", "implement")
|
||
|
|
tasks = ap._all_tasks(tmp_project)
|
||
|
|
names = [t[0] for t in tasks]
|
||
|
|
assert "proj-task" in names
|
||
|
|
|
||
|
|
def test_wrong_path_returns_empty(self, tmp_path):
|
||
|
|
"""If .automaton/tasks doesn't exist, return empty (not crash)."""
|
||
|
|
tasks = ap._all_tasks(tmp_path)
|
||
|
|
assert tasks == []
|
||
|
|
|
||
|
|
|
||
|
|
class TestIsTerminal:
|
||
|
|
"""Bug 7: human_intervention was treated as terminal."""
|
||
|
|
|
||
|
|
def test_complete_is_terminal(self):
|
||
|
|
assert ap.is_terminal({"phase": "complete"}) is True
|
||
|
|
|
||
|
|
def test_human_intervention_not_terminal(self):
|
||
|
|
assert ap.is_terminal({"phase": "human_intervention"}) is False
|
||
|
|
|
||
|
|
def test_none_phase_not_terminal(self):
|
||
|
|
assert ap.is_terminal({"phase": None}) is False
|
||
|
|
|
||
|
|
def test_new_not_terminal(self):
|
||
|
|
assert ap.is_terminal({"phase": "new"}) is False
|
||
|
|
|
||
|
|
|
||
|
|
class TestNeedsUserInput:
|
||
|
|
"""Bug 8: human_intervention wasn't flagged as needing user input."""
|
||
|
|
|
||
|
|
def test_awaiting_approval_needs_input(self, tmp_project):
|
||
|
|
tp = _make_task(tmp_project, "task-r", "research:awaiting_approval")
|
||
|
|
assert ap.needs_user_input({"phase": "research:awaiting_approval", "path": str(tp)}) is True
|
||
|
|
|
||
|
|
def test_human_intervention_needs_input(self, tmp_project):
|
||
|
|
tp = _make_task(tmp_project, "task-hi", "human_intervention")
|
||
|
|
assert ap.needs_user_input({"phase": "human_intervention", "path": str(tp)}) is True
|
||
|
|
|
||
|
|
def test_implement_no_input(self, tmp_project):
|
||
|
|
tp = _make_task(tmp_project, "task-i", "implement")
|
||
|
|
assert ap.needs_user_input({"phase": "implement", "path": str(tp)}) is False
|
||
|
|
|
||
|
|
def test_verdict_fail_needs_input(self, tmp_project):
|
||
|
|
tp = _make_task(tmp_project, "task-v", "referee")
|
||
|
|
_write_verdict(tp, "FAIL: bugs found")
|
||
|
|
assert ap.needs_user_input({"phase": "referee", "path": str(tp)}) is True
|
||
|
|
|
||
|
|
def test_verdict_pass_no_input(self, tmp_project):
|
||
|
|
tp = _make_task(tmp_project, "task-vp", "referee")
|
||
|
|
_write_verdict(tp, "PASS: all good")
|
||
|
|
assert ap.needs_user_input({"phase": "referee", "path": str(tp)}) is False
|
||
|
|
|
||
|
|
|
||
|
|
class TestSortByAdvancement:
|
||
|
|
"""Verify sort orders code_review correctly after the fix."""
|
||
|
|
|
||
|
|
def test_code_review_more_advanced_than_implement(self):
|
||
|
|
tasks = [
|
||
|
|
{"name": "impl-task", "base_phase": "implement"},
|
||
|
|
{"name": "cr-task", "base_phase": "code_review"},
|
||
|
|
]
|
||
|
|
sorted_tasks = ap.sort_by_advancement(tasks)
|
||
|
|
assert sorted_tasks[0]["name"] == "cr-task"
|
||
|
|
|
||
|
|
def test_bug_find_more_advanced_than_code_review(self):
|
||
|
|
tasks = [
|
||
|
|
{"name": "cr-task", "base_phase": "code_review"},
|
||
|
|
{"name": "bf-task", "base_phase": "bug_find"},
|
||
|
|
]
|
||
|
|
sorted_tasks = ap.sort_by_advancement(tasks)
|
||
|
|
assert sorted_tasks[0]["name"] == "bf-task"
|
||
|
|
|
||
|
|
|
||
|
|
class TestCmdDriveHumanIntervention:
|
||
|
|
"""Verify human_intervention is now drivable (not dead code)."""
|
||
|
|
|
||
|
|
def test_human_intervention_shows_in_blocked(self, tmp_project, capsys):
|
||
|
|
_make_task(tmp_project, "task-hi", "human_intervention")
|
||
|
|
args = _Args(project=str(tmp_project))
|
||
|
|
ap.cmd_drive(args)
|
||
|
|
out = capsys.readouterr().out
|
||
|
|
assert "ORCHESTRATION_BLOCKED" in out
|
||
|
|
assert "task-hi" in out
|
||
|
|
assert "--transition referee" in out
|
||
|
|
|
||
|
|
def test_human_intervention_not_in_unblocked(self, tmp_project, capsys):
|
||
|
|
"""human_intervention should be blocked, not selected as NEXT_TASK."""
|
||
|
|
_make_task(tmp_project, "task-hi", "human_intervention")
|
||
|
|
_make_task(tmp_project, "task-active", "implement")
|
||
|
|
args = _Args(project=str(tmp_project))
|
||
|
|
ap.cmd_drive(args)
|
||
|
|
out = capsys.readouterr().out
|
||
|
|
assert "NEXT_TASK: task-active" in out
|
||
|
|
assert "NEXT_TASK: task-hi" not in out
|
||
|
|
|
||
|
|
|
||
|
|
class TestCmdDriveSummary:
|
||
|
|
"""Integration: cmd_summary with mixed states."""
|
||
|
|
|
||
|
|
def test_summary_counts(self, tmp_project, capsys):
|
||
|
|
_make_task(tmp_project, "t-new", "new")
|
||
|
|
_make_task(tmp_project, "t-impl", "implement")
|
||
|
|
_make_task(tmp_project, "t-blocked", "research:awaiting_approval")
|
||
|
|
_make_task(tmp_project, "t-hi", "human_intervention")
|
||
|
|
complete_dir = tmp_project / ".automaton" / "tasks" / "complete"
|
||
|
|
complete_dir.mkdir()
|
||
|
|
(complete_dir / "t-done").mkdir()
|
||
|
|
(complete_dir / "t-done" / ".state").write_text("complete\n")
|
||
|
|
|
||
|
|
args = _Args(project=str(tmp_project))
|
||
|
|
ap.cmd_summary(args)
|
||
|
|
out = capsys.readouterr().out
|
||
|
|
assert "3" in out # 3 non-terminal (new, implement, blocked, hi... wait, hi is now non-terminal too)
|
||
|
|
assert "READY TO DRIVE" in out
|
||
|
|
assert "AWAITING USER" in out
|
||
|
|
assert "t-done" not in out # complete dir should not appear
|
||
|
|
assert "complete" not in out.split("READY")[0] # no "complete" as a task name
|
||
|
|
|
||
|
|
|
||
|
|
class TestCmdDriveAllPhases:
|
||
|
|
"""Every phase produces a transition command — no silent stalls."""
|
||
|
|
|
||
|
|
@pytest.mark.parametrize("phase,expected_fragment", [
|
||
|
|
("new", "--transition research"),
|
||
|
|
("research", "--transition research:awaiting_approval"),
|
||
|
|
("research:approved", "--transition decomposition"),
|
||
|
|
("decomposition", "--transition decomposition:awaiting_approval"),
|
||
|
|
("decomposition:approved", "--transition complete"),
|
||
|
|
("design", "--transition design:awaiting_approval"),
|
||
|
|
("design:approved", "--transition test_design"),
|
||
|
|
("test_design", "--transition test_design:awaiting_approval"),
|
||
|
|
("test_design:approved", "--transition implement"),
|
||
|
|
("implement", "--transition code_review"),
|
||
|
|
("code_review", "--transition code_review:awaiting_approval"),
|
||
|
|
("code_review:approved", "--transition bug_find"),
|
||
|
|
("bug_find", "--transition adversarial_bug_find"),
|
||
|
|
("adversarial_bug_find", "--transition doc_review"),
|
||
|
|
("doc_review", "--transition referee"),
|
||
|
|
("referee", "--transition complete"),
|
||
|
|
])
|
||
|
|
def test_phase_produces_transition(self, tmp_project, capsys, phase, expected_fragment):
|
||
|
|
_make_task(tmp_project, "test-task", phase)
|
||
|
|
args = _Args(project=str(tmp_project))
|
||
|
|
rc = ap.cmd_drive(args)
|
||
|
|
out = capsys.readouterr().out
|
||
|
|
assert rc == 0
|
||
|
|
assert expected_fragment in out, f"Phase '{phase}' should suggest '{expected_fragment}', got:\n{out}"
|