claude-bot
9070489968
The self-hosting orchestrator looped on deploy-staging -> development because
scripts/staging_check.py exited 1 on ANY failed check, so two infra-only checks
(C9a sandbox branch / C9b analyst-job — caused by SANDBOX bot accounts not being
members of the sandbox Plane project, NOT a pipeline regress) forced
staging_status: FAILED -> rollback -> loop, burning developer retries and tokens.
Direction (б) per ADR-001: classify staging checks as REAL (all pipeline checks,
fail-closed) vs SANDBOX_INFRA (narrow allowlist {C9a, C9b}, waivable). New leaf
module src/staging_verdict.py (stdlib-only, never-raise): classify_check +
compute_staging_verdict fold per-check results into a tolerant-but-fail-closed
verdict — any REAL failure -> FAILED/exit1 (safety net holds under any flag);
only C9a/C9b failed & tolerant -> SUCCESS/exit0 with waived list; only infra &
strict -> FAILED/exit1; any internal error -> FAILED/exit1 (never a false green).
staging_check.py now auto-classifies each check (public 3-tuple _items shape kept
as an ORCH-048 b6 regression guard), exposes categorized_items(), prints
INFRA-WAIVED/VERDICT lines, and exits via the verdict; new --strict flag forces
legacy strictness per-run. Kill-switch ORCH_STAGING_INFRA_TOLERANCE_ENABLED
(default true) restores legacy strict mode globally. launcher gains
action_stage_no_changes_note so "no changes to commit" on action stages is logged
as expected, not treated as under-delivery.
Contracts unchanged: STAGE_TRANSITIONS, QG_CHECKS registry, staging_status:/
deploy_status: frontmatter, hook exit-code (0/1/2), check_staging_status; no DB
migration. Docs: README, STAGING_CHECK.md, deployer.md, .env.example, CHANGELOG.
Refs: ORCH-061
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
757 lines
32 KiB
Python
757 lines
32 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,
|
|
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 <repo>".
|
|
|
|
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
|
|
|
|
# ------------------------------------------------------------------
|
|
# ORCH-061 / TC-08: the conditional staging gate is unchanged for
|
|
# non-self-hosting repos AND independent of the new tolerance flag (FR-5/AC-6).
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_tc08_non_self_na_independent_of_tolerance_flag(self, tmp_path, monkeypatch):
|
|
"""TC-08 / AC-6: for a non-self-hosting repo check_staging_status is the
|
|
byte-identical (True, "Staging gate N/A …") regardless of whether the
|
|
ORCH-061 infra-tolerance flag is on or off — the new behaviour never
|
|
activates off the self-hosting path."""
|
|
monkeypatch.setattr("src.qg.checks.settings.repos_dir", str(tmp_path))
|
|
from src.qg.checks import check_staging_status
|
|
for flag in (True, False):
|
|
monkeypatch.setattr(
|
|
"src.config.settings.staging_infra_tolerance_enabled", flag,
|
|
raising=False,
|
|
)
|
|
passed, reason = check_staging_status("enduro-trails", "ET-035")
|
|
assert passed is True
|
|
assert reason == "Staging gate N/A for enduro-trails"
|
|
|
|
# ------------------------------------------------------------------
|
|
# 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"
|
|
|