Jon Chery
b758a7c242
---ci--- project: acdl phase: 21 milestone: v1.6 status: execute ---/ci--- Rename the acdl_platform/ package to core/ across the directory, all imports in tests/scripts/pipelines/workflows, and doc references. The package is imported as core.confidence_signal / core.contract_resolver / core.outbox_writer. The deploy workflow's platform-repo checkout dir is renamed acdl-platform/ -> platform/ (workspace path, not the python package). Both .gitea + .github workflows stay byte-identical. Note: the original target name 'platform/' shadows Python's stdlib platform module (pytest's import uuid -> platform.system() fails when the repo root is on sys.path, which every test does). 'core/' avoids the clash while honoring the intent (drop the verbose acdl_platform). Tests: 154 pass. run_ci.sh green.
42 lines
1.8 KiB
Python
42 lines
1.8 KiB
Python
"""Check that qaApprover != prodApprover for a contract (ARCHITECTURE.md
|
|
§10.3, D-042). Reads `approver_qa` from the DynamoDB outbox for the
|
|
contractId, compares to the prod-dispatch `gitea.actor`. Blocks on
|
|
equality, emits `SEPARATION_OF_DUTIES_VIOLATION`, routes a halt artifact
|
|
to SRE on-call.
|
|
|
|
Spike scope (A-8.1): the spike is dev-only (REQ-27 contract has
|
|
environment: dev); HITL is not exercised. This module is authored to its
|
|
full v1.2 shape but the spike calls it with current_prod_approver=None
|
|
and a None outbox_client — the check returns (True, 'no QA approver
|
|
recorded (dev-only spike)').
|
|
"""
|
|
|
|
from typing import Optional, Tuple
|
|
|
|
|
|
def check(outbox_client, contract_id: str,
|
|
current_prod_approver: Optional[str]) -> Tuple[bool, str]:
|
|
"""Return (ok, reason). ok=False means block the prod promotion."""
|
|
if outbox_client is None:
|
|
return (True, "no outbox client (dev-only spike)")
|
|
item = outbox_client.get(contract_id)
|
|
if item is None:
|
|
return (True, "no prior approver (first promotion)")
|
|
qa_approver = item.get("approver_qa")
|
|
if not qa_approver:
|
|
return (True, "no QA approver recorded (dev-only spike)")
|
|
if current_prod_approver is None:
|
|
return (True, "no prod approver supplied (dev-only spike)")
|
|
if qa_approver == current_prod_approver:
|
|
return (False,
|
|
f"SEPARATION_OF_DUTIES_VIOLATION: "
|
|
f"qaApprover==prodApprover=={qa_approver}")
|
|
return (True, "distinct")
|
|
|
|
|
|
def route_halt_artifact(contract_id: str, violation_reason: str,
|
|
oncall_client) -> None:
|
|
"""Route a halt artifact to SRE on-call. Spike: stub that logs. v1.2
|
|
wires a real pager."""
|
|
print(f"[halt-artifact] contract={contract_id} reason={violation_reason} "
|
|
f"oncall={oncall_client}", flush=True) |