remove dynamo utils and add tests

Author: motola

lambdas/mns_publisher/poetry.lock

@@ -23,3 +23,8 @@ cache = "^1.0.3"
 [build-system]
 requires = ["poetry-core >= 1.5.0"]
 build-backend = "poetry.core.masonry.api"
+
+[dependency-groups]
+dev = [
+    "responses (>=0.26.0,<0.27.0)"
+]

lambdas/mns_publisher/pyproject.toml

@@ -1,14 +1,15 @@
+import json
 import os
 import uuid
 from datetime import datetime
+from typing import Any
 
 from aws_lambda_typing.events.sqs import SQSMessage
 
 from common.api_clients.get_pds_details import pds_get_patient_details
 from common.clients import logger
 from common.get_service_url import get_service_url
-from constants import IMMUNISATION_TYPE, SPEC_VERSION, MnsNotificationPayload
-from sqs_dynamo_utils import extract_sqs_imms_data
+from constants import DYNAMO_DB_TYPE_DESCRIPTORS, IMMUNISATION_TYPE, SPEC_VERSION, MnsNotificationPayload
 
 IMMUNIZATION_ENV = os.getenv("IMMUNIZATION_ENV")
 IMMUNIZATION_BASE_PATH = os.getenv("IMMUNIZATION_BASE_PATH")
@@ -18,28 +19,37 @@ def create_mns_notification(sqs_event: SQSMessage) -> MnsNotificationPayload:
     """Create a notification payload for MNS."""
     immunisation_url = get_service_url(IMMUNIZATION_ENV, IMMUNIZATION_BASE_PATH)
 
-    # Simple, direct extraction
-    imms_data = extract_sqs_imms_data(sqs_event)
+    body = json.loads(sqs_event.get("body", "{}"))
+    new_image = body.get("dynamodb", {}).get("NewImage", {})
+    imms_id = _unwrap_dynamodb_value(new_image.get("ImmsID", {}))
+    supplier_system = _unwrap_dynamodb_value(new_image.get("SupplierSystem", {}))
+    vaccine_type = _unwrap_dynamodb_value(new_image.get("VaccineType", {}))
+    operation = _unwrap_dynamodb_value(new_image.get("Operation", {}))
 
-    patient_age = calculate_age_at_vaccination(imms_data["person_dob"], imms_data["date_and_time"])
+    imms_map = new_image.get("Imms", {}).get("M", {})
+    nhs_number = _unwrap_dynamodb_value(imms_map.get("NHS_NUMBER", {}))
+    person_dob = _unwrap_dynamodb_value(imms_map.get("PERSON_DOB", {}))
+    date_and_time = _unwrap_dynamodb_value(imms_map.get("DATE_AND_TIME", {}))
+    site_code = _unwrap_dynamodb_value(imms_map.get("SITE_CODE", {}))
 
-    gp_ods_code = get_practitioner_details_from_pds(imms_data["nhs_number"])
+    patient_age = calculate_age_at_vaccination(person_dob, date_and_time)
+    gp_ods_code = get_practitioner_details_from_pds(nhs_number)
 
     return {
         "specversion": SPEC_VERSION,
         "id": str(uuid.uuid4()),
         "source": immunisation_url,
         "type": IMMUNISATION_TYPE,
-        "time": imms_data["date_and_time"],
-        "subject": imms_data["nhs_number"],
-        "dataref": f"{immunisation_url}/Immunization/{imms_data['imms_id']}",
+        "time": date_and_time,
+        "subject": nhs_number,
+        "dataref": f"{immunisation_url}/Immunization/{imms_id}",
         "filtering": {
             "generalpractitioner": gp_ods_code,
-            "sourceorganisation": imms_data["site_code"],
-            "sourceapplication": imms_data["supplier_system"],
+            "sourceorganisation": site_code,
+            "sourceapplication": supplier_system,
             "subjectage": str(patient_age),
-            "immunisationtype": imms_data["vaccine_type"],
-            "action": imms_data["operation"],
+            "immunisationtype": vaccine_type,
+            "action": operation,
         },
     }
 
@@ -95,3 +105,24 @@ def get_practitioner_details_from_pds(nhs_number: str) -> str | None:
             return None
 
     return gp_ods_code
+
+
+def _unwrap_dynamodb_value(value: dict) -> Any:
+    """
+    Unwrap DynamoDB type descriptor to get the actual value.
+    DynamoDB types: S (String), N (Number), BOOL, M (Map), L (List), NULL
+    """
+    if not isinstance(value, dict):
+        return value
+
+    # DynamoDB type descriptors
+    if "NULL" in value:
+        return None
+
+    # Check other DynamoDB types
+    for key in DYNAMO_DB_TYPE_DESCRIPTORS:
+        if key in value:
+            return value[key]
+
+    # Not a DynamoDB type, return as-is
+    return value

lambdas/mns_publisher/src/create_notification.py

@@ -27,7 +27,7 @@ def process_records(records: list[SQSMessage]) -> dict[str, list]:
         except Exception:
             message_id = record.get("messageId", "unknown")
             batch_item_failures.append({"itemIdentifier": message_id})
-            logger.exception("Failed to process record", trace_id={"message_id": message_id})
+            logger.exception("Failed to process record", extra={"message_id": message_id})
 
     if batch_item_failures:
         logger.warning(f"Batch completed with {len(batch_item_failures)} failures")
@@ -37,7 +37,7 @@ def process_records(records: list[SQSMessage]) -> dict[str, list]:
     return {"batchItemFailures": batch_item_failures}
 
 
-def process_record(record: SQSMessage, mns_service: MnsService) -> dict | None:
+def process_record(record: SQSMessage, mns_service: MnsService) -> None:
     """
     Process a single SQS record.
     Args:
@@ -48,13 +48,12 @@ def process_record(record: SQSMessage, mns_service: MnsService) -> dict | None:
     message_id, immunisation_id = extract_trace_ids(record)
     notification_id = None
 
-    # Create notification payload
     mns_notification_payload = create_mns_notification(record)
     notification_id = mns_notification_payload.get("id")
     action_flag = mns_notification_payload.get("filtering", {}).get("action")
     logger.info(
         "Processing message",
-        trace_ids={
+        extra={
             "notification_id": notification_id,
             "message_id": message_id,
             "immunisation_id": immunisation_id,
@@ -63,7 +62,7 @@ def process_record(record: SQSMessage, mns_service: MnsService) -> dict | None:
     )
 
     mns_service.publish_notification(mns_notification_payload)
-    logger.info("Successfully created MNS notification", trace_ids={"mns_notification_id": notification_id})
+    logger.info("Successfully created MNS notification", extra={"mns_notification_id": notification_id})
 
     return None
 

lambdas/mns_publisher/src/process_records.py

@@ -5,6 +5,7 @@
 
 from constants import IMMUNISATION_TYPE, SPEC_VERSION
 from create_notification import (
+    _unwrap_dynamodb_value,
     calculate_age_at_vaccination,
     create_mns_notification,
     get_practitioner_details_from_pds,
@@ -346,5 +347,45 @@ def test_get_practitioner_pds_exception(self, mock_logger, mock_pds_get):
         self.assertEqual(str(context.exception), "PDS API error")
 
 
+class TestUnwrapDynamodbValue(unittest.TestCase):
+    """Tests for _unwrap_dynamodb_value helper function."""
+
+    def test_unwrap_string_type(self):
+        """Test unwrapping DynamoDB String type."""
+        value = {"S": "test-value"}
+        result = _unwrap_dynamodb_value(value)
+        self.assertEqual(result, "test-value")
+
+    def test_unwrap_number_type(self):
+        """Test unwrapping DynamoDB Number type."""
+        value = {"N": "123"}
+        result = _unwrap_dynamodb_value(value)
+        self.assertEqual(result, "123")
+
+    def test_unwrap_boolean_type(self):
+        """Test unwrapping DynamoDB Boolean type."""
+        value = {"BOOL": True}
+        result = _unwrap_dynamodb_value(value)
+        self.assertTrue(result)
+
+    def test_unwrap_null_type(self):
+        """Test unwrapping DynamoDB NULL type."""
+        value = {"NULL": True}
+        result = _unwrap_dynamodb_value(value)
+        self.assertIsNone(result)
+
+    def test_unwrap_map_type(self):
+        """Test unwrapping DynamoDB Map type."""
+        value = {"M": {"key": {"S": "value"}}}
+        result = _unwrap_dynamodb_value(value)
+        self.assertEqual(result, {"key": {"S": "value"}})
+
+    def test_unwrap_list_type(self):
+        """Test unwrapping DynamoDB List type."""
+        value = {"L": [{"S": "item1"}, {"S": "item2"}]}
+        result = _unwrap_dynamodb_value(value)
+        self.assertEqual(result, [{"S": "item1"}, {"S": "item2"}])
+
+
 if __name__ == "__main__":
     unittest.main()

lambdas/mns_publisher/src/sqs_dynamo_utils.py

@@ -5,6 +5,7 @@
 
 from lambda_handler import lambda_handler
 from process_records import extract_trace_ids, process_record, process_records
+from test_utils import load_sample_sqs_event
 
 
 class TestExtractTraceIds(unittest.TestCase):
@@ -13,14 +14,7 @@ class TestExtractTraceIds(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
         """Load the sample SQS event once for all tests."""
-        sample_event_path = Path(__file__).parent.parent / "tests/sqs_event.json"
-        with open(sample_event_path, "r") as f:
-            raw_event = json.load(f)
-
-        if isinstance(raw_event.get("body"), dict):
-            raw_event["body"] = json.dumps(raw_event["body"])
-
-        cls.sample_sqs_event = raw_event
+        cls.sample_sqs_event = load_sample_sqs_event()
 
     def test_extract_trace_ids_success_from_real_payload(self):
         """Test successful extraction using real SQS event structure."""
@@ -72,14 +66,7 @@ class TestProcessRecord(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
         """Load the sample SQS event once for all tests."""
-        sample_event_path = Path(__file__).parent.parent / "tests/sqs_event.json"
-        with open(sample_event_path, "r") as f:
-            raw_event = json.load(f)
-
-        if isinstance(raw_event.get("body"), dict):
-            raw_event["body"] = json.dumps(raw_event["body"])
-
-        cls.sample_sqs_record = raw_event
+        cls.sample_sqs_record = load_sample_sqs_event()
 
     def setUp(self):
         """Set up test fixtures."""
@@ -181,7 +168,7 @@ def test_process_records_partial_failure(self, mock_process_record, mock_get_mns
 
         self.assertEqual(len(result["batchItemFailures"]), 1)
         self.assertEqual(result["batchItemFailures"][0]["itemIdentifier"], "msg-456")
-        mock_logger.warning.assert_called_with("Batch completed with 1 failures")
+        mock_logger.exception.assert_called_once()
 
     @patch("process_records.logger")
     @patch("process_records.get_mns_service")
@@ -219,14 +206,7 @@ class TestLambdaHandler(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
         """Load the sample SQS event once for all tests."""
-        sample_event_path = Path(__file__).parent.parent / "tests/sqs_event.json"
-        with open(sample_event_path, "r") as f:
-            raw_event = json.load(f)
-
-        if isinstance(raw_event.get("body"), dict):
-            raw_event["body"] = json.dumps(raw_event["body"])
-
-        cls.sample_sqs_record = raw_event
+        cls.sample_sqs_record = load_sample_sqs_event()
 
     @patch("lambda_handler.process_records")
     def test_lambda_handler_all_success(self, mock_process_records):