fix(pipeline): make deploy-staging gate conditional on self-hosting repo (ORCH-35)
All checks were successful
CI / test (push) Successful in 10s
CI / test (pull_request) Successful in 10s

This commit is contained in:
Dev Agent
2026-06-05 10:36:46 +03:00
parent e0b6e92b09
commit e0c14fae5f
2 changed files with 134 additions and 27 deletions

View File

@@ -441,6 +441,26 @@ def check_deploy_status(repo: str, work_item_id: str, branch: str | None = None)
# ---------------------------------------------------------------------------
# Self-hosting detection: staging-infra (localhost:8501) exists ONLY for the
# orchestrator repo itself (self-hosting). Other repos have no staging instance
# and their deployer prompts know nothing about it -- the gate must be a no-op
# for them. The repo value is the plain gitea repo name (ProjectConfig.repo),
# matching what _run_qg/advance_stage pass in. See ORCH-35 / PR #31.
# ---------------------------------------------------------------------------
SELF_HOSTING_REPO = "orchestrator"
def is_self_hosting_repo(repo: str) -> bool:
"""Return True iff repo is the self-hosted orchestrator (has staging infra).
Comparison is case-insensitive and strips whitespace for safety, but in
practice repo comes from the gitea webhook payload .repository.name which
is always lowercase (confirmed via projects.py registry entry).
"""
return (repo or "").strip().lower() == SELF_HOSTING_REPO.lower()
def _parse_staging_status(content: str) -> tuple[bool, str]:
"""Parse a 15-staging-log.md body and map its `staging_status:` frontmatter to a
quality-gate verdict. Reads ONLY the machine-readable YAML field, never prose.
@@ -505,16 +525,26 @@ def check_staging_status(repo: str, work_item_id: str, branch: str | None = None
Gate the deploy-staging -> deploy transition on the deployer's machine-readable
verdict in 15-staging-log.md frontmatter (staging_status: SUCCESS|FAILED).
Mirrors check_deploy_status (БАГ 8): reads ONLY the machine-readable YAML field,
never the body prose. The deployer runs the staging test suite against localhost:8501
and writes the verdict into 15-staging-log.md.
ORCH-35 conditional gate (Variant A):
- Non-self-hosting repos (anything other than "orchestrator") have no staging
instance and no deployer knowledge of it -> gate is an immediate pass.
- Self-hosting repo ("orchestrator") -> real check: reads ONLY the machine-
readable staging_status: field from YAML frontmatter, never body prose.
Lookup order: worktree -> origin/main -> not found.
Mirrors check_deploy_status (БАГ 8) for the self-hosting path.
Lookup order (self-hosting only): worktree -> origin/main -> not found.
Returns:
(True, ...) -> staging_status: SUCCESS
(True, "Staging gate N/A for <repo>") -> non-self-hosting repo (instant pass)
(True, ...) -> staging_status: SUCCESS (self-hosting path)
(False, ...) -> staging_status: FAILED, missing field, or no frontmatter
"""
# Variant A: non-self-hosting repos have no staging infra -- skip entirely.
if not is_self_hosting_repo(repo):
return True, f"Staging gate N/A for {repo}"
# Self-hosting (orchestrator) path: real verdict check.
repo_path = _repo_path(repo, branch)
log_path = os.path.join(repo_path, f"docs/work-items/{work_item_id}/15-staging-log.md")
@@ -526,7 +556,7 @@ def check_staging_status(repo: str, work_item_id: str, branch: str | None = None
return False, f"Error reading staging log: {e}"
return _parse_staging_status(content)
# Not in the feature worktree the deployer may have merged it into main.
# Not in the feature worktree -- the deployer may have merged it into main.
main_content = _staging_log_from_main(repo, work_item_id)
if main_content is not None:
return _parse_staging_status(main_content)

View File

@@ -452,58 +452,75 @@ class TestCheckTestsLocal:
class TestCheckStagingStatus:
"""ORCH-35: deploy-staging -> deploy gate reads machine-readable staging_status:
from 15-staging-log.md frontmatter. Mirrors check_deploy_status pattern."""
"""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>".
def _write_log(self, repo_dir, content):
wi_dir = repo_dir / "docs" / "work-items" / "ET-035"
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)
def test_success_verdict_passes(self, setup_work_item_dir):
# ------------------------------------------------------------------
# Self-hosting (orchestrator) path -- real file check
# ------------------------------------------------------------------
def test_success_verdict_passes(self, orch_dir):
self._write_log(
setup_work_item_dir,
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("enduro-trails", "ET-035")
passed, reason = check_staging_status("orchestrator", "ORCH-035")
assert passed is True
assert "SUCCESS" in reason
def test_failed_verdict_fails(self, setup_work_item_dir):
def test_failed_verdict_fails(self, orch_dir):
self._write_log(
setup_work_item_dir,
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("enduro-trails", "ET-035")
passed, reason = check_staging_status("orchestrator", "ORCH-035")
assert passed is False
assert "FAILED" in reason
def test_no_file_fails(self, setup_work_item_dir):
def test_no_file_fails_for_self_hosting(self, orch_dir):
from src.qg.checks import check_staging_status
passed, reason = check_staging_status("enduro-trails", "ET-035")
passed, reason = check_staging_status("orchestrator", "ORCH-035")
assert passed is False
assert "not found" in reason.lower()
def test_no_field_fails(self, setup_work_item_dir):
def test_no_field_fails(self, orch_dir):
# Frontmatter present but no staging_status field -> must NOT pass.
self._write_log(
setup_work_item_dir,
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("enduro-trails", "ET-035")
passed, reason = check_staging_status("orchestrator", "ORCH-035")
assert passed is False
def test_prose_only_no_frontmatter_fails(self, setup_work_item_dir):
def test_prose_only_no_frontmatter_fails(self, orch_dir):
# Prose mentioning SUCCESS but no machine-readable frontmatter -> fail.
self._write_log(
setup_work_item_dir,
orch_dir,
"# Staging Log\n\nStatus: SUCCESS (prose, not frontmatter).\n",
)
from src.qg.checks import check_staging_status
passed, reason = check_staging_status("enduro-trails", "ET-035")
passed, reason = check_staging_status("orchestrator", "ORCH-035")
assert passed is False
def test_origin_main_success_passes_when_absent_in_worktree(self, monkeypatch):
@@ -513,7 +530,7 @@ class TestCheckStagingStatus:
lambda repo, wi: "---\nstaging_status: SUCCESS\n---\n\nAll good.\n",
)
from src.qg.checks import check_staging_status
passed, reason = check_staging_status("enduro-trails", "ET-035-main")
passed, reason = check_staging_status("orchestrator", "ORCH-035-main")
assert passed is True
assert "SUCCESS" in reason
@@ -523,7 +540,7 @@ class TestCheckStagingStatus:
lambda repo, wi: "---\nstaging_status: FAILED\n---\n\nboom.\n",
)
from src.qg.checks import check_staging_status
passed, reason = check_staging_status("enduro-trails", "ET-035-main")
passed, reason = check_staging_status("orchestrator", "ORCH-035-main")
assert passed is False
assert "FAILED" in reason
@@ -532,10 +549,70 @@ class TestCheckStagingStatus:
"src.qg.checks._staging_log_from_main", lambda repo, wi: None
)
from src.qg.checks import check_staging_status
passed, reason = check_staging_status("enduro-trails", "ET-035-absent")
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"