Jon Chery
fda4564a7f
EXECUTE stage. Fixes the 4-VPC bug: adds a single shared VPC to
terraform/platform, drops the vpc child from the microservice composition
(references the platform VPC via data source), and makes state keys
env-aware (spike/{id}/{env}/terraform.tfstate — stable across lifecycle).
Platform VPC (terraform/platform/main.tf):
- aws_vpc.acdl_shared (10.0.0.0/16) + 2 subnets + IGW + route table + SG
- Outputs: vpc_id, subnet_ids, ecs_security_group_id
Microservice composition (modules/l2/microservice/composition.json):
- Dropped the vpc child (no per-contract VPC ever again).
- Added data_sources block: platform_vpc → terraform_remote_state (platform).
- Wires: vpc.outputs.subnet_ids → platform_vpc.outputs.subnet_ids.
- Wires: platform_vpc.outputs.vpc_id → alb.inputs.vpc_id.
- Wires: platform_vpc.outputs.ecs_security_group_id → service.inputs.security_group.
Contract resolver (core/contract_resolver.py):
- Added environment to the stack instance (stack.environment).
- Added data_sources handling: pseudo-children with outputs but no resources.
- data_sources propagated through fragment merge to the final stack instance.
Adapter (adapters/terraform/adapter.py):
- State key: spike/{stack_name}/{environment}/terraform.tfstate (env-aware).
- Emits data "terraform_remote_state" "platform" block when data_sources present.
- ref:platform_vpc.<output> → data.terraform_remote_state.platform.outputs.<output>.
Tests (tests/test_adapter.py):
- test_adapt_env_aware_state_key: spike/msvc/prod/terraform.tfstate.
- test_adapt_emits_data_source_block: data.terraform_remote_state.platform.
- test_adapt_no_vpc_for_microservice: no resource "aws_vpc" in microservice output.
- Updated existing state key assertion (spike/s3/dev/terraform.tfstate).
Regression: 467 passed, 0 skipped, 5 deselected. run_platform.sh --check-only
passes for both microservice (9 resources, no VPC) and static-assets (5 resources).
---ci---
project: acdl
phase: P58
milestone: v1.11
status: execute
---/ci---
244 lines
9.7 KiB
Python
244 lines
9.7 KiB
Python
import json
|
|
import os
|
|
import subprocess
|
|
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 adapt, _tf_value, _ref_expr, _module_name
|
|
|
|
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", "kms-key", "uptime"}
|
|
EXPECTED_L2_KEYS = {"static-assets", "microservice"}
|
|
|
|
def test_registry_has_14_entries(self, registry):
|
|
assert len(registry) == 14
|
|
assert set(registry.keys()) == (self.EXPECTED_L1_KEYS | self.EXPECTED_L2_KEYS)
|
|
|
|
def test_registry_has_12_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
|
|
|
|
def test_s3_has_terraform_dir(self, registry):
|
|
assert registry["s3"]["1.0.0"]["terraform_dir"] == "modules/l1/s3/terraform"
|
|
|
|
|
|
class TestModuleAssembly:
|
|
"""Assert the adapter ASSEMBLES module instantiations, not HCL strings."""
|
|
|
|
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_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_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_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_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_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_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_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_ref_translates_to_module_output(self):
|
|
assert _ref_expr("ref:kms.kms_key_arn") == "module.kms.kms_key_arn"
|
|
|
|
def test_non_ref_returns_none(self):
|
|
assert _ref_expr("plain-string") is None
|
|
assert _ref_expr(42) is None
|
|
|
|
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 TestTfValue:
|
|
def test_string(self):
|
|
assert _tf_value("hello") == '"hello"'
|
|
|
|
def test_bool(self):
|
|
assert _tf_value(True) == "true"
|
|
assert _tf_value(False) == "false"
|
|
|
|
def test_number(self):
|
|
assert _tf_value(42) == "42"
|
|
|
|
def test_ref(self):
|
|
assert _tf_value("ref:kms.kms_key_arn") == "module.kms.kms_key_arn"
|
|
|
|
def test_dict(self):
|
|
result = _tf_value({"key": "val"})
|
|
assert result.startswith("jsonencode(")
|
|
assert "key" in result
|
|
|
|
def test_list(self):
|
|
result = _tf_value(["a", "b"])
|
|
assert result.startswith("jsonencode(")
|
|
|
|
|
|
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 TestAdapterEmitsValidTerraform:
|
|
"""The adapter-emitted root main.tf must pass terraform validate."""
|
|
|
|
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}" |