feat(staging-check): ORCH-048 B6 reads registry via GET /projects, not local import
All checks were successful
CI / test (push) Successful in 13s
CI / test (pull_request) Successful in 12s

B6 built the project registry by importing src.projects locally (host-path hack
+ importlib.reload), so it evaluated ORCH_PROJECTS_JSON from the launcher's
process-env. On the deployer's canonical host run that var is unset → built-in
default (ET+ORCH) → false FAIL even when staging isolation is healthy.

- Add read-only additive endpoint GET /projects (src/main.py) returning
  known_plane_project_ids + {plane_project_id, repo, work_item_prefix, name}
  of the live process; no secrets. Existing routes unchanged.
- Rewrite B6 to fetch GET {base}/projects via the same stdlib _get helper as
  A/B4/B5/C; drop the host-path hack and importlib.reload (launch-invariant).
- Isolate the verdict in pure _evaluate_b6(known) -> (passed, detail); contract
  unchanged (PASS iff SANDBOX in known and prod ET/ORCH absent). Endpoint
  degradation (non-200 / missing key / bad body / network) → deterministic FAIL.
- src/projects.py and .env* untouched.

Docs (golden source): API table + staging-gate B6 mechanic in
docs/architecture/README.md; B6 description + isolation row in
docs/operations/STAGING_CHECK.md; CHANGELOG entry.

Tests: tests/test_staging_check_b6.py (TC-01..TC-07), tests/test_projects_endpoint.py.

Refs: ORCH-048

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
2026-06-06 05:25:45 +00:00
parent f77825b3c4
commit 2cf873a777
7 changed files with 369 additions and 27 deletions

View File

@@ -0,0 +1,55 @@
"""ORCH-048: tests for the read-only GET /projects diagnostics endpoint.
Added by ADR-001 so the staging-check suite (B6) can read the project registry of
the *live* instance over HTTP instead of importing src.projects into the script's
own process-env. The endpoint is read-only / additive and must expose only
id / repo / prefix / name — never secrets.
"""
from fastapi.testclient import TestClient
from src.main import app
from src import projects as P
from src.projects import reload_projects
client = TestClient(app)
def test_projects_endpoint_returns_known_ids():
"""GET /projects → 200 with known_plane_project_ids matching the registry."""
resp = client.get("/projects")
assert resp.status_code == 200
body = resp.json()
assert "known_plane_project_ids" in body
assert set(body["known_plane_project_ids"]) == P.known_plane_project_ids()
def test_projects_endpoint_lists_projects_with_expected_fields():
"""Each project entry exposes exactly id/repo/prefix/name (no secrets)."""
body = client.get("/projects").json()
assert isinstance(body["projects"], list)
assert len(body["projects"]) == len(P.PROJECTS)
allowed = {"plane_project_id", "repo", "work_item_prefix", "name"}
for entry in body["projects"]:
assert set(entry.keys()) == allowed
# No secret-looking keys leaked into the payload.
for key in entry:
assert "token" not in key.lower()
assert "secret" not in key.lower()
def test_projects_endpoint_reflects_custom_registry(monkeypatch):
"""The endpoint reflects the running process's registry, not a hardcoded one."""
custom = (
'[{"plane_project_id": "endpoint-uuid", "repo": "endpoint-repo", '
'"work_item_prefix": "EP", "name": "Endpoint"}]'
)
monkeypatch.setattr(P.settings, "projects_json", custom)
reload_projects()
try:
body = client.get("/projects").json()
assert body["known_plane_project_ids"] == ["endpoint-uuid"]
assert body["projects"][0]["repo"] == "endpoint-repo"
assert body["projects"][0]["work_item_prefix"] == "EP"
finally:
reload_projects()

View File

@@ -0,0 +1,203 @@
"""ORCH-048: tests for the staging-check B6 registry-isolation verdict.
Background / bug being fixed
----------------------------
B6 in ``scripts/staging_check.py`` used to build the project registry by importing
``src.projects`` locally (``sys.path.insert(0,"/repos/orchestrator")`` +
``importlib.reload``). That read the registry of WHOEVER ran the script: on the
deployer's canonical host run there is no ``ORCH_PROJECTS_JSON`` → the built-in
default registry (ET+ORCH) → a **false FAIL** even when staging isolation is fine.
ADR-001 reworks B6 to read the registry over HTTP from the live instance
(``GET /projects``) and isolates the verdict in the pure function
``_evaluate_b6(known) -> (passed, detail)`` so both outcomes are unit-testable
without a live staging instance or docker (02-trz §9, AC-2).
These tests cover:
* TC-01 clean registry (known={SANDBOX}) → PASS
* TC-02 polluted with prod-ET (known={SANDBOX, PROD_ET}) → FAIL
* TC-03 polluted with prod-ORCH (known={SANDBOX, PROD_ORCH})→ FAIL
* TC-04 sandbox absent (known=set()) → FAIL, deterministic
* TC-05 polluted with both prod projects → FAIL
* TC-06 no host-path hack / no local src.projects import (TR-6)
* TC-07 source degradation (HTTP error / non-200 / bad body) → deterministic FAIL (TR-4)
"""
import importlib.util
import pathlib
import pytest
# ---------------------------------------------------------------------------
# Load scripts/staging_check.py (scripts/ is not a Python package).
# ---------------------------------------------------------------------------
_SCRIPT_PATH = pathlib.Path(__file__).resolve().parents[1] / "scripts" / "staging_check.py"
_spec = importlib.util.spec_from_file_location("staging_check", _SCRIPT_PATH)
sc = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(sc)
SANDBOX = sc.SANDBOX_PROJECT_ID
PROD_ET = sc.PROD_ET_PROJECT_ID
PROD_ORCH = sc.PROD_ORCH_PROJECT_ID
# ---------------------------------------------------------------------------
# _evaluate_b6 — pure verdict (AC-1, AC-2)
# ---------------------------------------------------------------------------
def test_tc01_clean_registry_passes():
"""TC-01: known={SANDBOX} → PASS, detail shows sandbox=YES / prod absent."""
ok, detail = sc._evaluate_b6({SANDBOX})
assert ok is True
assert "sandbox=YES" in detail
assert "prod-ET=NO(good)" in detail
assert "prod-ORCH=NO(good)" in detail
def test_tc02_prod_et_pollution_fails():
"""TC-02: prod-ET leaked into the registry → FAIL, flagged in detail."""
ok, detail = sc._evaluate_b6({SANDBOX, PROD_ET})
assert ok is False
assert "prod-ET=YES(BAD!)" in detail
# sandbox is still present, only ET is the violation
assert "sandbox=YES" in detail
def test_tc03_prod_orch_pollution_fails():
"""TC-03: prod-ORCH leaked into the registry → FAIL, flagged in detail."""
ok, detail = sc._evaluate_b6({SANDBOX, PROD_ORCH})
assert ok is False
assert "prod-ORCH=YES(BAD!)" in detail
assert "sandbox=YES" in detail
def test_tc04_sandbox_absent_fails_deterministically():
"""TC-04: empty registry (no sandbox) → FAIL, no exception (TR-4)."""
ok, detail = sc._evaluate_b6(set())
assert ok is False
assert "sandbox=NO" in detail
# prod projects genuinely absent, but the missing sandbox alone fails it
assert "prod-ET=NO(good)" in detail
assert "prod-ORCH=NO(good)" in detail
def test_tc05_both_prod_pollution_fails():
"""TC-05: both prod projects present → FAIL."""
ok, detail = sc._evaluate_b6({SANDBOX, PROD_ET, PROD_ORCH})
assert ok is False
assert "prod-ET=YES(BAD!)" in detail
assert "prod-ORCH=YES(BAD!)" in detail
# ---------------------------------------------------------------------------
# TC-06 — registry source no longer depends on the host-path hack (TR-6)
# ---------------------------------------------------------------------------
def test_tc06_no_host_path_hack_in_source():
"""TC-06: B6 must not import src.projects from a hardcoded host path.
Static guard: the removed anti-pattern (``sys.path.insert(0,"/repos/orchestrator")``,
``importlib.reload(... src.projects ...)``, ``from src.projects import``) must
be gone from the script. B6 now reads the registry over HTTP (GET /projects).
"""
source = _SCRIPT_PATH.read_text()
assert 'sys.path.insert(0, "/repos/orchestrator")' not in source
assert "importlib.reload" not in source
assert "from src.projects import" not in source
# Positive assertion: B6 reads the registry endpoint over HTTP.
assert "/projects" in source
# ---------------------------------------------------------------------------
# TC-07 — source degradation → deterministic FAIL, never a false PASS (TR-4)
# ---------------------------------------------------------------------------
class _FakeResults:
"""Minimal Results stand-in capturing (label, passed, detail) tuples."""
def __init__(self):
self.items = []
def add(self, label, passed, detail=""):
self.items.append((label, passed, detail))
def _last(results: _FakeResults):
assert results.items, "expected _check_b6 to record exactly one result"
return results.items[-1]
def test_tc07_network_error_is_deterministic_fail(monkeypatch):
"""TC-07a: a network/transport error → FAIL, no unhandled exception."""
def _boom(url, *a, **k):
raise RuntimeError("GET http://x/projects → connection refused")
monkeypatch.setattr(sc, "_get", _boom)
results = _FakeResults()
sc._check_b6("http://staging:8501", results)
label, passed, detail = _last(results)
assert passed is False
assert "GET /projects failed" in detail
def test_tc07_non_200_is_fail(monkeypatch):
"""TC-07b: non-200 response → FAIL with the status in the detail."""
monkeypatch.setattr(sc, "_get", lambda url, *a, **k: (503, {"_raw": "down"}))
results = _FakeResults()
sc._check_b6("http://staging:8501", results)
label, passed, detail = _last(results)
assert passed is False
assert "503" in detail
def test_tc07_missing_key_is_fail(monkeypatch):
"""TC-07c: 200 but no known_plane_project_ids key → FAIL (no false PASS)."""
monkeypatch.setattr(sc, "_get", lambda url, *a, **k: (200, {"projects": []}))
results = _FakeResults()
sc._check_b6("http://staging:8501", results)
label, passed, detail = _last(results)
assert passed is False
assert "known_plane_project_ids" in detail
def test_tc07_malformed_value_is_fail(monkeypatch):
"""TC-07d: known_plane_project_ids is not a list → FAIL, no exception."""
monkeypatch.setattr(
sc, "_get",
lambda url, *a, **k: (200, {"known_plane_project_ids": "not-a-list"}),
)
results = _FakeResults()
sc._check_b6("http://staging:8501", results)
label, passed, detail = _last(results)
assert passed is False
assert "not a list" in detail
# ---------------------------------------------------------------------------
# Happy path through _check_b6 (HTTP wiring + verdict together)
# ---------------------------------------------------------------------------
def test_check_b6_clean_endpoint_passes(monkeypatch):
"""A healthy /projects response with only the sandbox id → PASS."""
monkeypatch.setattr(
sc, "_get",
lambda url, *a, **k: (200, {"known_plane_project_ids": [SANDBOX]}),
)
results = _FakeResults()
sc._check_b6("http://staging:8501", results)
label, passed, detail = _last(results)
assert passed is True
assert "sandbox=YES" in detail
def test_check_b6_polluted_endpoint_fails(monkeypatch):
"""A /projects response leaking a prod id → FAIL (defence works end-to-end)."""
monkeypatch.setattr(
sc, "_get",
lambda url, *a, **k: (200, {"known_plane_project_ids": [SANDBOX, PROD_ORCH]}),
)
results = _FakeResults()
sc._check_b6("http://staging:8501", results)
label, passed, detail = _last(results)
assert passed is False
assert "prod-ORCH=YES(BAD!)" in detail