"""Unit tests for core/lambda/contract_ingestor.py. The source file lives at ``core/lambda/contract_ingestor.py`` for repo organization, but ``lambda`` is a Python reserved word — so the package path ``core.lambda`` cannot be imported with normal ``import`` syntax. The Lambda runtime packages the handler as a top-level ``contract_ingestor.py`` (handler ``contract_ingestor.lambda_handler``), which is the name the Terraform ``handler`` attribute uses. The tests mirror that by loading the module from its file path under the name ``contract_ingestor``. Uses moto (already a test dependency — see requirements-test.txt) to mock DynamoDB, mirroring the pattern in tests/test_outbox_writer.py. """ import datetime import importlib.util import json import os import sys from pathlib import Path from unittest import mock import pytest sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) # Load core/lambda/contract_ingestor.py as a top-level module named # `contract_ingestor` (the name the Lambda runtime uses). _SOURCE_PATH = Path(__file__).resolve().parent.parent / "core" / "lambda" / "contract_ingestor.py" _spec = importlib.util.spec_from_file_location("contract_ingestor", _SOURCE_PATH) ingestor = importlib.util.module_from_spec(_spec) _spec.loader.exec_module(ingestor) # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture def sample_payload(): return { "consumerRepo": "acdl/consumer-a", "contractId": "contract-001", "contract": {"stack": "s3", "environment": "dev"}, "environment": "dev", "action": "submit_contract", } @pytest.fixture def function_url_event(sample_payload): return {"body": json.dumps(sample_payload)} @pytest.fixture def moto_contracts_table(monkeypatch): """Spin up a moto-backed DynamoDB and point the ingestor at it.""" from moto import mock_aws import boto3 monkeypatch.setenv("AWS_DEFAULT_REGION", "us-east-1") monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing") monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing") with mock_aws(): dyn = boto3.client("dynamodb", region_name="us-east-1") dyn.create_table( TableName="acdl-contracts", KeySchema=[ {"AttributeName": "consumerRepo", "KeyType": "HASH"}, {"AttributeName": "contractId#submittedAt", "KeyType": "RANGE"}, ], AttributeDefinitions=[ {"AttributeName": "consumerRepo", "AttributeType": "S"}, {"AttributeName": "contractId#submittedAt", "AttributeType": "S"}, ], BillingMode="PAY_PER_REQUEST", ) # Reset the cached boto3 clients so the ingestor picks up the moto # session, then yield with moto active. saved_dynamodb = ingestor._dynamodb saved_secrets = ingestor._secrets_client ingestor._dynamodb = None ingestor._secrets_client = None monkeypatch.setattr(ingestor, "TABLE_NAME", "acdl-contracts") yield dyn ingestor._dynamodb = saved_dynamodb ingestor._secrets_client = saved_secrets # --------------------------------------------------------------------------- # submit_contract # --------------------------------------------------------------------------- class TestSubmitContract: def test_submit_contract_writes_correct_pk_sk_attributes(self, moto_contracts_table, sample_payload): result = ingestor._submit_contract(sample_payload) assert result["status"] == "ok" assert result["contractId"] == "contract-001" assert result["action"] == "submit_contract" assert "submittedAt" in result # Verify what landed in DynamoDB. sk = f"contract-001#{result['submittedAt']}" resp = moto_contracts_table.get_item( TableName="acdl-contracts", Key={ "consumerRepo": {"S": "acdl/consumer-a"}, "contractId#submittedAt": {"S": sk}, }, ) assert "Item" in resp item = resp["Item"] assert item["consumerRepo"]["S"] == "acdl/consumer-a" assert item["contractId"]["S"] == "contract-001" assert item["status"]["S"] == "submitted" assert item["environment"]["S"] == "dev" assert item["submittedAt"]["S"] == result["submittedAt"] # The contract attribute holds the full contract object. boto3's # resource API serializes a dict as a DynamoDB Map (type "M"); each # leaf scalar is wrapped in its own type tag. expected_contract = sample_payload["contract"] actual_contract = item["contract"] # The resource API stores scalars inside the map with their own type # tags (e.g. {"S": ...}); unwrap one level for the two known leaves. unwrapped = { k: list(v.values())[0] if isinstance(v, dict) and len(v) == 1 else v for k, v in actual_contract["M"].items() } assert unwrapped == expected_contract def test_submit_contract_sk_contains_contract_id_and_timestamp(self, moto_contracts_table, sample_payload): result = ingestor._submit_contract(sample_payload) sk = f"contract-001#{result['submittedAt']}" # SK format is contractId#ISO8601 assert sk.split("#")[0] == "contract-001" # timestamp parses as ISO 8601 with a Z suffix. ts = sk.split("#", 1)[1] datetime.datetime.strptime(ts, "%Y-%m-%dT%H:%M:%SZ") # --------------------------------------------------------------------------- # report_error stub # --------------------------------------------------------------------------- class TestReportError: def test_report_error_returns_prepared_status(self): payload = { "consumerRepo": "acdl/consumer-a", "contractId": "contract-001", "error": "deploy failed", } result = ingestor._report_error(payload) assert result["status"] == "error_report_prepared" assert result["contractId"] == "contract-001" assert result["action"] == "report_error" def test_report_error_missing_field_raises(self): payload = {"consumerRepo": "acdl/consumer-a"} # missing contractId, error with pytest.raises(ValueError): ingestor._report_error(payload) # --------------------------------------------------------------------------- # lambda_handler wrapper (Function URL event) # --------------------------------------------------------------------------- class TestLambdaHandler: def test_submit_contract_event_returns_200(self, moto_contracts_table, function_url_event): resp = ingestor.lambda_handler(function_url_event, None) assert resp["statusCode"] == 200 body = json.loads(resp["body"]) assert body["status"] == "ok" assert body["contractId"] == "contract-001" assert body["action"] == "submit_contract" def test_body_can_be_dict_not_string(self, moto_contracts_table, sample_payload): # Some test harnesses pass body as a dict already. resp = ingestor.lambda_handler({"body": sample_payload}, None) assert resp["statusCode"] == 200 def test_missing_field_returns_400(self, moto_contracts_table): payload = { "consumerRepo": "acdl/consumer-a", # missing contractId, contract, environment } resp = ingestor.lambda_handler({"body": json.dumps(payload)}, None) assert resp["statusCode"] == 400 body = json.loads(resp["body"]) assert "missing field" in body["error"] def test_missing_required_field_contract(self, moto_contracts_table, sample_payload): del sample_payload["contract"] resp = ingestor.lambda_handler({"body": json.dumps(sample_payload)}, None) assert resp["statusCode"] == 400 assert "contract" in json.loads(resp["body"])["error"] def test_unknown_action_returns_400(self, moto_contracts_table): payload = { "consumerRepo": "acdl/consumer-a", "contractId": "contract-001", "contract": {}, "environment": "dev", "action": "do_something_else", } resp = ingestor.lambda_handler({"body": json.dumps(payload)}, None) assert resp["statusCode"] == 400 body = json.loads(resp["body"]) assert "unknown action" in body["error"] def test_default_action_is_submit_contract(self, moto_contracts_table, sample_payload): del sample_payload["action"] resp = ingestor.lambda_handler({"body": json.dumps(sample_payload)}, None) assert resp["statusCode"] == 200 body = json.loads(resp["body"]) assert body["action"] == "submit_contract" def test_internal_error_returns_500(self, moto_contracts_table, sample_payload): # Force _submit_contract to blow up after passing validation. with mock.patch.object(ingestor, "_submit_contract", side_effect=RuntimeError("boom")): resp = ingestor.lambda_handler( {"body": json.dumps(sample_payload)}, None ) assert resp["statusCode"] == 500 body = json.loads(resp["body"]) assert body["error"] == "boom"