Jon Chery
ad3cc5f129
acdl-ci / Lint (pull_request) Successful in 7s
acdl-ci / Test (pull_request) Successful in 4m3s
acdl-ci / Platform check-only (offline) (pull_request) Successful in 21s
acdl-modules-lifecycle / CI VPC apply (pull_request) Failing after 1m25s
acdl-modules-lifecycle / L1 lifecycle (alb) (pull_request) Has been skipped
acdl-modules-lifecycle / L1 lifecycle (cloudfront) (pull_request) Has been skipped
acdl-modules-lifecycle / L1 lifecycle (ecr) (pull_request) Has been skipped
acdl-modules-lifecycle / L1 lifecycle (ecs-cluster) (pull_request) Has been skipped
acdl-modules-lifecycle / L1 lifecycle (ecs-service) (pull_request) Has been skipped
acdl-modules-lifecycle / L1 lifecycle (iam-role) (pull_request) Has been skipped
acdl-modules-lifecycle / L1 lifecycle (kms-key) (pull_request) Has been skipped
acdl-modules-lifecycle / L1 lifecycle (rds) (pull_request) Has been skipped
acdl-modules-lifecycle / L1 lifecycle (s3) (pull_request) Has been skipped
acdl-modules-lifecycle / L1 lifecycle (uptime) (pull_request) Has been skipped
acdl-modules-lifecycle / L1 lifecycle (vpc) (pull_request) Has been skipped
acdl-modules-lifecycle / L1 lifecycle (waf) (pull_request) Has been skipped
acdl-modules-lifecycle / CI VPC destroy (pull_request) Successful in 44s
Two architectural changes: 1. Created terraform/ci-vpc/ — a short-lived VPC for L1 module lifecycle testing, separate from the long-lived platform VPC. Created before VPC-dependent modules (alb, ecs-service, rds, uptime) are tested, destroyed after. Outputs (vpc_id, subnet_ids, sg_id, cluster_arn) are passed to those modules via scripts/run_lifecycle_test.sh + run_lifecycle_destroy.sh wrappers that inject the CI VPC outputs into the example contracts. 2. Updated the workflow to use ci-vpc-apply → lifecycle (with artifact passing) → ci-vpc-destroy (always runs). 8 module-specific fixes: - s3: unique bucket names (acdl-ci-s3a-simple/complex) instead of globally-taken 'my-simple-bucket' - kms-key: alias name with no spaces (locals.tf → alias/acdl-ci-kms) - iam-role: example contract uses role_name (not name, which the interface doesn't declare) - ecs-service: example contract uses family (not name); VPC inputs (cluster_arn, subnets, security_group) injected by CI VPC wrapper - uptime: added subnets, security_group, cluster_arn to interface + module; network_configuration is dynamic (only when subnets provided) - rds: added subnet_ids input + db_subnet_group resource (conditional on subnet_ids being non-empty) - alb: removed hardcoded placeholder sg/subnet values from examples; vpc_id + subnets + security_group injected by CI VPC wrapper - cloudfront: removed invalid placeholder WAF ARN from complex example Regression: 479 passed, 0 skipped, 5 deselected. All 24 example contracts pass --check-only. ---ci--- project: acdl phase: P59 milestone: v1.11 status: execute ---/ci---
587 lines
25 KiB
Python
587 lines
25 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.yml")
|
|
jsonschema.validate(contract, schema)
|
|
|
|
def test_contract_has_three_stages(self):
|
|
contract = _load_yaml("pipelines/ci.yml")
|
|
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.yml")
|
|
assert contract["runner"] == "ubuntu-latest"
|
|
|
|
def test_contract_python_version(self):
|
|
contract = _load_yaml("pipelines/ci.yml")
|
|
assert contract["python_version"] == "3.12"
|
|
|
|
def test_contract_triggers_push_main(self):
|
|
contract = _load_yaml("pipelines/ci.yml")
|
|
assert "main" in contract["triggers"]["push"]
|
|
|
|
def test_contract_triggers_pr_main(self):
|
|
contract = _load_yaml("pipelines/ci.yml")
|
|
assert "main" in contract["triggers"]["pull_request"]
|
|
|
|
def test_contract_all_stages_required(self):
|
|
contract = _load_yaml("pipelines/ci.yml")
|
|
for stage in contract["stages"]:
|
|
assert stage["required"] is True
|
|
|
|
def test_contract_lint_command_compiles_python(self):
|
|
contract = _load_yaml("pipelines/ci.yml")
|
|
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.yml")
|
|
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.yml")
|
|
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.yml")
|
|
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.yml")
|
|
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.yml")
|
|
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.yml")
|
|
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" in result.stdout
|
|
assert "main.tf" in result.stdout
|
|
assert "module" 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" 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/contract.yml")
|
|
jsonschema.validate(contract, schema)
|
|
|
|
def test_deploy_contract_has_nine_stages(self):
|
|
contract = _load_yaml("pipelines/contract.yml")
|
|
stage_names = [s["name"] for s in contract["stages"]]
|
|
assert stage_names == [
|
|
"validate-contract",
|
|
"resolve-stack",
|
|
"terraform-plan",
|
|
"checkov",
|
|
"confidence",
|
|
"apply",
|
|
"publish-outputs",
|
|
"deploy-uptime",
|
|
"comment-outputs",
|
|
]
|
|
|
|
def test_deploy_contract_runner_is_ubuntu_latest(self):
|
|
contract = _load_yaml("pipelines/contract.yml")
|
|
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/contract.yml")
|
|
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.yml"
|
|
|
|
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/contract.yml")
|
|
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/contract.yml")
|
|
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"
|
|
|
|
def test_deploy_workflow_static_key_override_wired(self):
|
|
"""S1: the static-key override must be wired to configure-aws-credentials
|
|
inputs (access-key-id/secret-access-key), not inert env vars."""
|
|
wf = _load_workflow(".gitea/workflows/deploy.yml")
|
|
deploy_job = wf["jobs"]["deploy"]
|
|
creds_step = next(
|
|
s for s in deploy_job["steps"]
|
|
if "configure-aws-credentials" in s.get("uses", "")
|
|
)
|
|
with_block = creds_step.get("with", {})
|
|
assert "access-key-id" in with_block, "S1: access-key-id input must be wired"
|
|
assert "secret-access-key" in with_block, "S1: secret-access-key input must be wired"
|
|
assert "role-to-assume" in with_block, "S1: role-to-assume must still be present (conditional)"
|
|
|
|
|
|
class TestSampleContractVersioning:
|
|
def test_ci_workflow_uses_versioned_tag(self):
|
|
"""The consumer CI workflow (the runtime dispatch) uses a versioned @vX.Y tag.
|
|
The contract no longer carries a `uses:` field (removed in P57); the
|
|
version pin lives in the consumer's CI workflow reference."""
|
|
import yaml
|
|
wf = yaml.safe_load((ROOT / ".github/workflows/deploy.yml").read_text())
|
|
# The workflow itself doesn't have a top-level uses; check the checkout
|
|
# ref of the platform repo (the versioned tag the consumer pins to).
|
|
deploy_job = wf["jobs"]["deploy"]
|
|
checkout_steps = [s for s in deploy_job["steps"]
|
|
if "checkout" in s.get("uses", "")]
|
|
platform_checkout = next(
|
|
(s for s in checkout_steps if s.get("with", {}).get("path") == "platform"),
|
|
None)
|
|
assert platform_checkout is not None, "must have a platform repo checkout"
|
|
ref = platform_checkout["with"]["ref"]
|
|
assert ref.startswith("v"), f"platform ref must be a versioned tag, got {ref}"
|
|
assert "@main" not in ref and ref != "main", "must not pin to @main"
|
|
|
|
|
|
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/*.yml" 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
|
|
|
|
|
|
class TestModulesLifecyclePipeline:
|
|
"""P59: modules-lifecycle pipeline — schema, byte-identical, matrix."""
|
|
|
|
def test_schema_is_valid_json_schema(self):
|
|
schema = json.load(open(ROOT / "schemas/modules-lifecycle-pipeline.schema.json"))
|
|
jsonschema.Draft202012Validator.check_schema(schema)
|
|
|
|
def test_contract_validates_against_schema(self):
|
|
schema = json.load(open(ROOT / "schemas/modules-lifecycle-pipeline.schema.json"))
|
|
contract = _load_yaml("pipelines/modules-lifecycle.yml")
|
|
jsonschema.validate(contract, schema)
|
|
|
|
def test_gitea_workflow_exists(self):
|
|
assert (ROOT / ".gitea/workflows/modules-lifecycle.yml").is_file()
|
|
|
|
def test_github_workflow_exists(self):
|
|
assert (ROOT / ".github/workflows/modules-lifecycle.yml").is_file()
|
|
|
|
def test_workflows_are_byte_identical(self):
|
|
gitea = open(ROOT / ".gitea/workflows/modules-lifecycle.yml", "rb").read()
|
|
github = open(ROOT / ".github/workflows/modules-lifecycle.yml", "rb").read()
|
|
assert gitea == github, "Gitea and GitHub workflows must be byte-identical"
|
|
|
|
def test_workflow_name_matches_contract(self):
|
|
wf = _load_workflow(".gitea/workflows/modules-lifecycle.yml")
|
|
contract = _load_yaml("pipelines/modules-lifecycle.yml")
|
|
assert wf["name"] == contract["name"]
|
|
|
|
def test_workflow_has_three_jobs(self):
|
|
wf = _load_workflow(".gitea/workflows/modules-lifecycle.yml")
|
|
assert set(wf["jobs"].keys()) == {"ci-vpc-apply", "lifecycle", "ci-vpc-destroy"}
|
|
|
|
def test_workflow_triggers_match_contract(self):
|
|
wf = _load_workflow(".gitea/workflows/modules-lifecycle.yml")
|
|
contract = _load_yaml("pipelines/modules-lifecycle.yml")
|
|
assert wf["on"]["pull_request"]["branches"] == contract["triggers"]["pull_request"]
|
|
assert "workflow_dispatch" in wf["on"]
|
|
|
|
def test_matrix_lists_all_12_l1_modules(self):
|
|
wf = _load_workflow(".gitea/workflows/modules-lifecycle.yml")
|
|
matrix_modules = wf["jobs"]["lifecycle"]["strategy"]["matrix"]["module"]
|
|
expected = {"s3", "kms-key", "ecr", "ecs-cluster", "iam-role", "cloudfront",
|
|
"waf", "vpc", "alb", "ecs-service", "rds", "uptime"}
|
|
assert set(matrix_modules) == expected
|
|
|
|
def test_contract_matrix_lists_all_12_l1_modules(self):
|
|
contract = _load_yaml("pipelines/modules-lifecycle.yml")
|
|
assert set(contract["matrix"]["modules"]) == {
|
|
"s3", "kms-key", "ecr", "ecs-cluster", "iam-role", "cloudfront",
|
|
"waf", "vpc", "alb", "ecs-service", "rds", "uptime"
|
|
}
|
|
|
|
def test_lifecycle_job_has_apply_modify_destroy_steps(self):
|
|
wf = _load_workflow(".gitea/workflows/modules-lifecycle.yml")
|
|
steps = wf["jobs"]["lifecycle"]["steps"]
|
|
step_names = [s.get("name", "") for s in steps]
|
|
assert any("Apply" in n for n in step_names), "Missing apply step"
|
|
assert any("Modify" in n for n in step_names), "Missing modify step"
|
|
assert any("Destroy" in n for n in step_names), "Missing destroy step"
|
|
|
|
def test_platform_vpc_destroy_always_runs(self):
|
|
wf = _load_workflow(".gitea/workflows/modules-lifecycle.yml")
|
|
destroy_job = wf["jobs"]["ci-vpc-destroy"]
|
|
assert destroy_job.get("if") == "always()", "ci-vpc-destroy must always run (cleanup)" |