check logs to pinpoint pds bad request

Author: motola

lambdas/mns_publisher/src/process_records.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.error(Exception)
+            logger.exception("Failed to process record", trace_id={"message_id": message_id})
 
     if batch_item_failures:
         logger.warning(f"Batch completed with {len(batch_item_failures)} failures")
@@ -62,7 +62,6 @@ def process_record(record: SQSMessage, mns_service: MnsService) -> dict | None:
         },
     )
 
-    # Publish to MNS
     mns_service.publish_notification(mns_notification_payload)
     logger.info("Successfully created MNS notification", trace_ids={"mns_notification_id": notification_id})
 

lambdas/mns_publisher/src/sqs_dynamo_utils.py

@@ -1,6 +1,7 @@
 import json
 from typing import Any
 
+from common.clients import logger
 from constants import DYNAMO_DB_TYPE_DESCRIPTORS, ImmsData
 
 
@@ -18,6 +19,7 @@ def extract_sqs_imms_data(sqs_record: dict) -> ImmsData:
     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", {}))
+    logger.info("Top Level Fields retrieved successfully")
 
     imms_map = new_image.get("Imms", {}).get("M", {})
 

lambdas/mns_publisher/tests/test_create_notification.py

@@ -1,7 +1,6 @@
 import copy
 import json
 import unittest
-from pathlib import Path
 from unittest.mock import MagicMock, patch
 
 from constants import IMMUNISATION_TYPE, SPEC_VERSION
@@ -10,21 +9,7 @@
     create_mns_notification,
     get_practitioner_details_from_pds,
 )
-
-
-def _load_sample_sqs_event() -> dict:
-    """
-    Loads the sample SQS event and normalises body to a JSON string (as SQS delivers it).
-    Expects: lambdas/mns_publisher/tests/sqs_event.json
-    """
-    sample_event_path = Path(__file__).parent / "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"])
-
-    return raw_event
+from test_utils import load_sample_sqs_event
 
 
 class TestCalculateAgeAtVaccination(unittest.TestCase):
@@ -78,7 +63,7 @@ class TestCreateMnsNotification(unittest.TestCase):
 
     @classmethod
     def setUpClass(cls):
-        cls.sample_sqs_event = _load_sample_sqs_event()
+        cls.sample_sqs_event = load_sample_sqs_event()
 
     def setUp(self):
         self.expected_gp_ods_code = "Y12345"

lambdas/mns_publisher/tests/test_sqs_dynamo_utils.py

@@ -0,0 +1,87 @@
+import json
+import unittest
+
+from sqs_dynamo_utils import _unwrap_dynamodb_value, extract_sqs_imms_data
+from test_utils import load_sample_sqs_event
+
+
+class TestExtractSqsImmsData(unittest.TestCase):
+    """
+    Test SQS Event extraction utility
+    """
+
+    @classmethod
+    def setUpClass(cls):
+        cls.sample_sqs_event = load_sample_sqs_event()
+
+    def test_extract_sqs_imms_data(self):
+        result = extract_sqs_imms_data(self.sample_sqs_event)
+
+        self.assertEqual(result["imms_id"], "d058014c-b0fd-4471-8db9-3316175eb825")
+        self.assertEqual(result["supplier_system"], "TPP")
+        self.assertEqual(result["vaccine_type"], "hib")
+        self.assertEqual(result["operation"], "CREATE")
+        self.assertEqual(result["nhs_number"], "9481152782")
+        self.assertEqual(result["person_dob"], "20040609")
+        self.assertEqual(result["date_and_time"], "20260212T17443700")
+        self.assertEqual(result["site_code"], "B0C4P")
+
+    def test_extract_imms_data_field_types(self):
+        """Test that extracted fields are the correct types."""
+        result = extract_sqs_imms_data(self.sample_sqs_event)
+
+        self.assertIsInstance(result["imms_id"], str)
+        self.assertIsInstance(result["supplier_system"], str)
+        self.assertIsInstance(result["vaccine_type"], str)
+        self.assertIsInstance(result["operation"], str)
+        self.assertIsInstance(result["nhs_number"], str)
+        self.assertIsInstance(result["person_dob"], str)
+        self.assertIsInstance(result["date_and_time"], str)
+        self.assertIsInstance(result["site_code"], str)
+
+    def test_extract_imms_data_invalid_json_body(self):
+        """Test extraction when body is invalid JSON."""
+        event = {"body": "invalid json {"}
+
+        with self.assertRaises(json.JSONDecodeError):
+            extract_sqs_imms_data(event)
+
+
+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"}])

lambdas/mns_publisher/tests/test_utils.py

@@ -0,0 +1,17 @@
+import json
+from pathlib import Path
+
+
+def load_sample_sqs_event() -> dict:
+    """
+    Loads the sample SQS event and normalises body to a JSON string (as SQS delivers it).
+    Expects: lambdas/mns_publisher/tests/sqs_event.json
+    """
+    sample_event_path = Path(__file__).parent / "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"])
+
+    return raw_event