gitea
3d4c0926b4
CI / build (push) Has been cancelled
- Insert code_review phase between implement and bug_find - Approval gate: code_review:awaiting_approval → code_review:approved - Read-only phase — no edits, no fixes, no returning to implement - Reviewer≠implementer: .state.implementer tracking + --claim enforcement - Structured CODE_REVIEW.md: spec compliance, design conformance, quality scorecard, items found (severity/category/location/resolution), test coverage - Updated status.py (10 data structures), dashboard (4 files), prompts (3 files), agent routing, tests (6 new test classes, 19 new tests)
398 lines
15 KiB
Python
398 lines
15 KiB
Python
"""Tests for automaton.dashboard.core.task."""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from automaton.dashboard.core.task import (
|
|
determine_task_state,
|
|
discover_tasks,
|
|
parse_sub_tasks,
|
|
parse_waves,
|
|
parse_vram_config,
|
|
WaveGroup,
|
|
TaskState,
|
|
)
|
|
|
|
|
|
def _make_task(tmp_path: Path, name: str, artifacts: dict[str, str]) -> Path:
|
|
"""Create a task folder with the given artifact contents."""
|
|
task_dir = tmp_path / name
|
|
task_dir.mkdir(parents=True)
|
|
for filename, content in artifacts.items():
|
|
(task_dir / filename).write_text(content)
|
|
return task_dir
|
|
|
|
|
|
def test_backlog_state(tmp_path: Path) -> None:
|
|
task_dir = _make_task(tmp_path, "backlog-task", {})
|
|
state, artifacts = determine_task_state(task_dir)
|
|
assert state == TaskState.BACKLOG
|
|
assert not artifacts
|
|
|
|
|
|
def test_research_state(tmp_path: Path) -> None:
|
|
task_dir = _make_task(tmp_path, "research-task", {"SPEC.md": "# Spec"})
|
|
state, artifacts = determine_task_state(task_dir)
|
|
assert state == TaskState.RESEARCH
|
|
assert "SPEC.md" in artifacts
|
|
|
|
|
|
def test_implementation_state(tmp_path: Path) -> None:
|
|
task_dir = _make_task(
|
|
tmp_path,
|
|
"impl-task",
|
|
{"SPEC.md": "# Spec", "TEST_PLAN.md": "# Tests", "IMPLEMENTATION.md": "# Impl"},
|
|
)
|
|
state, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.IMPLEMENT
|
|
|
|
|
|
def test_implementation_from_test_plan(tmp_path: Path) -> None:
|
|
task_dir = _make_task(
|
|
tmp_path,
|
|
"impl-test-plan",
|
|
{"SPEC.md": "# Spec", "TEST_PLAN.md": "# Tests"},
|
|
)
|
|
state, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.IMPLEMENT
|
|
|
|
|
|
def test_bug_find_state(tmp_path: Path) -> None:
|
|
task_dir = _make_task(
|
|
tmp_path,
|
|
"bug-task",
|
|
{"SPEC.md": "# Spec", "IMPLEMENTATION.md": "# Impl", "BUG_REPORT.md": "# Bugs"},
|
|
)
|
|
state, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.BUG_FIND
|
|
|
|
|
|
def test_adv_bug_find_state(tmp_path: Path) -> None:
|
|
task_dir = _make_task(
|
|
tmp_path,
|
|
"adv-task",
|
|
{
|
|
"SPEC.md": "# Spec",
|
|
"IMPLEMENTATION.md": "# Impl",
|
|
"BUG_REPORT.md": "# Bugs",
|
|
"ADVERSARIAL_BUG_REPORT.md": "# Adv",
|
|
},
|
|
)
|
|
state, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.ADV_BUG_FIND
|
|
|
|
|
|
def test_done_state(tmp_path: Path) -> None:
|
|
task_dir = _make_task(
|
|
tmp_path,
|
|
"done-task",
|
|
{"SPEC.md": "# Spec", "VERDICT.md": "## Status: PASS"},
|
|
)
|
|
state, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.DONE
|
|
|
|
|
|
def test_blocked_state(tmp_path: Path) -> None:
|
|
task_dir = _make_task(
|
|
tmp_path,
|
|
"blocked-task",
|
|
{"SPEC.md": "# Spec", "VERDICT.md": "## Status: FAIL"},
|
|
)
|
|
state, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.BLOCKED
|
|
|
|
|
|
def test_empty_verdict_is_blocked(tmp_path: Path) -> None:
|
|
task_dir = _make_task(tmp_path, "empty-verdict", {"VERDICT.md": ""})
|
|
state, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.BLOCKED
|
|
|
|
|
|
def test_parse_sub_tasks(tmp_path: Path) -> None:
|
|
parent = tmp_path / "parent"
|
|
parent.mkdir()
|
|
sub = parent / "subtasks" / "sub-a"
|
|
sub.mkdir(parents=True)
|
|
(sub / "SPEC.md").write_text("# Spec")
|
|
(sub / "VERDICT.md").write_text("## Status: PASS")
|
|
|
|
sub_tasks = parse_sub_tasks(parent)
|
|
assert len(sub_tasks) == 1
|
|
assert sub_tasks[0].name == "sub-a"
|
|
assert sub_tasks[0].has_verdict is True
|
|
assert sub_tasks[0].verdict_status == "PASS"
|
|
|
|
|
|
def test_discover_tasks(tmp_path: Path) -> None:
|
|
_make_task(tmp_path, "aaa", {"SPEC.md": "# Spec"})
|
|
_make_task(tmp_path, "zzz", {"VERDICT.md": "## Status: PASS"})
|
|
tasks = discover_tasks(tmp_path)
|
|
assert len(tasks) == 2
|
|
# Done task should be first (sorted by advancement).
|
|
assert tasks[0].state == TaskState.DONE
|
|
assert tasks[1].state == TaskState.RESEARCH
|
|
|
|
|
|
def test_discover_tasks_skips_subtasks_root(tmp_path: Path) -> None:
|
|
parent = tmp_path / "parent"
|
|
parent.mkdir()
|
|
(parent / "SPEC.md").write_text("# Spec")
|
|
(parent / "subtasks" / "sub-a").mkdir(parents=True)
|
|
(parent / "subtasks" / "sub-a" / "SPEC.md").write_text("# Sub")
|
|
|
|
tasks = discover_tasks(tmp_path)
|
|
assert len(tasks) == 1
|
|
assert tasks[0].name == "parent"
|
|
assert len(tasks[0].sub_tasks) == 1
|
|
|
|
|
|
class TestVerdictParsing:
|
|
"""Tests for structured verdict status parsing (R1-R3 of fix-verdict-parsing SPEC)."""
|
|
|
|
def test_pass_verdict_with_fail_in_findings(self, tmp_path: Path) -> None:
|
|
ver = "## Status: PASS\n\nThe previous FAIL finding was resolved."
|
|
task_dir = _make_task(tmp_path, "pass-with-fail", {"SPEC.md": "# Spec", "VERDICT.md": ver})
|
|
state, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.DONE
|
|
|
|
def test_pass_verdict_with_needs_review_in_body(self, tmp_path: Path) -> None:
|
|
ver = "## Status: PASS\n\nNote: NEEDS_REVIEW was discussed but resolved."
|
|
task_dir = _make_task(tmp_path, "pass-with-nr", {"SPEC.md": "# Spec", "VERDICT.md": ver})
|
|
state, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.DONE
|
|
|
|
def test_fail_verdict_structured(self, tmp_path: Path) -> None:
|
|
ver = "## Status: FAIL\n\n2 tests PASS, 1 test FAIL."
|
|
task_dir = _make_task(tmp_path, "fail-mentions-pass", {"SPEC.md": "# Spec", "VERDICT.md": ver})
|
|
state, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.BLOCKED
|
|
|
|
def test_needs_review_verdict_structured(self, tmp_path: Path) -> None:
|
|
ver = "## Status: NEEDS_REVIEW\n\nSome items PASS but need review."
|
|
task_dir = _make_task(tmp_path, "nr-mentions-pass", {"SPEC.md": "# Spec", "VERDICT.md": ver})
|
|
state, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.BLOCKED
|
|
|
|
def test_verdict_with_bold_status(self, tmp_path: Path) -> None:
|
|
ver = "# Verdict\n\n- **Status**: PASS\n- **Timestamp**: 2025-01-01"
|
|
task_dir = _make_task(tmp_path, "bold-status", {"SPEC.md": "# Spec", "VERDICT.md": ver})
|
|
state, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.DONE
|
|
|
|
def test_verdict_no_status_line(self, tmp_path: Path) -> None:
|
|
ver = "# Verdict\nEverything looks good, PASS!"
|
|
task_dir = _make_task(tmp_path, "no-status-line", {"SPEC.md": "# Spec", "VERDICT.md": ver})
|
|
state, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.REFEREE # no structured ## Status: header → ambiguous
|
|
|
|
def test_verdict_no_status_no_keywords(self, tmp_path: Path) -> None:
|
|
ver = "# Verdict\n\nNeeds further discussion."
|
|
task_dir = _make_task(tmp_path, "no-status-no-keywords", {"SPEC.md": "# Spec", "VERDICT.md": ver})
|
|
state, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.REFEREE
|
|
|
|
|
|
class TestStateMachineAlignment:
|
|
"""Tests for state machine alignment with orchestrate.md (R4)."""
|
|
|
|
def test_implementation_alone_shows_implement(self, tmp_path: Path) -> None:
|
|
task_dir = _make_task(tmp_path, "impl-only", {"IMPLEMENTATION.md": "# Impl"})
|
|
state, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.IMPLEMENT
|
|
|
|
def test_bug_report_without_adversarial(self, tmp_path: Path) -> None:
|
|
task_dir = _make_task(
|
|
tmp_path, "bug-only",
|
|
{"SPEC.md": "# Spec", "IMPLEMENTATION.md": "# Impl", "BUG_REPORT.md": "# Bugs"},
|
|
)
|
|
state, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.BUG_FIND
|
|
|
|
def test_both_bug_reports_shows_adv_bug_find(self, tmp_path: Path) -> None:
|
|
task_dir = _make_task(
|
|
tmp_path, "both-bugs",
|
|
{"SPEC.md": "# Spec", "IMPLEMENTATION.md": "# Impl",
|
|
"BUG_REPORT.md": "# Bugs", "ADVERSARIAL_BUG_REPORT.md": "# Adv"},
|
|
)
|
|
state, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.ADV_BUG_FIND
|
|
|
|
def test_adv_bug_report_alone_shows_bug_find(self, tmp_path: Path) -> None:
|
|
task_dir = _make_task(
|
|
tmp_path, "adv-only",
|
|
{"SPEC.md": "# Spec", "IMPLEMENTATION.md": "# Impl",
|
|
"ADVERSARIAL_BUG_REPORT.md": "# Adv bugs only"},
|
|
)
|
|
state, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.BUG_FIND
|
|
|
|
def test_spec_alone_shows_research(self, tmp_path: Path) -> None:
|
|
task_dir = _make_task(tmp_path, "spec-only", {"SPEC.md": "# Spec"})
|
|
state, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.RESEARCH
|
|
|
|
def test_test_plan_shows_implement(self, tmp_path: Path) -> None:
|
|
task_dir = _make_task(tmp_path, "testplan", {"SPEC.md": "# Spec", "TEST_PLAN.md": "# Tests"})
|
|
state, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.IMPLEMENT
|
|
|
|
def test_design_with_spec_shows_design(self, tmp_path: Path) -> None:
|
|
task_dir = _make_task(tmp_path, "design-spec", {"SPEC.md": "# Spec", "DESIGN.md": "# Design"})
|
|
state, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.DESIGN
|
|
|
|
def test_doc_review_shows_doc_review(self, tmp_path: Path) -> None:
|
|
task_dir = _make_task(
|
|
tmp_path, "doc-review",
|
|
{"SPEC.md": "# Spec", "IMPLEMENTATION.md": "# Impl",
|
|
"BUG_REPORT.md": "# Bugs", "ADVERSARIAL_BUG_REPORT.md": "# Adv",
|
|
"DOC_REVIEW.md": "# Docs"},
|
|
)
|
|
state, _ = determine_task_state(task_dir)
|
|
assert state == TaskState.DOC_REVIEW
|
|
|
|
|
|
class TestTaskNameValidation:
|
|
"""Tests for filesystem-sourced task name validation."""
|
|
|
|
def test_valid_task_names(self, tmp_path: Path) -> None:
|
|
for name in ["my-task", "task_1", "Task-Name-123"]:
|
|
d = tmp_path / name
|
|
d.mkdir()
|
|
(d / "SPEC.md").write_text("# Spec")
|
|
tasks = discover_tasks(tmp_path)
|
|
assert len(tasks) == 3
|
|
|
|
def test_invalid_task_name_skipped(self, tmp_path: Path) -> None:
|
|
valid = tmp_path / "good-task"
|
|
valid.mkdir()
|
|
(valid / "SPEC.md").write_text("# Spec")
|
|
bad = tmp_path / "task with spaces"
|
|
bad.mkdir()
|
|
(bad / "SPEC.md").write_text("# Bad")
|
|
tasks = discover_tasks(tmp_path)
|
|
assert len(tasks) == 1
|
|
assert tasks[0].name == "good-task"
|
|
|
|
|
|
class TestParseWaves:
|
|
def test_parse_waves_from_template(self) -> None:
|
|
content = "# Decomposition: Test\n\n## Sub-Tasks\n\n### Wave 1 (Parallel)\n- subtask-a: Implement core\n- subtask-b: Implement model\n\n### Wave 2 (Dependent)\n- subtask-c: Implement UI\n- subtask-d: Implement tests\n"
|
|
waves = parse_waves(content)
|
|
assert len(waves) == 2
|
|
assert waves[0].wave_number == 1
|
|
assert waves[0].label == "Parallel"
|
|
assert waves[0].sub_task_names == ["subtask-a", "subtask-b"]
|
|
assert waves[1].wave_number == 2
|
|
assert waves[1].label == "Dependent"
|
|
assert waves[1].sub_task_names == ["subtask-c", "subtask-d"]
|
|
|
|
def test_parse_waves_empty(self) -> None:
|
|
assert parse_waves("") == []
|
|
assert parse_waves(None) == []
|
|
|
|
def test_parse_waves_no_waves(self) -> None:
|
|
content = "# No waves here\nJust text\n"
|
|
assert parse_waves(content) == []
|
|
|
|
def test_parse_waves_colon_format(self) -> None:
|
|
content = "### Wave 1: Setup\n- task-alpha: Do setup\n### Wave 2: Execution\n- task-beta: Do execution\n"
|
|
waves = parse_waves(content)
|
|
assert len(waves) == 2
|
|
assert waves[0].label == "Setup"
|
|
assert waves[0].sub_task_names == ["task-alpha"]
|
|
assert waves[1].label == "Execution"
|
|
assert waves[1].sub_task_names == ["task-beta"]
|
|
|
|
|
|
class TestDecompositionContent:
|
|
def test_task_has_decomposition_content(self, tmp_path: Path) -> None:
|
|
task_dir = tmp_path / "my-task"
|
|
task_dir.mkdir()
|
|
(task_dir / "SPEC.md").write_text("# Spec")
|
|
(task_dir / "DECOMPOSITION.md").write_text("### Wave 1 (Build)\n- sub-a: Build core\n")
|
|
tasks = discover_tasks(tmp_path)
|
|
assert len(tasks) == 1
|
|
assert tasks[0].decomposition_content is not None
|
|
assert "Wave 1" in tasks[0].decomposition_content
|
|
assert len(tasks[0].waves) == 1
|
|
assert tasks[0].waves[0].sub_task_names == ["sub-a"]
|
|
|
|
|
|
class TestParentSpecAndVramConfig:
|
|
def test_parent_spec_content(self, tmp_path: Path) -> None:
|
|
task_dir = tmp_path / "my-task"
|
|
task_dir.mkdir()
|
|
(task_dir / "SPEC.md").write_text("# Spec")
|
|
(task_dir / "PARENT_SPEC.md").write_text("# Parent Context\nDetails here")
|
|
tasks = discover_tasks(tmp_path)
|
|
assert tasks[0].parent_spec_content is not None
|
|
assert "Parent Context" in tasks[0].parent_spec_content
|
|
|
|
def test_vram_config_content(self, tmp_path: Path) -> None:
|
|
task_dir = tmp_path / "my-task"
|
|
task_dir.mkdir()
|
|
(task_dir / "SPEC.md").write_text("# Spec")
|
|
(task_dir / "VRAM_CONFIG.md").write_text("# VRAM\nmodel: llama-3")
|
|
tasks = discover_tasks(tmp_path)
|
|
assert tasks[0].vram_config_content is not None
|
|
assert "llama-3" in tasks[0].vram_config_content
|
|
|
|
def test_subtask_has_parent_spec(self, tmp_path: Path) -> None:
|
|
tasks_dir = tmp_path
|
|
parent = tasks_dir / "parent-task"
|
|
parent.mkdir()
|
|
(parent / "SPEC.md").write_text("# Parent")
|
|
subtasks_dir = parent / "subtasks"
|
|
subtasks_dir.mkdir()
|
|
sub = subtasks_dir / "child-a"
|
|
sub.mkdir()
|
|
(sub / "SPEC.md").write_text("# Child")
|
|
(sub / "PARENT_SPEC.md").write_text("# Parent Spec for child")
|
|
(sub / "VRAM_CONFIG.md").write_text("# VRAM config")
|
|
sub_tasks = parse_sub_tasks(parent)
|
|
assert len(sub_tasks) == 1
|
|
# discover_tasks doesn't recurse into subtasks for content, but parse_sub_tasks returns SubTask objects
|
|
# Verify the files exist
|
|
assert (sub / "PARENT_SPEC.md").exists()
|
|
|
|
|
|
class TestStatusReason:
|
|
def test_done_task_reason(self, tmp_path: Path) -> None:
|
|
task_dir = tmp_path / "done-task"
|
|
task_dir.mkdir()
|
|
(task_dir / "VERDICT.md").write_text("## Status: PASS\nAll good.\n")
|
|
(task_dir / "SPEC.md").write_text("# Spec\n")
|
|
tasks = discover_tasks(tmp_path)
|
|
assert tasks[0].status_reason == "Verdict: PASS"
|
|
|
|
def test_blocked_fail_reason(self, tmp_path: Path) -> None:
|
|
task_dir = tmp_path / "blocked-task"
|
|
task_dir.mkdir()
|
|
(task_dir / "VERDICT.md").write_text("## Status: FAIL\nBroken.\n")
|
|
(task_dir / "SPEC.md").write_text("# Spec\n")
|
|
tasks = discover_tasks(tmp_path)
|
|
assert "FAIL" in tasks[0].status_reason
|
|
|
|
def test_blocked_needs_review_reason(self, tmp_path: Path) -> None:
|
|
task_dir = tmp_path / "review-task"
|
|
task_dir.mkdir()
|
|
(task_dir / "VERDICT.md").write_text("## Status: NEEDS_REVIEW\nUnclear.\n")
|
|
(task_dir / "SPEC.md").write_text("# Spec\n")
|
|
tasks = discover_tasks(tmp_path)
|
|
assert "NEEDS_REVIEW" in tasks[0].status_reason
|
|
|
|
def test_bug_find_implementation_reason(self, tmp_path: Path) -> None:
|
|
task_dir = tmp_path / "impl-task"
|
|
task_dir.mkdir()
|
|
(task_dir / "IMPLEMENTATION.md").write_text("# Implementation\n")
|
|
tasks = discover_tasks(tmp_path)
|
|
assert "Implementation complete" in tasks[0].status_reason
|
|
|
|
def test_backlog_reason(self, tmp_path: Path) -> None:
|
|
task_dir = tmp_path / "empty-task"
|
|
task_dir.mkdir()
|
|
tasks = discover_tasks(tmp_path)
|
|
assert "No artifacts" in tasks[0].status_reason
|