Files
acdl/tests/test_pipeline_contract.py
T
Jon Chery a9d8b31595 feat(P27): RDS primitive + adapter expansion (D-059)
Add modules/l1/rds/ with interface.json (engine enum for postgres, mysql,
mariadb, sqlserver, oracle), instance.json, README.md (full template with
compliance extension points). Register in registry.json. Expand the
adapter TYPE_MAP/INPUT_MAP/OUTPUT_MAP for aws:rds:instance -> aws_db_instance;
emit backup_retention_period, deletion_protection, storage_encrypted, and
skip_final_snapshot from NFRs/inputs. Add RDS to the primitives-plan matrix.
Update tests for the new registry entry count (12) + RDS adapter emission.

---ci---
project: acdl
phase: 27
milestone: v1.7
status: execute
---/ci---
2026-07-22 20:21:45 +00:00

493 lines
20 KiB
Python

import json
import os
import subprocess
import sys
from pathlib import Path
import jsonschema
import pytest
import yaml
ROOT = Path(__file__).resolve().parent.parent
def _load_yaml(path):
with open(ROOT / path) as f:
return yaml.safe_load(f)
def _load_workflow(path):
wf = _load_yaml(path)
if True in wf:
wf["on"] = wf[True]
return wf
class TestPipelineSchema:
def test_schema_is_valid_json_schema(self):
schema = json.load(open(ROOT / "schemas/pipeline.schema.json"))
jsonschema.Draft202012Validator.check_schema(schema)
def test_schema_has_required_fields(self):
schema = json.load(open(ROOT / "schemas/pipeline.schema.json"))
assert "name" in schema["required"]
assert "triggers" in schema["required"]
assert "runner" in schema["required"]
assert "stages" in schema["required"]
def test_schema_stage_def_has_command_and_required(self):
schema = json.load(open(ROOT / "schemas/pipeline.schema.json"))
stage_def = schema["$defs"]["stage"]
assert "command" in stage_def["required"]
assert "required" in stage_def["required"]
class TestPipelineContract:
def test_contract_validates_against_schema(self):
schema = json.load(open(ROOT / "schemas/pipeline.schema.json"))
contract = _load_yaml("pipelines/ci.yaml")
jsonschema.validate(contract, schema)
def test_contract_has_three_stages(self):
contract = _load_yaml("pipelines/ci.yaml")
stage_names = [s["name"] for s in contract["stages"]]
assert stage_names == ["lint", "test", "check-only"]
def test_contract_runner_is_ubuntu_latest(self):
contract = _load_yaml("pipelines/ci.yaml")
assert contract["runner"] == "ubuntu-latest"
def test_contract_python_version(self):
contract = _load_yaml("pipelines/ci.yaml")
assert contract["python_version"] == "3.12"
def test_contract_triggers_push_main(self):
contract = _load_yaml("pipelines/ci.yaml")
assert "main" in contract["triggers"]["push"]
def test_contract_triggers_pr_main(self):
contract = _load_yaml("pipelines/ci.yaml")
assert "main" in contract["triggers"]["pull_request"]
def test_contract_all_stages_required(self):
contract = _load_yaml("pipelines/ci.yaml")
for stage in contract["stages"]:
assert stage["required"] is True
def test_contract_lint_command_compiles_python(self):
contract = _load_yaml("pipelines/ci.yaml")
lint = next(s for s in contract["stages"] if s["name"] == "lint")
assert "py_compile" in lint["command"]
assert "core/confidence_signal.py" in lint["command"]
assert "adapters/terraform/adapter.py" in lint["command"]
def test_contract_test_command_runs_pytest(self):
contract = _load_yaml("pipelines/ci.yaml")
test_stage = next(s for s in contract["stages"] if s["name"] == "test")
assert "pytest" in test_stage["command"]
def test_contract_check_only_runs_platform(self):
contract = _load_yaml("pipelines/ci.yaml")
check = next(s for s in contract["stages"] if s["name"] == "check-only")
assert "run_platform.sh" in check["command"]
assert "--check-only" in check["command"]
class TestWorkflowConformance:
def test_gitea_workflow_exists(self):
assert (ROOT / ".gitea/workflows/ci.yml").is_file()
def test_github_workflow_exists(self):
assert (ROOT / ".github/workflows/ci.yml").is_file()
def test_workflows_are_byte_identical(self):
gitea = open(ROOT / ".gitea/workflows/ci.yml", "rb").read()
github = open(ROOT / ".github/workflows/ci.yml", "rb").read()
assert gitea == github, "Gitea and GitHub workflows must be byte-identical"
def test_gitea_workflow_name_matches_contract(self):
wf = _load_workflow(".gitea/workflows/ci.yml")
contract = _load_yaml("pipelines/ci.yaml")
assert wf["name"] == contract["name"]
def test_gitea_workflow_has_three_jobs(self):
wf = _load_workflow(".gitea/workflows/ci.yml")
assert set(wf["jobs"].keys()) == {"lint", "test", "check-only"}
def test_gitea_workflow_triggers_match_contract(self):
wf = _load_workflow(".gitea/workflows/ci.yml")
contract = _load_yaml("pipelines/ci.yaml")
assert wf["on"]["push"]["branches"] == contract["triggers"]["push"]
assert wf["on"]["pull_request"]["branches"] == contract["triggers"]["pull_request"]
def test_gitea_workflow_runner_matches_contract(self):
wf = _load_workflow(".gitea/workflows/ci.yml")
contract = _load_yaml("pipelines/ci.yaml")
for job in wf["jobs"].values():
assert job["runs-on"] == contract["runner"]
def test_gitea_workflow_python_version_matches_contract(self):
wf = _load_workflow(".gitea/workflows/ci.yml")
contract = _load_yaml("pipelines/ci.yaml")
for job in wf["jobs"].values():
setup_step = next(
s for s in job["steps"] if "setup-python" in s.get("uses", "")
)
assert setup_step["with"]["python-version"] == contract["python_version"]
def test_gitea_lint_command_matches_contract(self):
wf = _load_workflow(".gitea/workflows/ci.yml")
lint_job = wf["jobs"]["lint"]
run_step = next(s for s in lint_job["steps"] if "run" in s)
assert "py_compile" in run_step["run"]
for py_file in [
"core/confidence_signal.py",
"core/outbox_writer.py",
"core/contract_resolver.py",
"adapters/terraform/adapter.py",
"adapters/terraform/policy/checkov_adapter.py",
"scripts/push_consumer_image.py",
]:
assert py_file in run_step["run"], f"{py_file} missing from lint command"
def test_gitea_test_command_matches_contract(self):
wf = _load_workflow(".gitea/workflows/ci.yml")
test_job = wf["jobs"]["test"]
run_step = next(s for s in test_job["steps"] if "run" in s and "pytest" in s["run"])
assert "pytest" in run_step["run"]
def test_gitea_check_only_command_matches_contract(self):
wf = _load_workflow(".gitea/workflows/ci.yml")
check_job = wf["jobs"]["check-only"]
run_step = next(
s for s in check_job["steps"] if "run" in s and "run_platform" in s["run"]
)
assert "run_platform.sh" in run_step["run"]
assert "--check-only" in run_step["run"]
class TestRunCiScript:
def test_run_ci_script_exists_and_executable(self):
path = ROOT / "scripts/run_ci.sh"
assert path.is_file()
assert os.access(path, os.X_OK)
def test_run_ci_script_contains_lint_stage(self):
content = open(ROOT / "scripts/run_ci.sh").read()
assert "py_compile" in content
assert "core/confidence_signal.py" in content
assert "core/contract_resolver.py" in content
assert "adapters/terraform/adapter.py" in content
def test_run_ci_script_contains_test_stage(self):
content = open(ROOT / "scripts/run_ci.sh").read()
assert "pytest" in content
assert "tests/" in content
def test_run_ci_script_contains_check_only_stage(self):
content = open(ROOT / "scripts/run_ci.sh").read()
assert "run_platform.sh" in content
assert "--check-only" in content
def test_run_ci_script_has_success_message(self):
content = open(ROOT / "scripts/run_ci.sh").read()
assert "CI PIPELINE OK" in content
def test_run_ci_lint_and_check_only_pass(self):
result = subprocess.run(
["bash", "-c",
f"cd {ROOT} && "
"python3 -m py_compile "
"core/confidence_signal.py "
"core/outbox_writer.py "
"core/contract_resolver.py "
"adapters/terraform/adapter.py "
"adapters/terraform/policy/checkov_adapter.py "
"scripts/push_consumer_image.py && "
"echo 'lint: OK' && "
"bash scripts/run_platform.sh --check-only && "
"echo 'check-only: OK'"],
capture_output=True, text=True, cwd=str(ROOT),
timeout=30,
)
assert result.returncode == 0, f"stdout: {result.stdout}\nstderr: {result.stderr}"
assert "lint: OK" in result.stdout
assert "check-only: OK" in result.stdout
assert "PLATFORM CHECK OK" in result.stdout
class TestRunPlatformStreaming:
def test_check_only_streams_emitted_terraform(self):
result = subprocess.run(
["bash", str(ROOT / "scripts/run_platform.sh"), "--check-only"],
capture_output=True, text=True, cwd=str(ROOT),
timeout=30,
)
assert result.returncode == 0
assert "PLATFORM CHECK OK" in result.stdout
assert "--- emitted terraform/spike/main.tf ---" in result.stdout
assert "aws_s3_bucket" in result.stdout
def test_check_only_quiet_suppresses_terraform(self):
result = subprocess.run(
["bash", str(ROOT / "scripts/run_platform.sh"), "--check-only", "--quiet"],
capture_output=True, text=True, cwd=str(ROOT),
timeout=30,
)
assert result.returncode == 0
assert "PLATFORM CHECK OK" in result.stdout
assert "--- emitted terraform/spike/main.tf ---" not in result.stdout
class TestDeployPipelineSchema:
def test_deploy_schema_is_valid_json_schema(self):
schema = json.load(open(ROOT / "schemas/deploy-pipeline.schema.json"))
jsonschema.Draft202012Validator.check_schema(schema)
def test_deploy_schema_has_required_fields(self):
schema = json.load(open(ROOT / "schemas/deploy-pipeline.schema.json"))
assert "name" in schema["required"]
assert "triggers" in schema["required"]
assert "runner" in schema["required"]
assert "stages" in schema["required"]
def test_deploy_schema_stage_def_has_command_and_required(self):
schema = json.load(open(ROOT / "schemas/deploy-pipeline.schema.json"))
stage_def = schema["$defs"]["stage"]
assert "command" in stage_def["required"]
assert "required" in stage_def["required"]
class TestDeployPipelineContract:
def test_deploy_contract_validates_against_schema(self):
schema = json.load(open(ROOT / "schemas/deploy-pipeline.schema.json"))
contract = _load_yaml("pipelines/deploy.yaml")
jsonschema.validate(contract, schema)
def test_deploy_contract_has_eight_stages(self):
contract = _load_yaml("pipelines/deploy.yaml")
stage_names = [s["name"] for s in contract["stages"]]
assert stage_names == [
"validate-contract",
"resolve-stack",
"terraform-plan",
"checkov",
"confidence",
"apply",
"publish-outputs",
"comment-outputs",
]
def test_deploy_contract_runner_is_ubuntu_latest(self):
contract = _load_yaml("pipelines/deploy.yaml")
assert contract["runner"] == "ubuntu-latest"
class TestDeployWorkflowConformance:
def test_gitea_deploy_workflow_exists(self):
assert (ROOT / ".gitea/workflows/deploy.yml").is_file()
def test_github_deploy_workflow_exists(self):
assert (ROOT / ".github/workflows/deploy.yml").is_file()
def test_deploy_workflows_are_byte_identical(self):
gitea = open(ROOT / ".gitea/workflows/deploy.yml", "rb").read()
github = open(ROOT / ".github/workflows/deploy.yml", "rb").read()
assert gitea == github, "Gitea and GitHub deploy workflows must be byte-identical"
def test_deploy_workflow_name_matches_contract(self):
wf = _load_workflow(".gitea/workflows/deploy.yml")
contract = _load_yaml("pipelines/deploy.yaml")
assert wf["name"] == contract["name"]
def test_deploy_workflow_is_reusable(self):
wf = _load_workflow(".gitea/workflows/deploy.yml")
assert "workflow_call" in wf["on"]
def test_deploy_workflow_has_contract_input(self):
wf = _load_workflow(".gitea/workflows/deploy.yml")
inputs = wf["on"]["workflow_call"]["inputs"]
assert "contract" in inputs
assert inputs["contract"]["default"] == ".acdl/contract.yaml"
def test_deploy_workflow_has_mode_input(self):
wf = _load_workflow(".gitea/workflows/deploy.yml")
inputs = wf["on"]["workflow_call"]["inputs"]
assert "mode" in inputs
assert inputs["mode"]["default"] == "full"
def test_deploy_workflow_runner_matches_contract(self):
wf = _load_workflow(".gitea/workflows/deploy.yml")
contract = _load_yaml("pipelines/deploy.yaml")
for job in wf["jobs"].values():
assert job["runs-on"] == contract["runner"]
def test_deploy_workflow_python_version_matches_contract(self):
wf = _load_workflow(".gitea/workflows/deploy.yml")
contract = _load_yaml("pipelines/deploy.yaml")
for job in wf["jobs"].values():
setup_step = next(
s for s in job["steps"] if "setup-python" in s.get("uses", "")
)
assert setup_step["with"]["python-version"] == contract["python_version"]
def test_deploy_workflow_invokes_run_platform(self):
wf = _load_workflow(".gitea/workflows/deploy.yml")
deploy_job = wf["jobs"]["deploy"]
run_step = next(
s for s in deploy_job["steps"] if "run" in s and "run_platform" in s["run"]
)
assert "run_platform.sh" in run_step["run"]
def test_deploy_workflow_checks_out_platform_repo(self):
wf = _load_workflow(".gitea/workflows/deploy.yml")
deploy_job = wf["jobs"]["deploy"]
platform_checkout = next(
s for s in deploy_job["steps"]
if "checkout" in s.get("uses", "") and s.get("with", {}).get("path") == "platform"
)
assert platform_checkout["with"]["repository"] == "acdl/acdl"
def test_deploy_workflow_permissions_id_token_write(self):
wf = _load_workflow(".gitea/workflows/deploy.yml")
assert wf["permissions"]["id-token"] == "write"
assert wf["permissions"]["contents"] == "read"
class TestSampleContractVersioning:
def test_sample_contract_uses_versioned_tag(self):
contract = _load_yaml("contracts/static-assets.yaml")
uses = contract["uses"]
assert "@v" in uses, "sample contract must use a versioned @vX.Y tag"
assert "@main" not in uses, "sample contract must not use @main"
assert uses == "acdl/pipelines/deploy.yaml@v1.6"
class TestPlatformWorkflows:
"""Validate the Phase 26 platform pipelines exist and conform."""
def test_platform_test_workflow_exists(self):
assert (ROOT / ".github/workflows/platform-test.yml").is_file()
def test_primitives_plan_workflow_exists(self):
assert (ROOT / ".github/workflows/primitives-plan.yml").is_file()
def test_patterns_plan_workflow_exists(self):
assert (ROOT / ".github/workflows/patterns-plan.yml").is_file()
def test_release_workflow_exists(self):
assert (ROOT / ".github/workflows/release.yml").is_file()
def test_platform_test_has_four_stages(self):
wf = _load_workflow(".github/workflows/platform-test.yml")
job_names = set(wf["jobs"].keys())
assert job_names == {"lint", "unit-test", "integration-test", "schema-validation"}
def test_platform_test_lint_compiles_python(self):
wf = _load_workflow(".github/workflows/platform-test.yml")
lint_job = wf["jobs"]["lint"]
run_step = next(s for s in lint_job["steps"] if "run" in s)
assert "py_compile" in run_step["run"]
for py_file in [
"core/confidence_signal.py",
"core/outbox_writer.py",
"core/contract_resolver.py",
"core/environment_check.py",
"core/output_publisher.py",
"core/lambda/contract_ingestor.py",
"adapters/terraform/adapter.py",
"adapters/terraform/policy/checkov_adapter.py",
"adapters/wiz/wiz_adapter.py",
"adapters/kyverno/kyverno_adapter.py",
"scripts/push_consumer_image.py",
]:
assert py_file in run_step["run"], f"{py_file} missing from platform-test lint"
def test_platform_test_unit_test_runs_pytest(self):
wf = _load_workflow(".github/workflows/platform-test.yml")
test_job = wf["jobs"]["unit-test"]
run_step = next(s for s in test_job["steps"] if "run" in s and "pytest" in s["run"])
assert "pytest" in run_step["run"]
def test_platform_test_integration_runs_all_contracts(self):
wf = _load_workflow(".github/workflows/platform-test.yml")
integ_job = wf["jobs"]["integration-test"]
run_step = next(
s for s in integ_job["steps"] if "run" in s and "run_platform" in s["run"]
)
assert "run_platform.sh" in run_step["run"]
assert "--check-only" in run_step["run"]
assert "contracts/*.yaml" in run_step["run"]
def test_platform_test_schema_validation_validates_schemas(self):
wf = _load_workflow(".github/workflows/platform-test.yml")
schema_job = wf["jobs"]["schema-validation"]
steps_text = " ".join(s.get("run", "") for s in schema_job["steps"])
assert "jsonschema" in steps_text
assert "stack.schema.json" in steps_text
def test_platform_test_triggers_pr_only(self):
wf = _load_workflow(".github/workflows/platform-test.yml")
assert "pull_request" in wf["on"]
assert "main" in wf["on"]["pull_request"]["branches"]
# platform-test should NOT trigger on push (ci.yml handles push-to-main)
assert "push" not in wf["on"]
def test_primitives_plan_has_matrix_with_all_l1_primitives(self):
wf = _load_workflow(".github/workflows/primitives-plan.yml")
job = wf["jobs"]["primitive-plan"]
matrix = job["strategy"]["matrix"]
expected = ["s3", "vpc", "ecs-cluster", "ecs-service", "iam-role", "alb", "ecr", "cloudfront", "waf", "rds"]
assert sorted(matrix["primitive"]) == sorted(expected)
def test_primitives_plan_runs_run_primitive_plan(self):
wf = _load_workflow(".github/workflows/primitives-plan.yml")
job = wf["jobs"]["primitive-plan"]
run_step = next(s for s in job["steps"] if "run" in s and "run_primitive_plan" in s["run"])
assert "run_primitive_plan.sh" in run_step["run"]
assert "--check-only" in run_step["run"]
def test_primitives_plan_triggers_pr_only(self):
wf = _load_workflow(".github/workflows/primitives-plan.yml")
assert "pull_request" in wf["on"]
assert "main" in wf["on"]["pull_request"]["branches"]
assert "push" not in wf["on"]
def test_patterns_plan_has_matrix_with_all_l2_modules(self):
wf = _load_workflow(".github/workflows/patterns-plan.yml")
job = wf["jobs"]["pattern-plan"]
matrix = job["strategy"]["matrix"]
expected = ["static-assets", "microservice"]
assert sorted(matrix["module"]) == sorted(expected)
def test_patterns_plan_runs_run_pattern_plan(self):
wf = _load_workflow(".github/workflows/patterns-plan.yml")
job = wf["jobs"]["pattern-plan"]
run_step = next(s for s in job["steps"] if "run" in s and "run_pattern_plan" in s["run"])
assert "run_pattern_plan.sh" in run_step["run"]
assert "--check-only" in run_step["run"]
def test_patterns_plan_triggers_pr_only(self):
wf = _load_workflow(".github/workflows/patterns-plan.yml")
assert "pull_request" in wf["on"]
assert "main" in wf["on"]["pull_request"]["branches"]
assert "push" not in wf["on"]
def test_release_workflow_triggers_push_main(self):
wf = _load_workflow(".github/workflows/release.yml")
assert "push" in wf["on"]
assert "main" in wf["on"]["push"]["branches"]
def test_release_workflow_has_contents_write_permission(self):
wf = _load_workflow(".github/workflows/release.yml")
# permissions are declared at the job level (the release job)
release_job = wf["jobs"]["release"]
assert release_job["permissions"]["contents"] == "write"
def test_release_workflow_fetch_depth_zero(self):
wf = _load_workflow(".github/workflows/release.yml")
release_job = wf["jobs"]["release"]
checkout = next(
s for s in release_job["steps"] if "checkout" in s.get("uses", "")
)
assert checkout["with"]["fetch-depth"] == 0