228 lines
8.3 KiB
Python
228 lines
8.3 KiB
Python
import pytest
|
|
import os
|
|
import tempfile
|
|
from unittest.mock import patch, MagicMock
|
|
import httpx
|
|
|
|
# Override DB path before importing app
|
|
_test_db = os.path.join(tempfile.gettempdir(), "test_orchestrator.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.qg.checks import (
|
|
check_analysis_complete,
|
|
check_architecture_done,
|
|
check_ci_green,
|
|
check_review_approved,
|
|
check_tests_passed,
|
|
check_tests_local,
|
|
)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def setup_work_item_dir(tmp_path, monkeypatch):
|
|
"""Create temp repo structure for filesystem checks."""
|
|
monkeypatch.setattr("src.qg.checks.settings.repos_dir", str(tmp_path))
|
|
repo_dir = tmp_path / "enduro-trails"
|
|
repo_dir.mkdir()
|
|
return repo_dir
|
|
|
|
|
|
class TestCheckAnalysisComplete:
|
|
def test_all_files_present(self, setup_work_item_dir):
|
|
repo_dir = setup_work_item_dir
|
|
wi_dir = repo_dir / "docs" / "work-items" / "ET-001"
|
|
wi_dir.mkdir(parents=True)
|
|
(wi_dir / "01-brd.md").write_text("# BRD")
|
|
(wi_dir / "02-trz.md").write_text("# TRZ")
|
|
(wi_dir / "03-acceptance-criteria.md").write_text("# AC")
|
|
(wi_dir / "04-test-plan.yaml").write_text("tests: []")
|
|
|
|
passed, reason = check_analysis_complete("enduro-trails", "ET-001")
|
|
assert passed is True
|
|
|
|
def test_missing_files(self, setup_work_item_dir):
|
|
repo_dir = setup_work_item_dir
|
|
wi_dir = repo_dir / "docs" / "work-items" / "ET-002"
|
|
wi_dir.mkdir(parents=True)
|
|
(wi_dir / "01-brd.md").write_text("# BRD")
|
|
|
|
passed, reason = check_analysis_complete("enduro-trails", "ET-002")
|
|
assert passed is False
|
|
assert "Missing files" in reason
|
|
|
|
def test_no_directory(self, setup_work_item_dir):
|
|
passed, reason = check_analysis_complete("enduro-trails", "ET-999")
|
|
assert passed is False
|
|
|
|
|
|
class TestCheckArchitectureDone:
|
|
def test_adr_directory_with_files(self, setup_work_item_dir):
|
|
repo_dir = setup_work_item_dir
|
|
adr_dir = repo_dir / "docs" / "work-items" / "ET-001" / "06-adr"
|
|
adr_dir.mkdir(parents=True)
|
|
(adr_dir / "001-use-postgres.md").write_text("# ADR")
|
|
|
|
passed, reason = check_architecture_done("enduro-trails", "ET-001")
|
|
assert passed is True
|
|
|
|
def test_infra_requirements(self, setup_work_item_dir):
|
|
repo_dir = setup_work_item_dir
|
|
wi_dir = repo_dir / "docs" / "work-items" / "ET-001"
|
|
wi_dir.mkdir(parents=True)
|
|
(wi_dir / "07-infra-requirements.md").write_text("# Infra")
|
|
|
|
passed, reason = check_architecture_done("enduro-trails", "ET-001")
|
|
assert passed is True
|
|
|
|
def test_empty_adr_directory(self, setup_work_item_dir):
|
|
repo_dir = setup_work_item_dir
|
|
adr_dir = repo_dir / "docs" / "work-items" / "ET-001" / "06-adr"
|
|
adr_dir.mkdir(parents=True)
|
|
|
|
passed, reason = check_architecture_done("enduro-trails", "ET-001")
|
|
assert passed is False
|
|
|
|
def test_nothing_present(self, setup_work_item_dir):
|
|
passed, reason = check_architecture_done("enduro-trails", "ET-001")
|
|
assert passed is False
|
|
|
|
|
|
class TestCheckCIGreen:
|
|
@patch("src.qg.checks.httpx.get")
|
|
def test_ci_success(self, mock_get):
|
|
mock_resp = MagicMock()
|
|
mock_resp.status_code = 200
|
|
mock_resp.json.return_value = {"state": "success"}
|
|
mock_resp.raise_for_status = MagicMock()
|
|
mock_get.return_value = mock_resp
|
|
|
|
passed, reason = check_ci_green("enduro-trails", "feature/ET-001-test")
|
|
assert passed is True
|
|
assert "green" in reason.lower()
|
|
|
|
@patch("src.qg.checks.httpx.get")
|
|
def test_ci_pending(self, mock_get):
|
|
mock_resp = MagicMock()
|
|
mock_resp.status_code = 200
|
|
mock_resp.json.return_value = {"state": "pending"}
|
|
mock_resp.raise_for_status = MagicMock()
|
|
mock_get.return_value = mock_resp
|
|
|
|
passed, reason = check_ci_green("enduro-trails", "feature/ET-001-test")
|
|
assert passed is False
|
|
|
|
@patch("src.qg.checks.httpx.get")
|
|
def test_ci_branch_not_found(self, mock_get):
|
|
mock_resp = MagicMock()
|
|
mock_resp.status_code = 404
|
|
mock_get.return_value = mock_resp
|
|
|
|
passed, reason = check_ci_green("enduro-trails", "nonexistent")
|
|
assert passed is False
|
|
|
|
|
|
class TestCheckReviewApproved:
|
|
@patch("src.qg.checks.httpx.get")
|
|
def test_approved(self, mock_get):
|
|
mock_resp = MagicMock()
|
|
mock_resp.status_code = 200
|
|
mock_resp.json.return_value = [
|
|
{"state": "APPROVED", "user": {"login": "reviewer1"}}
|
|
]
|
|
mock_resp.raise_for_status = MagicMock()
|
|
mock_get.return_value = mock_resp
|
|
|
|
passed, reason = check_review_approved("enduro-trails", 1)
|
|
assert passed is True
|
|
|
|
@patch("src.qg.checks.httpx.get")
|
|
def test_changes_requested(self, mock_get):
|
|
mock_resp = MagicMock()
|
|
mock_resp.status_code = 200
|
|
mock_resp.json.return_value = [
|
|
{"state": "REQUEST_CHANGES", "user": {"login": "reviewer1"}}
|
|
]
|
|
mock_resp.raise_for_status = MagicMock()
|
|
mock_get.return_value = mock_resp
|
|
|
|
passed, reason = check_review_approved("enduro-trails", 1)
|
|
assert passed is False
|
|
assert "Changes requested" in reason
|
|
|
|
@patch("src.qg.checks.httpx.get")
|
|
def test_no_reviews(self, mock_get):
|
|
mock_resp = MagicMock()
|
|
mock_resp.status_code = 200
|
|
mock_resp.json.return_value = []
|
|
mock_resp.raise_for_status = MagicMock()
|
|
mock_get.return_value = mock_resp
|
|
|
|
passed, reason = check_review_approved("enduro-trails", 1)
|
|
assert passed is False
|
|
|
|
|
|
class TestCheckTestsPassed:
|
|
def test_report_with_pass(self, setup_work_item_dir):
|
|
repo_dir = setup_work_item_dir
|
|
wi_dir = repo_dir / "docs" / "work-items" / "ET-001"
|
|
wi_dir.mkdir(parents=True)
|
|
(wi_dir / "13-test-report.md").write_text("# Test Report\n\nResult: PASS\n")
|
|
|
|
passed, reason = check_tests_passed("enduro-trails", "ET-001")
|
|
assert passed is True
|
|
|
|
def test_report_without_pass(self, setup_work_item_dir):
|
|
repo_dir = setup_work_item_dir
|
|
wi_dir = repo_dir / "docs" / "work-items" / "ET-001"
|
|
wi_dir.mkdir(parents=True)
|
|
(wi_dir / "13-test-report.md").write_text("# Test Report\n\nResult: FAIL\n")
|
|
|
|
passed, reason = check_tests_passed("enduro-trails", "ET-001")
|
|
assert passed is False
|
|
|
|
def test_no_report(self, setup_work_item_dir):
|
|
passed, reason = check_tests_passed("enduro-trails", "ET-001")
|
|
assert passed is False
|
|
assert "not found" in reason.lower()
|
|
|
|
|
|
class TestCheckTestsLocal:
|
|
"""BUG 5: check_tests_local must run pytest directly (not make, which is
|
|
not installed in the orchestrator container)."""
|
|
|
|
@patch("src.qg.checks.ensure_worktree")
|
|
@patch("subprocess.run")
|
|
def test_passes_on_returncode_zero(self, mock_run, mock_wt, tmp_path):
|
|
mock_wt.return_value = str(tmp_path)
|
|
mock_run.return_value = MagicMock(returncode=0, stdout="ok", stderr="")
|
|
passed, reason = check_tests_local("enduro-trails", "feature/ET-001-x")
|
|
assert passed is True
|
|
assert reason == "Local tests passed"
|
|
|
|
@patch("src.qg.checks.ensure_worktree")
|
|
@patch("subprocess.run")
|
|
def test_fails_on_nonzero_returncode(self, mock_run, mock_wt, tmp_path):
|
|
mock_wt.return_value = str(tmp_path)
|
|
mock_run.return_value = MagicMock(returncode=1, stdout="boom", stderr="trace")
|
|
passed, reason = check_tests_local("enduro-trails", "feature/ET-001-x")
|
|
assert passed is False
|
|
assert "Local tests failed" in reason
|
|
|
|
@patch("src.qg.checks.ensure_worktree")
|
|
@patch("subprocess.run")
|
|
def test_invokes_pytest_not_make(self, mock_run, mock_wt, tmp_path):
|
|
"""The subprocess call must be pytest, from src/api, against ../../tests/."""
|
|
mock_wt.return_value = str(tmp_path)
|
|
mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="")
|
|
check_tests_local("enduro-trails", "feature/ET-001-x")
|
|
args, kwargs = mock_run.call_args
|
|
cmd = args[0]
|
|
assert "make" not in cmd
|
|
assert cmd[:3] == ["python", "-m", "pytest"]
|
|
assert "../../tests/" in cmd
|
|
assert kwargs["cwd"] == os.path.join(str(tmp_path), "src", "api")
|
|
|