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, _child_id

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", "dynamodb"}
    EXPECTED_L2_KEYS = {"static-assets", "microservice"}

    def test_registry_has_15_entries(self, registry):
        assert len(registry) == 15
        assert set(registry.keys()) == (self.EXPECTED_L1_KEYS | self.EXPECTED_L2_KEYS)

    def test_registry_has_13_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_nova_state_bucket(self, tmp_path):
        """P1 (REQ-165): the emitted backend references nova-tfstate-*
        (not acdl-tfstate-*); the live bucket was renamed in v1.15 P4."""
        instance = json.load(open(ROOT / "modules/l1/s3/instance.json"))
        adapt(instance, str(tmp_path))
        terraform_tf = (tmp_path / "terraform.tf").read_text()
        assert "nova-tfstate-" in terraform_tf
        assert "acdl-tfstate-" not 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_250_lines(self):
        # P03 W3 (REQ-319): the adapter now loads the env onboarding JSON to
        # source env.state_backend.bucket + env.account_id + env.region for
        # the S3 backend block (two small helpers). The bound is 250 (was
        # 200) — still a tight statelessness guardrail against type-specific
        # logic / constant tables creeping back in.
        adapter_path = ROOT / "adapters/terraform/adapter.py"
        line_count = len(adapter_path.read_text().splitlines())
        assert line_count < 250, f"adapter is {line_count} lines, expected < 250"


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


class TestAdapterDedupMultiResourceL1:
    """CAP-013 (v1.12, REQ-129): multi-resource L1s (ecs-service, alb) with
    stack outputs + cross-module refs must dedup to ONE module block named by
    the composition child id, with all expanded sub-ids rewritten. This is the
    regression test that locks in the fix — a future refactor that re-breaks
    the dedup must turn this red."""

    def test_microservice_dedup_names_modules_by_child_id(self, tmp_path):
        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()
        # The merged modules are named by the child id (alb, service), not the
        # expanded sub-ids (alb-loadbalancer, service-service, ...).
        assert 'module "alb" {' in main_tf
        assert 'module "service" {' in main_tf
        # Expanded sub-ids must NOT appear as module names.
        assert 'module "alb-loadbalancer"' not in main_tf
        assert 'module "alb-targetgroup"' not in main_tf
        assert 'module "alb-listener"' not in main_tf
        assert 'module "service-task-definition"' not in main_tf
        assert 'module "service-service"' not in main_tf

    def test_microservice_dedup_rewrites_stack_outputs(self, tmp_path):
        """Stack outputs reference the expanded sub-ids; after dedup they
        must resolve to the child-id module name."""
        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()
        # service_arn output references service-service -> module.service
        assert 'output "service_arn" {\n  value = module.service.service_arn' in main_tf
        # lb_arn output references alb-loadbalancer -> module.alb
        assert 'output "lb_arn" {\n  value = module.alb.lb_arn' in main_tf

    def test_microservice_dedup_rewrites_cross_module_refs(self, tmp_path):
        """A ref: input that targets an expanded sub-id (e.g.
        ref:alb-targetgroup.target_group_arn) must be rewritten to the
        child-id module (module.alb.target_group_arn)."""
        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 "lb_target_group_arn = module.alb.target_group_arn" in main_tf
        # The un-rewritten expanded form must NOT appear.
        assert "module.alb-targetgroup" not in main_tf
        assert "module.service-service" not in main_tf

    def test_microservice_emits_valid_terraform(self, tmp_path):
        """CAP-013 end-to-end: the microservice (multi-resource L1s) main.tf
        passes terraform init + validate. This is the real regression — the
        v1.11 dedup produced 'No module call name' here."""
        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))
        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 (CAP-013 regression): {result.stderr}"


class TestChildIdHelper:
    """Unit tests for _child_id (the common-prefix heuristic)."""

    def test_single_resource_returns_id_verbatim(self):
        assert _child_id(["cluster"]) == "cluster"
        assert _child_id(["kms-key"]) == "kms-key"

    def test_multi_resource_returns_common_prefix(self):
        # ecs-service expands to service-task-definition + service-service
        assert _child_id(["service-task-definition", "service-service"]) == "service"
        # alb expands to alb-loadbalancer + alb-targetgroup + alb-listener
        assert _child_id(["alb-loadbalancer", "alb-targetgroup", "alb-listener"]) == "alb"


class TestAdapterDedupRejectsUnregisteredModule:
    """P1-1 (v1.14, REQ-135): a resource whose module is not in the
    registry must raise ValueError, not be silently dropped from the
    dedup merge. A typo'd module field (e.g. 'iam-role' vs 'iam_roles')
    must surface as a diagnostic, not vanish."""

    def test_unregistered_module_raises_valueerror(self, tmp_path):
        stack = {
            "resources": [
                {"id": "bad", "type": "aws:bogus:thing", "module": "nonexistent@1.0.0", "inputs": {}}
            ],
            "outputs": {},
            "data_sources": [],
        }
        with pytest.raises(ValueError, match="no terraform_dir for module 'nonexistent'"):
            adapt(stack, str(tmp_path))

    def test_registered_module_still_works(self, tmp_path):
        """A registered module (s3) must still emit valid terraform — the
        ValueError guard must not break the happy path."""
        stack = {
            "resources": [
                {"id": "s3", "type": "aws:s3:bucket", "module": "s3@1.0.0", "inputs": {"bucket_name": "test"}}
            ],
            "outputs": {},
            "data_sources": [],
        }
        adapt(stack, str(tmp_path))
        assert (tmp_path / "main.tf").exists()


class TestAdapterDedupMergesSameModule:
    """P2-2 (v1.14, REQ-139): two resources with the same module collapse
    to one module block named by the child id, with merged inputs. This
    locks in the dedup-merge behavior at the unit level."""

    def test_two_resources_same_module_collapse_to_one_block(self, tmp_path):
        """Two resources sharing the same terraform dir (e.g. cloudfront
        distribution + OAC) must produce ONE module block, not two."""
        stack = {
            "resources": [
                {"id": "cloudfront-distribution", "type": "aws:cloudfront:distribution", "module": "cloudfront@1.0.0", "inputs": {"price_class": "PriceClass_100"}},
                {"id": "cloudfront-originaccesscontrol", "type": "aws:cloudfront:originaccesscontrol", "module": "cloudfront@1.0.0", "inputs": {"viewer_protocol_policy": "redirect-to-https"}},
            ],
            "outputs": {},
            "data_sources": [],
        }
        adapt(stack, str(tmp_path))
        main_tf = (tmp_path / "main.tf").read_text()
        # Exactly one module block for cloudfront (deduped to child id "cloudfront")
        assert main_tf.count('module "cloudfront" {') == 1
        # No separate module blocks for the expanded sub-ids
        assert 'module "cloudfront-distribution"' not in main_tf
        assert 'module "cloudfront-originaccesscontrol"' not in main_tf

    def test_dedup_merges_inputs_from_both_resources(self, tmp_path):
        """When two resources share a module, their inputs are merged into
        the single module block (first resource's inputs + second's, with
        first-wins for overlapping keys)."""
        stack = {
            "resources": [
                {"id": "cloudfront-distribution", "type": "aws:cloudfront:distribution", "module": "cloudfront@1.0.0", "inputs": {"price_class": "PriceClass_100", "region": "us-east-1"}},
                {"id": "cloudfront-originaccesscontrol", "type": "aws:cloudfront:originaccesscontrol", "module": "cloudfront@1.0.0", "inputs": {"viewer_protocol_policy": "redirect-to-https"}},
            ],
            "outputs": {},
            "data_sources": [],
        }
        adapt(stack, str(tmp_path))
        main_tf = (tmp_path / "main.tf").read_text()
        # Both inputs present in the merged module block
        assert "PriceClass_100" in main_tf
        assert "redirect-to-https" in main_tf


class TestAdapterRemoteStateKeyOverride:
    """P2-2 (v1.14, REQ-139): NOVA_REMOTE_STATE_KEY env var (P2 renamed from
    NOVA_REMOTE_STATE_KEY; dual-read NOVA_* preferred, ACDL_* fallback until
    P5) overrides the default 'platform/terraform.tfstate' key in the emitted
    data terraform_remote_state block. This is the load-bearing correctness
    mechanism for the microservice L2 lifecycle (remote state points at the
    CI VPC, not the platform VPC)."""

    def test_default_remote_state_key(self, tmp_path, monkeypatch):
        """When NOVA_REMOTE_STATE_KEY is unset, the default key is used."""
        # P5 (REQ-164): ACDL_* fallback removed — NOVA_* only.
        monkeypatch.delenv("NOVA_REMOTE_STATE_KEY", raising=False)
        stack = {
            "resources": [
                {"id": "s3", "type": "aws:s3:bucket", "module": "s3@1.0.0", "inputs": {"bucket_name": "test", "region": "us-east-1"}}
            ],
            "outputs": {},
            "data_sources": ["platform"],
        }
        adapt(stack, str(tmp_path))
        terraform_tf = (tmp_path / "terraform.tf").read_text()
        main_tf = (tmp_path / "main.tf").read_text()
        # The remote state data block uses the default key
        assert "platform/terraform.tfstate" in main_tf

    def test_env_override_remote_state_key(self, tmp_path, monkeypatch):
        """When NOVA_REMOTE_STATE_KEY is set, the emitted data block uses
        the overridden key (e.g. 'spike/ci-vpc/terraform.tfstate')."""
        monkeypatch.setenv("NOVA_REMOTE_STATE_KEY", "spike/ci-vpc/terraform.tfstate")
        stack = {
            "resources": [
                {"id": "s3", "type": "aws:s3:bucket", "module": "s3@1.0.0", "inputs": {"bucket_name": "test", "region": "us-east-1"}}
            ],
            "outputs": {},
            "data_sources": ["platform"],
        }
        adapt(stack, str(tmp_path))
        main_tf = (tmp_path / "main.tf").read_text()
        # The remote state data block uses the overridden key
        assert "spike/ci-vpc/terraform.tfstate" in main_tf
        assert "platform/terraform.tfstate" not in main_tf