tests work

nhsdevws · nhsdevws · commit c0c16286da7f · 2025-09-24T16:02:19.000+01:00
diff --git a/ack_backend/src/ack_processor.py b/ack_backend/src/ack_processor.py
@@ -4,7 +4,6 @@
 from logging_decorators import ack_lambda_handler_logging_decorator
 from update_ack_file import update_ack_file
 from convert_message_to_ack_row import convert_message_to_ack_row
-from clients import logger
 
 
 @ack_lambda_handler_logging_decorator
@@ -14,7 +13,6 @@ def lambda_handler(event, context):
     For each record: each message in the array of messages is converted to an ack row,
     then all of the ack rows for that array of messages are uploaded to the ack file in one go.
     """
-    logger.info("SAW DEBUG ack_processor.lambda_handler")
     if not event.get("Records"):
         raise ValueError("Error in ack_processor_lambda_handler: No records found in the event")
 
@@ -25,10 +23,8 @@ def lambda_handler(event, context):
     ack_data_rows = []
 
     for i, record in enumerate(event["Records"]):
-        logger.info(f"SAW DEBUG record {i}")
         try:
             incoming_message_body = json.loads(record["body"])
-            logger.info(f"SAW DEBUG incoming_message_body {i}: {incoming_message_body}")
         except Exception as body_json_error:
             raise ValueError("Could not load incoming message body") from body_json_error
 
@@ -44,8 +40,6 @@ def lambda_handler(event, context):
             created_at_formatted_string = incoming_message_body[0].get("created_at_formatted_string")
 
         for message in incoming_message_body:
-            msg = message if message else 'EMPTY MESSAGE'
-            logger.info(f"SAW DEBUG message {i}: {msg}")
             ack_data_rows.append(convert_message_to_ack_row(message, created_at_formatted_string))
 
     update_ack_file(file_key, message_id, supplier, vaccine_type, created_at_formatted_string, ack_data_rows)
diff --git a/ack_backend/src/convert_message_to_ack_row.py b/ack_backend/src/convert_message_to_ack_row.py
@@ -3,7 +3,6 @@
 from typing import Union
 from logging_decorators import convert_message_to_ack_row_logging_decorator
 from update_ack_file import create_ack_data
-from clients import logger
 
 def get_error_message_for_ack_file(message_diagnostics) -> Union[None, str]:
     """Determines and returns the error message to be displayed in the ack file"""
@@ -26,8 +25,6 @@ def convert_message_to_ack_row(message, created_at_formatted_string):
     A value error is raised if the file_key or created_at_formatted_string for the message do not match the
     expected values.
     """
-    logger.info("SAW DEBUG convert_message_to_ack_row")
-    logger.info(f"SAW DEBUG convert_message_to_ack_row: {message}")
     diagnostics = message.get("diagnostics")
     return create_ack_data(
         created_at_formatted_string=created_at_formatted_string,
diff --git a/ack_backend/src/update_ack_file.py b/ack_backend/src/update_ack_file.py
@@ -20,8 +20,6 @@ def create_ack_data(
 ) -> dict:
     """Returns a dictionary containing the ack headers as keys, along with the relevant values."""
     # Pack multi-line diagnostics down to single line (because Imms API diagnostics may be multi-line)
-    logger.info("SAW DEBUG create_ack_data")
-    logger.info(f"SAW DEBUG create_ack_data: {diagnostics}")
     diagnostics = (
         " ".join(diagnostics.replace("\r", " ").replace("\n", " ").replace("\t", " ").replace("\xa0", " ").split())
         if diagnostics is not None
diff --git a/batch_processor_filter/src/batch_audit_repository.py b/batch_processor_filter/src/batch_audit_repository.py
@@ -42,19 +42,10 @@ def is_event_processing_or_failed_for_supplier_and_vacc_type(self, supplier: str
         return False
 
     def update_status(self, message_id: str, updated_status: str) -> None:
-        try:
-            print(f"Updating status for message_id: {message_id} to {updated_status}")
-            print(f"Table name: {AUDIT_TABLE_NAME}")
-            print(f"Region: {REGION_NAME}")
-            print(f"Batch audit table: {self._batch_audit_table.table_name}")
-            self._batch_audit_table.update_item(
-                Key={AuditTableKeys.MESSAGE_ID: message_id},
-                UpdateExpression="SET #status = :status",
-                ExpressionAttributeNames={"#status": "status"},
-                ExpressionAttributeValues={":status": updated_status},
-                ConditionExpression="attribute_exists(message_id)"
-            )
-            print(f"Successfully updated status for message_id: {message_id} to {updated_status}")
-        except Exception as e:
-            print(f"Error updating status for message_id: {message_id} to {updated_status}: {e}")
-            raise
+        self._batch_audit_table.update_item(
+            Key={AuditTableKeys.MESSAGE_ID: message_id},
+            UpdateExpression="SET #status = :status",
+            ExpressionAttributeNames={"#status": "status"},
+            ExpressionAttributeValues={":status": updated_status},
+            ConditionExpression="attribute_exists(message_id)"
+        )
diff --git a/batch_processor_filter/src/batch_processor_filter_service.py b/batch_processor_filter/src/batch_processor_filter_service.py
@@ -27,57 +27,40 @@ def _is_duplicate_file(self, file_key: str) -> bool:
         return self._batch_audit_repository.is_duplicate_file(file_key)
 
     def apply_filter(self, batch_file_created_event: BatchFileCreatedEvent) -> None:
-        try:
-            filename = batch_file_created_event["filename"]
-            message_id = batch_file_created_event["message_id"]
-            supplier = batch_file_created_event["supplier"]
-            vaccine_type = batch_file_created_event["vaccine_type"]
+        filename = batch_file_created_event["filename"]
+        message_id = batch_file_created_event["message_id"]
+        supplier = batch_file_created_event["supplier"]
+        vaccine_type = batch_file_created_event["vaccine_type"]
 
-            # debug
-            print("SAW ----------- apply_filter DEBUG -------------")
-            print(f"filename: {filename}  ")
-            print(f"message_id: {message_id}  ")
-            print(f"supplier: {supplier}  ")
-            print(f"vaccine_type: {vaccine_type}  ")
+        logger.info("Received batch file event for filename: %s with message id: %s", filename, message_id)
 
-            print("apply_filter...checking for duplicate file...")
-            if self._is_duplicate_file(filename):
-                print("apply_filter...duplicate file found")
-                # Mark as processed and return without error so next event will be picked up from queue
-                logger.error("A duplicate file has already been processed. Filename: %s", filename)
-                self._batch_audit_repository.update_status(
-                    message_id,
-                    f"{FileStatus.NOT_PROCESSED} - {FileNotProcessedReason.DUPLICATE}"
-                )
-                self._batch_file_repo.upload_failure_ack(batch_file_created_event)
-                self._batch_file_repo.move_source_file_to_archive(filename)
-                return
+        if self._is_duplicate_file(filename):
+            # Mark as processed and return without error so next event will be picked up from queue
+            logger.error("A duplicate file has already been processed. Filename: %s", filename)
+            self._batch_audit_repository.update_status(
+                message_id,
+                f"{FileStatus.NOT_PROCESSED} - {FileNotProcessedReason.DUPLICATE}"
+            )
+            self._batch_file_repo.upload_failure_ack(batch_file_created_event)
+            self._batch_file_repo.move_source_file_to_archive(filename)
+            return
 
-            print("apply_filter...check for event already processing for supplier and vacc type...")
-            if self._batch_audit_repository.is_event_processing_or_failed_for_supplier_and_vacc_type(
-                supplier,
-                vaccine_type
-            ):
-                print("apply_filter...event already processing for supplier and vacc type found")
-                # Raise error so event is returned to queue and retried again later
-                logger.info("Batch event already processing for supplier and vacc type. Filename: %s", filename)
-                raise EventAlreadyProcessingForSupplierAndVaccTypeError(f"Batch event already processing for supplier: "
-                                                                        f"{supplier} and vacc type: {vaccine_type}")
+        if self._batch_audit_repository.is_event_processing_or_failed_for_supplier_and_vacc_type(
+            supplier,
+            vaccine_type
+        ):
+            # Raise error so event is returned to queue and retried again later
+            logger.info("Batch event already processing for supplier and vacc type. Filename: %s", filename)
+            raise EventAlreadyProcessingForSupplierAndVaccTypeError(f"Batch event already processing for supplier: "
+                                                                    f"{supplier} and vacc type: {vaccine_type}")
 
-            print("apply_filter...forwarding file for processing...")
-            self._queue_client.send_message(
-                QueueUrl=QUEUE_URL,
-                MessageBody=json.dumps(batch_file_created_event),
-                MessageGroupId=f"{supplier}_{vaccine_type}"
-            )
-            print("apply_filter...updating status to processing...")
-            self._batch_audit_repository.update_status(message_id, FileStatus.PROCESSING)
+        self._batch_audit_repository.update_status(message_id, FileStatus.PROCESSING)
+        self._queue_client.send_message(
+            QueueUrl=QUEUE_URL,
+            MessageBody=json.dumps(batch_file_created_event),
+            MessageGroupId=f"{supplier}_{vaccine_type}"
+        )
 
-            print("apply_filter...sending log to firehose...")
-            successful_log_message = f"File forwarded for processing by ECS. Filename: {filename}"
-            logger.info(successful_log_message)
-            send_log_to_firehose({**batch_file_created_event, "message": successful_log_message})
-            print("apply_filter...done")
-        except Exception as ex:
-            logger.error("Error in batch processor filter service: %s", str(ex))
-            raise ex
+        successful_log_message = f"File forwarded for processing by ECS. Filename: {filename}"
+        logger.info(successful_log_message)
+        send_log_to_firehose({**batch_file_created_event, "message": successful_log_message})
diff --git a/e2e_batch/Makefile b/e2e_batch/Makefile
@@ -1,4 +1,4 @@
 -include .env
 
-run-immunization-batch:
-	ENVIRONMENT=$(environment) poetry run python -m unittest -v -c
+test:
+	ENVIRONMENT=$(ENVIRONMENT) poetry run python -m unittest -v -c
diff --git a/e2e_batch/README.md b/e2e_batch/README.md
@@ -0,0 +1,35 @@
+# End-to-End Batch Test Suite (test_e2e_batch.py)
+
+This test suite provides automated end-to-end (E2E) testing for the Immunisation FHIR API batch processing pipeline. It verifies that batch file submissions are correctly processed, acknowledged, and validated across the system.
+
+## Overview
+- Framework: Python unittest
+- Purpose: Simulate real-world batch file submissions, poll for acknowledgements, and validate processing results.
+- Test Scenarios: Defined in the scenarios module and enabled in setUp().
+- Key Features:
+- - Uploads test batch files to S3.
+- - Waits for and validates ACK (acknowledgement) files.
+- - Cleans up SQS queues and test artifacts after each run.
+
+## Test Flow
+1. Setup (setUp)
+- Loads and enables a set of test scenarios.
+- Prepares test data for batch submission.
+2. Test Execution (test_batch_submission)
+- Uploads ALL enabled test files to S3.
+- Polls for ALL ACK responses and forwarded files.
+- Validates the content and structure of ACK files.
+3. Teardown (tearDown)
+- Cleans up SQS queues and any generated test files.
+
+## Key Functions
+- send_files(tests): Uploads enabled test files to the S3 input bucket.
+- poll_for_responses(tests, max_timeout): Polls for ACKs and processed files, with a timeout.
+- validate_responses(tests): Validates the content of ACK files and checks for expected outcomes.
+
+## How to Run
+1. Ensure all dependencies and environment variables are set (see project root README).
+2. Run tests from vscode debugger or from makefile using 
+```
+make test
+```
diff --git a/e2e_batch/clients.py b/e2e_batch/clients.py
@@ -3,7 +3,10 @@
 """
 
 import logging
-from constants import (environment, REGION)
+from constants import (
+    environment, REGION,
+    batch_fifo_queue_name, ack_metadata_queue_name, audit_table_name
+    )
 from boto3 import client as boto3_client, resource as boto3_resource
 
 
@@ -16,10 +19,9 @@
 sqs_client = boto3_client('sqs', region_name=REGION)
 events_table_name = f"imms-{environment}-imms-events"
 events_table = dynamodb.Table(events_table_name)
-audit_table_name = f"immunisation-batch-{environment}-audit-table"
 audit_table = dynamodb.Table(audit_table_name)
-batch_fifo_queue_name = f"imms-{environment}-batch-file-created-queue.fifo"
 batch_fifo_queue_url = sqs_client.get_queue_url(QueueName=batch_fifo_queue_name)['QueueUrl']
+ack_metadata_queue_url = sqs_client.get_queue_url(QueueName=ack_metadata_queue_name)['QueueUrl']
 # Logger
 logging.basicConfig(level="INFO")
 logger = logging.getLogger()
diff --git a/e2e_batch/constants.py b/e2e_batch/constants.py
@@ -12,11 +12,15 @@
 POST_VALIDATION_ERROR = "Validation errors: contained[?(@.resourceType=='Patient')].name[0].given is a mandatory field"
 DUPLICATE = "The provided identifier:"
 ACK_PREFIX = "ack/"
+TEMP_ACK_PREFIX = "TempAck/"
 HEADER_RESPONSE_CODE_COLUMN = "HEADER_RESPONSE_CODE"
 FILE_NAME_VAL_ERROR = "Infrastructure Level Response Value - Processing Error"
 CONFIG_BUCKET = "imms-internal-dev-supplier-config"
 PERMISSIONS_CONFIG_FILE_KEY = "permissions_config.json"
 RAVS_URI = "https://www.ravs.england.nhs.uk/"
+batch_fifo_queue_name = f"imms-{environment}-batch-file-created-queue.fifo"
+ack_metadata_queue_name = f"imms-{environment}-ack-metadata-queue.fifo"
+audit_table_name = f"immunisation-batch-{environment}-audit-table"
 
 
 class EventName():
@@ -40,6 +44,38 @@ class ActionFlag():
     NONE = "NONE"
 
 
+class InfResult:
+    SUCCESS = "Success"
+    PARTIAL_SUCCESS = "Partial Success"
+    FATAL_ERROR = "Fatal Error"
+
+
+class BusRowResult():
+    SUCCESS = "OK"
+    FATAL_ERROR = "Fatal Error"
+    IMMS_NOT_FOUND = "Immunization resource does not exist"
+    NONE = "NONE"
+
+
+class OperationOutcome:
+    IMMS_NOT_FOUND = "Immunization resource does not exist"
+    TEST = "TEST"
+
+
+class OpMsgs:
+    VALIDATION_ERROR = "Validation errors"
+    MISSING_MANDATORY_FIELD = "is a mandatory field"
+    DOSE_QUANTITY_NOT_NUMBER = "doseQuantity.value must be a number"
+    IMM_NOT_EXIST = "Immunization resource does not exist"
+    IDENTIFIER_PROVIDED = "The provided identifier:"
+    INVALID_DATE_FORMAT = "is not in the correct format"
+
+
+class DestinationType:
+    INF = ACK_PREFIX
+    BUS = FORWARDEDFILE_PREFIX
+
+
 class ActionSequence():
     def __init__(self, desc: str, actions: list[ActionFlag], outcome: ActionFlag = None):
         self.actions = actions
@@ -62,7 +98,7 @@ class TestSet():
     UPDATE_FAIL = ActionSequence("Update without Create. Fail", [ActionFlag.UPDATE], outcome=ActionFlag.NONE)
 
 
-def create_row(unique_id, dose_amount, action_flag: ActionFlag, header, inject_char=None):
+def create_row(unique_id, dose_amount, action_flag: str, header, inject_char=None):
     """Helper function to create a single row with the specified UNIQUE_ID and ACTION_FLAG."""
 
     name = "James" if not inject_char else b'Jam\xe9s'
diff --git a/e2e_batch/scenarios.py b/e2e_batch/scenarios.py
diff --git a/e2e_batch/test_e2e_batch.py b/e2e_batch/test_e2e_batch.py
diff --git a/e2e_batch/utils.py b/e2e_batch/utils.py
diff --git a/e2e_batch/vax_suppliers.py b/e2e_batch/vax_suppliers.py
