Jon Chery
1598c54a8b
90 offline tests covering adapter, confidence_signal, checkov_adapter, outbox_writer, and pipeline integration. Identical CI/CD workflows for Gitea Actions (dev) and GitHub Actions (production). New --check-only mode for run_platform.sh (offline, no AWS). ---ci--- project: acdl phase: 18 milestone: v1.3 status: verify ---/ci---
174 lines
6.1 KiB
Python
174 lines
6.1 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 TestSpikeInstance:
|
|
def test_spike_instance_validates_against_ir_schema(self, spike_ir, ir_schema):
|
|
jsonschema.validate(spike_ir, ir_schema)
|
|
|
|
def test_spike_instance_has_one_resource(self, spike_ir):
|
|
assert len(spike_ir["resources"]) == 1
|
|
r = spike_ir["resources"][0]
|
|
assert r["id"] == "s3"
|
|
assert r["type"] == "aws:s3:bucket"
|
|
|
|
def test_spike_instance_stack_is_l1_s3(self, spike_ir):
|
|
assert spike_ir["stack"]["name"] == "l1-s3"
|
|
assert spike_ir["stack"]["kind"] == "l1"
|
|
|
|
|
|
class TestRegistry:
|
|
def test_registry_has_7_l1_entries(self, registry):
|
|
assert len(registry) == 7
|
|
for key in registry:
|
|
assert key.startswith("l1-")
|
|
|
|
def test_registry_has_no_l2_entries(self, registry):
|
|
l2 = [k for k in registry if k.startswith("l2")]
|
|
assert l2 == []
|
|
|
|
def test_all_l1_interfaces_exist(self, registry, repo_root):
|
|
for name, versions in registry.items():
|
|
for ver, entry in versions.items():
|
|
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"
|
|
|
|
|
|
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 IR resource id"):
|
|
_ref_expr("ref:nonexistent.output", {"s3": "aws:s3:bucket"})
|
|
|
|
|
|
class TestAdapt:
|
|
def test_adapt_emits_three_files(self, spike_ir, tmp_path):
|
|
out_dir = str(tmp_path / "tf_out")
|
|
adapt(spike_ir, 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, spike_ir, tmp_path):
|
|
out_dir = str(tmp_path / "tf_out")
|
|
adapt(spike_ir, 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, spike_ir, tmp_path):
|
|
out_dir = str(tmp_path / "tf_out")
|
|
adapt(spike_ir, 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, spike_ir, tmp_path):
|
|
out_dir = str(tmp_path / "tf_out")
|
|
adapt(spike_ir, 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, spike_ir, tmp_path):
|
|
out_dir = str(tmp_path / "tf_out")
|
|
adapt(spike_ir, 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, spike_ir, tmp_path):
|
|
out_dir = str(tmp_path / "tf_out")
|
|
adapt(spike_ir, 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, spike_ir, tmp_path):
|
|
out_dir = str(tmp_path / "tf_out")
|
|
adapt(spike_ir, out_dir)
|
|
terraform_tf = open(os.path.join(out_dir, "terraform.tf")).read()
|
|
assert "spike/l1-s3/terraform.tfstate" in terraform_tf |