create temporary sqs queue for mns publish notification

Author: motola

infrastructure/instance/.terraform.lock.hcl

@@ -11,6 +11,7 @@ module "mns_publisher" {
   imms_base_path                     = strcontains(var.sub_environment, "pr-") ? "immunisation-fhir-api/FHIR/R4-${var.sub_environment}" : "immunisation-fhir-api/FHIR/R4"
   lambda_kms_encryption_key_arn      = data.aws_kms_key.existing_lambda_encryption_key.arn
   mns_publisher_resource_name_prefix = "${local.resource_scope}-mns-outbound-events"
+  mns_test_notifcation_name_prefix   = "${local.resource_scope}-mns-test-notification"
   secrets_manager_policy_path        = "${local.policy_path}/secret_manager.json"
   account_id                         = data.aws_caller_identity.current.account_id
   pds_environment                    = var.pds_environment

infrastructure/instance/mns_publisher.tf

@@ -115,4 +115,4 @@ resource "aws_pipes_pipe" "mns_outbound_events" {
       log_group_arn = aws_cloudwatch_log_group.mns_outbound_events_eb_pipe.arn
     }
   }
-}
+}

infrastructure/instance/modules/mns_publisher/mns_outbound_events_eb_pipe.tf

@@ -192,6 +192,7 @@ resource "aws_lambda_function" "mns_publisher_lambda" {
   environment {
     variables = {
       SPLUNK_FIREHOSE_NAME   = var.splunk_firehose_stream_name
+      MNS_TEST_QUEUE_URL     = aws_sqs_queue.mns_test_notification.url
       IMMUNIZATION_ENV       = var.resource_scope,
       IMMUNIZATION_BASE_PATH = var.imms_base_path
       PDS_ENV                = var.pds_environment

infrastructure/instance/modules/mns_publisher/mns_publisher_lambda.tf

@@ -0,0 +1,33 @@
+# Main queue for MNS notification testing
+resource "aws_sqs_queue" "mns_test_notification" {
+  name                       = "${var.mns_test_notifcation_name_prefix}-queue"
+  fifo_queue                 = false
+  kms_master_key_id          = aws_kms_key.mns_outbound_events.arn
+  visibility_timeout_seconds = 300
+}
+
+
+data "aws_iam_policy_document" "mns_test_notification_sqs_policy" {
+  statement {
+    sid    = "mns-test-notification-allow-lambda-access"
+    effect = "Allow"
+
+    principals {
+      type        = "AWS"
+      identifiers = [aws_iam_role.mns_publisher_lambda_exec_role.arn]
+    }
+
+    actions = [
+      "sqs:SendMessage",
+    ]
+
+    resources = [
+      aws_sqs_queue.mns_test_notification.arn
+    ]
+  }
+}
+
+resource "aws_sqs_queue_policy" "mns_test_notification_sqs" {
+  queue_url = aws_sqs_queue.mns_test_notification.id
+  policy    = data.aws_iam_policy_document.mns_test_notification_sqs_policy.json
+}

infrastructure/instance/modules/mns_publisher/sqs_test_publish_mns.tf

@@ -102,4 +102,9 @@ variable "account_id" {
 variable "secrets_manager_policy_path" {
   type        = string
   description = "Path to the IAM policy JSON template for Secrets Manager access (e.g., ./policies/secret_manager.json)."
+}
+
+variable "mns_test_notifcation_name_prefix" {
+  type        = string
+  description = "The prefix for the name of resources for testing mns notification"
 }

infrastructure/instance/modules/mns_publisher/variables.tf

@@ -0,0 +1,29 @@
+import json
+import os
+
+import boto3
+
+from common.clients import logger
+
+MNS_TEST_QUEUE_URL = os.getenv("MNS_TEST_QUEUE_URL")
+sqs_client = boto3.client("sqs", region_name="eu-west-2")
+
+
+def send_notification_to_test_queue(mns_payload: dict) -> None:
+    """
+    Send MNS notification payload to test SQS queue as fallback.
+    Args: payload: MNS notification payload
+    """
+    if not MNS_TEST_QUEUE_URL:
+        logger.error("MNS_TEST_QUEUE_URL environment variable is not set")
+        return
+
+    try:
+        response = sqs_client.send_message(
+            QueueUrl=MNS_TEST_QUEUE_URL,
+            MessageBody=json.dumps(mns_payload),
+            MessageAttributes={"source": {"StringValue": "mns-publisher-lambda", "DataType": "String"}},
+        )
+        logger.info("Successfully sent notification to test queue", extra={"message_id": response["MessageId"]})
+    except Exception:
+        logger.exception("Failed to send to test SQS queue")

lambdas/mns_publisher/src/mns_test_queue.py

@@ -8,6 +8,7 @@
 from common.api_clients.mns_setup import get_mns_service
 from common.clients import logger
 from create_notification import create_mns_notification
+from mns_test_queue import send_notification_to_test_queue
 
 mns_env = os.getenv("MNS_ENV", "int")
 
@@ -62,10 +63,12 @@ def process_record(record: SQSMessage, mns_service: MnsService) -> None:
         },
     )
 
-    mns_service.publish_notification(mns_notification_payload)
-    logger.info("Successfully created MNS notification", extra={"mns_notification_id": notification_id})
-
-    return None
+    try:
+        mns_service.publish_notification(mns_notification_payload)
+        logger.info("Successfully created MNS notification", extra={"mns_notification_id": notification_id})
+    except Exception:
+        send_notification_to_test_queue(mns_notification_payload)
+        raise
 
 
 def extract_trace_ids(record: SQSMessage) -> Tuple[str, str | None]:

lambdas/mns_publisher/src/process_records.py

@@ -0,0 +1,59 @@
+import json
+import unittest
+from unittest.mock import patch
+
+import boto3
+from moto import mock_aws
+
+from mns_test_queue import send_notification_to_test_queue
+
+
+@mock_aws
+class TestMnsSqsQueue(unittest.TestCase):
+    def setUp(self):
+        self.sqs = boto3.client("sqs", region_name="eu-west-2")
+        response = self.sqs.create_queue(QueueName="mns-test-notifications-int")
+        self.queue_url = response["QueueUrl"]
+
+        self.mns_payload = {
+            "specversion": "1.0",
+            "id": "236a1d4a-5d69-4fa9-9c7f-e72bf505aa5b",
+            "source": "https://int.api.service.nhs.uk/immunisation-fhir-api",
+            "type": "imms-vaccinations-2",
+            "time": "20260212T174437+00:00",
+            "subject": "9481152782",
+            "dataref": "https://int.api.service.nhs.uk/immunisation-fhir-api/Immunization/d058014c-b0fd-4471-8db9-3316175eb825",
+            "filtering": {
+                "generalpractitioner": "Y12345",
+                "sourceorganisation": "B0C4P",
+                "sourceapplication": "TPP",
+                "subjectage": 21,
+                "immunisationtype": "HIB",
+                "action": "CREATE",
+            },
+        }
+
+    def test_send_notification_to_queue_success(self):
+        """Test successful send to SQS queue with complete payload."""
+        with patch("mns_test_queue.MNS_TEST_QUEUE_URL", self.queue_url):
+            send_notification_to_test_queue(self.mns_payload)
+
+        messages = self.sqs.receive_message(
+            QueueUrl=self.queue_url, MaxNumberOfMessages=1, MessageAttributeNames=["All"]
+        )
+
+        self.assertIn("Messages", messages)
+        self.assertEqual(len(messages["Messages"]), 1)
+
+        # Verify message body
+        mns_payload = json.loads(messages["Messages"][0]["Body"])
+        attributes = messages["Messages"][0]["MessageAttributes"]
+        self.assertEqual(attributes["source"]["StringValue"], "mns-publisher-lambda")
+
+        self.assertEqual(mns_payload["subject"], "9481152782")
+        self.assertEqual(mns_payload["filtering"]["generalpractitioner"], "Y12345")
+        self.assertEqual(mns_payload["filtering"]["sourceorganisation"], "B0C4P")
+        self.assertEqual(mns_payload["filtering"]["sourceapplication"], "TPP")
+        self.assertEqual(mns_payload["filtering"]["immunisationtype"], "HIB")
+        self.assertEqual(mns_payload["filtering"]["action"], "CREATE")
+        self.assertEqual(mns_payload["filtering"]["subjectage"], 21)