acdl/tests/test_adapter.py

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/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


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}"