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()) == {"platform-vpc-apply", "lifecycle", "platform-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"]["platform-vpc-destroy"] assert destroy_job.get("if") == "always()", "platform-vpc-destroy must always run (cleanup)"