4e4cc6c724
ET-013: deployer writes 14-deploy-log.md and merges deploy artifacts into main via a separate PR, so the log lands in origin/main, not the feature branch worktree that check_deploy_status reads via _repo_path(repo, branch). Result: every successful deploy was falsely failed (Deploy log not found) and rolled back deploy->development. Fix: when the log is absent in the worktree, fall back to reading it from origin/main on the shared clone (git fetch origin main + git show origin/main:docs/work-items/<WI>/14-deploy-log.md). Lookup order: worktree -> origin/main -> not found. Fetch/show failures degrade to not found (never raise). Does not touch the merge-gate in gitea.py. Tests: origin/main SUCCESS->PASS (ET-013 case), origin/main FAILED->FAILED, absent everywhere->not found, fetch failure->degrades no exception, worktree log short-circuits main lookup.
364 lines
14 KiB
Python
364 lines
14 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,
|
|
)
|
|
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
|
|
|
|
# --- 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")
|
|
|