1 Commits

Author SHA1 Message Date
CIAgent 14be01d971 docs(P02): complete terraform-adapter-engine-boundary phase
---ci---
project: nova-platform
phase: 2
milestone: v1.0
status: complete
---/ci---

P2 complete: adapter (loads registry D-037), engine-boundary test (D-034).
58 tests pass. REQ-07,08,09,25,26 + REQ-10 (registry) covered.
2026-08-24 17:33:22 +00:00
7 changed files with 474 additions and 15 deletions
+9 -15
View File
@@ -1,25 +1,19 @@
{
"phase": 1,
"phase": 2,
"stage": "verify",
"milestone": "v1.0",
"phase_role": "execution",
"attempts": 0,
"updated_at": "2026-08-20T19:00:00Z",
"updated_at": "2026-08-20T19:15:00Z",
"project": "nova-platform",
"milestone_branch": "milestone/v1.0-nova-platform",
"phase_branch": "phase/01-contract-surface-schemas-resolver",
"phase_0_ship": {
"tag": "v0.1.0",
"merged_to_milestone": true,
"merge_commit": "7835c2a",
"phase_branch_deleted": true,
"local_only": true,
"release_pending": "NOVA_FORGE_TOKEN blank"
},
"phase_1_verify": {
"phase_branch": "phase/02-terraform-adapter-engine-boundary",
"phase_0_ship": {"tag": "v0.1.0", "merged_to_milestone": true, "merge_commit": "7835c2a", "phase_branch_deleted": true, "local_only": true},
"phase_1_ship": {"tag": "v0.1.1", "merged_to_milestone": true, "phase_branch_deleted": true, "local_only": true},
"phase_2_verify": {
"tests_pass": true,
"tests_count": 36,
"reqs_covered": ["REQ-01", "REQ-02", "REQ-03", "REQ-04", "REQ-05", "REQ-06", "REQ-23", "REQ-24", "REQ-27", "REQ-28"]
"tests_count": 58,
"reqs_covered": ["REQ-07", "REQ-08", "REQ-09", "REQ-25", "REQ-26", "REQ-10"]
},
"next_phase": "phase/02-terraform-adapter-engine-boundary"
"next_phase": "phase/03-l1-primitives-registry"
}
View File
+3
View File
@@ -0,0 +1,3 @@
from adapters.terraform.adapter import adapt
__all__ = ["adapt"]
+102
View File
@@ -0,0 +1,102 @@
"""Nova Platform — Terraform Adapter.
The ONLY engine-specific code in the platform (per REQ-09, verified by
tests/test_engine_boundary.py). Loads modules/registry.json internally
to map module -> terraform_dir (per D-037/C-1 grill fix — the resolver
does NOT put a `source` field in the stack; the adapter resolves it
here, inside the engine boundary).
Stateless assembler: no `terraform` CLI invocation, no state files, no
plan files. Emits Terraform HCL: one `module "x" { source = ...; <inputs> }`
block per stack resource.
"""
import json
import os
from pathlib import Path
def _load_registry(repo_root):
"""Load modules/registry.json -> {module_name: terraform_dir}."""
registry_path = os.path.join(str(repo_root), "modules", "registry.json")
with open(registry_path) as fh:
registry = json.load(fh)
return {name: list(versions.values())[0].get("terraform_dir")
for name, versions in registry.items()
if list(versions.values())[0].get("terraform_dir")}
def _tf_value(value):
"""Render a Python value as an HCL expression."""
if isinstance(value, bool):
return "true" if value else "false"
if isinstance(value, (int, float)):
return str(value)
if isinstance(value, list):
return "[" + ", ".join(_tf_value(v) for v in value) + "]"
if isinstance(value, dict):
return "{ " + ", ".join(f"{k} = {_tf_value(v)}" for k, v in value.items()) + " }"
return json.dumps(str(value))
def _emit_module_block(resource, terraform_dirs, repo_root):
"""Emit one `module "x" { source = ...; <inputs> }` block."""
module_name = resource["module"]
rid = module_name.replace("-", "_")
tf_dir = terraform_dirs.get(module_name)
if tf_dir is None:
raise ValueError(f"module '{module_name}' has no terraform_dir in registry")
source = os.path.join(str(repo_root), tf_dir)
lines = [f'module "{rid}" {{', f' source = "{source}"']
for key, val in resource.get("inputs", {}).items():
if key == "region":
continue
lines.append(f" {key} = {_tf_value(val)}")
lines.append("}")
return "\n".join(lines)
def adapt(stack, repo_root):
"""Compile a flat stack dict to Terraform HCL.
Args:
stack: a flat stack dict conforming to schemas/stack.schema.json
(NO `source` field per D-037 — the adapter resolves
module -> terraform_dir via the registry).
repo_root: Path to the repo root (the adapter loads
modules/registry.json from here).
Returns:
A string of Terraform HCL with one `module "x" {}` block per
stack resource.
"""
terraform_dirs = _load_registry(repo_root)
blocks = []
for resource in stack.get("resources", []):
blocks.append(_emit_module_block(resource, terraform_dirs, repo_root))
return "\n\n".join(blocks) + "\n"
def main(argv=None):
import sys
argv = argv or sys.argv[1:]
if len(argv) < 1:
print("usage: adapter.py <stack.json> [out.tf]", file=sys.stderr)
return 2
stack_path = argv[0]
out_path = argv[1] if len(argv) > 1 else None
repo_root = Path(__file__).resolve().parent.parent.parent
with open(stack_path) as fh:
stack = json.load(fh)
hcl = adapt(stack, repo_root)
if out_path:
with open(out_path, "w") as fh:
fh.write(hcl)
else:
print(hcl)
return 0
if __name__ == "__main__":
import sys
sys.exit(main())
+137
View File
@@ -0,0 +1,137 @@
{
"s3": {
"1.0.0": {
"interface": "modules/l1/s3/interface.json",
"terraform_dir": "modules/l1/s3/terraform",
"published_at": "2026-08-20T00:00:00Z",
"deprecated": false,
"kind": "l1"
}
},
"vpc": {
"1.0.0": {
"interface": "modules/l1/vpc/interface.json",
"terraform_dir": "modules/l1/vpc/terraform",
"published_at": "2026-08-20T00:00:00Z",
"deprecated": false,
"kind": "l1"
}
},
"ecs-cluster": {
"1.0.0": {
"interface": "modules/l1/ecs-cluster/interface.json",
"terraform_dir": "modules/l1/ecs-cluster/terraform",
"published_at": "2026-08-20T00:00:00Z",
"deprecated": false,
"kind": "l1"
}
},
"ecs-service": {
"1.0.0": {
"interface": "modules/l1/ecs-service/interface.json",
"terraform_dir": "modules/l1/ecs-service/terraform",
"published_at": "2026-08-20T00:00:00Z",
"deprecated": false,
"kind": "l1"
}
},
"iam-role": {
"1.0.0": {
"interface": "modules/l1/iam-role/interface.json",
"terraform_dir": "modules/l1/iam-role/terraform",
"published_at": "2026-08-20T00:00:00Z",
"deprecated": false,
"kind": "l1"
}
},
"alb": {
"1.0.0": {
"interface": "modules/l1/alb/interface.json",
"terraform_dir": "modules/l1/alb/terraform",
"published_at": "2026-08-20T00:00:00Z",
"deprecated": false,
"kind": "l1"
}
},
"ecr": {
"1.0.0": {
"interface": "modules/l1/ecr/interface.json",
"terraform_dir": "modules/l1/ecr/terraform",
"published_at": "2026-08-20T00:00:00Z",
"deprecated": false,
"kind": "l1"
}
},
"cloudfront": {
"1.0.0": {
"interface": "modules/l1/cloudfront/interface.json",
"terraform_dir": "modules/l1/cloudfront/terraform",
"published_at": "2026-08-20T00:00:00Z",
"deprecated": false,
"kind": "l1"
}
},
"waf": {
"1.0.0": {
"interface": "modules/l1/waf/interface.json",
"terraform_dir": "modules/l1/waf/terraform",
"published_at": "2026-08-20T00:00:00Z",
"deprecated": false,
"kind": "l1"
}
},
"rds": {
"1.0.0": {
"interface": "modules/l1/rds/interface.json",
"terraform_dir": "modules/l1/rds/terraform",
"published_at": "2026-08-20T00:00:00Z",
"deprecated": false,
"kind": "l1"
}
},
"kms-key": {
"1.0.0": {
"interface": "modules/l1/kms-key/interface.json",
"terraform_dir": "modules/l1/kms-key/terraform",
"published_at": "2026-08-20T00:00:00Z",
"deprecated": false,
"kind": "l1"
}
},
"dynamodb": {
"1.0.0": {
"interface": "modules/l1/dynamodb/interface.json",
"terraform_dir": "modules/l1/dynamodb/terraform",
"published_at": "2026-08-20T00:00:00Z",
"deprecated": false,
"kind": "l1"
}
},
"uptime": {
"1.0.0": {
"interface": "modules/l1/uptime/interface.json",
"terraform_dir": "modules/l1/uptime/terraform",
"published_at": "2026-08-20T00:00:00Z",
"deprecated": false,
"kind": "l1"
}
},
"microservice": {
"1.0.0": {
"interface": "modules/l2/microservice/interface.json",
"terraform_dir": "modules/l2/microservice/terraform",
"published_at": "2026-08-20T00:00:00Z",
"deprecated": false,
"kind": "l2"
}
},
"static-assets": {
"1.0.0": {
"interface": "modules/l2/static-assets/interface.json",
"terraform_dir": "modules/l2/static-assets/terraform",
"published_at": "2026-08-20T00:00:00Z",
"deprecated": false,
"kind": "l2"
}
}
}
+128
View File
@@ -0,0 +1,128 @@
"""Engine-boundary test — REQ-09, D-034.
Verifies no engine-specific *logic* (HCL strings, aws_ resource types,
terraform CLI calls, module/provider/resource block declarations) leaks
outside adapters/terraform/.
Scans .py files in core/, schemas/, contracts/, tests/, scripts/, root.
EXCLUDES adapters/terraform/ (the boundary), modules/, .tf/.md/.json data
files (per D-034).
To avoid false positives on docstrings/comments that *mention* "Terraform"
conceptually, the test strips comments + docstrings before scanning.
The forbidden terms are checked as *code-level* tokens: aws_<word>,
`module "`, `provider "`, `resource "` (HCL block declarations that
would indicate actual HCL emission outside the adapter). The bare word
"terraform" is NOT forbidden (it appears in import paths like
`adapters.terraform` and docstrings); only `terraform ` followed by a
block brace or CLI invocation is.
"""
import ast
import re
from pathlib import Path
import pytest
from tests.conftest import ROOT
# Forbidden as code-level patterns (not in strings/comments):
# - aws_<word>: AWS resource type prefixes (e.g. aws_s3_bucket)
# - 'module "': HCL module block declaration
# - 'provider "': HCL provider block declaration
# - 'resource "': HCL resource block declaration
# - terraform init/plan/apply: CLI invocations
FORBIDDEN_PATTERNS = [
re.compile(r'\baws_[a-z_]+'),
re.compile(r'module\s+"'),
re.compile(r'provider\s+"'),
re.compile(r'resource\s+"'),
re.compile(r'\bterraform\s+(init|plan|apply|validate|destroy)\b'),
]
EXCLUDE_DIRS = {
"adapters/terraform",
".venv",
"__pycache__",
".ciagent",
".git",
"modules",
"terraform",
"docs",
}
# Test files that legitimately reference engine terms to verify the
# boundary/adapter (they assert HCL output contains 'module "' etc.).
# These are part of the boundary enforcement, not engine logic leaks.
EXCLUDE_FILES = {
"tests/test_terraform_adapter.py",
"tests/test_engine_boundary.py",
}
def _strip_docstrings_and_comments(source):
"""Remove docstrings + comments from Python source, return code only."""
try:
tree = ast.parse(source)
except SyntaxError:
return source
lines = source.splitlines(keepends=True)
# Collect line ranges of docstring nodes
docstring_ranges = set()
for node in ast.walk(tree):
if isinstance(node, (ast.Expr,)) and isinstance(node.value, ast.Constant) and isinstance(node.value.value, str):
for ln in range(node.lineno, node.end_lineno + 1):
docstring_ranges.add(ln)
if isinstance(node, ast.Constant) and isinstance(node.value, str) and node.lineno == getattr(node, "end_lineno", None):
# standalone string used as docstring at module/class level
pass
out = []
for i, line in enumerate(lines, start=1):
if i in docstring_ranges:
continue
# strip inline comments
stripped = re.sub(r'#.*$', '', line)
out.append(stripped)
return "".join(out)
def _scan_files():
for path in ROOT.rglob("*.py"):
rel = path.relative_to(ROOT)
rel_str = str(rel)
if any(rel_str.startswith(ex) for ex in EXCLUDE_DIRS):
continue
if any(part in EXCLUDE_DIRS for part in rel.parts):
continue
if rel_str in EXCLUDE_FILES:
continue
yield path
def _forbidden_matches(code):
matches = []
for pat in FORBIDDEN_PATTERNS:
found = pat.findall(code)
if found:
matches.extend(found)
return matches
class TestEngineBoundary:
@pytest.mark.parametrize("path", list(_scan_files()),
ids=[str(p.relative_to(ROOT)) for p in _scan_files()])
def test_no_engine_logic(self, path):
source = path.read_text()
code = _strip_docstrings_and_comments(source)
matches = _forbidden_matches(code)
assert not matches, (
f"{path.relative_to(ROOT)} contains forbidden engine logic: {matches}. "
f"Engine-specific code must live ONLY in adapters/terraform/."
)
def test_boundary_scans_files(self):
files = list(_scan_files())
assert len(files) > 0, "engine-boundary test must scan at least one .py file"
for f in files:
assert "adapters/terraform" not in str(f.relative_to(ROOT)), \
f"adapters/terraform/ must be excluded but found {f}"
+95
View File
@@ -0,0 +1,95 @@
"""Tests for adapters/terraform/adapter.py — REQ-25."""
import sys
from pathlib import Path
import pytest
from adapters.terraform.adapter import adapt, _tf_value
def _stack(resources):
return {
"contract_id": "test", "contract_name": "Test", "environment": "dev",
"resources": resources,
}
class TestAdapt:
def test_single_resource(self, repo_root):
stack = _stack([
{"module": "s3", "version": "1.0.0",
"inputs": {"bucket_name": "test-bucket", "enabled": True}},
])
hcl = adapt(stack, repo_root)
assert 'module "s3" {' in hcl
assert 'source = ' in hcl
assert 'bucket_name = "test-bucket"' in hcl
assert "enabled = true" in hcl
def test_multi_resource(self, repo_root):
stack = _stack([
{"module": "s3", "version": "1.0.0",
"inputs": {"bucket_name": "b1", "enabled": True}},
{"module": "vpc", "version": "1.0.0",
"inputs": {"cidr": "10.0.0.0/16"}},
])
hcl = adapt(stack, repo_root)
assert 'module "s3" {' in hcl
assert 'module "vpc" {' in hcl
def test_input_passthrough_scalar(self, repo_root):
stack = _stack([
{"module": "s3", "version": "1.0.0",
"inputs": {"bucket_name": "my-bucket", "region": "us-east-1"}},
])
hcl = adapt(stack, repo_root)
assert 'bucket_name = "my-bucket"' in hcl
assert "us-east-1" not in hcl.split("inputs")[0] if "inputs" in hcl else True
# region is skipped (provider-level)
def test_input_passthrough_list(self, repo_root):
stack = _stack([
{"module": "vpc", "version": "1.0.0",
"inputs": {"azs": ["us-east-1a", "us-east-1b"]}},
])
hcl = adapt(stack, repo_root)
assert 'azs = ["us-east-1a", "us-east-1b"]' in hcl
def test_input_passthrough_number(self, repo_root):
stack = _stack([
{"module": "vpc", "version": "1.0.0",
"inputs": {"desired_count": 3}},
])
hcl = adapt(stack, repo_root)
assert "desired_count = 3" in hcl
def test_hcl_validity_balanced_braces(self, repo_root):
stack = _stack([
{"module": "s3", "version": "1.0.0",
"inputs": {"bucket_name": "b", "enabled": True}},
])
hcl = adapt(stack, repo_root)
assert hcl.count("{") == hcl.count("}")
def test_unknown_module_raises(self, repo_root):
stack = _stack([
{"module": "nonexistent", "version": "1.0.0", "inputs": {}},
])
with pytest.raises(ValueError, match="no terraform_dir"):
adapt(stack, repo_root)
class TestTfValue:
def test_bool(self):
assert _tf_value(True) == "true"
assert _tf_value(False) == "false"
def test_int(self):
assert _tf_value(42) == "42"
def test_string(self):
assert _tf_value("hello") == '"hello"'
def test_list(self):
assert _tf_value(["a", "b"]) == '["a", "b"]'