"""Tests for the local emulating adapters (D-092, REQ-113). Verifies the four local adapters and the headline E2E run against the local tier with no cloud credentials: 1. FlatFileOutbox - flat-file DynamoDB outbox emulator 2. LocalEcsEmulator - local ECS Fargate HTTP 200 emulator 3. LocalS3StateBackend - terraform S3 -> local backend rewrite 4. LocalLambdaStub - in-process contract_ingestor invocation 5. run_local_e2e - the full headline E2E against the local tier """ import json import os import sys import tempfile from pathlib import Path import pytest ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(ROOT)) import core.local_emulators as le # noqa: E402 # --------------------------------------------------------------------------- # 1. FlatFileOutbox # --------------------------------------------------------------------------- def test_flat_file_outbox_writes_hash_chained_event(tmp_path): outbox = le.FlatFileOutbox.create(dir=tmp_path) event = { "contractId": "c1", "eventType": "CONFIDENCE_COMPUTED", "ts": "2026-07-27T00:00:00Z", "environment": "dev", "stack": "s1", "score": 0.9, "band": "pass", "prev_event_hash": "GENESIS", } item = outbox.write_event(event) assert item["hash"] assert len(item["hash"]) == 64 # SHA-256 hex assert item["prev_event_hash"] == "GENESIS" events = outbox.read_all() assert len(events) == 1 assert events[0]["hash"] == item["hash"] def test_flat_file_outbox_chain_links_prior_hash(tmp_path): outbox = le.FlatFileOutbox.create(dir=tmp_path) e1 = {"contractId": "c1", "eventType": "E1", "ts": "t1", "environment": "dev", "stack": "s", "score": 0.9, "band": "pass", "prev_event_hash": "GENESIS"} item1 = outbox.write_event(e1) e2 = {"contractId": "c1", "eventType": "E2", "ts": "t2", "environment": "dev", "stack": "s", "score": 0.9, "band": "pass", "prev_event_hash": item1["hash"]} item2 = outbox.write_event(e2) assert item2["prev_event_hash"] == item1["hash"] assert outbox.verify_chain() def test_flat_file_outbox_detects_broken_chain(tmp_path): outbox = le.FlatFileOutbox.create(dir=tmp_path) e1 = {"contractId": "c1", "eventType": "E1", "ts": "t1", "environment": "dev", "stack": "s", "score": 0.9, "band": "pass", "prev_event_hash": "GENESIS"} item1 = outbox.write_event(e1) # Tamper: write a second event claiming the wrong prev hash. e2 = {"contractId": "c1", "eventType": "E2", "ts": "t2", "environment": "dev", "stack": "s", "score": 0.9, "band": "pass", "prev_event_hash": "WRONG"} outbox.write_event(e2) assert outbox.verify_chain() is False def test_flat_file_outbox_resumes_chain_across_instances(tmp_path): outbox1 = le.FlatFileOutbox.create(dir=tmp_path) e1 = {"contractId": "c1", "eventType": "E1", "ts": "t1", "environment": "dev", "stack": "s", "score": 0.9, "band": "pass", "prev_event_hash": "GENESIS"} item1 = outbox1.write_event(e1) # New instance pointing at the same dir must resume from item1's hash. outbox2 = le.FlatFileOutbox.create(dir=tmp_path) assert outbox2._chain_tail_hash == item1["hash"] # --------------------------------------------------------------------------- # 2. LocalEcsEmulator # --------------------------------------------------------------------------- def test_local_ecs_emulator_returns_http_200(): ecs = le.LocalEcsEmulator( service_name="test-svc", service_definition={"desired_count": 1}, ) try: meta = ecs.deploy() assert meta["status"] == "RUNNING" assert meta["endpoint"].startswith("http://127.0.0.1:") ok, status = ecs.health_check(meta["endpoint"]) assert ok is True assert status == 200 finally: ecs.destroy() def test_local_ecs_emulator_destroy_stops_server(): ecs = le.LocalEcsEmulator("svc", {"desired_count": 1}) meta = ecs.deploy() ecs.destroy() # After destroy, the health check must fail (server stopped). ok, status = ecs.health_check(meta["endpoint"], timeout_s=1.0) assert ok is False # --------------------------------------------------------------------------- # 3. LocalS3StateBackend # --------------------------------------------------------------------------- def test_local_s3_backend_rewrites_s3_to_local(tmp_path): backend = le.LocalS3StateBackend.create(dir=tmp_path / "state") tf = tmp_path / "terraform.tf" tf.write_text( 'terraform {\n required_version = ">= 1.9"\n backend "s3" {\n' ' bucket = "nova-tfstate-x"\n key = "spike/s.tfstate"\n' ' region = "us-east-1"\n }\n}\n' ) backend.rewrite_terraform_tf(tf, "test-stack") content = tf.read_text() assert 'backend "local"' in content assert 'backend "s3"' not in content assert "test-stack.tfstate" in content def test_local_s3_backend_state_path_is_unique_per_stack(tmp_path): backend = le.LocalS3StateBackend.create(dir=tmp_path / "state") p1 = backend.state_path("stack-a") p2 = backend.state_path("stack-b") assert p1 != p2 assert p1.name == "stack-a.tfstate" # --------------------------------------------------------------------------- # 4. LocalLambdaStub # --------------------------------------------------------------------------- def test_local_lambda_stub_invokes_contract_ingestor(tmp_path): outbox = le.FlatFileOutbox.create(dir=tmp_path / "outbox") stub = le.LocalLambdaStub(outbox=outbox) result = stub.invoke({ "action": "submit_contract", "consumerRepo": "local-test/consumer", "contractId": "lambda-test", "contract": {"module": "microservice", "environment": "dev"}, "environment": "dev", }) assert result["statusCode"] == 200 body = json.loads(result["body"]) assert "consumerRepo" in body or "contractId" in body def test_local_lambda_stub_rejects_missing_field(tmp_path): outbox = le.FlatFileOutbox.create(dir=tmp_path / "outbox") stub = le.LocalLambdaStub(outbox=outbox) result = stub.invoke({ "action": "submit_contract", "consumerRepo": "local-test/consumer", # contractId intentionally missing "contract": {"module": "microservice", "environment": "dev"}, "environment": "dev", }) assert result["statusCode"] == 400 # --------------------------------------------------------------------------- # 5. run_local_e2e (the headline E2E against the local tier) # --------------------------------------------------------------------------- @pytest.mark.slow def test_run_local_e2e_microservice(): """Headline E2E: contract -> resolver -> adapter -> local S3 backend -> local ECS (HTTP 200) -> flat-file outbox -> local Lambda. No AWS.""" os.environ["NOVA_LOCAL_TIER"] = "1" try: result = le.run_local_e2e("contracts/microservice.yml") finally: os.environ.pop("NOVA_LOCAL_TIER", None); os.environ.pop("ACDL_LOCAL_TIER", None) assert result["tier"] == "local-emulator" assert result["backend"] == "local" assert result["ecs"] is not None assert result["ecs"]["status"] == "RUNNING" assert result["outbox_chain_verified"] is True assert result["lambda_status"] == 200 @pytest.mark.slow def test_run_local_e2e_static_assets(): """Static-assets stack has no ECS service; the local E2E must still complete (ecs=None) and the outbox chain + Lambda stub must pass.""" os.environ["NOVA_LOCAL_TIER"] = "1" try: result = le.run_local_e2e("contracts/static-assets.yml") finally: os.environ.pop("NOVA_LOCAL_TIER", None); os.environ.pop("ACDL_LOCAL_TIER", None) assert result["tier"] == "local-emulator" assert result["ecs"] is None # no ECS service in this stack assert result["outbox_chain_verified"] is True assert result["lambda_status"] == 200 def test_is_local_tier_flag(): assert le.is_local_tier() is False os.environ["NOVA_LOCAL_TIER"] = "1" try: assert le.is_local_tier() is True finally: os.environ.pop("NOVA_LOCAL_TIER", None); os.environ.pop("ACDL_LOCAL_TIER", None) assert le.is_local_tier() is False