392 lines
23 KiB
Python
392 lines
23 KiB
Python
#!/usr/bin/env python3
|
|
"""Contract regression tests for trusted artifact/review/workflow mutators."""
|
|
import json
|
|
import hashlib
|
|
import os
|
|
import shutil
|
|
import sys
|
|
|
|
import yaml
|
|
|
|
ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
HOOKS = os.path.join(ROOT, ".claude", "hooks")
|
|
WS = os.path.join(ROOT, ".claude", "tests", "fixtures", "workflow-contract-ws")
|
|
os.environ["CLAUDE_PROJECT_DIR"] = ROOT
|
|
os.environ["ORGOS_WORKSPACE"] = WS
|
|
sys.path.insert(0, HOOKS)
|
|
|
|
import acceptance_log as AL # noqa: E402
|
|
import context_package as CP # noqa: E402
|
|
import state_engine as SE # noqa: E402
|
|
|
|
shutil.rmtree(WS, ignore_errors=True)
|
|
os.makedirs(WS, exist_ok=True)
|
|
|
|
passed = failed = 0
|
|
|
|
|
|
def check(name, condition):
|
|
global passed, failed
|
|
if condition:
|
|
passed += 1
|
|
print(f" PASS {name}")
|
|
else:
|
|
failed += 1
|
|
print(f" FAIL {name}")
|
|
|
|
|
|
def init(wf, plan="cascade", tier="standard"):
|
|
SE.init_ledger(wf, plan=plan, tier=tier)
|
|
|
|
|
|
def write_report(wf, artifact_id, kind, producer, payload, stage="intake"):
|
|
directory = os.path.join(WS, "completion-records", wf)
|
|
os.makedirs(directory, exist_ok=True)
|
|
path = os.path.join(directory, artifact_id + ".report.yaml")
|
|
report = {
|
|
"report-type": "workflow-artifact",
|
|
"artifact-kind": kind,
|
|
"artifact-version": 1,
|
|
"tier": (SE.read_ledger(wf) or {}).get("tier", "standard"),
|
|
"identity": {
|
|
"artifact-id": artifact_id,
|
|
"workflow-id": wf,
|
|
"stage": stage,
|
|
"producer-role-id": producer,
|
|
},
|
|
"payload": payload,
|
|
"report-header": {
|
|
"bottom-line": f"{kind} fixture",
|
|
"decision-needed": {"needed": False},
|
|
"confidence": {"value": "Med", "derived-from": "evidence"},
|
|
"risks": [],
|
|
"evidence": [{"source-uri": "README.md", "grade": "E3"}],
|
|
},
|
|
}
|
|
with open(path, "w", encoding="utf-8") as fh:
|
|
yaml.safe_dump(report, fh, allow_unicode=True, sort_keys=False)
|
|
return path
|
|
|
|
|
|
def submit(wf, path, actor="OPS-ORCH"):
|
|
return SE.submit_report(wf, path, actor)
|
|
|
|
|
|
def open_stage(wf, stage):
|
|
"""Test fixture helper: open a stage without re-testing unrelated upstream gates."""
|
|
current = SE.read_ledger(wf).get("stage")
|
|
event = {
|
|
"state-event-id": f"fixture-{wf}-{stage}", "event-type": "state-transition",
|
|
"workflow-id": wf, "from": current, "to": stage, "actor": "OPS-ORCH",
|
|
"effective-at": SE._now(),
|
|
}
|
|
return SE._atomic_event_transaction(wf, workflow_event=event)[0]
|
|
|
|
|
|
def add_receipt(wf, receipt_id, command="fixture-check", subject=None,
|
|
assertion_status="passed", category="test"):
|
|
directory = os.path.join(WS, "evidence")
|
|
os.makedirs(directory, exist_ok=True)
|
|
with open(os.path.join(directory, "ledger.jsonl"), "a", encoding="utf-8") as fh:
|
|
fh.write(json.dumps({
|
|
"tool_use_id": receipt_id, "tool_name": "Bash", "command": command,
|
|
"exit_code": 0 if assertion_status == "passed" else 1,
|
|
"workflow_id": wf, "session_id": "fixture-session",
|
|
"agent_id": "fixture-agent", "ts": SE._now(),
|
|
"receipt_type": "verification-run", "verification_category": category,
|
|
"verification_subject": subject or receipt_id,
|
|
"assertion_status": assertion_status,
|
|
}) + "\n")
|
|
|
|
|
|
def file_sha(path):
|
|
return hashlib.sha256(open(path, "rb").read()).hexdigest()
|
|
|
|
|
|
print("== trusted submit derives gate facts ==")
|
|
init("wf-contract", tier="light")
|
|
brief = write_report("wf-contract", "brief-1", "decision-brief", "EXEC-CEO",
|
|
{"mode": "converge", "tier": "light",
|
|
"candidate-families": ["FAM-CPO", "FAM-CTO", "FAM-CFO"]})
|
|
profile = write_report("wf-contract", "profile-1", "workload-profile", "EXEC-CEO", {
|
|
"surfaces": {"ui": False, "public-api": False, "persistence": False, "infrastructure": False},
|
|
"risk": {"security-bearing": False, "data-migration": False, "external-side-effect": False,
|
|
"risk-level": "Low", "reversibility": "two-way-door", "blast-radius": "single-role",
|
|
"privacy": False, "regulatory": False, "slo-impact": False},
|
|
"required-capabilities": ["product"], "product-feature": False,
|
|
})
|
|
check("submit decision brief", submit("wf-contract", brief)[0])
|
|
check("submit workload profile", submit("wf-contract", profile)[0])
|
|
facts = SE._facts("wf-contract", SE.read_ledger("wf-contract"))
|
|
check("intake facts come from artifacts", facts["decision_brief_present"] and facts["workload_profile_present"])
|
|
|
|
check("fixture opens discovery stage", open_stage("wf-contract", "discovery"))
|
|
ground = write_report("wf-contract", "ground-1", "grounding-package", "STR-ANALYST",
|
|
{"problem-structure": {"question": "target scope"},
|
|
"analysis-synthesis": {"insight": "scope trades off against cost"},
|
|
"options": [
|
|
{"id": "a", "problem": "narrow", "tradeoffs": ["scope"], "evidence-refs": ["README.md"]},
|
|
{"id": "b", "problem": "broad", "tradeoffs": ["cost"], "evidence-refs": ["README.md"]}],
|
|
"evidence": ["README.md"],
|
|
"source-contributions": [
|
|
{"report-id": f"fixture-source-{index}",
|
|
"report-ref": f"fixture/source-{index}.report.yaml",
|
|
"report-sha256": str(index) * 64, "producer-role-id": "EXEC-CEO",
|
|
"context-package-ref": f"fixture/context-{index}.yaml",
|
|
"context-package-sha256": str(index + 3) * 64,
|
|
"assigned-lens": lens, "producer-run-id": f"fixture-run-{index}"}
|
|
for index, lens in enumerate(
|
|
["LENS-VALUE", "LENS-PRODUCT", "LENS-TECH"], 1)],
|
|
"lens-coverage": {"required-min": 3,
|
|
"covered": ["LENS-VALUE", "LENS-PRODUCT", "LENS-TECH"],
|
|
"contrarian-report-id": None}}, stage="discovery")
|
|
check("grounding-package submit", submit("wf-contract", ground)[0])
|
|
check("option count derived from report", SE._facts("wf-contract", SE.read_ledger("wf-contract"))["option_set_count"] == 2)
|
|
ok, _ = SE.record_artifact("wf-contract", design_type="completion-record", option_count=99, evidence_grade="E5")
|
|
check("unsafe record API removed", not ok)
|
|
|
|
print("== exact effective decision + authorization ==")
|
|
check("fixture opens decide stage", open_stage("wf-contract", "decide"))
|
|
packet = write_report("wf-contract", "packet-1", "executive-decision-packet", "EXEC-CEO",
|
|
{"recommendation": "A"}, stage="decide")
|
|
check("decision packet submit", submit("wf-contract", packet)[0])
|
|
bad_stage_packet = write_report("wf-contract", "packet-bad-stage", "executive-decision-packet",
|
|
"EXEC-CEO", {"recommendation": "A"}, stage="discovery")
|
|
check("artifact kind cannot claim an undeclared output stage",
|
|
not submit("wf-contract", bad_stage_packet)[0])
|
|
check("unrelated accepted artifact cannot approve packet",
|
|
not SE._facts("wf-contract", SE.read_ledger("wf-contract"))["decision_packet_accepted"])
|
|
check("self-review rejected", not SE.review_artifact("wf-contract", packet, "accepted", "EXEC-CEO")[0])
|
|
check("unauthorized reviewer rejected", not SE.review_artifact("wf-contract", packet, "accepted", "ENG-BE")[0])
|
|
ok, accepted = SE.review_artifact("wf-contract", packet, "accepted", "HUMAN-001")
|
|
check("authorized exact revision accepted", ok and accepted.get("artifact-sha256"))
|
|
check("packet gate accepted", SE._facts("wf-contract", SE.read_ledger("wf-contract"))["decision_packet_accepted"])
|
|
check("later changes-requested invalidates acceptance",
|
|
SE.review_artifact("wf-contract", packet, "changes-requested", "HUMAN-001")[0]
|
|
and not SE._facts("wf-contract", SE.read_ledger("wf-contract"))["decision_packet_accepted"])
|
|
check("re-accept latest decision", SE.review_artifact("wf-contract", packet, "accepted", "HUMAN-001")[0])
|
|
|
|
packet2 = write_report("wf-contract", "packet-2", "executive-decision-packet", "EXEC-CEO",
|
|
{"recommendation": "B"}, stage="decide")
|
|
check("new packet revision submitted", submit("wf-contract", packet2)[0])
|
|
check("newer unreviewed packet makes old accepted revision stale",
|
|
not SE._facts("wf-contract", SE.read_ledger("wf-contract"))["decision_packet_accepted"])
|
|
check("new packet supersedes old on acceptance",
|
|
SE.review_artifact("wf-contract", packet2, "accepted", "HUMAN-001", supersedes="packet-1")[0])
|
|
old_sha = next(a["artifact-sha256"] for a in SE._trusted_artifacts("wf-contract") if a["artifact-id"] == "packet-1")
|
|
check("old revision effective decision is Superseded",
|
|
AL.effective_decision("wf-contract", "packet-1", old_sha) == "Superseded")
|
|
|
|
print("== current-stage lifecycle + event projection ==")
|
|
direction_brief = os.path.join(WS, "direction-input-brief.yaml")
|
|
with open(direction_brief, "w", encoding="utf-8") as fh:
|
|
yaml.safe_dump({
|
|
"product-goal": "ship a coherent interface", "core-users": ["operator"],
|
|
"core-tasks": ["review"], "information-density": "high",
|
|
"required-accessibility": "WCAG AA", "brand-constraints": ["legible"],
|
|
"avoid-cliches": ["generic dashboard"], "representative-screen-requirement": "core-task",
|
|
"tech-platform-constraints": ["web"],
|
|
}, fh, allow_unicode=True, sort_keys=False)
|
|
SE.init_ledger("dd-contract", plan="design-direction", parent_workflow="wf-contract",
|
|
product_decision="packet-2", direction_input_brief=direction_brief)
|
|
check("new workflow starts with current stage running",
|
|
SE.read_ledger("dd-contract").get("stage") == "design-direction-intake"
|
|
and SE.read_ledger("dd-contract").get("stage-status") == "running")
|
|
check("cannot enter next stage before current is completed",
|
|
not SE.enter_stage("dd-contract", "design-direction-discovery", actor="OPS-ORCH")[0])
|
|
check("complete-stage validates selected exit gate",
|
|
SE.complete_stage("dd-contract", "OPS-ORCH", to="design-direction-discovery")[0]
|
|
and SE.read_ledger("dd-contract").get("stage-status") == "completed")
|
|
check("enter-stage opens next stage as running",
|
|
SE.enter_stage("dd-contract", "design-direction-discovery", actor="OPS-ORCH")[0]
|
|
and SE.read_ledger("dd-contract").get("stage") == "design-direction-discovery"
|
|
and SE.read_ledger("dd-contract").get("stage-status") == "running")
|
|
ledger_path = SE._ledger_path("dd-contract", create=False)
|
|
tampered = yaml.safe_load(open(ledger_path, encoding="utf-8"))
|
|
tampered["stage"] = "design-direction-approved"
|
|
with open(ledger_path, "w", encoding="utf-8") as fh:
|
|
yaml.safe_dump(tampered, fh, allow_unicode=True, sort_keys=False)
|
|
check("workflow.yaml stage edit cannot override canonical events",
|
|
SE.read_ledger("dd-contract").get("stage") == "design-direction-discovery")
|
|
|
|
print("== design/spec bundles are all-of ==")
|
|
check("fixture opens design stage", open_stage("wf-contract", "design"))
|
|
packet_basis = next(a for a in SE._trusted_artifacts("wf-contract") if a["artifact-id"] == "packet-2")
|
|
design = write_report("wf-contract", "design-1", "overall-design", "ARCH-SOLUTION",
|
|
{"summary": "overall design", "basis-artifact-id": "packet-2",
|
|
"basis-artifact-sha256": packet_basis["artifact-sha256"]}, stage="design")
|
|
check("overall design submitted", submit("wf-contract", design)[0])
|
|
check("design bundle not accepted on presence", not SE._facts("wf-contract", SE.read_ledger("wf-contract"))["design_accepted"])
|
|
check("design bundle accepted after exact architecture review", SE.review_artifact("wf-contract", design, "accepted", "ARCH-EA")[0]
|
|
and SE._facts("wf-contract", SE.read_ledger("wf-contract"))["design_accepted"])
|
|
design_basis = next(a for a in SE._trusted_artifacts("wf-contract") if a["artifact-id"] == "design-1")
|
|
check("fixture opens spec stage", open_stage("wf-contract", "spec"))
|
|
api = write_report("wf-contract", "api-1", "api-contract", "ARCH-TECH",
|
|
{"summary": "api contract", "basis-artifact-id": "design-1",
|
|
"basis-artifact-sha256": design_basis["artifact-sha256"]}, stage="spec")
|
|
check("api contract alone accepted", submit("wf-contract", api)[0]
|
|
and SE.review_artifact("wf-contract", api, "accepted", "ARCH-APP")[0]
|
|
and not SE._facts("wf-contract", SE.read_ledger("wf-contract"))["spec_accepted"])
|
|
criteria = write_report("wf-contract", "criteria-1", "acceptance-criteria", "PROD-PO",
|
|
{"basis-artifact-id": "design-1", "basis-artifact-sha256": design_basis["artifact-sha256"],
|
|
"criteria": [{"criterion-id": "AC-1", "preconditions": [], "input": {},
|
|
"expected-result": "works", "risk-level": "Low",
|
|
"verification-method": "automated-test"}]}, stage="spec")
|
|
check("required spec bundle accepted", submit("wf-contract", criteria)[0]
|
|
and SE.review_artifact("wf-contract", criteria, "accepted", "EXEC-CPO")[0]
|
|
and SE._facts("wf-contract", SE.read_ledger("wf-contract"))["spec_accepted"])
|
|
|
|
print("== quality/release event mutators ==")
|
|
check("fixture opens build stage", open_stage("wf-contract", "build"))
|
|
implementation = os.path.join(WS, "implemented.txt")
|
|
with open(implementation, "w", encoding="utf-8") as fh:
|
|
fh.write("implemented\n")
|
|
add_receipt("wf-contract", "build-test-receipt")
|
|
completion = write_report("wf-contract", "build-1", "completion-record", "ENG-BE",
|
|
{"summary": "implementation complete",
|
|
"source-revision": {"kind": "workspace-tree", "sha256": "a" * 64},
|
|
"primary-artifacts": [{"path": implementation, "kind": "code", "sha256": file_sha(implementation)}],
|
|
"acceptance-criteria-coverage": [{"criterion-id": "AC-1", "status": "Passed",
|
|
"evidence-receipt-ids": ["build-test-receipt"]}],
|
|
"verification-receipt-ids": ["build-test-receipt"],
|
|
"remaining-risks": []}, stage="build")
|
|
check("completion submitted", submit("wf-contract", completion)[0])
|
|
target = next(a for a in SE._trusted_artifacts("wf-contract") if a["artifact-id"] == "build-1")
|
|
check("fixture opens verification stage", open_stage("wf-contract", "verification"))
|
|
add_receipt("wf-contract", "quality-test-receipt", subject="AC-1")
|
|
quality = write_report("wf-contract", "quality-1", "quality-gate-review", "QA", {
|
|
"quality-gate": {"status": "Passed"}, "blocker-open": False,
|
|
"reviewed-artifact-id": "build-1", "reviewed-artifact-sha256": target["artifact-sha256"],
|
|
"checks": [{"check-id": "AC-1", "category": "test", "status": "Passed",
|
|
"evidence-receipt-ids": ["quality-test-receipt"]}],
|
|
"findings": [],
|
|
}, stage="verification")
|
|
check("quality gate event records protected facts", SE.record_quality_gate("wf-contract", quality, "QA")[0]
|
|
and SE._facts("wf-contract", SE.read_ledger("wf-contract"))["quality_gate_status"] == "Passed")
|
|
check("Passed quality gate cannot trigger verification rework",
|
|
not SE.can_transition("wf-contract", "build", actor="OPS-ORCH")[0])
|
|
check("caller fact injection cannot trigger verification rework",
|
|
not SE.can_transition("wf-contract", "build", actor="OPS-ORCH",
|
|
ctx={"facts": {"quality-gate-failed": True}})[0])
|
|
|
|
add_receipt("wf-contract", "quality-failure-receipt", subject="AC-1",
|
|
assertion_status="failed")
|
|
quality_failure = write_report("wf-contract", "quality-2", "quality-gate-review", "QA", {
|
|
"quality-gate": {"status": "Failed"}, "blocker-open": True,
|
|
"reviewed-artifact-id": "build-1", "reviewed-artifact-sha256": target["artifact-sha256"],
|
|
"checks": [{"check-id": "AC-1", "category": "test", "status": "Failed",
|
|
"evidence-receipt-ids": ["quality-failure-receipt"]}],
|
|
"findings": [{"finding-id": "QF-REWORK", "severity": "High", "blocking": True,
|
|
"resolved": False, "summary": "verification regression requires build rework"}],
|
|
}, stage="verification")
|
|
check("trusted current Failed quality gate enables rework predicate",
|
|
SE.record_quality_gate("wf-contract", quality_failure, "QA")[0]
|
|
and SE.can_transition("wf-contract", "build", actor="OPS-ORCH")[0])
|
|
check("non-executor cannot enter rework transition",
|
|
not SE.can_transition("wf-contract", "build", actor="QA")[0])
|
|
check("verification rework transition returns workflow to running build",
|
|
SE.transition("wf-contract", "build", actor="OPS-ORCH")[0]
|
|
and SE.read_ledger("wf-contract").get("stage") == "build"
|
|
and SE.read_ledger("wf-contract").get("stage-status") == "running")
|
|
|
|
check("fixture reopens verification after corrected build", open_stage("wf-contract", "verification"))
|
|
add_receipt("wf-contract", "quality-recovery-receipt", subject="AC-1")
|
|
quality_recovery = write_report("wf-contract", "quality-3", "quality-gate-review", "QA", {
|
|
"quality-gate": {"status": "Passed"}, "blocker-open": False,
|
|
"reviewed-artifact-id": "build-1", "reviewed-artifact-sha256": target["artifact-sha256"],
|
|
"checks": [{"check-id": "AC-1", "category": "test", "status": "Passed",
|
|
"evidence-receipt-ids": ["quality-recovery-receipt"]}],
|
|
"findings": [{"finding-id": "QF-REWORK", "severity": "High", "blocking": True,
|
|
"resolved": True, "summary": "verification regression corrected and rerun"}],
|
|
}, stage="verification")
|
|
check("fresh Passed quality gate supersedes the failed rework review",
|
|
SE.record_quality_gate("wf-contract", quality_recovery, "QA")[0]
|
|
and SE._facts("wf-contract", SE.read_ledger("wf-contract"))["quality_gate_status"] == "Passed"
|
|
and not SE._facts("wf-contract", SE.read_ledger("wf-contract"))["blocker_open"])
|
|
check("fixture opens acceptance stage", open_stage("wf-contract", "acceptance"))
|
|
quality_state = SE.read_ledger("wf-contract")
|
|
release = write_report("wf-contract", "release-1", "release-decision", "HUMAN-001", {
|
|
"release-decision": {"status": "Approved"}, "unresolved-critical-risks": False,
|
|
"reviewed-completion-artifact-id": quality_state["current-completion-artifact-id"],
|
|
"reviewed-completion-artifact-sha256": quality_state["current-completion-artifact-sha256"],
|
|
"reviewed-quality-event-id": quality_state["current-quality-event-id"],
|
|
"reviewed-quality-artifact-id": quality_state["current-quality-artifact-id"],
|
|
"reviewed-quality-artifact-sha256": quality_state["current-quality-artifact-sha256"],
|
|
}, stage="acceptance")
|
|
check("release decision event records protected facts", SE.record_release_decision("wf-contract", release, "HUMAN-001")[0]
|
|
and SE._facts("wf-contract", SE.read_ledger("wf-contract"))["release_acceptance_status"] == "Approved")
|
|
|
|
print("== blocked/resume and actors ==")
|
|
blocked = write_report("wf-contract", "blocked-1", "blocked-report", "OPS-ORCH",
|
|
{"blocker": "dependency", "resume-condition": "receipt exists"}, stage="blocked")
|
|
check("block command requires valid report and enters blocked", SE.block_workflow("wf-contract", blocked)[0]
|
|
and SE.current_stage("wf-contract") == "blocked")
|
|
resume = write_report("wf-contract", "resume-1", "resume-evidence", "OPS-ORCH",
|
|
{"resume-condition-satisfied": True}, stage="blocked")
|
|
check("resume command restores blocked-from", SE.resume_workflow("wf-contract", resume)[0]
|
|
and SE.current_stage("wf-contract") != "blocked")
|
|
check("invented actor rejected", not SE._actor_allowed({"allowed-by": ["OPS-ORCH"]}, "fake-role-agent")[0])
|
|
check("placeholder-only authorization rejected",
|
|
not SE._actor_allowed({"allowed-by": ["design-lead-role-agent"]}, "OPS-ORCH")[0])
|
|
|
|
print("== family candidate metadata only ==")
|
|
resolved = SE.resolve_family("FAM-DESIGN")
|
|
check("family resolves to concrete workers", resolved and "DES-DIRECTOR" in resolved["resolved-workers"])
|
|
pkg = {"target-role-agent": "fam-design"}
|
|
check("family target rejected by context-package", any("candidate metadata" in error for error in CP._semantic_errors(pkg)))
|
|
|
|
print("== contract-derived invariants ==")
|
|
contract = SE.load_contracts()
|
|
kinds = contract.get("artifact-kinds", {})
|
|
bundled = set()
|
|
for bundle in (contract.get("artifact-bundles", {}) or {}).values():
|
|
bundled.update(bundle.get("always") or [])
|
|
for conditional in bundle.get("conditional") or []:
|
|
bundled.update(conditional.get("require") or [])
|
|
check("every bundle artifact-kind is declared", bundled.issubset(set(kinds)))
|
|
mapped_must_reads = {
|
|
kind
|
|
for mapping in (SE.load_cmap().get("design-to-build-contract", {}).get("mappings", []) or [])
|
|
for kind in (mapping.get("must-read-designs") or [])
|
|
}
|
|
check("compatibility must-read vocabulary contains only canonical artifact-kinds",
|
|
mapped_must_reads.issubset(set(kinds)))
|
|
default_schema = contract.get("default-payload-schema-ref")
|
|
schema_dir = os.path.join(ROOT, ".claude", "schemas")
|
|
check("every artifact-kind resolves to an existing payload schema", all(
|
|
os.path.isfile(os.path.join(schema_dir, definition.get("payload-schema-ref") or default_schema))
|
|
for definition in kinds.values()))
|
|
cascade_pairs = {("intake", "discovery"), ("discovery", "decide"), ("decide", "design"),
|
|
("design", "spec"), ("spec", "build"), ("build", "verification"),
|
|
("verification", "acceptance"), ("acceptance", "released")}
|
|
runtime = {(t.get("from"), t.get("to")): t for t in SE._ws_transitions()}
|
|
check("every cascade transition has concrete OPS-ORCH executor", all(
|
|
pair in runtime and "OPS-ORCH" in ((runtime[pair].get("allowed-by") or {}).get("executor") or [])
|
|
for pair in cascade_pairs))
|
|
check("every protected mutable gate has an official mutator", all(callable(getattr(SE, name, None)) for name in (
|
|
"init_workflow", "submit_artifact", "review_artifact", "record_human_signoff",
|
|
"record_quality_gate", "record_release_decision", "block_workflow", "resume_workflow",
|
|
"complete_stage", "enter_stage")))
|
|
workflow_outputs = set()
|
|
workflow_conditions = set()
|
|
for workflow in (contract.get("workflows", {}) or {}).values():
|
|
for stage in (workflow.get("stages", {}) or {}).values():
|
|
outputs = stage.get("outputs") or {}
|
|
workflow_outputs.update(outputs.get("bundle") or [])
|
|
if outputs.get("dynamic-bundle"):
|
|
bundle = (contract.get("artifact-bundles", {}) or {}).get(outputs["dynamic-bundle"], {})
|
|
workflow_outputs.update(bundle.get("always") or [])
|
|
for condition in bundle.get("conditional") or []:
|
|
workflow_outputs.update(condition.get("require") or [])
|
|
workflow_conditions.update(stage.get("exit-gate") or [])
|
|
for transition in workflow.get("additional-transitions") or []:
|
|
workflow_conditions.update(transition.get("required-conditions") or [])
|
|
for transition in contract.get("side-transitions") or []:
|
|
workflow_conditions.update(transition.get("required-conditions") or [])
|
|
check("every declared stage output has an artifact contract", workflow_outputs.issubset(set(kinds)))
|
|
check("every declared exit gate has a runtime predicate", workflow_conditions.issubset(set(SE._PREDICATES)))
|
|
|
|
print(f"\n{passed} passed, {failed} failed")
|
|
shutil.rmtree(WS, ignore_errors=True)
|
|
sys.exit(0 if failed == 0 else 1)
|