import json import os import sys from pathlib import Path import jsonschema import pytest import yaml sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) ROOT = Path(__file__).resolve().parent.parent class TestContractSchema: def test_schema_is_valid_json_schema(self): schema = json.load(open(ROOT / "schemas/contract.schema.json")) jsonschema.Draft202012Validator.check_schema(schema) def test_schema_requires_id_name_environment_infrastructure(self): schema = json.load(open(ROOT / "schemas/contract.schema.json")) for field in ["id", "name", "environment", "infrastructure"]: assert field in schema["required"] class TestResolveStaticAsset: def test_resolve_static_asset_contract(self, tmp_path): from core.contract_resolver import resolve stack = resolve(str(ROOT / "contracts/static-assets.yml"), str(ROOT)) assert stack["stack"]["name"] == "assets" assert stack["stack"]["title"] == "static-assets" assert stack["stack"]["kind"] == "l2" assert stack["stack"]["depth"] == 1 assert len(stack["resources"]) >= 1 def test_resolve_static_asset_has_s3_cloudfront_waf_resources(self): 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:s3:bucket" in types assert "aws:cloudfront:distribution" in types assert "aws:cloudfront:originaccesscontrol" in types assert "aws:wafv2:webacl" in types def test_resolve_static_asset_has_s3_resource(self): from core.contract_resolver import resolve stack = resolve(str(ROOT / "contracts/static-assets.yml"), str(ROOT)) s3_res = [r for r in stack["resources"] if r["type"] == "aws:s3:bucket"] assert len(s3_res) == 1 assert s3_res[0]["inputs"]["bucket_name"] == "acdl-dev-assets-000000000000-us-east-1" assert s3_res[0]["inputs"]["region"] == "us-east-1" def test_resolve_static_asset_validates_against_stack_schema(self): from core.contract_resolver import resolve stack = resolve(str(ROOT / "contracts/static-assets.yml"), str(ROOT)) schema = json.load(open(ROOT / "schemas/stack.schema.json")) jsonschema.validate(stack, schema) class TestResolveMicroservice: def test_resolve_microservice_contract(self): contract = { "id": "msvc", "name": "microservice-test", "environment": "dev", "infrastructure": { "microservice": { "version": "1.0.0", "inputs": { "image": "581513795199.dkr.ecr.us-east-1.amazonaws.com/nova-microservice:latest", "port": 8080, "region": "us-east-1", }, } }, } contract_path = ROOT / "contracts" / "test-microservice.yml" with open(contract_path, "w") as fh: yaml.dump(contract, fh) try: from core.contract_resolver import resolve stack = resolve(str(contract_path), str(ROOT)) assert stack["stack"]["name"] == "msvc" assert stack["stack"]["title"] == "microservice-test" assert stack["stack"]["kind"] == "l2" assert len(stack["resources"]) >= 6 finally: os.remove(contract_path) class TestResolveL1Direct: def test_resolve_s3_direct(self, tmp_path): contract = { "id": "s3a", "name": "s3-direct-test", "environment": "dev", "infrastructure": { "s3": { "version": "1.0.0", "inputs": { "bucket_name": "my-test-bucket", "region": "us-east-1", }, } }, } contract_path = tmp_path / "test-s3.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)) assert stack["stack"]["name"] == "s3a" assert stack["stack"]["kind"] == "l1" assert len(stack["resources"]) == 1 assert stack["resources"][0]["type"] == "aws:s3:bucket" assert stack["resources"][0]["inputs"]["bucket_name"] == "my-test-bucket" def test_resolve_s3_validates_against_stack_schema(self, tmp_path): contract = { "id": "s3a", "name": "s3-schema-test", "environment": "dev", "infrastructure": { "s3": { "version": "1.0.0", "inputs": {"bucket_name": "test", "region": "us-east-1"}, } }, } contract_path = tmp_path / "test-s3-schema.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)) schema = json.load(open(ROOT / "schemas/stack.schema.json")) jsonschema.validate(stack, schema) class TestResolveErrors: def test_unknown_module_raises(self, tmp_path): contract = { "id": "bad1", "name": "unknown-module-test", "environment": "dev", "infrastructure": { "nonexistent": { "version": "1.0.0", "inputs": {}, } }, } contract_path = tmp_path / "bad.yml" with open(contract_path, "w") as fh: yaml.dump(contract, fh) from core.contract_resolver import resolve with pytest.raises(ValueError, match="not found in registry"): resolve(str(contract_path), str(ROOT)) def test_missing_required_field_fails_validation(self, tmp_path): contract = {"id": "bad2", "name": "incomplete", "environment": "dev"} contract_path = tmp_path / "incomplete.yml" with open(contract_path, "w") as fh: yaml.dump(contract, fh) from core.contract_resolver import resolve with pytest.raises(jsonschema.ValidationError): resolve(str(contract_path), str(ROOT)) class TestDeployPipelineContract: def test_deploy_pipeline_validates_against_schema(self): schema = json.load(open(ROOT / "schemas/deploy-pipeline.schema.json")) with open(ROOT / "pipelines/contract.yml") as fh: contract = yaml.safe_load(fh) jsonschema.validate(contract, schema) def test_deploy_pipeline_has_required_stages(self): with open(ROOT / "pipelines/contract.yml") as fh: contract = yaml.safe_load(fh) stage_names = [s["name"] for s in contract["stages"]] assert "validate-contract" in stage_names assert "resolve-stack" in stage_names assert "checkov-static" in stage_names, "REQ-250: checkov-static stage missing" assert "terraform-plan" in stage_names assert "runtime-policy-scan" in stage_names, "REQ-250: runtime-policy-scan stage missing" assert "confidence" in stage_names assert "apply" in stage_names # The old single 'checkov' stage is gone (split into checkov-static + runtime-policy-scan) assert "checkov" not in stage_names, "old 'checkov' stage should be replaced by checkov-static + runtime-policy-scan" class TestL2OutputsResolution: """P1-7: L2 composition outputs[] is resolved into stack.outputs.""" def test_static_assets_outputs_present(self): from core.contract_resolver import resolve stack = resolve(str(ROOT / "contracts/static-assets.yml"), str(ROOT)) assert "outputs" in stack, "stack.outputs must be present for L2 modules (P1-7)" assert "distribution_domain_name" in stack["outputs"] assert "bucket_arn" in stack["outputs"] assert "web_acl_arn" in stack["outputs"] def test_static_assets_output_from_field_resolves_to_resource_id(self): from core.contract_resolver import resolve stack = resolve(str(ROOT / "contracts/static-assets.yml"), str(ROOT)) dist = stack["outputs"]["distribution_domain_name"] assert "from" in dist assert "output" in dist assert dist["output"] == "distribution_domain_name" def test_microservice_outputs_present(self): from core.contract_resolver import resolve stack = resolve(str(ROOT / "contracts/microservice.yml"), str(ROOT)) assert "outputs" in stack, "stack.outputs must be present for L2 modules (P1-7)" assert "lb_arn" in stack["outputs"] assert "service_arn" in stack["outputs"]