acdl/tests/test_adapter.py

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


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