131 lines
5.6 KiB
Python
131 lines
5.6 KiB
Python
"""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()
|