Jon Chery
0eb578c606
---ci---
project: acdl
phase: 28
milestone: v1.8
status: execute
---/ci---
P1-4: WAF custom rules now emit nested `rules { ... }` blocks per rule
instead of `rules = [...]` attribute syntax (invalid HCL).
P1-5: WAF default_action input is honored (allow/block) instead of
hardcoding `allow {}`. Default is `allow` when absent (backward compat).
P1-7: L2 composition outputs[] array is now processed by resolve_l2().
The resolver builds stack.outputs from the composition outputs wires.
The adapter emits `output` blocks from stack.outputs.
Tests: +10 (275 -> 285). All pass. run_platform.sh --check-only green.
404 lines
16 KiB
Python
404 lines
16 KiB
Python
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import jsonschema
|
|
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,
|
|
)
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
|
|
|
|
class TestInstance:
|
|
def test_instance_validates_against_stack_schema(self, stack_instance, stack_schema):
|
|
jsonschema.validate(stack_instance, stack_schema)
|
|
|
|
def test_instance_has_one_resource(self, stack_instance):
|
|
assert len(stack_instance["resources"]) == 1
|
|
r = stack_instance["resources"][0]
|
|
assert r["id"] == "s3"
|
|
assert r["type"] == "aws:s3:bucket"
|
|
|
|
def test_instance_stack_is_s3(self, stack_instance):
|
|
assert stack_instance["stack"]["name"] == "s3"
|
|
assert stack_instance["stack"]["kind"] == "l1"
|
|
|
|
|
|
class TestRegistry:
|
|
EXPECTED_L1_KEYS = {"s3", "vpc", "ecs-cluster", "ecs-service", "iam-role", "alb", "ecr", "cloudfront", "waf", "rds"}
|
|
EXPECTED_L2_KEYS = {"static-assets", "microservice"}
|
|
|
|
def test_registry_has_12_entries(self, registry):
|
|
assert len(registry) == 12
|
|
assert set(registry.keys()) == (self.EXPECTED_L1_KEYS | self.EXPECTED_L2_KEYS)
|
|
|
|
def test_registry_has_10_l1_entries(self, registry):
|
|
l1 = {k for k in registry if registry[k]["1.0.0"]["interface"].startswith("modules/l1/")}
|
|
assert l1 == self.EXPECTED_L1_KEYS
|
|
|
|
def test_registry_has_2_l2_entries(self, registry):
|
|
l2 = {k for k in registry if registry[k]["1.0.0"]["interface"].startswith("modules/l2/")}
|
|
assert l2 == self.EXPECTED_L2_KEYS
|
|
|
|
def test_all_l1_interfaces_exist(self, registry, repo_root):
|
|
for name in self.EXPECTED_L1_KEYS:
|
|
entry = registry[name]["1.0.0"]
|
|
iface_path = os.path.join(repo_root, entry["interface"])
|
|
assert os.path.isfile(iface_path), f"{iface_path} missing"
|
|
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"
|
|
|
|
|
|
class TestTfValue:
|
|
def test_string_quoted(self):
|
|
assert _tf_value("hello") == '"hello"'
|
|
|
|
def test_bool_true(self):
|
|
assert _tf_value(True) == "true"
|
|
|
|
def test_bool_false(self):
|
|
assert _tf_value(False) == "false"
|
|
|
|
def test_int(self):
|
|
assert _tf_value(42) == "42"
|
|
|
|
def test_float(self):
|
|
assert _tf_value(3.14) == "3.14"
|
|
|
|
def test_dict_jsonencoded(self):
|
|
result = _tf_value({"key": "val"})
|
|
assert "jsonencode" in result
|
|
assert '"key"' in result
|
|
|
|
def test_list_jsonencoded(self):
|
|
result = _tf_value([1, 2])
|
|
assert "jsonencode" in result
|
|
|
|
def test_json_string_jsonencoded(self):
|
|
result = _tf_value('{"k":"v"}')
|
|
assert "jsonencode" in result
|
|
|
|
def test_ref_raises(self):
|
|
with pytest.raises(ValueError, match="ref: values"):
|
|
_tf_value("ref:s3.bucket_arn")
|
|
|
|
|
|
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_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_unknown_id_raises(self):
|
|
with pytest.raises(ValueError, match="unknown stack resource id"):
|
|
_ref_expr("ref:nonexistent.output", {"s3": "aws:s3:bucket"})
|
|
|
|
|
|
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"))
|
|
|
|
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_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_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_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_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
|
|
|
|
|
|
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 TestRdsPrimitive:
|
|
@pytest.fixture
|
|
def rds_stack(self):
|
|
return json.load(open(ROOT / "modules/l1/rds/instance.json"))
|
|
|
|
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.yaml"), 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 = [...]`."""
|
|
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 "rules {" 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.yaml"), 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.yaml"), 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.yaml"), 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 |