feat(P03 W5): CAP-025 live-pilot-apply regression check (REQ-316)
CAP-025 (local tier) asserts the pilot-apply pipeline is structurally ready: run_platform.sh steps present, core pipeline modules importable, dev env bound to 581513795199 (D-203), dynamodb L1 registered (REQ-322), pilot policies authored (REQ-315/320), outcome backfill present (REQ-317). Returns Verified on the current branch (all W2/W3/W4 dependencies in place). Added to CAPABILITY_REGISTRY. The live apply (P4) exercises this end-to-end against AWS. ---ci--- project: acdl phase: 3 milestone: v1.26 status: execute wave: W5 ---
This commit is contained in:
Jon Chery
@@ -620,6 +620,98 @@ def _check_cap_024_deck_structure() -> Tuple[Status, str]:
|
||||
return "Verified", f"deck has {slide_count} slides, recap+ask present, per-slide benefits present"
|
||||
|
||||
|
||||
def _check_cap_025_live_pilot_apply() -> Tuple[Status, str]:
|
||||
"""CAP-025 (REQ-316): live-pilot-apply pipeline readiness — structural
|
||||
check that the pilot-apply end-to-end pipeline is wired (NOT a live
|
||||
apply; the live apply lands in P4).
|
||||
|
||||
The pilot-apply round-trip is:
|
||||
contract resolve -> adapter compile -> terraform plan -> policy scan
|
||||
-> confidence signal -> terraform apply -> outbox write
|
||||
|
||||
For P3 this is a LOCAL-tier structural-readiness check: the scripts
|
||||
exist + are wired, the core pipeline modules import, the pilot env is
|
||||
bound to a real account (D-203), the DynamoDB L1 primitive is
|
||||
registered (REQ-322), the pilot policies are authored (REQ-315/320),
|
||||
and the outcome-backfill module exists (REQ-317). The live apply
|
||||
against AWS is P4's live-verify (D-093 / G-111 steady state aside).
|
||||
"""
|
||||
import json
|
||||
|
||||
# 1. scripts/run_platform.sh exists + contains the pipeline step markers.
|
||||
run_platform = ROOT / "scripts" / "run_platform.sh"
|
||||
if not run_platform.is_file():
|
||||
return "Broken", "scripts/run_platform.sh missing (pilot-apply pipeline driver)"
|
||||
script_text = run_platform.read_text()
|
||||
# Step markers mirrored from the script's own comments + Step headers.
|
||||
required_markers = [
|
||||
"resolve contract", # Step 2: contract_resolver
|
||||
"adapter compiles stack", # Step 3: terraform adapter
|
||||
"terraform init", # Step 4: terraform plan
|
||||
"terraform plan", # Step 4: terraform plan
|
||||
"policy scan", # Step 5: runtime policy scan (Wiz/Checkov)
|
||||
"confidence signal", # Step 7: confidence_signal compute
|
||||
"terraform apply", # Step 5: terraform apply (--apply mode)
|
||||
"outbox", # outbox write (Step 8)
|
||||
]
|
||||
missing_markers = [m for m in required_markers if m not in script_text]
|
||||
if missing_markers:
|
||||
return "Broken", f"run_platform.sh missing step markers: {missing_markers}"
|
||||
|
||||
# 2. core pipeline modules importable.
|
||||
for mod_name in (
|
||||
"core.contract_resolver",
|
||||
"adapters.terraform.adapter",
|
||||
"core.confidence_signal",
|
||||
"core.outbox_writer",
|
||||
):
|
||||
try:
|
||||
importlib.import_module(mod_name)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
return "Broken", f"pipeline module not importable: {mod_name} ({type(exc).__name__}: {exc})"[:200]
|
||||
|
||||
# 3. dev env bound to the real pilot account (D-203).
|
||||
dev_env_path = ROOT / "core" / "environments" / "dev.json"
|
||||
if not dev_env_path.is_file():
|
||||
return "Broken", "core/environments/dev.json missing"
|
||||
try:
|
||||
dev_env = json.loads(dev_env_path.read_text())
|
||||
except Exception as exc: # noqa: BLE001
|
||||
return "Broken", f"dev.json parse failed: {exc}"[:200]
|
||||
account_id = dev_env.get("account_id")
|
||||
if account_id != "581513795199":
|
||||
return "Broken", f"dev env not bound to real account (D-203): account_id={account_id!r}"
|
||||
|
||||
# 4. DynamoDB L1 primitive registered (REQ-322).
|
||||
registry_path = ROOT / "modules" / "registry.json"
|
||||
if not registry_path.is_file():
|
||||
return "Broken", "modules/registry.json missing"
|
||||
try:
|
||||
registry = json.loads(registry_path.read_text())
|
||||
except Exception as exc: # noqa: BLE001
|
||||
return "Broken", f"registry.json parse failed: {exc}"[:200]
|
||||
if "dynamodb" not in registry:
|
||||
return "Broken", "dynamodb L1 primitive not registered (REQ-322)"
|
||||
|
||||
# 5. pilot policies authored (REQ-315/320).
|
||||
pilot_policies = [
|
||||
ROOT / "adapters" / "kyverno-json" / "policies" / "pilot-readiness" / "no-placeholder-account.json",
|
||||
ROOT / "adapters" / "kyverno-json" / "policies" / "settlement-finality" / "all-matches-committed.json",
|
||||
]
|
||||
missing_policies = [str(p.relative_to(ROOT)) for p in pilot_policies if not p.is_file()]
|
||||
if missing_policies:
|
||||
return "Broken", f"pilot policies not authored (REQ-315/320): {missing_policies}"
|
||||
|
||||
# 6. outcome-backfill module exists (REQ-317).
|
||||
outcome_backfill = ROOT / "core" / "metrics" / "outcome_backfill.py"
|
||||
if not outcome_backfill.is_file():
|
||||
return "Broken", "outcome backfill not implemented (REQ-317)"
|
||||
|
||||
return ("Verified",
|
||||
"pilot-apply pipeline structurally ready "
|
||||
"(contract->adapter->plan->policy->confidence->apply->outbox)")
|
||||
|
||||
|
||||
# Registry: ordered, each entry is (capability_id, name, tier, check_fn).
|
||||
# Phase 52 seeds this with 10 local-tier checks; Phase 54 expands it to
|
||||
# cover every v1.1->v1.8 advertised capability and adds the live-AWS tier
|
||||
@@ -673,6 +765,8 @@ CAPABILITY_REGISTRY: List[Tuple[str, str, str, Callable[[], Tuple[Status, str]]]
|
||||
_check_cap_023_metrics_collector),
|
||||
("CAP-024", "unified deck structure (slide count, x3, per-slide benefits)", "local",
|
||||
_check_cap_024_deck_structure),
|
||||
("CAP-025", "live-pilot-apply pipeline readiness (contract->apply->outbox)", "local",
|
||||
_check_cap_025_live_pilot_apply),
|
||||
]
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,75 @@
|
||||
"""Tests for CAP-025 (live-pilot-apply pipeline readiness) — P3 W5, REQ-316.
|
||||
|
||||
CAP-025 is a LOCAL-tier structural-readiness check: the pilot-apply
|
||||
pipeline (contract resolve -> adapter compile -> terraform plan -> policy
|
||||
scan -> confidence signal -> terraform apply -> outbox) must be wired and
|
||||
all its dependencies present. The live apply against AWS is P4's
|
||||
live-verify; P3 only asserts the pipeline is structurally ready.
|
||||
"""
|
||||
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
import core.regression_verify as rv # noqa: E402
|
||||
|
||||
|
||||
def test_cap_025_pipeline_ready():
|
||||
"""CAP-025: on the current branch (W2/W3/W4/W5 deps in place), the
|
||||
pilot-apply pipeline is structurally ready -> Verified."""
|
||||
status, detail = rv._check_cap_025_live_pilot_apply()
|
||||
assert status == "Verified", f"CAP-025 {status}: {detail}"
|
||||
|
||||
|
||||
def test_cap_025_in_registry():
|
||||
"""CAP-025 is in the seeded CAPABILITY_REGISTRY."""
|
||||
cap_ids = [entry[0] for entry in rv.CAPABILITY_REGISTRY]
|
||||
assert "CAP-025" in cap_ids
|
||||
# the entry's check fn must be the one we wrote
|
||||
cap_025 = [e for e in rv.CAPABILITY_REGISTRY if e[0] == "CAP-025"][0]
|
||||
assert cap_025[2] == "local" # tier
|
||||
assert cap_025[3] is rv._check_cap_025_live_pilot_apply
|
||||
|
||||
|
||||
def test_cap_025_detects_missing_primitive(tmp_path, monkeypatch):
|
||||
"""CAP-025 detects an absent DynamoDB L1 primitive (REQ-322): if the
|
||||
modules/registry.json lacks a `dynamodb` entry, the check returns
|
||||
Broken (not Verified). Uses monkeypatch to redirect the registry path
|
||||
to a tmp copy without the dynamodb key."""
|
||||
# Snapshot the real registry so we can restore after the check runs.
|
||||
real_registry = ROOT / "modules" / "registry.json"
|
||||
real_data = json.loads(real_registry.read_text())
|
||||
|
||||
# Build a fake registry without `dynamodb`.
|
||||
fake_data = {k: v for k, v in real_data.items() if k != "dynamodb"}
|
||||
assert "dynamodb" not in fake_data, "test setup: dynamodb must be removed"
|
||||
|
||||
fake_registry = tmp_path / "registry.json"
|
||||
fake_registry.write_text(json.dumps(fake_data))
|
||||
|
||||
# Point ROOT at a tmp dir that mirrors only the files the check reads
|
||||
# after the registry step. The check reads (in order):
|
||||
# scripts/run_platform.sh, core.contract_resolver, adapters.terraform.adapter,
|
||||
# core.confidence_signal, core.outbox_writer, core/environments/dev.json,
|
||||
# modules/registry.json, adapters/kyverno-json/policies/..., core/metrics/outcome_backfill.py
|
||||
# Simpler approach: monkeypatch the registry_path by inlining the check
|
||||
# logic against a fake ROOT. We re-run the check with a patched
|
||||
# `Path.read_text` scoped to the registry file via monkeypatch.
|
||||
original_read_text = Path.read_text
|
||||
|
||||
def fake_read_text(self, *args, **kwargs):
|
||||
if self == real_registry:
|
||||
return json.dumps(fake_data)
|
||||
return original_read_text(self, *args, **kwargs)
|
||||
|
||||
monkeypatch.setattr(Path, "read_text", fake_read_text)
|
||||
|
||||
status, detail = rv._check_cap_025_live_pilot_apply()
|
||||
assert status == "Broken", f"expected Broken for missing dynamodb, got {status}: {detail}"
|
||||
assert "dynamodb" in detail
|
||||
assert "REQ-322" in detail
|
||||
Reference in New Issue
Block a user