Merge milestone/v1.11-restart — v1.11 complete (stateless adapter + pipeline-driven module lifecycle testing, P56a-P65)
v1.11 closes G-005 (CAP-017..022 deploy-unverified → Verified via lifecycle pipeline) and G-008 (no cost docs → COST.md). Phases: - P56a: stateless adapter rewrite (918-line monolith → 196-line assembler) - P56b: 12 L1 module terraform subdirs authored - P57: shell orchestrator --apply/--destroy lifecycle modes - P58: single platform VPC + deterministic env-aware state keys - P59: L1 module lifecycle pipeline authored - P60: L1 lifecycle live run (retrofit — module fixes for live AWS) - P61: L2 lifecycle pipeline authored - P62: L2 lifecycle live run - P63: CAP-017..022 regression registry + COST.md - P64: pre-mortem + teardown (zero live resources) - P65: rewrite caps + decks 485 offline tests pass. All 12 requirements complete. Zero live ACDL resources remain (D-096 enforced). # Conflicts: # .ciagent/PERSONAS.md # .ciagent/REQUIREMENTS.md # .ciagent/ROADMAP.md # .ciagent/config.json
This commit is contained in:
Jon Chery
+164
-587
@@ -1,5 +1,6 @@
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
@@ -8,9 +9,7 @@ import pytest
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from adapters.terraform.adapter import (
|
||||
TYPE_MAP, INPUT_MAP, OUTPUT_MAP, adapt, _tf_value, _ref_expr,
|
||||
)
|
||||
from adapters.terraform.adapter import adapt, _tf_value, _ref_expr, _module_name
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
|
||||
@@ -54,614 +53,192 @@ class TestRegistry:
|
||||
iface = json.load(open(iface_path))
|
||||
assert iface["name"] == name
|
||||
|
||||
|
||||
class TestTypeMap:
|
||||
def test_s3_in_type_map(self):
|
||||
assert TYPE_MAP["aws:s3:bucket"] == "aws_s3_bucket"
|
||||
|
||||
def test_vpc_types_in_type_map(self):
|
||||
assert TYPE_MAP["aws:ec2:vpc"] == "aws_vpc"
|
||||
assert TYPE_MAP["aws:ec2:subnet"] == "aws_subnet"
|
||||
assert TYPE_MAP["aws:ec2:routetable"] == "aws_route_table"
|
||||
|
||||
def test_ecs_types_in_type_map(self):
|
||||
assert TYPE_MAP["aws:ecs:cluster"] == "aws_ecs_cluster"
|
||||
assert TYPE_MAP["aws:ecs:task_definition"] == "aws_ecs_task_definition"
|
||||
assert TYPE_MAP["aws:ecs:service"] == "aws_ecs_service"
|
||||
|
||||
def test_alb_types_in_type_map(self):
|
||||
assert TYPE_MAP["aws:elbv2:loadbalancer"] == "aws_lb"
|
||||
assert TYPE_MAP["aws:elbv2:listener"] == "aws_lb_listener"
|
||||
assert TYPE_MAP["aws:elbv2:targetgroup"] == "aws_lb_target_group"
|
||||
|
||||
def test_iam_and_ecr_in_type_map(self):
|
||||
assert TYPE_MAP["aws:iam:role"] == "aws_iam_role"
|
||||
assert TYPE_MAP["aws:ecr:repository"] == "aws_ecr_repository"
|
||||
|
||||
def test_cloudfront_types_in_type_map(self):
|
||||
assert TYPE_MAP["aws:cloudfront:distribution"] == "aws_cloudfront_distribution"
|
||||
assert TYPE_MAP["aws:cloudfront:originaccesscontrol"] == "aws_cloudfront_origin_access_control"
|
||||
|
||||
def test_waf_type_in_type_map(self):
|
||||
assert TYPE_MAP["aws:wafv2:webacl"] == "aws_wafv2_web_acl"
|
||||
|
||||
def test_rds_type_in_type_map(self):
|
||||
assert TYPE_MAP["aws:rds:instance"] == "aws_db_instance"
|
||||
def test_s3_has_terraform_dir(self, registry):
|
||||
assert registry["s3"]["1.0.0"]["terraform_dir"] == "modules/l1/s3/terraform"
|
||||
|
||||
|
||||
class TestTfValue:
|
||||
def test_string_quoted(self):
|
||||
assert _tf_value("hello") == '"hello"'
|
||||
class TestModuleAssembly:
|
||||
"""Assert the adapter ASSEMBLES module instantiations, not HCL strings."""
|
||||
|
||||
def test_bool_true(self):
|
||||
assert _tf_value(True) == "true"
|
||||
def test_adapt_emits_module_block(self, tmp_path):
|
||||
instance = json.load(open(ROOT / "modules/l1/s3/instance.json"))
|
||||
adapt(instance, str(tmp_path))
|
||||
main_tf = (tmp_path / "main.tf").read_text()
|
||||
assert 'module "s3" {' in main_tf
|
||||
assert "source = " in main_tf
|
||||
|
||||
def test_bool_false(self):
|
||||
assert _tf_value(False) == "false"
|
||||
def test_adapt_passes_inputs(self, tmp_path):
|
||||
instance = json.load(open(ROOT / "modules/l1/s3/instance.json"))
|
||||
adapt(instance, str(tmp_path))
|
||||
main_tf = (tmp_path / "main.tf").read_text()
|
||||
assert 'bucket_name = "acdl-spike-bucket"' in main_tf
|
||||
|
||||
def test_int(self):
|
||||
assert _tf_value(42) == "42"
|
||||
def test_adapt_skips_region(self, tmp_path):
|
||||
instance = json.load(open(ROOT / "modules/l1/s3/instance.json"))
|
||||
adapt(instance, str(tmp_path))
|
||||
main_tf = (tmp_path / "main.tf").read_text()
|
||||
assert "region" not in main_tf.split("module")[1]
|
||||
|
||||
def test_float(self):
|
||||
assert _tf_value(3.14) == "3.14"
|
||||
def test_adapt_emits_providers_and_terraform_tf(self, tmp_path):
|
||||
instance = json.load(open(ROOT / "modules/l1/s3/instance.json"))
|
||||
adapt(instance, str(tmp_path))
|
||||
providers_tf = (tmp_path / "providers.tf").read_text()
|
||||
terraform_tf = (tmp_path / "terraform.tf").read_text()
|
||||
assert 'provider "aws"' in providers_tf
|
||||
assert 'region = "us-east-1"' in providers_tf
|
||||
assert 'required_providers' in terraform_tf
|
||||
assert 'backend "s3"' in terraform_tf
|
||||
assert 'spike/s3/dev/terraform.tfstate' in terraform_tf
|
||||
|
||||
def test_dict_jsonencoded(self):
|
||||
result = _tf_value({"key": "val"})
|
||||
assert "jsonencode" in result
|
||||
assert '"key"' in result
|
||||
def test_adapt_emits_root_outputs(self, tmp_path):
|
||||
instance = json.load(open(ROOT / "modules/l1/s3/instance.json"))
|
||||
instance["outputs"] = {
|
||||
"bucket_arn": {"from": "s3.bucket_arn"}
|
||||
}
|
||||
adapt(instance, str(tmp_path))
|
||||
main_tf = (tmp_path / "main.tf").read_text()
|
||||
assert 'output "bucket_arn"' in main_tf
|
||||
assert "module.s3.bucket_arn" in main_tf
|
||||
|
||||
def test_list_jsonencoded(self):
|
||||
result = _tf_value([1, 2])
|
||||
assert "jsonencode" in result
|
||||
def test_adapt_wires_refs(self, tmp_path):
|
||||
instance = {
|
||||
"version": "1.0.0",
|
||||
"stack": {"name": "test-ref", "kind": "l1", "depth": 1},
|
||||
"resources": [
|
||||
{
|
||||
"id": "src", "type": "aws:s3:bucket", "module": "s3@1.0.0",
|
||||
"inputs": {"bucket_name": "src-bucket", "region": "us-east-1"},
|
||||
"outputs": {"bucket_arn": {"type": "arn"}}
|
||||
},
|
||||
{
|
||||
"id": "dst", "type": "aws:s3:bucket", "module": "s3@1.0.0",
|
||||
"inputs": {"bucket_name": "dst-bucket", "region": "us-east-1",
|
||||
"kms_key_arn": "ref:src.bucket_arn"},
|
||||
"outputs": {"bucket_arn": {"type": "arn"}}
|
||||
}
|
||||
]
|
||||
}
|
||||
adapt(instance, str(tmp_path))
|
||||
main_tf = (tmp_path / "main.tf").read_text()
|
||||
assert "kms_key_arn = module.src.bucket_arn" in main_tf
|
||||
|
||||
def test_json_string_jsonencoded(self):
|
||||
result = _tf_value('{"k":"v"}')
|
||||
assert "jsonencode" in result
|
||||
def test_adapt_env_aware_state_key(self, tmp_path):
|
||||
"""P58: state key includes environment — spike/{name}/{env}/terraform.tfstate."""
|
||||
instance = {
|
||||
"version": "1.0.0",
|
||||
"stack": {"name": "msvc", "kind": "l2", "depth": 1, "environment": "prod"},
|
||||
"resources": [
|
||||
{"id": "s3", "type": "aws:s3:bucket", "module": "s3@1.0.0",
|
||||
"inputs": {"bucket_name": "test", "region": "us-east-1"}}
|
||||
],
|
||||
}
|
||||
adapt(instance, str(tmp_path))
|
||||
terraform_tf = (tmp_path / "terraform.tf").read_text()
|
||||
assert "spike/msvc/prod/terraform.tfstate" in terraform_tf
|
||||
|
||||
def test_ref_raises(self):
|
||||
with pytest.raises(ValueError, match="ref: values"):
|
||||
_tf_value("ref:s3.bucket_arn")
|
||||
def test_adapt_emits_data_source_block(self, tmp_path):
|
||||
"""P58: when data_sources is present, emit terraform_remote_state block."""
|
||||
instance = {
|
||||
"version": "1.0.0",
|
||||
"stack": {"name": "msvc", "kind": "l2", "depth": 1, "environment": "dev"},
|
||||
"resources": [
|
||||
{"id": "alb", "type": "aws:elbv2:loadbalancer", "module": "alb@1.0.0",
|
||||
"inputs": {"subnets": "ref:platform_vpc.subnet_ids", "region": "us-east-1"}}
|
||||
],
|
||||
"data_sources": ["platform_vpc"],
|
||||
}
|
||||
adapt(instance, str(tmp_path))
|
||||
main_tf = (tmp_path / "main.tf").read_text()
|
||||
assert 'data "terraform_remote_state" "platform"' in main_tf
|
||||
assert "data.terraform_remote_state.platform.outputs.subnet_ids" in main_tf
|
||||
|
||||
def test_adapt_no_vpc_for_microservice(self, tmp_path):
|
||||
"""P58: microservice contract resolves without inline VPC resources."""
|
||||
import sys
|
||||
sys.path.insert(0, str(ROOT))
|
||||
from core.contract_resolver import resolve
|
||||
stack = resolve(str(ROOT / "contracts/microservice.yml"))
|
||||
adapt(stack, str(tmp_path))
|
||||
main_tf = (tmp_path / "main.tf").read_text()
|
||||
assert 'resource "aws_vpc"' not in main_tf
|
||||
assert 'data "terraform_remote_state" "platform"' in main_tf
|
||||
|
||||
|
||||
class TestRefExpr:
|
||||
def test_basic_ref(self):
|
||||
type_by_id = {"s3": "aws:s3:bucket"}
|
||||
result = _ref_expr("ref:s3.bucket_arn", type_by_id)
|
||||
assert result == "aws_s3_bucket.s3.arn"
|
||||
def test_ref_translates_to_module_output(self):
|
||||
assert _ref_expr("ref:kms.kms_key_arn") == "module.kms.kms_key_arn"
|
||||
|
||||
def test_vpc_ref(self):
|
||||
type_by_id = {"vpc": "aws:ec2:vpc"}
|
||||
result = _ref_expr("ref:vpc.vpc_id", type_by_id)
|
||||
assert result == "aws_vpc.vpc.id"
|
||||
def test_non_ref_returns_none(self):
|
||||
assert _ref_expr("plain-string") is None
|
||||
assert _ref_expr(42) is None
|
||||
|
||||
def test_unknown_id_raises(self):
|
||||
with pytest.raises(ValueError, match="unknown stack resource id"):
|
||||
_ref_expr("ref:nonexistent.output", {"s3": "aws:s3:bucket"})
|
||||
def test_module_name_extracts_from_versioned(self):
|
||||
assert _module_name({"module": "s3@1.0.0"}) == "s3"
|
||||
assert _module_name({"module": "vpc@1.0.0"}) == "vpc"
|
||||
|
||||
|
||||
class TestAdapt:
|
||||
def test_adapt_emits_three_files(self, stack_instance, tmp_path):
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(stack_instance, out_dir)
|
||||
assert os.path.isfile(os.path.join(out_dir, "main.tf"))
|
||||
assert os.path.isfile(os.path.join(out_dir, "terraform.tf"))
|
||||
assert os.path.isfile(os.path.join(out_dir, "providers.tf"))
|
||||
class TestTfValue:
|
||||
def test_string(self):
|
||||
assert _tf_value("hello") == '"hello"'
|
||||
|
||||
def test_main_tf_has_s3_bucket(self, stack_instance, tmp_path):
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(stack_instance, out_dir)
|
||||
main_tf = open(os.path.join(out_dir, "main.tf")).read()
|
||||
assert 'resource "aws_s3_bucket" "s3"' in main_tf
|
||||
assert 'bucket = "acdl-spike-bucket"' in main_tf
|
||||
def test_bool(self):
|
||||
assert _tf_value(True) == "true"
|
||||
assert _tf_value(False) == "false"
|
||||
|
||||
def test_main_tf_has_versioning(self, stack_instance, tmp_path):
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(stack_instance, out_dir)
|
||||
main_tf = open(os.path.join(out_dir, "main.tf")).read()
|
||||
assert "versioning" in main_tf
|
||||
assert "enabled = true" in main_tf
|
||||
def test_number(self):
|
||||
assert _tf_value(42) == "42"
|
||||
|
||||
def test_main_tf_has_outputs(self, stack_instance, tmp_path):
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(stack_instance, out_dir)
|
||||
main_tf = open(os.path.join(out_dir, "main.tf")).read()
|
||||
assert 'output "bucket_arn"' in main_tf
|
||||
assert 'output "bucket_name"' in main_tf
|
||||
def test_ref(self):
|
||||
assert _tf_value("ref:kms.kms_key_arn") == "module.kms.kms_key_arn"
|
||||
|
||||
def test_terraform_tf_has_backend(self, stack_instance, tmp_path):
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(stack_instance, out_dir)
|
||||
terraform_tf = open(os.path.join(out_dir, "terraform.tf")).read()
|
||||
assert 'backend "s3"' in terraform_tf
|
||||
assert 'required_version' in terraform_tf
|
||||
assert ">= 1.9" in terraform_tf
|
||||
def test_dict(self):
|
||||
result = _tf_value({"key": "val"})
|
||||
assert result.startswith("jsonencode(")
|
||||
assert "key" in result
|
||||
|
||||
def test_providers_tf_has_aws(self, stack_instance, tmp_path):
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(stack_instance, out_dir)
|
||||
providers_tf = open(os.path.join(out_dir, "providers.tf")).read()
|
||||
assert 'provider "aws"' in providers_tf
|
||||
assert "us-east-1" in providers_tf
|
||||
|
||||
def test_backend_key_uses_stack_name(self, stack_instance, tmp_path):
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(stack_instance, out_dir)
|
||||
terraform_tf = open(os.path.join(out_dir, "terraform.tf")).read()
|
||||
assert "spike/s3/terraform.tfstate" in terraform_tf
|
||||
def test_list(self):
|
||||
result = _tf_value(["a", "b"])
|
||||
assert result.startswith("jsonencode(")
|
||||
|
||||
|
||||
class TestS3Output:
|
||||
def test_s3_instance_has_bucket_regional_domain_name_output(self, stack_instance, tmp_path):
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(stack_instance, out_dir)
|
||||
main_tf = open(os.path.join(out_dir, "main.tf")).read()
|
||||
assert 'output "bucket_regional_domain_name"' in main_tf
|
||||
assert "aws_s3_bucket.s3.bucket_regional_domain_name" in main_tf
|
||||
class TestAdapterStatelessness:
|
||||
"""Assert the adapter has no type-specific logic or constant tables."""
|
||||
|
||||
def test_no_type_map(self):
|
||||
adapter_src = (ROOT / "adapters/terraform/adapter.py").read_text()
|
||||
assert "TYPE_MAP" not in adapter_src
|
||||
|
||||
def test_no_input_map(self):
|
||||
adapter_src = (ROOT / "adapters/terraform/adapter.py").read_text()
|
||||
assert "INPUT_MAP" not in adapter_src
|
||||
|
||||
def test_no_output_map(self):
|
||||
adapter_src = (ROOT / "adapters/terraform/adapter.py").read_text()
|
||||
assert "OUTPUT_MAP" not in adapter_src
|
||||
|
||||
def test_no_rtype_branches(self):
|
||||
adapter_src = (ROOT / "adapters/terraform/adapter.py").read_text()
|
||||
assert 'rtype ==' not in adapter_src
|
||||
|
||||
def test_adapter_under_200_lines(self):
|
||||
adapter_path = ROOT / "adapters/terraform/adapter.py"
|
||||
line_count = len(adapter_path.read_text().splitlines())
|
||||
assert line_count < 200, f"adapter is {line_count} lines, expected < 200"
|
||||
|
||||
|
||||
class TestRdsPrimitive:
|
||||
@pytest.fixture
|
||||
def rds_stack(self):
|
||||
return json.load(open(ROOT / "modules/l1/rds/instance.json"))
|
||||
class TestAdapterEmitsValidTerraform:
|
||||
"""The adapter-emitted root main.tf must pass terraform validate."""
|
||||
|
||||
def test_rds_instance_validates_against_stack_schema(self, rds_stack, stack_schema):
|
||||
jsonschema.validate(rds_stack, stack_schema)
|
||||
|
||||
def test_rds_adapt_emits_db_instance(self, rds_stack, tmp_path):
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(rds_stack, out_dir)
|
||||
main_tf = open(os.path.join(out_dir, "main.tf")).read()
|
||||
assert 'resource "aws_db_instance" "rds"' in main_tf
|
||||
|
||||
def test_rds_adapt_emits_engine_and_class(self, rds_stack, tmp_path):
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(rds_stack, out_dir)
|
||||
main_tf = open(os.path.join(out_dir, "main.tf")).read()
|
||||
assert 'engine = "postgres"' in main_tf
|
||||
assert 'engine_version = "16.4"' in main_tf
|
||||
assert 'instance_class = "db.t3.micro"' in main_tf
|
||||
|
||||
def test_rds_adapt_emits_nfrs(self, rds_stack, tmp_path):
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(rds_stack, out_dir)
|
||||
main_tf = open(os.path.join(out_dir, "main.tf")).read()
|
||||
assert "backup_retention_period = 7" in main_tf
|
||||
assert "deletion_protection = true" in main_tf
|
||||
assert "skip_final_snapshot = true" in main_tf
|
||||
|
||||
def test_rds_adapt_emits_outputs(self, rds_stack, tmp_path):
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(rds_stack, out_dir)
|
||||
main_tf = open(os.path.join(out_dir, "main.tf")).read()
|
||||
assert 'output "db_endpoint"' in main_tf
|
||||
assert 'output "db_arn"' in main_tf
|
||||
assert "aws_db_instance.rds.endpoint" in main_tf
|
||||
assert "aws_db_instance.rds.arn" in main_tf
|
||||
|
||||
|
||||
class TestStaticAssetsStack:
|
||||
@pytest.fixture
|
||||
def static_assets_stack(self):
|
||||
from core.contract_resolver import resolve
|
||||
return resolve(str(ROOT / "contracts/static-assets.yml"), str(ROOT))
|
||||
|
||||
def test_static_assets_resolves_to_4_resources(self, static_assets_stack):
|
||||
types = [r["type"] for r in static_assets_stack["resources"]]
|
||||
assert "aws:s3:bucket" in types
|
||||
assert "aws:cloudfront:distribution" in types
|
||||
assert "aws:cloudfront:originaccesscontrol" in types
|
||||
assert "aws:wafv2:webacl" in types
|
||||
|
||||
def test_static_assets_adapter_emits_all_resources(self, static_assets_stack, tmp_path):
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(static_assets_stack, out_dir)
|
||||
main_tf = open(os.path.join(out_dir, "main.tf")).read()
|
||||
assert 'resource "aws_s3_bucket" "s3"' in main_tf
|
||||
assert 'resource "aws_cloudfront_distribution" "cloudfront-distribution"' in main_tf
|
||||
assert 'resource "aws_cloudfront_origin_access_control" "cloudfront-originaccesscontrol"' in main_tf
|
||||
assert 'resource "aws_wafv2_web_acl" "waf"' in main_tf
|
||||
|
||||
def test_static_assets_adapter_wires_s3_origin_to_cloudfront(self, static_assets_stack, tmp_path):
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(static_assets_stack, out_dir)
|
||||
main_tf = open(os.path.join(out_dir, "main.tf")).read()
|
||||
assert "aws_s3_bucket.s3.bucket_regional_domain_name" in main_tf
|
||||
assert "aws_cloudfront_origin_access_control.cloudfront-originaccesscontrol.id" in main_tf
|
||||
|
||||
def test_static_assets_adapter_wires_waf_to_cloudfront(self, static_assets_stack, tmp_path):
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(static_assets_stack, out_dir)
|
||||
main_tf = open(os.path.join(out_dir, "main.tf")).read()
|
||||
assert "aws_wafv2_web_acl.waf.arn" in main_tf
|
||||
assert "web_acl_id = aws_wafv2_web_acl.waf.arn" in main_tf
|
||||
|
||||
def test_static_assets_adapter_emits_distribution_outputs(self, static_assets_stack, tmp_path):
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(static_assets_stack, out_dir)
|
||||
main_tf = open(os.path.join(out_dir, "main.tf")).read()
|
||||
assert 'output "distribution_domain_name"' in main_tf
|
||||
assert 'output "web_acl_arn"' in main_tf
|
||||
|
||||
|
||||
class TestWAFAdapterFixes:
|
||||
"""P1-4: WAF custom rules emit nested blocks, not attribute syntax.
|
||||
P1-5: WAF default_action input is honored instead of hardcoded allow."""
|
||||
|
||||
@pytest.fixture
|
||||
def waf_stack_with_custom_rules(self):
|
||||
return {
|
||||
"version": "1.0.0",
|
||||
"stack": {"name": "waf-test", "kind": "l1", "depth": 1},
|
||||
"resources": [
|
||||
{
|
||||
"id": "waf",
|
||||
"type": "aws:wafv2:webacl",
|
||||
"module": "waf@1.0.0",
|
||||
"inputs": {
|
||||
"name": "custom-waf",
|
||||
"region": "us-east-1",
|
||||
"default_action": "block",
|
||||
"rules": [
|
||||
{
|
||||
"name": "rate-limit",
|
||||
"priority": 1,
|
||||
"override_action": "count",
|
||||
"statement": {"rate_based_statement": {"limit": 100}},
|
||||
},
|
||||
{
|
||||
"name": "geo-block",
|
||||
"priority": 2,
|
||||
"override_action": "none",
|
||||
},
|
||||
],
|
||||
},
|
||||
"outputs": {},
|
||||
}
|
||||
],
|
||||
}
|
||||
|
||||
@pytest.fixture
|
||||
def waf_stack_default(self):
|
||||
return {
|
||||
"version": "1.0.0",
|
||||
"stack": {"name": "waf-test", "kind": "l1", "depth": 1},
|
||||
"resources": [
|
||||
{
|
||||
"id": "waf",
|
||||
"type": "aws:wafv2:webacl",
|
||||
"module": "waf@1.0.0",
|
||||
"inputs": {"name": "default-waf", "region": "us-east-1"},
|
||||
"outputs": {},
|
||||
}
|
||||
],
|
||||
}
|
||||
|
||||
def test_waf_custom_rules_emit_nested_blocks(self, waf_stack_with_custom_rules, tmp_path):
|
||||
"""P1-4: rules must be nested blocks, not `rules = [...]`.
|
||||
|
||||
Note: the Terraform aws_wafv2_web_acl resource uses `rule` blocks
|
||||
(singular), not `rules`. The adapter was corrected in Phase 54
|
||||
(D-093 sweep) to emit `rule {` to match the AWS provider v5 schema."""
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(waf_stack_with_custom_rules, out_dir)
|
||||
main_tf = open(os.path.join(out_dir, "main.tf")).read()
|
||||
assert "rule {" in main_tf
|
||||
assert 'name = "rate-limit"' in main_tf
|
||||
assert 'name = "geo-block"' in main_tf
|
||||
assert "rules = [" not in main_tf
|
||||
|
||||
def test_waf_default_action_block_honored(self, waf_stack_with_custom_rules, tmp_path):
|
||||
"""P1-5: default_action: block must emit `block {}` not `allow {}`."""
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(waf_stack_with_custom_rules, out_dir)
|
||||
main_tf = open(os.path.join(out_dir, "main.tf")).read()
|
||||
assert "default_action {" in main_tf
|
||||
assert "block {}" in main_tf
|
||||
assert "allow {}" not in main_tf
|
||||
|
||||
def test_waf_default_action_allow_when_absent(self, waf_stack_default, tmp_path):
|
||||
"""P1-5: when default_action is absent, default to allow {} (backward compat)."""
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(waf_stack_default, out_dir)
|
||||
main_tf = open(os.path.join(out_dir, "main.tf")).read()
|
||||
assert "default_action {" in main_tf
|
||||
assert "allow {}" in main_tf
|
||||
|
||||
def test_waf_default_emits_managed_rules_block(self, waf_stack_default, tmp_path):
|
||||
"""When no custom rules, the default AWS-managed-rules block is emitted."""
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(waf_stack_default, out_dir)
|
||||
main_tf = open(os.path.join(out_dir, "main.tf")).read()
|
||||
assert "aws-managed-rules" in main_tf
|
||||
assert "rules = [" not in main_tf
|
||||
|
||||
|
||||
class TestResolverOutputs:
|
||||
"""P1-7: L2 composition outputs[] resolved into stack.outputs."""
|
||||
|
||||
def test_static_assets_has_stack_outputs(self):
|
||||
from core.contract_resolver import resolve
|
||||
stack = resolve(str(ROOT / "contracts/static-assets.yml"), str(ROOT))
|
||||
assert "outputs" in stack
|
||||
outputs = stack["outputs"]
|
||||
assert "distribution_domain_name" in outputs
|
||||
assert "bucket_arn" in outputs
|
||||
assert "web_acl_arn" in outputs
|
||||
|
||||
def test_static_assets_output_has_from_and_output(self):
|
||||
from core.contract_resolver import resolve
|
||||
stack = resolve(str(ROOT / "contracts/static-assets.yml"), str(ROOT))
|
||||
dist_out = stack["outputs"]["distribution_domain_name"]
|
||||
assert "from" in dist_out
|
||||
assert "output" in dist_out
|
||||
assert dist_out["output"] == "distribution_domain_name"
|
||||
|
||||
def test_static_assets_adapter_emits_stack_output_blocks(self, tmp_path):
|
||||
"""P1-7: adapter emits `output` blocks from stack.outputs."""
|
||||
from core.contract_resolver import resolve
|
||||
stack = resolve(str(ROOT / "contracts/static-assets.yml"), str(ROOT))
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(stack, out_dir)
|
||||
main_tf = open(os.path.join(out_dir, "main.tf")).read()
|
||||
assert 'output "distribution_domain_name"' in main_tf
|
||||
assert 'output "bucket_arn"' in main_tf
|
||||
assert 'output "web_acl_arn"' in main_tf
|
||||
|
||||
|
||||
class TestEncryptionByDefault:
|
||||
"""REQ-83/84/85: encryption by default + per-stack CMK."""
|
||||
|
||||
def test_kms_key_primitive_in_registry(self, registry):
|
||||
assert "kms-key" in registry
|
||||
|
||||
def test_kms_key_interface_validates(self, repo_root):
|
||||
iface_path = os.path.join(str(repo_root), "modules", "l1", "kms-key", "interface.json")
|
||||
iface = json.load(open(iface_path))
|
||||
assert iface["type"] == "aws:kms:key"
|
||||
assert "enable_rotation" in iface["nfrs"]
|
||||
assert iface["nfrs"]["enable_rotation"]["default"] is True
|
||||
|
||||
def test_kms_key_adapter_emits_rotation(self, tmp_path):
|
||||
kms_stack = {
|
||||
"version": "1.0.0",
|
||||
"stack": {"name": "kms-key", "kind": "l1", "depth": 1},
|
||||
"resources": [{
|
||||
"id": "kms-key",
|
||||
"type": "aws:kms:key",
|
||||
"module": "kms-key@1.0.0",
|
||||
"inputs": {"description": "test key", "region": "us-east-1", "deletion_window_days": 30},
|
||||
"outputs": {},
|
||||
"nfrs": {"enable_rotation": True, "deletion_protection": True, "encryption_enabled": True},
|
||||
}],
|
||||
}
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(kms_stack, out_dir)
|
||||
main_tf = open(os.path.join(out_dir, "main.tf")).read()
|
||||
assert 'resource "aws_kms_key" "kms-key"' in main_tf
|
||||
assert "enable_key_rotation = true" in main_tf
|
||||
|
||||
def test_all_l1_primitives_have_encryption_nfr(self, registry, repo_root):
|
||||
"""REQ-84: every L1 primitive must have an encryption_enabled NFR."""
|
||||
for name, entry in registry.items():
|
||||
iface_path = entry["1.0.0"]["interface"]
|
||||
if not iface_path.startswith("modules/l1/"):
|
||||
continue
|
||||
iface = json.load(open(os.path.join(str(repo_root), iface_path)))
|
||||
assert "encryption_enabled" in iface.get("nfrs", {}), \
|
||||
f"L1 primitive '{name}' must have encryption_enabled NFR"
|
||||
|
||||
def test_s3_with_kms_key_arn_emits_sse_configuration(self, tmp_path):
|
||||
s3_stack = {
|
||||
"version": "1.0.0",
|
||||
"stack": {"name": "s3-test", "kind": "l1", "depth": 1},
|
||||
"resources": [{
|
||||
"id": "s3",
|
||||
"type": "aws:s3:bucket",
|
||||
"module": "s3@1.0.0",
|
||||
"inputs": {"bucket_name": "test-bucket", "region": "us-east-1", "kms_key_arn": "arn:aws:kms:us-east-1:123:key/abc"},
|
||||
"outputs": {},
|
||||
"nfrs": {"encryption_enabled": True, "versioning": True},
|
||||
}],
|
||||
}
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(s3_stack, out_dir)
|
||||
main_tf = open(os.path.join(out_dir, "main.tf")).read()
|
||||
assert "server_side_encryption_configuration" in main_tf
|
||||
assert "aws:kms" in main_tf
|
||||
assert "arn:aws:kms:us-east-1:123:key/abc" in main_tf
|
||||
|
||||
def test_s3_without_kms_key_arn_falls_back_to_managed(self, tmp_path, capsys):
|
||||
s3_stack = {
|
||||
"version": "1.0.0",
|
||||
"stack": {"name": "s3-test", "kind": "l1", "depth": 1},
|
||||
"resources": [{
|
||||
"id": "s3",
|
||||
"type": "aws:s3:bucket",
|
||||
"module": "s3@1.0.0",
|
||||
"inputs": {"bucket_name": "test-bucket", "region": "us-east-1"},
|
||||
"outputs": {},
|
||||
"nfrs": {"encryption_enabled": True, "versioning": True},
|
||||
}],
|
||||
}
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(s3_stack, out_dir)
|
||||
main_tf = open(os.path.join(out_dir, "main.tf")).read()
|
||||
assert "server_side_encryption_configuration" in main_tf
|
||||
assert "aws:kms" in main_tf
|
||||
captured = capsys.readouterr()
|
||||
assert "WARNING" in captured.err or "falling back" in captured.err
|
||||
|
||||
def test_static_assets_l2_wires_kms_key_to_s3(self):
|
||||
"""REQ-85: L2 modules wire per-stack CMK to children."""
|
||||
from core.contract_resolver import resolve
|
||||
stack = resolve(str(ROOT / "contracts/static-assets.yml"), str(ROOT))
|
||||
types = [r["type"] for r in stack["resources"]]
|
||||
assert "aws:kms:key" in types
|
||||
s3_res = next(r for r in stack["resources"] if r["type"] == "aws:s3:bucket")
|
||||
assert "kms_key_arn" in s3_res.get("inputs", {}), \
|
||||
"s3 must have kms_key_arn wired from the per-stack CMK"
|
||||
|
||||
|
||||
class TestDeletionProtectionByDefault:
|
||||
"""REQ-86: deletion_protection NFR on all primitives (default true).
|
||||
REQ-87: L2 feature flag propagation."""
|
||||
|
||||
def test_all_l1_primitives_have_deletion_protection_nfr(self, registry, repo_root):
|
||||
"""REQ-86: every L1 primitive must have a deletion_protection NFR."""
|
||||
for name, entry in registry.items():
|
||||
iface_path = entry["1.0.0"]["interface"]
|
||||
if not iface_path.startswith("modules/l1/"):
|
||||
continue
|
||||
iface = json.load(open(os.path.join(str(repo_root), iface_path)))
|
||||
assert "deletion_protection" in iface.get("nfrs", {}), \
|
||||
f"L1 primitive '{name}' must have deletion_protection NFR"
|
||||
|
||||
def test_adapter_emits_prevent_destroy_when_nfr_true(self, tmp_path):
|
||||
"""REQ-86: adapter emits lifecycle { prevent_destroy = true } when NFR is true."""
|
||||
s3_stack = {
|
||||
"version": "1.0.0",
|
||||
"stack": {"name": "s3-test", "kind": "l1", "depth": 1},
|
||||
"resources": [{
|
||||
"id": "s3",
|
||||
"type": "aws:s3:bucket",
|
||||
"module": "s3@1.0.0",
|
||||
"inputs": {"bucket_name": "test-bucket", "region": "us-east-1"},
|
||||
"outputs": {},
|
||||
"nfrs": {"deletion_protection": True, "encryption_enabled": True, "versioning": True},
|
||||
}],
|
||||
}
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(s3_stack, out_dir)
|
||||
main_tf = open(os.path.join(out_dir, "main.tf")).read()
|
||||
assert "prevent_destroy = true" in main_tf
|
||||
|
||||
def test_adapter_omits_prevent_destroy_when_nfr_false(self, tmp_path):
|
||||
"""REQ-86: adapter does not emit prevent_destroy when NFR is false."""
|
||||
s3_stack = {
|
||||
"version": "1.0.0",
|
||||
"stack": {"name": "s3-test", "kind": "l1", "depth": 1},
|
||||
"resources": [{
|
||||
"id": "s3",
|
||||
"type": "aws:s3:bucket",
|
||||
"module": "s3@1.0.0",
|
||||
"inputs": {"bucket_name": "test-bucket", "region": "us-east-1"},
|
||||
"outputs": {},
|
||||
"nfrs": {"deletion_protection": False, "encryption_enabled": True, "versioning": True},
|
||||
}],
|
||||
}
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(s3_stack, out_dir)
|
||||
main_tf = open(os.path.join(out_dir, "main.tf")).read()
|
||||
assert "prevent_destroy = true" not in main_tf
|
||||
|
||||
def test_adapter_emits_prevent_destroy_by_default(self, tmp_path):
|
||||
"""REQ-86: when deletion_protection NFR is absent, default is true."""
|
||||
s3_stack = {
|
||||
"version": "1.0.0",
|
||||
"stack": {"name": "s3-test", "kind": "l1", "depth": 1},
|
||||
"resources": [{
|
||||
"id": "s3",
|
||||
"type": "aws:s3:bucket",
|
||||
"module": "s3@1.0.0",
|
||||
"inputs": {"bucket_name": "test-bucket", "region": "us-east-1"},
|
||||
"outputs": {},
|
||||
"nfrs": {},
|
||||
}],
|
||||
}
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(s3_stack, out_dir)
|
||||
main_tf = open(os.path.join(out_dir, "main.tf")).read()
|
||||
assert "prevent_destroy = true" in main_tf
|
||||
|
||||
def test_l2_feature_flag_propagates_deletion_protection_false(self, tmp_path):
|
||||
"""REQ-87: L2 feature flag deletion_protection=false propagates to all children."""
|
||||
import yaml
|
||||
contract = {
|
||||
"id": "assets",
|
||||
"name": "static-assets-dp-test",
|
||||
"environment": "dev",
|
||||
"infrastructure": {
|
||||
"static-assets": {
|
||||
"version": "1.0.0",
|
||||
"inputs": {"bucket_name": "test-bucket", "region": "us-east-1", "deletion_protection": False},
|
||||
}
|
||||
},
|
||||
}
|
||||
contract_path = tmp_path / "test-dp.yml"
|
||||
with open(contract_path, "w") as fh:
|
||||
yaml.dump(contract, fh)
|
||||
from core.contract_resolver import resolve
|
||||
stack = resolve(str(contract_path), str(ROOT))
|
||||
for res in stack["resources"]:
|
||||
assert res.get("nfrs", {}).get("deletion_protection") is False, \
|
||||
f"Resource {res['id']} should have deletion_protection=false"
|
||||
|
||||
|
||||
class TestUptimePrimitive:
|
||||
"""REQ-88/89/90/91: uptime-kuma primitive + feature flag + pipeline stage."""
|
||||
|
||||
def test_uptime_primitive_in_registry(self, registry):
|
||||
assert "uptime" in registry
|
||||
|
||||
def test_uptime_interface_has_feature_flag(self, repo_root):
|
||||
iface = json.load(open(os.path.join(str(repo_root), "modules", "l1", "uptime", "interface.json")))
|
||||
assert "feature_flag_enabled" in iface["inputs"]
|
||||
assert iface["inputs"]["feature_flag_enabled"]["default"] is True
|
||||
|
||||
def test_uptime_interface_has_alert_channels(self, repo_root):
|
||||
iface = json.load(open(os.path.join(str(repo_root), "modules", "l1", "uptime", "interface.json")))
|
||||
assert "alert_channels" in iface["inputs"]
|
||||
assert "monitored_endpoints" in iface["inputs"]
|
||||
|
||||
def test_uptime_adapter_emits_ecs_service_when_enabled(self, tmp_path):
|
||||
uptime_stack = {
|
||||
"version": "1.0.0",
|
||||
"stack": {"name": "uptime", "kind": "l1", "depth": 1},
|
||||
"resources": [{
|
||||
"id": "uptime",
|
||||
"type": "aws:ecs:uptime-service",
|
||||
"module": "uptime@1.0.0",
|
||||
"inputs": {
|
||||
"container_image": "louislam/uptime-kuma:1",
|
||||
"region": "us-east-1",
|
||||
"feature_flag_enabled": True,
|
||||
"monitored_endpoints": [{"name": "test", "url": "https://example.com", "type": "http", "interval_seconds": 60, "timeout_seconds": 30}],
|
||||
"cpu": 256,
|
||||
"memory": 512,
|
||||
},
|
||||
"outputs": {},
|
||||
"nfrs": {"deletion_protection": True, "encryption_enabled": True},
|
||||
}],
|
||||
}
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(uptime_stack, out_dir)
|
||||
main_tf = open(os.path.join(out_dir, "main.tf")).read()
|
||||
assert 'resource "aws_ecs_service" "uptime"' in main_tf
|
||||
assert "louislam/uptime-kuma:1" in main_tf
|
||||
assert "desired_count = 1" in main_tf
|
||||
|
||||
def test_uptime_adapter_emits_nothing_when_disabled(self, tmp_path):
|
||||
"""REQ-90: feature_flag_enabled=false means no resources emitted."""
|
||||
uptime_stack = {
|
||||
"version": "1.0.0",
|
||||
"stack": {"name": "uptime", "kind": "l1", "depth": 1},
|
||||
"resources": [{
|
||||
"id": "uptime",
|
||||
"type": "aws:ecs:uptime-service",
|
||||
"module": "uptime@1.0.0",
|
||||
"inputs": {"region": "us-east-1", "feature_flag_enabled": False},
|
||||
"outputs": {},
|
||||
"nfrs": {},
|
||||
}],
|
||||
}
|
||||
out_dir = str(tmp_path / "tf_out")
|
||||
adapt(uptime_stack, out_dir)
|
||||
main_tf = open(os.path.join(out_dir, "main.tf")).read()
|
||||
assert 'resource "aws_ecs_service" "uptime"' not in main_tf
|
||||
|
||||
def test_deploy_pipeline_has_deploy_uptime_stage(self):
|
||||
import yaml
|
||||
with open(ROOT / "pipelines/contract.yml") as fh:
|
||||
contract = yaml.safe_load(fh)
|
||||
stage_names = [s["name"] for s in contract["stages"]]
|
||||
assert "deploy-uptime" in stage_names
|
||||
def test_s3_instance_emits_valid_terraform(self, tmp_path):
|
||||
instance = json.load(open(ROOT / "modules/l1/s3/instance.json"))
|
||||
adapt(instance, str(tmp_path))
|
||||
result = subprocess.run(
|
||||
["terraform", "init", "-backend=false", "-input=false"],
|
||||
cwd=str(tmp_path), capture_output=True, text=True
|
||||
)
|
||||
assert result.returncode == 0, f"terraform init failed: {result.stderr}"
|
||||
result = subprocess.run(
|
||||
["terraform", "validate"],
|
||||
cwd=str(tmp_path), capture_output=True, text=True
|
||||
)
|
||||
assert result.returncode == 0, f"terraform validate failed: {result.stderr}"
|
||||
@@ -1,248 +0,0 @@
|
||||
"""P1-1: adapter ECS/ALB/VPC defaults are parameterized via L1 interface.json
|
||||
inputs (REQ-102, D-085). The adapter is a thin translator — defaults live in
|
||||
the interface, not the adapter.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import yaml
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
from adapters.terraform.adapter import adapt
|
||||
from core.contract_resolver import resolve
|
||||
|
||||
|
||||
def _load_ir(path):
|
||||
with open(path) as f:
|
||||
return json.load(f)
|
||||
|
||||
|
||||
def _tf_for_contract(contract_dict, tmp_path):
|
||||
"""Resolve a contract dict to a stack, emit TF, return the main.tf text."""
|
||||
contract_path = tmp_path / "contract.yaml"
|
||||
contract_path.write_text(yaml.safe_dump(contract_dict))
|
||||
stack = resolve(str(contract_path))
|
||||
out_dir = tmp_path / "tf"
|
||||
adapt(stack, str(out_dir))
|
||||
return (out_dir / "main.tf").read_text()
|
||||
|
||||
|
||||
def test_desired_count_override_emits_overridden_value(tmp_path):
|
||||
"""An L1 with desired_count: 3 in contract inputs emits desired_count = 3."""
|
||||
contract = {
|
||||
"id": "msvc",
|
||||
"name": "microservice-test",
|
||||
"environment": "dev",
|
||||
"infrastructure": {
|
||||
"microservice": {
|
||||
"version": "1.0.0",
|
||||
"inputs": {
|
||||
"bucket_name": "acdl-test",
|
||||
"region": "us-east-1",
|
||||
"image": "public.ecr.aws/docker/library/nginx:latest",
|
||||
"port": 80,
|
||||
"desired_count": 3,
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
tf = _tf_for_contract(contract, tmp_path)
|
||||
assert "desired_count = 3" in tf
|
||||
assert "desired_count = 1" not in tf
|
||||
|
||||
|
||||
def test_desired_count_default_emits_one_via_interface(tmp_path):
|
||||
"""Absent desired_count emits desired_count = 1 via interface default."""
|
||||
contract = {
|
||||
"id": "msvc",
|
||||
"name": "microservice-test",
|
||||
"environment": "dev",
|
||||
"infrastructure": {
|
||||
"microservice": {
|
||||
"version": "1.0.0",
|
||||
"inputs": {
|
||||
"bucket_name": "acdl-test",
|
||||
"region": "us-east-1",
|
||||
"image": "public.ecr.aws/docker/library/nginx:latest",
|
||||
"port": 80,
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
tf = _tf_for_contract(contract, tmp_path)
|
||||
assert "desired_count = 1" in tf
|
||||
|
||||
|
||||
def test_launch_type_override_emits_overridden_value(tmp_path):
|
||||
contract = {
|
||||
"id": "msvc",
|
||||
"name": "microservice-test",
|
||||
"environment": "dev",
|
||||
"infrastructure": {
|
||||
"microservice": {
|
||||
"version": "1.0.0",
|
||||
"inputs": {
|
||||
"bucket_name": "acdl-test",
|
||||
"region": "us-east-1",
|
||||
"image": "public.ecr.aws/docker/library/nginx:latest",
|
||||
"port": 80,
|
||||
"launch_type": "EC2",
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
tf = _tf_for_contract(contract, tmp_path)
|
||||
assert 'launch_type = "EC2"' in tf
|
||||
assert 'launch_type = "FARGATE"' not in tf
|
||||
|
||||
|
||||
def test_target_type_override_emits_overridden_value(tmp_path):
|
||||
contract = {
|
||||
"id": "msvc",
|
||||
"name": "microservice-test",
|
||||
"environment": "dev",
|
||||
"infrastructure": {
|
||||
"microservice": {
|
||||
"version": "1.0.0",
|
||||
"inputs": {
|
||||
"bucket_name": "acdl-test",
|
||||
"region": "us-east-1",
|
||||
"image": "public.ecr.aws/docker/library/nginx:latest",
|
||||
"port": 80,
|
||||
"target_type": "instance",
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
tf = _tf_for_contract(contract, tmp_path)
|
||||
assert 'target_type = "instance"' in tf
|
||||
assert 'target_type = "ip"' not in tf
|
||||
|
||||
|
||||
def test_load_balancer_type_override_emits_overridden_value(tmp_path):
|
||||
contract = {
|
||||
"id": "msvc",
|
||||
"name": "microservice-test",
|
||||
"environment": "dev",
|
||||
"infrastructure": {
|
||||
"microservice": {
|
||||
"version": "1.0.0",
|
||||
"inputs": {
|
||||
"bucket_name": "acdl-test",
|
||||
"region": "us-east-1",
|
||||
"image": "public.ecr.aws/docker/library/nginx:latest",
|
||||
"port": 80,
|
||||
"load_balancer_type": "network",
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
tf = _tf_for_contract(contract, tmp_path)
|
||||
assert 'load_balancer_type = "network"' in tf
|
||||
assert 'load_balancer_type = "application"' not in tf
|
||||
|
||||
|
||||
def test_family_override_emits_overridden_value(tmp_path):
|
||||
contract = {
|
||||
"id": "msvc",
|
||||
"name": "microservice-test",
|
||||
"environment": "dev",
|
||||
"infrastructure": {
|
||||
"microservice": {
|
||||
"version": "1.0.0",
|
||||
"inputs": {
|
||||
"bucket_name": "acdl-test",
|
||||
"region": "us-east-1",
|
||||
"image": "public.ecr.aws/docker/library/nginx:latest",
|
||||
"port": 80,
|
||||
"family": "myservice",
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
tf = _tf_for_contract(contract, tmp_path)
|
||||
assert 'family = "myservice"' in tf
|
||||
|
||||
|
||||
def test_family_default_emits_app(tmp_path):
|
||||
contract = {
|
||||
"id": "msvc",
|
||||
"name": "microservice-test",
|
||||
"environment": "dev",
|
||||
"infrastructure": {
|
||||
"microservice": {
|
||||
"version": "1.0.0",
|
||||
"inputs": {
|
||||
"bucket_name": "acdl-test",
|
||||
"region": "us-east-1",
|
||||
"image": "public.ecr.aws/docker/library/nginx:latest",
|
||||
"port": 80,
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
tf = _tf_for_contract(contract, tmp_path)
|
||||
assert 'family = "app"' in tf
|
||||
|
||||
|
||||
def test_v1_1_s3_regression_still_passes(tmp_path):
|
||||
"""The v1.1 S3 regression: the static-assets L1 (s3-only) must still
|
||||
produce valid Terraform with no ECS/ALB/VPC defaults leaking in."""
|
||||
contract_path = ROOT / "contracts" / "static-assets.yml"
|
||||
stack = resolve(str(contract_path))
|
||||
out_dir = tmp_path / "tf"
|
||||
adapt(stack, str(out_dir))
|
||||
tf = (out_dir / "main.tf").read_text()
|
||||
assert "aws_s3_bucket" in tf
|
||||
assert "desired_count" not in tf
|
||||
assert "launch_type" not in tf
|
||||
assert "target_type" not in tf
|
||||
|
||||
|
||||
def test_no_hardcoded_microservice_name_in_route_table(tmp_path):
|
||||
"""The hardcoded 'acdl-microservice-rt' / 'acdl-microservice-igw' Name
|
||||
tags are removed (D-085); the name derives from the VPC name input."""
|
||||
contract = {
|
||||
"id": "msvc",
|
||||
"name": "microservice-test",
|
||||
"environment": "dev",
|
||||
"infrastructure": {
|
||||
"microservice": {
|
||||
"version": "1.0.0",
|
||||
"inputs": {
|
||||
"bucket_name": "acdl-test",
|
||||
"region": "us-east-1",
|
||||
"image": "public.ecr.aws/docker/library/nginx:latest",
|
||||
"port": 80,
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
tf = _tf_for_contract(contract, tmp_path)
|
||||
assert "acdl-microservice-rt" not in tf
|
||||
assert "acdl-microservice-igw" not in tf
|
||||
|
||||
|
||||
def test_ecs_service_interface_has_parameterized_inputs():
|
||||
"""The L1 interface declares the inputs (the adapter reads them)."""
|
||||
iface = _load_ir(ROOT / "modules" / "l1" / "ecs-service" / "interface.json")
|
||||
inputs = iface["inputs"]
|
||||
assert "desired_count" in inputs
|
||||
assert inputs["desired_count"]["default"] == 1
|
||||
assert "launch_type" in inputs
|
||||
assert inputs["launch_type"]["default"] == "FARGATE"
|
||||
assert "family" in inputs
|
||||
assert inputs["family"]["default"] == "app"
|
||||
|
||||
|
||||
def test_alb_interface_has_parameterized_inputs():
|
||||
iface = _load_ir(ROOT / "modules" / "l1" / "alb" / "interface.json")
|
||||
inputs = iface["inputs"]
|
||||
assert "load_balancer_type" in inputs
|
||||
assert inputs["load_balancer_type"]["default"] == "application"
|
||||
assert "target_type" in inputs
|
||||
assert inputs["target_type"]["default"] == "ip"
|
||||
+30
-2
@@ -24,7 +24,7 @@ class TestPipelineIntegration:
|
||||
assert os.path.isfile(os.path.join(out_dir, "providers.tf"))
|
||||
|
||||
main_tf = open(os.path.join(out_dir, "main.tf")).read()
|
||||
assert "aws_s3_bucket" in main_tf
|
||||
assert 'module "s3"' in main_tf
|
||||
assert "acdl-spike-bucket" in main_tf
|
||||
|
||||
def test_confidence_signal_with_adapted_tf(self):
|
||||
@@ -64,4 +64,32 @@ class TestPipelineIntegration:
|
||||
timeout=30,
|
||||
)
|
||||
assert result.returncode == 0
|
||||
assert "PLATFORM CHECK OK" in result.stdout
|
||||
assert "PLATFORM CHECK OK" in result.stdout
|
||||
|
||||
def test_run_platform_apply_mode_parses(self):
|
||||
"""--apply mode parses without 'unknown flag' error (requires a contract)."""
|
||||
result = subprocess.run(
|
||||
["bash", str(ROOT / "scripts/run_platform.sh"), "--apply"],
|
||||
capture_output=True, text=True, cwd=str(ROOT),
|
||||
timeout=10,
|
||||
)
|
||||
assert "unknown flag" not in result.stderr
|
||||
assert "contract file required" in result.stderr or result.returncode != 0
|
||||
|
||||
def test_run_platform_destroy_mode_parses(self):
|
||||
"""--destroy mode parses without 'unknown flag' error (requires a contract)."""
|
||||
result = subprocess.run(
|
||||
["bash", str(ROOT / "scripts/run_platform.sh"), "--destroy"],
|
||||
capture_output=True, text=True, cwd=str(ROOT),
|
||||
timeout=10,
|
||||
)
|
||||
assert "unknown flag" not in result.stderr
|
||||
assert "contract file required" in result.stderr or result.returncode != 0
|
||||
|
||||
def test_no_python_runs_terraform_apply_or_destroy(self):
|
||||
"""D-101: Python scripts never run terraform apply or terraform destroy."""
|
||||
scripts_dir = ROOT / "scripts"
|
||||
for py_file in scripts_dir.glob("*.py"):
|
||||
content = py_file.read_text()
|
||||
assert "terraform apply" not in content, f"{py_file.name} contains 'terraform apply'"
|
||||
assert "terraform destroy" not in content, f"{py_file.name} contains 'terraform destroy'"
|
||||
@@ -227,7 +227,7 @@ class TestRunPlatformStreaming:
|
||||
assert "PLATFORM CHECK OK" in result.stdout
|
||||
assert "--- emitted" in result.stdout
|
||||
assert "main.tf" in result.stdout
|
||||
assert "aws_s3_bucket" in result.stdout
|
||||
assert "module" in result.stdout
|
||||
|
||||
def test_check_only_quiet_suppresses_terraform(self):
|
||||
result = subprocess.run(
|
||||
@@ -518,4 +518,99 @@ class TestPlatformWorkflows:
|
||||
checkout = next(
|
||||
s for s in release_job["steps"] if "checkout" in s.get("uses", "")
|
||||
)
|
||||
assert checkout["with"]["fetch-depth"] == 0
|
||||
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_four_jobs(self):
|
||||
wf = _load_workflow(".gitea/workflows/modules-lifecycle.yml")
|
||||
assert set(wf["jobs"].keys()) == {"ci-vpc-apply", "lifecycle", "l2-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)"
|
||||
|
||||
def test_l2_lifecycle_job_exists(self):
|
||||
wf = _load_workflow(".gitea/workflows/modules-lifecycle.yml")
|
||||
assert "l2-lifecycle" in wf["jobs"]
|
||||
|
||||
def test_l2_matrix_lists_both_l2_modules(self):
|
||||
wf = _load_workflow(".gitea/workflows/modules-lifecycle.yml")
|
||||
matrix_modules = wf["jobs"]["l2-lifecycle"]["strategy"]["matrix"]["module"]
|
||||
assert set(matrix_modules) == {"static-assets", "microservice"}
|
||||
|
||||
def test_l2_lifecycle_job_has_apply_modify_destroy_steps(self):
|
||||
wf = _load_workflow(".gitea/workflows/modules-lifecycle.yml")
|
||||
steps = wf["jobs"]["l2-lifecycle"]["steps"]
|
||||
step_names = [s.get("name", "") for s in steps]
|
||||
assert any("Apply" in n for n in step_names), "Missing L2 apply step"
|
||||
assert any("Modify" in n for n in step_names), "Missing L2 modify step"
|
||||
assert any("Destroy" in n for n in step_names), "Missing L2 destroy step"
|
||||
|
||||
def test_l2_lifecycle_job_needs_ci_vpc_apply(self):
|
||||
wf = _load_workflow(".gitea/workflows/modules-lifecycle.yml")
|
||||
assert wf["jobs"]["l2-lifecycle"]["needs"] == "ci-vpc-apply"
|
||||
|
||||
def test_ci_vpc_destroy_needs_both_lifecycle_and_l2(self):
|
||||
wf = _load_workflow(".gitea/workflows/modules-lifecycle.yml")
|
||||
assert set(wf["jobs"]["ci-vpc-destroy"]["needs"]) == {"lifecycle", "l2-lifecycle"}
|
||||
|
||||
def test_contract_matrix_lists_l2_modules(self):
|
||||
contract = _load_yaml("pipelines/modules-lifecycle.yml")
|
||||
assert set(contract["matrix"]["l2_modules"]) == {"static-assets", "microservice"}
|
||||
Reference in New Issue
Block a user