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, check_staging_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 def _ci_status_resp(state, status_code=200): """Build a MagicMock httpx response for the Gitea combined-status endpoint.""" mock_resp = MagicMock() mock_resp.status_code = status_code mock_resp.json.return_value = {"state": state} mock_resp.raise_for_status = MagicMock() return mock_resp class TestCheckCIGreen: """ORCH-045: check_ci_green now polls with retry to ride out a transient `pending` right after the developer push (race fix, see ORCH-017).""" @patch("src.qg.checks.time.sleep") @patch("src.qg.checks.httpx.get") def test_ci_success_first_attempt(self, mock_get, mock_sleep): mock_get.return_value = _ci_status_resp("success") passed, reason = check_ci_green("enduro-trails", "feature/ET-001-test") assert passed is True assert "green" in reason.lower() assert mock_get.call_count == 1 mock_sleep.assert_not_called() @patch("src.qg.checks.time.sleep") @patch("src.qg.checks.httpx.get") def test_ci_pending_then_success(self, mock_get, mock_sleep): # pending on the 1st poll, green on the 2nd -> success after one retry. mock_get.side_effect = [ _ci_status_resp("pending"), _ci_status_resp("success"), ] passed, reason = check_ci_green("enduro-trails", "feature/ET-001-test") assert passed is True assert "green" in reason.lower() assert mock_get.call_count == 2 assert mock_sleep.call_count == 1 # slept once between the two polls @patch("src.qg.checks.time.sleep") @patch("src.qg.checks.httpx.get") def test_ci_failure_no_retry(self, mock_get, mock_sleep): # CI is red -> terminal, return immediately without sleeping/retrying. mock_get.return_value = _ci_status_resp("failure") passed, reason = check_ci_green("enduro-trails", "feature/ET-001-test") assert passed is False assert "failure" in reason assert mock_get.call_count == 1 mock_sleep.assert_not_called() @patch("src.qg.checks.time.sleep") @patch("src.qg.checks.httpx.get") def test_ci_pending_exhausts_attempts(self, mock_get, mock_sleep): # Always pending -> after ci_poll_max_attempts polls return an explicit # (False, "...pending...") so the operator sees the reason (no silent stall). from src.qg.checks import settings as checks_settings mock_get.return_value = _ci_status_resp("pending") passed, reason = check_ci_green("enduro-trails", "feature/ET-001-test") assert passed is False assert "pending" in reason.lower() assert mock_get.call_count == checks_settings.ci_poll_max_attempts @patch("src.qg.checks.time.sleep") @patch("src.qg.checks.httpx.get") def test_ci_branch_not_found(self, mock_get, mock_sleep): 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 assert "not found" in reason.lower() assert mock_get.call_count == 1 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: """ET-013 fix: testing -> deploy gate reads the tester's MACHINE-READABLE verdict in 13-test-report.md frontmatter (verdict:/status:), NOT a substring of the body. Mirrors check_reviewer_verdict / check_deploy_status. The old `if "PASS" in content` let a `verdict: BLOCKED` report whose prose said "23 passed"/"✅ PASS" pass the gate, shipping an unfinished feature to Done.""" def _write(self, repo_dir, content, wi="ET-001"): wi_dir = repo_dir / "docs" / "work-items" / wi wi_dir.mkdir(parents=True) (wi_dir / "13-test-report.md").write_text(content) def test_verdict_pass_passes(self, setup_work_item_dir): # Most common real form (ET-001/002/005/009/011/012/014). self._write( setup_work_item_dir, "---\ntype: test-report\nverdict: PASS\nstatus: pass\n---\n\n# Test Report\n", ) passed, reason = check_tests_passed("enduro-trails", "ET-001") assert passed is True assert "PASS" in reason def test_verdict_pass_ready_to_deploy_passes(self, setup_work_item_dir): # ET-007 real form: "PASS — ready-to-deploy". self._write( setup_work_item_dir, "---\nverdict: PASS — ready-to-deploy\nstatus: PASS\n---\n\nbody\n", ) passed, reason = check_tests_passed("enduro-trails", "ET-001") assert passed is True def test_verdict_ready_to_deploy_with_status_passed_passes(self, setup_work_item_dir): # ET-006 real form: verdict has no PASS word, but status: PASSED. self._write( setup_work_item_dir, "---\nverdict: ready-to-deploy\nstatus: PASSED\n---\n\nbody\n", ) passed, reason = check_tests_passed("enduro-trails", "ET-001") assert passed is True def test_verdict_stage_ready_to_deploy_with_status_pass_passes(self, setup_work_item_dir): # ET-008 real form: verdict: stage:ready-to-deploy, status: pass. self._write( setup_work_item_dir, "---\nverdict: stage:ready-to-deploy\nstatus: pass\n---\n\nbody\n", ) passed, reason = check_tests_passed("enduro-trails", "ET-001") assert passed is True def test_blocked_verdict_with_pass_in_body_fails(self, setup_work_item_dir): # THE ET-013 BUG: verdict BLOCKED but body is full of "PASS"/"passed". self._write( setup_work_item_dir, "---\ntype: test-report\nstatus: blocked\nverdict: BLOCKED\n---\n\n" "23 passed\n✅ PASS (часть AC-18)\nAll checks passed\n", ) passed, reason = check_tests_passed("enduro-trails", "ET-001") assert passed is False assert "BLOCKED" in reason def test_failed_verdict_fails(self, setup_work_item_dir): self._write( setup_work_item_dir, "---\nverdict: FAILED\nstatus: failed\n---\n\nbody\n", ) passed, reason = check_tests_passed("enduro-trails", "ET-001") assert passed is False assert "FAILED" in reason def test_passed_count_in_body_but_blocked_verdict_fails(self, setup_work_item_dir): # Body says "23 passed" but frontmatter verdict BLOCKED -> substring no longer fools. self._write( setup_work_item_dir, "---\nverdict: BLOCKED\n---\n\nTests: 23 passed, 0 failed.\n", ) passed, reason = check_tests_passed("enduro-trails", "ET-001") assert passed is False def test_no_frontmatter_fails(self, setup_work_item_dir): # Old format / prose only -> no machine verdict -> fail. self._write( setup_work_item_dir, "# Test Report\n\nResult: PASS\nAll tests passed.\n", ) passed, reason = check_tests_passed("enduro-trails", "ET-001") assert passed is False def test_no_verdict_field_fails(self, setup_work_item_dir): # Frontmatter present but neither verdict nor status -> fail. self._write( setup_work_item_dir, "---\ntype: test-report\nversion: 1\n---\n\nResult: PASS\n", ) passed, reason = check_tests_passed("enduro-trails", "ET-001") assert passed is False def test_invalid_yaml_fails_no_exception(self, setup_work_item_dir): # Broken YAML frontmatter -> False with reason, never raises. self._write( setup_work_item_dir, "---\nverdict: [unclosed\n : : :\n---\n\nbody PASS\n", ) passed, reason = check_tests_passed("enduro-trails", "ET-001") assert passed is False assert "YAML" in reason or "frontmatter" in reason.lower() 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() # --- ORCH-047: `result:` is read as an equal-rank machine field --- def test_result_pass_passes(self, setup_work_item_dir): # TC-01 / AC-01: canonical tester field `result: PASS` (no verdict/status). self._write( setup_work_item_dir, "---\ntype: test-report\nresult: PASS\n---\n\n# Test Report\n", ) passed, reason = check_tests_passed("enduro-trails", "ET-001") assert passed is True assert "PASS" in reason def test_result_fail_fails(self, setup_work_item_dir): # TC-02 / AC-02: `result: FAIL` (no verdict/status) -> rollback, reason has FAIL. self._write(setup_work_item_dir, "---\nresult: FAIL\n---\n\nbody\n") passed, reason = check_tests_passed("enduro-trails", "ET-001") assert passed is False assert "FAIL" in reason def test_result_pass_but_verdict_blocked_fails(self, setup_work_item_dir): # TC-03 / AC-03: negative in another field is authoritative over result: PASS. self._write( setup_work_item_dir, "---\nresult: PASS\nverdict: BLOCKED\n---\n\n23 passed\n", ) passed, reason = check_tests_passed("enduro-trails", "ET-001") assert passed is False assert "BLOCKED" in reason def test_result_pass_but_status_failed_fails(self, setup_work_item_dir): # TC-04 / AC-03: status: failed authoritative over result: PASS. self._write( setup_work_item_dir, "---\nresult: PASS\nstatus: failed\n---\n\nbody\n", ) passed, reason = check_tests_passed("enduro-trails", "ET-001") assert passed is False assert "FAILED" in reason def test_result_ready_to_deploy_passes(self, setup_work_item_dir): # TC-05 / AC-04: positive token without the word PASS, in result field. self._write( setup_work_item_dir, "---\nresult: ready-to-deploy\n---\n\nbody\n", ) passed, reason = check_tests_passed("enduro-trails", "ET-001") assert passed is True def test_no_machine_field_reason_mentions_result(self, setup_work_item_dir): # AC-06: none of result/verdict/status -> fail; reason now lists result too. self._write( setup_work_item_dir, "---\ntype: test-report\nversion: 1\n---\n\nResult: PASS\n", ) passed, reason = check_tests_passed("enduro-trails", "ET-001") assert passed is False assert "result" 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 # --- ET-013 path-sync fix: log written to origin/main via separate PR --- def test_origin_main_success_passes_when_absent_in_worktree(self, monkeypatch): # Deployer merged 14-deploy-log.md into main via a separate PR; it is NOT # in the feature worktree. Gate must recover it from origin/main -> PASS. # (This is the exact ET-013 regression.) monkeypatch.setattr( "src.qg.checks._deploy_log_from_main", lambda repo, wi: "---\ndeploy_status: SUCCESS\nversion: v0.0.5\n---\n\nLive.\n", ) passed, reason = check_deploy_status("enduro-trails", "ET-013") assert passed is True assert "SUCCESS" in reason def test_origin_main_failed_fails(self, monkeypatch): # A genuine FAILED log in main must still fail. monkeypatch.setattr( "src.qg.checks._deploy_log_from_main", lambda repo, wi: "---\ndeploy_status: FAILED\nversion: v0.0.5\n---\n\nboom.\n", ) passed, reason = check_deploy_status("enduro-trails", "ET-013") assert passed is False assert "FAILED" in reason def test_absent_everywhere_fails(self, monkeypatch): # Not in worktree and origin/main lookup yields nothing -> not found. monkeypatch.setattr( "src.qg.checks._deploy_log_from_main", lambda repo, wi: None ) passed, reason = check_deploy_status("enduro-trails", "ET-013") assert passed is False assert "not found" in reason.lower() @patch("src.qg.checks.subprocess.run") @patch("src.qg.checks.os.path.isdir", return_value=True) def test_fetch_failure_degrades_no_exception(self, mock_isdir, mock_run): # git fetch/show raising (e.g. network) must degrade to "not found", # never propagate an exception out of the gate. import subprocess as _sp mock_run.side_effect = _sp.TimeoutExpired(cmd="git", timeout=30) passed, reason = check_deploy_status("enduro-trails", "ET-013") assert passed is False assert "not found" in reason.lower() def test_worktree_log_short_circuits_main_lookup(self, setup_work_item_dir, monkeypatch): # If the log IS present in the worktree, origin/main must NOT be consulted. self._write_log( setup_work_item_dir, "---\ndeploy_status: SUCCESS\nversion: v0.0.3\n---\n\nDeployed OK.\n", ) called = {"n": 0} def _boom(repo, wi): called["n"] += 1 return None monkeypatch.setattr("src.qg.checks._deploy_log_from_main", _boom) passed, reason = check_deploy_status("enduro-trails", "ET-011") assert passed is True assert called["n"] == 0 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") class TestCheckStagingStatus: """ORCH-35 conditional gate (Variant A): deploy-staging gate is active ONLY for the self-hosting orchestrator repo (has staging infra on localhost:8501). All other repos pass immediately with "Staging gate N/A for ". Self-hosting path: reads machine-readable staging_status: from 15-staging-log.md frontmatter. Mirrors check_deploy_status pattern. """ @pytest.fixture() def orch_dir(self, tmp_path, monkeypatch): """Temp orchestrator repo dir (self-hosting).""" monkeypatch.setattr("src.qg.checks.settings.repos_dir", str(tmp_path)) d = tmp_path / "orchestrator" d.mkdir(exist_ok=True) return d def _write_log(self, repo_dir, content, wi="ORCH-035"): wi_dir = repo_dir / "docs" / "work-items" / wi wi_dir.mkdir(parents=True, exist_ok=True) (wi_dir / "15-staging-log.md").write_text(content) # ------------------------------------------------------------------ # Self-hosting (orchestrator) path -- real file check # ------------------------------------------------------------------ def test_success_verdict_passes(self, orch_dir): self._write_log( orch_dir, "---\nstaging_status: SUCCESS\ntimestamp: 2026-06-05T00:00:00Z\n---\n\nAll staging tests passed.\n", ) from src.qg.checks import check_staging_status passed, reason = check_staging_status("orchestrator", "ORCH-035") assert passed is True assert "SUCCESS" in reason def test_failed_verdict_fails(self, orch_dir): self._write_log( orch_dir, "---\nstaging_status: FAILED\ntimestamp: 2026-06-05T00:00:00Z\n---\n\n2 tests failed.\n", ) from src.qg.checks import check_staging_status passed, reason = check_staging_status("orchestrator", "ORCH-035") assert passed is False assert "FAILED" in reason def test_no_file_fails_for_self_hosting(self, orch_dir): from src.qg.checks import check_staging_status passed, reason = check_staging_status("orchestrator", "ORCH-035") assert passed is False assert "not found" in reason.lower() def test_no_field_fails(self, orch_dir): # Frontmatter present but no staging_status field -> must NOT pass. self._write_log( orch_dir, "---\nversion: v0.0.3\n---\n\nStatus: all good (prose only).\n", ) from src.qg.checks import check_staging_status passed, reason = check_staging_status("orchestrator", "ORCH-035") assert passed is False def test_prose_only_no_frontmatter_fails(self, orch_dir): # Prose mentioning SUCCESS but no machine-readable frontmatter -> fail. self._write_log( orch_dir, "# Staging Log\n\nStatus: SUCCESS (prose, not frontmatter).\n", ) from src.qg.checks import check_staging_status passed, reason = check_staging_status("orchestrator", "ORCH-035") assert passed is False def test_origin_main_success_passes_when_absent_in_worktree(self, monkeypatch): # Deployer merged 15-staging-log.md into main; not in worktree -> recover from main. monkeypatch.setattr( "src.qg.checks._staging_log_from_main", lambda repo, wi: "---\nstaging_status: SUCCESS\n---\n\nAll good.\n", ) from src.qg.checks import check_staging_status passed, reason = check_staging_status("orchestrator", "ORCH-035-main") assert passed is True assert "SUCCESS" in reason def test_origin_main_failed_fails(self, monkeypatch): monkeypatch.setattr( "src.qg.checks._staging_log_from_main", lambda repo, wi: "---\nstaging_status: FAILED\n---\n\nboom.\n", ) from src.qg.checks import check_staging_status passed, reason = check_staging_status("orchestrator", "ORCH-035-main") assert passed is False assert "FAILED" in reason def test_absent_everywhere_fails(self, monkeypatch): monkeypatch.setattr( "src.qg.checks._staging_log_from_main", lambda repo, wi: None ) from src.qg.checks import check_staging_status passed, reason = check_staging_status("orchestrator", "ORCH-035-absent") assert passed is False assert "not found" in reason.lower() # ------------------------------------------------------------------ # Non-self-hosting path -- instant pass, no file dependency # ------------------------------------------------------------------ def test_non_self_hosting_passes_immediately_no_file(self, tmp_path, monkeypatch): """Non-self-hosting repo: gate is N/A even without a staging log file.""" monkeypatch.setattr("src.qg.checks.settings.repos_dir", str(tmp_path)) from src.qg.checks import check_staging_status passed, reason = check_staging_status("enduro-trails", "ET-035") assert passed is True assert "N/A" in reason assert "enduro-trails" in reason def test_non_self_hosting_passes_regardless_of_file_content(self, tmp_path, monkeypatch): """Even a FAILED staging log must not block a non-self-hosting repo.""" monkeypatch.setattr("src.qg.checks.settings.repos_dir", str(tmp_path)) et_dir = tmp_path / "enduro-trails" / "docs" / "work-items" / "ET-035" et_dir.mkdir(parents=True) (et_dir / "15-staging-log.md").write_text( "---\nstaging_status: FAILED\n---\nShould be ignored.\n" ) from src.qg.checks import check_staging_status passed, reason = check_staging_status("enduro-trails", "ET-035") assert passed is True assert "N/A" in reason def test_unknown_repo_also_passes_immediately(self, tmp_path, monkeypatch): """Any repo that is not orchestrator gets N/A gate.""" monkeypatch.setattr("src.qg.checks.settings.repos_dir", str(tmp_path)) from src.qg.checks import check_staging_status passed, reason = check_staging_status("some-other-project", "XY-001") assert passed is True assert "N/A" in reason # ------------------------------------------------------------------ # is_self_hosting_repo helper # ------------------------------------------------------------------ def test_is_self_hosting_true_for_orchestrator(self): from src.qg.checks import is_self_hosting_repo assert is_self_hosting_repo("orchestrator") is True def test_is_self_hosting_case_insensitive(self): from src.qg.checks import is_self_hosting_repo assert is_self_hosting_repo("Orchestrator") is True assert is_self_hosting_repo("ORCHESTRATOR") is True def test_is_self_hosting_false_for_enduro_trails(self): from src.qg.checks import is_self_hosting_repo assert is_self_hosting_repo("enduro-trails") is False def test_is_self_hosting_false_for_empty(self): from src.qg.checks import is_self_hosting_repo assert is_self_hosting_repo("") is False assert is_self_hosting_repo(None) is False # ------------------------------------------------------------------ # Stage machinery (regression: must not be broken) # ------------------------------------------------------------------ def test_deploy_staging_qg_is_check_staging_status(self): assert get_qg_for_stage("deploy-staging") == "check_staging_status" def test_registered_in_qg_checks(self): from src.qg.checks import QG_CHECKS, check_staging_status assert QG_CHECKS.get("check_staging_status") is check_staging_status def test_deploy_stage_qg_still_check_deploy_status(self): """Regression: existing deploy QG must not be broken.""" assert get_qg_for_stage("deploy") == "check_deploy_status" def test_stage_chain(self): """Full chain: testing->deploy-staging->deploy->done.""" from src.stages import get_next_stage assert get_next_stage("testing") == "deploy-staging" assert get_next_stage("deploy-staging") == "deploy" assert get_next_stage("deploy") == "done"