test(launcher): cover _write_task_file and reviewer verdict parsing (L-5)
This commit is contained in:
130
tests/test_launcher.py
Normal file
130
tests/test_launcher.py
Normal file
@@ -0,0 +1,130 @@
|
||||
"""Tests for launcher critical functions and reviewer verdict parsing.
|
||||
|
||||
Covers the audit-2026-06-02 fixes:
|
||||
- B-1: _write_task_file writes directly to /repos/<repo>/<task_file> (no docker),
|
||||
and raises on write failure instead of failing silently.
|
||||
- S-5: check_reviewer_verdict reads the machine-readable `verdict:` field from
|
||||
the YAML frontmatter only (no fragile substring matching).
|
||||
"""
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
import pytest
|
||||
|
||||
# Override env before importing app modules (same convention as test_qg.py)
|
||||
_test_db = os.path.join(tempfile.gettempdir(), "test_orchestrator_launcher.db")
|
||||
os.environ["ORCH_DB_PATH"] = _test_db
|
||||
os.environ["ORCH_REPOS_DIR"] = tempfile.gettempdir()
|
||||
os.environ["ORCH_GITEA_TOKEN"] = "test-token"
|
||||
os.environ["ORCH_PLANE_API_TOKEN"] = "test-token"
|
||||
|
||||
from src.agents.launcher import AgentLauncher
|
||||
from src.qg.checks import check_reviewer_verdict
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# B-1: _write_task_file
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestWriteTaskFile:
|
||||
def test_writes_to_repos_volume_path(self, tmp_path, monkeypatch):
|
||||
"""Task file is written to <repos_dir>/<repo>/<task_file>, content matches."""
|
||||
monkeypatch.setattr("src.agents.launcher.settings.repos_dir", str(tmp_path))
|
||||
repo_dir = tmp_path / "enduro-trails"
|
||||
repo_dir.mkdir()
|
||||
|
||||
launcher = AgentLauncher()
|
||||
launcher._write_task_file("enduro-trails", ".task-dev.md", "hello-content")
|
||||
|
||||
written = repo_dir / ".task-dev.md"
|
||||
assert written.is_file()
|
||||
assert written.read_text() == "hello-content"
|
||||
|
||||
def test_does_not_use_docker(self, tmp_path, monkeypatch):
|
||||
"""No subprocess/docker call: if subprocess.run were used it would error here."""
|
||||
monkeypatch.setattr("src.agents.launcher.settings.repos_dir", str(tmp_path))
|
||||
repo_dir = tmp_path / "enduro-trails"
|
||||
repo_dir.mkdir()
|
||||
|
||||
called = {"run": False}
|
||||
|
||||
def _fail_run(*a, **k):
|
||||
called["run"] = True
|
||||
raise AssertionError("subprocess.run must not be called by _write_task_file")
|
||||
|
||||
monkeypatch.setattr("src.agents.launcher.subprocess.run", _fail_run)
|
||||
|
||||
launcher = AgentLauncher()
|
||||
launcher._write_task_file("enduro-trails", ".task.md", "x")
|
||||
assert called["run"] is False
|
||||
|
||||
def test_raises_on_write_failure(self, tmp_path, monkeypatch):
|
||||
"""If the target dir does not exist, raise RuntimeError (do not fail silently)."""
|
||||
monkeypatch.setattr("src.agents.launcher.settings.repos_dir", str(tmp_path))
|
||||
# repo dir intentionally NOT created -> open() raises OSError
|
||||
|
||||
launcher = AgentLauncher()
|
||||
with pytest.raises(RuntimeError):
|
||||
launcher._write_task_file("nonexistent-repo", ".task.md", "x")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# S-5: check_reviewer_verdict (frontmatter-only)
|
||||
# ---------------------------------------------------------------------------
|
||||
@pytest.fixture
|
||||
def review_repo(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr("src.qg.checks.settings.repos_dir", str(tmp_path))
|
||||
wi_dir = tmp_path / "enduro-trails" / "docs" / "work-items" / "ET-001"
|
||||
wi_dir.mkdir(parents=True)
|
||||
return wi_dir
|
||||
|
||||
|
||||
def _write_review(wi_dir, text):
|
||||
(wi_dir / "12-review.md").write_text(text)
|
||||
|
||||
|
||||
class TestCheckReviewerVerdict:
|
||||
def test_approved_in_frontmatter(self, review_repo):
|
||||
_write_review(review_repo, "---\ntype: review\nverdict: APPROVED\n---\n# Review\nbody\n")
|
||||
passed, reason = check_reviewer_verdict("enduro-trails", "ET-001")
|
||||
assert passed is True
|
||||
assert "APPROVED" in reason
|
||||
|
||||
def test_request_changes_in_frontmatter(self, review_repo):
|
||||
_write_review(review_repo, "---\ntype: review\nverdict: REQUEST_CHANGES\n---\n# Review\n")
|
||||
passed, reason = check_reviewer_verdict("enduro-trails", "ET-001")
|
||||
assert passed is False
|
||||
assert "REQUEST_CHANGES" in reason
|
||||
|
||||
def test_lowercase_verdict_normalized(self, review_repo):
|
||||
_write_review(review_repo, "---\nverdict: approved\n---\nbody\n")
|
||||
passed, _ = check_reviewer_verdict("enduro-trails", "ET-001")
|
||||
assert passed is True
|
||||
|
||||
def test_no_verdict_field_is_not_approved(self, review_repo):
|
||||
# Frontmatter present but no verdict -> must NOT approve.
|
||||
_write_review(review_repo, "---\ntype: review\nstatus: done\n---\nbody\n")
|
||||
passed, reason = check_reviewer_verdict("enduro-trails", "ET-001")
|
||||
assert passed is False
|
||||
assert "verdict" in reason.lower()
|
||||
|
||||
def test_no_frontmatter_is_not_approved(self, review_repo):
|
||||
# APPROVED appears only in body/table text -> must NOT cause false positive (S-5).
|
||||
_write_review(review_repo, "# Review\n| Finding | Status |\n|---|---|\n| F-01 | APPROVED |\n")
|
||||
passed, _ = check_reviewer_verdict("enduro-trails", "ET-001")
|
||||
assert passed is False
|
||||
|
||||
def test_request_changes_in_body_does_not_block_approved_frontmatter(self, review_repo):
|
||||
# Body mentions REQUEST_CHANGES in a table, but frontmatter verdict is APPROVED.
|
||||
_write_review(
|
||||
review_repo,
|
||||
"---\nverdict: APPROVED\n---\n# Review\n"
|
||||
"| Item | Old verdict |\n|---|---|\n| x | REQUEST_CHANGES |\n",
|
||||
)
|
||||
passed, reason = check_reviewer_verdict("enduro-trails", "ET-001")
|
||||
assert passed is True
|
||||
assert "APPROVED" in reason
|
||||
|
||||
def test_missing_file(self, review_repo):
|
||||
passed, reason = check_reviewer_verdict("enduro-trails", "ET-999")
|
||||
assert passed is False
|
||||
assert "not found" in reason.lower()
|
||||
Reference in New Issue
Block a user