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, check_deploy_status, ) 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 TestCheckDeployStatus: """BUG 8: deploy -> done must be gated on the deployer's machine-readable deploy_status verdict in 14-deploy-log.md frontmatter, NOT the LLM exit code (always 0). Mirrors check_reviewer_verdict (reads ONLY the frontmatter field).""" def _write_log(self, repo_dir, content): wi_dir = repo_dir / "docs" / "work-items" / "ET-011" wi_dir.mkdir(parents=True) (wi_dir / "14-deploy-log.md").write_text(content) def test_success_verdict_passes(self, setup_work_item_dir): self._write_log( setup_work_item_dir, "---\ndeploy_status: SUCCESS\nversion: v0.0.3\n---\n\nDeployed OK.\n", ) passed, reason = check_deploy_status("enduro-trails", "ET-011") assert passed is True assert "SUCCESS" in reason def test_failed_verdict_fails(self, setup_work_item_dir): self._write_log( setup_work_item_dir, "---\ndeploy_status: FAILED\nversion: v0.0.3\n---\n\npermission denied.\n", ) passed, reason = check_deploy_status("enduro-trails", "ET-011") assert passed is False assert "FAILED" in reason def test_no_file_fails(self, setup_work_item_dir): passed, reason = check_deploy_status("enduro-trails", "ET-011") assert passed is False assert "not found" in reason.lower() def test_no_field_fails(self, setup_work_item_dir): # Frontmatter present but no deploy_status field -> must NOT pass. self._write_log( setup_work_item_dir, "---\nversion: v0.0.3\n---\n\nStatus: FAILED (prose only).\n", ) passed, reason = check_deploy_status("enduro-trails", "ET-011") assert passed is False def test_prose_only_no_frontmatter_fails(self, setup_work_item_dir): # Prose mentioning SUCCESS but no machine-readable frontmatter -> fail. self._write_log( setup_work_item_dir, "# Deploy log\n\nStatus: SUCCESS (prose, not frontmatter).\n", ) passed, reason = check_deploy_status("enduro-trails", "ET-011") assert passed is False def test_deploy_stage_qg_is_check_deploy_status(self): assert get_qg_for_stage("deploy") == "check_deploy_status" def test_registered_in_qg_checks(self): from src.qg.checks import QG_CHECKS assert QG_CHECKS.get("check_deploy_status") is check_deploy_status 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")