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, ) from src.stages import get_qg_for_stage @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 TestDevelopmentStageQG: """BUG 6: development stage QG is now check_ci_green (CI is the authoritative gate), not the deprecated check_tests_local.""" def test_development_qg_is_check_ci_green(self): assert get_qg_for_stage("development") == "check_ci_green" def test_check_tests_local_is_deprecated_and_unwired(self): # Kept in the registry for backward-compat, but not wired to any stage. from src.qg.checks import QG_CHECKS from src.stages import STAGE_TRANSITIONS assert "check_tests_local" in QG_CHECKS wired = {t.get("qg") for t in STAGE_TRANSITIONS.values()} assert "check_tests_local" not in wired 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")