2 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
CIAgent 97691fd752 docs(P01): complete contract-surface-schemas-resolver phase
---ci---
project: nova-platform
phase: 1
milestone: v1.0
status: complete
phase_role: execution
---/ci---

P1 complete: contract schema, stack schema, environment schema,
resolver, env_check, 10 sample contracts, 36 tests. REQ-01..06,
23..28 covered.
2026-08-24 17:31:00 +00:00
30 changed files with 1303 additions and 13 deletions
+13 -13
View File
@@ -1,19 +1,19 @@
{
"phase": 0,
"stage": "grill",
"phase": 2,
"stage": "verify",
"milestone": "v1.0",
"phase_role": "pre_execution",
"phase_role": "execution",
"attempts": 0,
"updated_at": "2026-08-20T18:30:00Z",
"updated_at": "2026-08-20T19:15:00Z",
"project": "nova-platform",
"milestone_branch": "milestone/v1.0-nova-platform",
"phase_branch": "phase/00-pre-execution",
"specify_validated": true,
"requirements_count": 38,
"clarify_decisions": "D-011..D-038",
"clarify_escalated": ["D-017", "D-025", "D-030"],
"grill_verdict": "PROCEED (conditions resolved)",
"grill_confidence": 0.82,
"grill_conditions": ["C-1 D-037", "C-2 D-038", "C-3 docs-grep", "C-4 reorder", "C-5 concurrency-note", "C-6 deviation-claim-fix"],
"mvp_ux_check": "PASS"
"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": 58,
"reqs_covered": ["REQ-07", "REQ-08", "REQ-09", "REQ-25", "REQ-26", "REQ-10"]
},
"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())
+9
View File
@@ -0,0 +1,9 @@
id: msvc
name: Microservice
environment: dev
infrastructure:
- module: microservice
version: "1.0.0"
inputs:
service_name: "${env.environment}-${contract.id}-svc"
desired_count: 2
+9
View File
@@ -0,0 +1,9 @@
id: msvc
name: Microservice
environment: dr
infrastructure:
- module: microservice
version: "1.0.0"
inputs:
service_name: "${env.environment}-${contract.id}-svc"
desired_count: 2
+9
View File
@@ -0,0 +1,9 @@
id: msvc
name: Microservice
environment: prod
infrastructure:
- module: microservice
version: "1.0.0"
inputs:
service_name: "${env.environment}-${contract.id}-svc"
desired_count: 2
+9
View File
@@ -0,0 +1,9 @@
id: msvc
name: Microservice
environment: qa
infrastructure:
- module: microservice
version: "1.0.0"
inputs:
service_name: "${env.environment}-${contract.id}-svc"
desired_count: 2
+9
View File
@@ -0,0 +1,9 @@
id: msvc
name: Microservice
environment: dev
infrastructure:
- module: microservice
version: "1.0.0"
inputs:
service_name: "${env.environment}-${contract.id}-svc"
desired_count: 2
+9
View File
@@ -0,0 +1,9 @@
id: stsi
name: Static Assets Site
environment: dev
infrastructure:
- module: static-assets
version: "1.0.0"
inputs:
bucket_name: "${env.environment}-${contract.id}-assets"
index_document: index.html
+9
View File
@@ -0,0 +1,9 @@
id: stsi
name: Static Assets Site
environment: dr
infrastructure:
- module: static-assets
version: "1.0.0"
inputs:
bucket_name: "${env.environment}-${contract.id}-assets"
index_document: index.html
+9
View File
@@ -0,0 +1,9 @@
id: stsi
name: Static Assets Site
environment: prod
infrastructure:
- module: static-assets
version: "1.0.0"
inputs:
bucket_name: "${env.environment}-${contract.id}-assets"
index_document: index.html
+9
View File
@@ -0,0 +1,9 @@
id: stsi
name: Static Assets Site
environment: qa
infrastructure:
- module: static-assets
version: "1.0.0"
inputs:
bucket_name: "${env.environment}-${contract.id}-assets"
index_document: index.html
+9
View File
@@ -0,0 +1,9 @@
id: stsi
name: Static Assets Site
environment: dev
infrastructure:
- module: static-assets
version: "1.0.0"
inputs:
bucket_name: "${env.environment}-${contract.id}-assets"
index_document: index.html
View File
+180
View File
@@ -0,0 +1,180 @@
"""Nova Platform — Contract Resolver.
Resolves a validated consumer contract to a Stack instance (a flat dict
conforming to schemas/stack.schema.json).
Flow:
1. Validate the contract dict against schemas/contract.schema.json.
2. Load the environment via core.environment_check.check().
3. Build an interpolation context {'env': env, 'contract': contract}.
4. For each infrastructure entry: look up the module + version in the
registry, interpolate ${env.*} / ${contract.*} tokens in inputs,
and emit a flat stack resource {module, version, inputs}.
5. Return the stack dict.
Engine-agnostic: no aws_*, no Terraform terms, no module paths. The stack
carries NO 'source' field (D-037/C-1 grill fix) — the adapter loads the
registry to map module -> terraform_dir. L2 is opaque (D-012): a single
stack resource, no children/wires expansion.
"""
import json
import re
from pathlib import Path
import jsonschema
import yaml
_TOKEN_RE = re.compile(r"\$\{([a-zA-Z_][a-zA-Z0-9_.]*)\}")
class ModuleNotFoundError(KeyError):
"""Raised when a contract references a module not in the registry."""
class VersionNotFoundError(KeyError):
"""Raised when a contract references a version not in the registry."""
def _lookup_dotted(context, dotted):
parts = dotted.split(".")
cur = context
for part in parts:
if isinstance(cur, dict) and part in cur:
cur = cur[part]
else:
raise KeyError(dotted)
return cur
def _expand_vars(value, context):
if isinstance(value, str):
def _replace(match):
token = match.group(1)
try:
resolved = _lookup_dotted(context, token)
except KeyError:
raise ValueError(f"unresolved interpolation token: ${{{token}}}")
if isinstance(resolved, (dict, list)):
return json.dumps(resolved)
return str(resolved)
return _TOKEN_RE.sub(_replace, value)
if isinstance(value, dict):
return {k: _expand_vars(v, context) for k, v in value.items()}
if isinstance(value, list):
return [_expand_vars(v, context) for v in value]
return value
def _latest_version(registry, module_name):
versions = registry[module_name]
non_deprecated = [(v, e) for v, e in versions.items()
if not e.get("deprecated", False)]
if not non_deprecated:
non_deprecated = list(versions.items())
non_deprecated.sort(key=lambda x: [int(p) for p in x[0].split(".")],
reverse=True)
return non_deprecated[0][0]
def _load_schema(path):
with open(path) as fh:
return json.load(fh)
def resolve(contract, registry, modules_dir, environments_dir=None,
repo_root=None):
"""Resolve a validated contract dict to a flat Stack dict.
Args:
contract: validated contract dict (must conform to
schemas/contract.schema.json).
registry: modules/registry.json loaded as a dict.
modules_dir: Path to the modules/ directory (unused for L2-opaque
resolution but kept per D-011 for future interface.json reads).
environments_dir: Path to core/environments/. If None, derived from
repo_root / 'core' / 'environments'.
repo_root: Path to the repo root. If None, derived from modules_dir
parent's parent (modules_dir is <root>/modules).
Returns:
A flat stack dict conforming to schemas/stack.schema.json:
{contract_id, contract_name, environment, resources: [{module,
version, inputs}]}.
Raises:
ModuleNotFoundError: contract references an unknown module.
VersionNotFoundError: contract references an unknown version.
jsonschema.ValidationError: contract does not conform to schema.
ValueError: unresolved interpolation token.
"""
if repo_root is None:
repo_root = Path(modules_dir).parent.parent
if environments_dir is None:
environments_dir = Path(repo_root) / "core" / "environments"
contract_schema_path = Path(repo_root) / "schemas" / "contract.schema.json"
contract_schema = _load_schema(contract_schema_path)
jsonschema.validate(contract, contract_schema)
from core import environment_check
env = environment_check.check(contract["environment"], environments_dir)
# Expose 'environment' as an alias for the env's 'name' field so
# ${env.environment} resolves (the env JSON uses 'name', but contracts
# reference the environment by ${env.environment}).
env["environment"] = env.get("name", contract["environment"])
context = {"env": env, "contract": contract}
resources = []
for item in contract["infrastructure"]:
module_name = item["module"]
if module_name not in registry:
raise ModuleNotFoundError(module_name)
version = item.get("version")
if version is None:
version = _latest_version(registry, module_name)
elif version not in registry[module_name]:
raise VersionNotFoundError(f"{module_name}@{version}")
inputs = _expand_vars(item.get("inputs", {}), context)
resources.append({
"module": module_name,
"version": version,
"inputs": inputs,
})
return {
"contract_id": contract["id"],
"contract_name": contract["name"],
"environment": contract["environment"],
"resources": resources,
}
def main(argv=None):
import sys
argv = argv or sys.argv[1:]
if len(argv) < 2:
print("usage: contract_resolver.py <contract.yaml> [out.json]",
file=sys.stderr)
return 2
contract_path = argv[0]
out_path = argv[1] if len(argv) > 1 else None
repo_root = Path(__file__).resolve().parent.parent
with open(contract_path) as fh:
contract = yaml.safe_load(fh)
with open(repo_root / "modules" / "registry.json") as fh:
registry = json.load(fh)
stack = resolve(contract, registry, repo_root / "modules",
repo_root=repo_root)
if out_path:
with open(out_path, "w") as fh:
json.dump(stack, fh, indent=2)
else:
print(json.dumps(stack, indent=2))
return 0
if __name__ == "__main__":
import sys
sys.exit(main())
+37
View File
@@ -0,0 +1,37 @@
"""Nova Platform — Environment Check.
Loads and validates a platform-managed environment JSON file.
Simplified per D-019: check(env_name, environments_dir) -> dict, raises
EnvironmentNotFoundError on missing env. Drops the reference's
(ok, message) tuple, _onboarding_message, and main() CLI.
"""
import json
from pathlib import Path
class EnvironmentNotFoundError(FileNotFoundError):
"""Raised when a named environment has no JSON file."""
def check(env_name, environments_dir):
"""Load and return the environment dict for env_name.
Args:
env_name: environment name (dev, qa, prod, dr).
environments_dir: Path to the core/environments/ directory.
Returns:
The parsed environment dict.
Raises:
EnvironmentNotFoundError: no <env_name>.json in environments_dir.
"""
env_path = Path(environments_dir) / f"{env_name}.json"
if not env_path.exists():
raise EnvironmentNotFoundError(
f"environment '{env_name}' not found at {env_path}")
with open(env_path) as fh:
env = json.load(fh)
return env
+14
View File
@@ -0,0 +1,14 @@
{
"name": "dev",
"description": "Sample dev environment for offline/local testing. account_id placeholder (000000000000) for offline mode.",
"account_id": "000000000000",
"region": "us-east-1",
"state_backend": {
"bucket": "nova-tfstate-dev-us-east-1",
"lock_table": "nova-tfstate-locks"
},
"network": {
"vpc_cidr": "10.0.0.0/16",
"azs": ["us-east-1a", "us-east-1b"]
}
}
+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"
}
}
}
+52
View File
@@ -0,0 +1,52 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://nova.cloudinit.dev/schemas/contract.schema.json",
"title": "Nova Consumer Contract",
"description": "A consumer's declaration of infrastructure intent. Engine-agnostic: no aws_* or Terraform terms.",
"type": "object",
"required": ["id", "name", "environment", "infrastructure"],
"additionalProperties": false,
"properties": {
"id": {
"type": "string",
"pattern": "^[a-z][a-z0-9-]{2,5}$",
"description": "Stable operational acronym (3-6 chars). Used in state keys + resource naming."
},
"name": {
"type": "string",
"minLength": 3,
"description": "Human-readable contract name."
},
"environment": {
"type": "string",
"enum": ["dev", "qa", "prod", "dr"],
"description": "Target environment."
},
"infrastructure": {
"type": "array",
"minItems": 1,
"description": "List of modules to deploy. Array (not map) per D-015.",
"items": {
"type": "object",
"required": ["module", "inputs"],
"additionalProperties": false,
"properties": {
"module": {
"type": "string",
"pattern": "^[a-z][a-z0-9-]*$",
"description": "Module name from the registry."
},
"version": {
"type": "string",
"pattern": "^\\d+\\.\\d+\\.\\d+$",
"description": "Semver version. Optional — defaults to latest non-deprecated."
},
"inputs": {
"type": "object",
"description": "Module-specific inputs. No aws_* keys (engine-agnostic)."
}
}
}
}
}
}
+60
View File
@@ -0,0 +1,60 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://nova.cloudinit.dev/schemas/environment.schema.json",
"title": "Nova Platform-Managed Environment",
"description": "Environment record. Simplified per D-017/D-018 (drops runner_role_arn/autonomy/confidence_threshold — OOS).",
"type": "object",
"required": ["name", "account_id", "region", "state_backend", "network"],
"additionalProperties": false,
"properties": {
"name": {
"type": "string",
"description": "Environment name (dev, qa, prod, dr)."
},
"description": {
"type": "string",
"description": "Optional human-readable description."
},
"account_id": {
"type": "string",
"pattern": "^[0-9]{12}$",
"description": "12-digit AWS account ID. Placeholder 000000000000 allowed for offline/dev."
},
"region": {
"type": "string",
"description": "AWS region (e.g. us-east-1)."
},
"state_backend": {
"type": "object",
"required": ["bucket", "lock_table"],
"additionalProperties": false,
"properties": {
"bucket": {
"type": "string",
"description": "S3 state bucket name."
},
"lock_table": {
"type": "string",
"description": "DynamoDB lock table name."
}
}
},
"network": {
"type": "object",
"required": ["vpc_cidr", "azs"],
"additionalProperties": false,
"properties": {
"vpc_cidr": {
"type": "string",
"description": "VPC CIDR block."
},
"azs": {
"type": "array",
"maxItems": 6,
"items": {"type": "string"},
"description": "Availability zones."
}
}
}
}
}
+49
View File
@@ -0,0 +1,49 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://nova.cloudinit.dev/schemas/stack.schema.json",
"title": "Nova Target Stack",
"description": "Resolved stack instance. Engine-agnostic: no source/Terraform paths, no aws_* terms (per D-037/C-1 — the adapter loads the registry to resolve module → terraform_dir).",
"type": "object",
"required": ["contract_id", "contract_name", "environment", "resources"],
"additionalProperties": false,
"properties": {
"contract_id": {
"type": "string",
"description": "Echoed from the contract."
},
"contract_name": {
"type": "string",
"description": "Echoed from the contract."
},
"environment": {
"type": "string",
"enum": ["dev", "qa", "prod", "dr"],
"description": "Echoed from the contract."
},
"resources": {
"type": "array",
"minItems": 1,
"description": "One entry per contract infrastructure item. Flat per D-012 (L2 is opaque — no children/wires expansion).",
"items": {
"type": "object",
"required": ["module", "version", "inputs"],
"additionalProperties": false,
"properties": {
"module": {
"type": "string",
"description": "Module name."
},
"version": {
"type": "string",
"pattern": "^\\d+\\.\\d+\\.\\d+$",
"description": "Resolved semver version."
},
"inputs": {
"type": "object",
"description": "Interpolated module inputs."
}
}
}
}
}
}
View File
+61
View File
@@ -0,0 +1,61 @@
"""Pytest fixtures for nova-platform tests."""
import json
import sys
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
@pytest.fixture
def repo_root():
return ROOT
@pytest.fixture
def registry():
"""Minimal registry fixture for resolver/adapter tests (P3 lands the real one)."""
return {
"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",
}
},
"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",
}
},
"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",
}
},
}
@pytest.fixture
def contract_schema():
with open(ROOT / "schemas" / "contract.schema.json") as fh:
return json.load(fh)
@pytest.fixture
def stack_schema():
with open(ROOT / "schemas" / "stack.schema.json") as fh:
return json.load(fh)
+118
View File
@@ -0,0 +1,118 @@
"""Tests for core/contract_resolver.py — REQ-23."""
import json
from pathlib import Path
import jsonschema
import pytest
from core.contract_resolver import (
resolve,
ModuleNotFoundError,
VersionNotFoundError,
)
def _contract(infra=None, env="dev", cid="stsi", name="Test"):
return {
"id": cid,
"name": name,
"environment": env,
"infrastructure": infra if infra is not None else [
{"module": "static-assets", "version": "1.0.0",
"inputs": {"bucket_name": "test-bucket", "index_document": "index.html"}},
],
}
class TestResolveHappyPath:
def test_resolves_static_assets(self, registry, repo_root):
contract = _contract()
stack = resolve(contract, registry, repo_root / "modules",
repo_root=repo_root)
assert stack["contract_id"] == "stsi"
assert stack["contract_name"] == "Test"
assert stack["environment"] == "dev"
assert len(stack["resources"]) == 1
r = stack["resources"][0]
assert r["module"] == "static-assets"
assert r["version"] == "1.0.0"
assert r["inputs"]["bucket_name"] == "test-bucket"
def test_stack_validates_against_schema(self, registry, repo_root, stack_schema):
contract = _contract()
stack = resolve(contract, registry, repo_root / "modules",
repo_root=repo_root)
jsonschema.validate(stack, stack_schema)
def test_no_source_field(self, registry, repo_root):
contract = _contract()
stack = resolve(contract, registry, repo_root / "modules",
repo_root=repo_root)
assert "source" not in stack["resources"][0]
class TestResolveInterpolation:
def test_env_token_expanded(self, registry, repo_root):
contract = _contract(infra=[
{"module": "static-assets", "version": "1.0.0",
"inputs": {"bucket_name": "${env.environment}-${contract.id}-assets",
"index_document": "index.html"}},
])
stack = resolve(contract, registry, repo_root / "modules",
repo_root=repo_root)
assert stack["resources"][0]["inputs"]["bucket_name"] == "dev-stsi-assets"
def test_env_dotted_token(self, registry, repo_root):
contract = _contract(infra=[
{"module": "static-assets", "version": "1.0.0",
"inputs": {"bucket_name": "${env.region}-bucket",
"index_document": "index.html"}},
])
stack = resolve(contract, registry, repo_root / "modules",
repo_root=repo_root)
assert stack["resources"][0]["inputs"]["bucket_name"] == "us-east-1-bucket"
def test_unresolved_token_raises(self, registry, repo_root):
contract = _contract(infra=[
{"module": "static-assets", "version": "1.0.0",
"inputs": {"bucket_name": "${env.nonexistent}-bucket",
"index_document": "index.html"}},
])
with pytest.raises(ValueError, match="unresolved interpolation token"):
resolve(contract, registry, repo_root / "modules", repo_root=repo_root)
class TestResolveVersioning:
def test_version_defaults_to_latest(self, registry, repo_root):
contract = _contract(infra=[
{"module": "static-assets", "inputs": {"bucket_name": "b",
"index_document": "index.html"}},
])
stack = resolve(contract, registry, repo_root / "modules",
repo_root=repo_root)
assert stack["resources"][0]["version"] == "1.0.0"
class TestResolveErrors:
def test_unknown_module(self, registry, repo_root):
contract = _contract(infra=[
{"module": "nope", "version": "1.0.0",
"inputs": {"x": "y"}},
])
with pytest.raises(ModuleNotFoundError):
resolve(contract, registry, repo_root / "modules", repo_root=repo_root)
def test_unknown_version(self, registry, repo_root):
contract = _contract(infra=[
{"module": "static-assets", "version": "9.9.9",
"inputs": {"bucket_name": "b", "index_document": "index.html"}},
])
with pytest.raises(VersionNotFoundError):
resolve(contract, registry, repo_root / "modules", repo_root=repo_root)
def test_invalid_contract_missing_id(self, registry, repo_root):
contract = {"name": "Test", "environment": "dev",
"infrastructure": [{"module": "s3", "inputs": {}}]}
with pytest.raises(jsonschema.ValidationError):
resolve(contract, registry, repo_root / "modules", repo_root=repo_root)
+60
View File
@@ -0,0 +1,60 @@
"""Tests for schemas/contract.schema.json — REQ-27.
Validates all 10 sample contracts against the contract schema.
"""
import glob
import jsonschema
import pytest
import yaml
from tests.conftest import ROOT
SAMPLE_CONTRACTS = sorted(glob.glob(str(ROOT / "contracts" / "*.yml")))
class TestSampleContracts:
@pytest.mark.parametrize("path", SAMPLE_CONTRACTS,
ids=[p.split("/")[-1] for p in SAMPLE_CONTRACTS])
def test_validates(self, path, contract_schema):
with open(path) as fh:
contract = yaml.safe_load(fh)
jsonschema.validate(contract, contract_schema)
class TestSchemaNegativeCases:
def _validate(self, instance, schema):
jsonschema.validate(instance, schema)
def test_missing_id_rejected(self, contract_schema):
with pytest.raises(jsonschema.ValidationError):
self._validate({"name": "X", "environment": "dev",
"infrastructure": []}, contract_schema)
def test_bad_id_pattern_rejected(self, contract_schema):
with pytest.raises(jsonschema.ValidationError):
self._validate({"id": "UPPER", "name": "X", "environment": "dev",
"infrastructure": []}, contract_schema)
def test_bad_environment_rejected(self, contract_schema):
with pytest.raises(jsonschema.ValidationError):
self._validate({"id": "abc", "name": "X", "environment": "staging",
"infrastructure": []}, contract_schema)
def test_empty_infrastructure_rejected(self, contract_schema):
with pytest.raises(jsonschema.ValidationError):
self._validate({"id": "abc", "name": "X", "environment": "dev",
"infrastructure": []}, contract_schema)
def test_infrastructure_missing_module_rejected(self, contract_schema):
with pytest.raises(jsonschema.ValidationError):
self._validate({"id": "abc", "name": "X", "environment": "dev",
"infrastructure": [{"inputs": {}}]}, contract_schema)
def test_additional_top_level_rejected(self, contract_schema):
with pytest.raises(jsonschema.ValidationError):
self._validate({"id": "abc", "name": "X", "environment": "dev",
"infrastructure": [{"module": "s3", "inputs": {}}],
"extra": True}, contract_schema)
+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}"
+34
View File
@@ -0,0 +1,34 @@
"""Tests for core/environment_check.py — REQ-24."""
from pathlib import Path
import pytest
from core.environment_check import check, EnvironmentNotFoundError
class TestCheckHappyPath:
def test_dev_returns_dict(self, repo_root):
env = check("dev", repo_root / "core" / "environments")
assert env["name"] == "dev"
assert env["region"] == "us-east-1"
assert env["state_backend"]["bucket"] == "nova-tfstate-dev-us-east-1"
assert env["state_backend"]["lock_table"] == "nova-tfstate-locks"
def test_returns_all_required_fields(self, repo_root):
env = check("dev", repo_root / "core" / "environments")
for f in ("name", "account_id", "region", "state_backend", "network"):
assert f in env, f"missing {f}"
class TestCheckErrors:
def test_missing_env_raises(self, repo_root, tmp_path):
with pytest.raises(EnvironmentNotFoundError):
check("nonexistent", tmp_path)
def test_missing_env_message_names_env(self, repo_root, tmp_path):
try:
check("qa", tmp_path)
assert False, "should have raised"
except EnvironmentNotFoundError as e:
assert "qa" in str(e)
+70
View File
@@ -0,0 +1,70 @@
"""Tests for schemas/stack.schema.json — REQ-28.
Validates resolved stacks against the stack schema. Builds stacks via
resolve() using a registry fixture.
"""
import jsonschema
import pytest
from core.contract_resolver import resolve
def _contract():
return {
"id": "stsi",
"name": "Test",
"environment": "dev",
"infrastructure": [
{"module": "static-assets", "version": "1.0.0",
"inputs": {"bucket_name": "test-bucket", "index_document": "index.html"}},
],
}
class TestStackSchema:
def test_resolved_stack_validates(self, registry, repo_root, stack_schema):
stack = resolve(_contract(), registry, repo_root / "modules",
repo_root=repo_root)
jsonschema.validate(stack, stack_schema)
def test_multi_resource_stack_validates(self, registry, repo_root, stack_schema):
contract = {
"id": "stsi", "name": "Test", "environment": "dev",
"infrastructure": [
{"module": "static-assets", "version": "1.0.0",
"inputs": {"bucket_name": "b1", "index_document": "index.html"}},
{"module": "microservice", "version": "1.0.0",
"inputs": {"service_name": "s1", "desired_count": 2}},
],
}
stack = resolve(contract, registry, repo_root / "modules",
repo_root=repo_root)
assert len(stack["resources"]) == 2
jsonschema.validate(stack, stack_schema)
class TestStackSchemaNegative:
def test_missing_resources_rejected(self, stack_schema):
bad = {"contract_id": "x", "contract_name": "X", "environment": "dev"}
with pytest.raises(jsonschema.ValidationError):
jsonschema.validate(bad, stack_schema)
def test_empty_resources_rejected(self, stack_schema):
bad = {"contract_id": "x", "contract_name": "X", "environment": "dev",
"resources": []}
with pytest.raises(jsonschema.ValidationError):
jsonschema.validate(bad, stack_schema)
def test_resource_missing_module_rejected(self, stack_schema):
bad = {"contract_id": "x", "contract_name": "X", "environment": "dev",
"resources": [{"version": "1.0.0", "inputs": {}}]}
with pytest.raises(jsonschema.ValidationError):
jsonschema.validate(bad, stack_schema)
def test_resource_extra_source_rejected(self, stack_schema):
bad = {"contract_id": "x", "contract_name": "X", "environment": "dev",
"resources": [{"module": "s3", "version": "1.0.0", "inputs": {},
"source": "modules/l1/s3/terraform"}]}
with pytest.raises(jsonschema.ValidationError):
jsonschema.validate(bad, stack_schema)
+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"]'