debug

Author: nhsdevws

ack_backend/src/ack_processor.py

@@ -4,6 +4,7 @@
 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
@@ -13,7 +14,7 @@ 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")
 
@@ -24,7 +25,7 @@ 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"])
         except Exception as body_json_error:

ack_backend/src/convert_message_to_ack_row.py

@@ -3,7 +3,7 @@
 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,6 +26,8 @@ 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,

ack_backend/src/update_ack_file.py

@@ -20,6 +20,8 @@ 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

e2e_batch/clients.py

@@ -13,10 +13,13 @@
 s3_client = boto3_client("s3", region_name=REGION)
 
 dynamodb = boto3_resource("dynamodb", region_name=REGION)
+sqs_client = boto3_client('sqs', region_name=REGION)
 table_name = f"imms-{environment}-imms-events"
 table = dynamodb.Table(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']
 # Logger
 logging.basicConfig(level="INFO")
 logger = logging.getLogger()

e2e_batch/constants.py

@@ -66,7 +66,6 @@ def create_row(unique_id, dose_amount, action_flag: ActionFlag, header, inject_c
     """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'
-
     return {
         header: "9732928395",
         "PERSON_FORENAME": "PHYLIS",
@@ -79,7 +78,7 @@ def create_row(unique_id, dose_amount, action_flag: ActionFlag, header, inject_c
         "SITE_CODE_TYPE_URI": "https://fhir.nhs.uk/Id/ods-organization-code",
         "UNIQUE_ID": unique_id,
         "UNIQUE_ID_URI": "https://www.ravs.england.nhs.uk/",
-        "ACTION_FLAG": action_flag if action_flag != "NEW" else "CREATE",  # TODO check if this is needed
+        "ACTION_FLAG": action_flag,
         "PERFORMING_PROFESSIONAL_FORENAME": "PHYLIS",
         "PERFORMING_PROFESSIONAL_SURNAME": name,
         "RECORDED_DATE": datetime.now(timezone.utc).strftime("%Y%m%d"),

e2e_batch/test_e2e_batch.py

@@ -34,17 +34,7 @@
 class TestE2EBatch(unittest.TestCase):
 
     def setUp(self):
-        self.seed_datas = [
-            TestData("V0V8L", ActionSet.CREATE),
-            # TestData("8HK48", ActionSet.UPDATE),
-            # TestData("8HA94", ActionSet.DELETE),
-            # TestData("Reinstate", "X26", [NEW, DELETE, UPDATE]),
-            # TestData("Update no Create", "YGM41", [UPDATE], expected=BusRowResult.FATAL_ERROR,
-            #          operation_outcome=OpMsgs.IMM_NOT_EXIST),
-            # TestData("Delete no Create", "YGJ", [DELETE], expected=BusRowResult.FATAL_ERROR,
-            #          operation_outcome=OpMsgs.IMM_NOT_EXIST),
-            # TestData("YGA", ActionSet.CREATE, inject_char=True, name="Create with 1252 char"),
-        ]
+        self.seed_datas = create_seed_data()
 
     def tearDown(self):
         # loop through all files and delete them from s3
@@ -99,31 +89,49 @@ def validate_responses(tests: list[TestData]):
     start_time = time.time()
     count = 0
     expected_count = len(tests) * 2
-    for test in tests:
-        logger.info(f"Validation for Test: {test.name} ")
-        # Validate the ACK file
-        if test.ack_keys[DestinationType.INF]:
-            count += 1
-            inf_ack_content = get_file_content_from_s3(ACK_BUCKET, test.ack_keys[DestinationType.INF])
-        else:
-            logger.error(f"INF ACK file not found for test: {test.name}")
-
-        check_ack_file_content(test.name, inf_ack_content, "Success", None, test.actions)
-        if test.ack_keys[DestinationType.BUS]:
-            count += 1
-            validate_row_count(f"{test.name} - inf", test.file_name, test.ack_keys[DestinationType.BUS])
-            # check row after header
-            bus_ack_content = get_file_content_from_s3(ACK_BUCKET, test.ack_keys[DestinationType.BUS])
-            # loop through each line in the bus ack content
-
-            # sometimes OK and sometimes CREATE
-            check_inf_file_content(f"{test.name} - bus", bus_ack_content, "OK", test.operation_outcome,
-                                   test.getOperations())
+    try:
+        for test in tests:
+            logger.info(f"Validation for Test: {test.name} ")
+            # Validate the ACK file
+            if test.ack_keys[DestinationType.INF]:
+                count += 1
+                inf_ack_content = get_file_content_from_s3(ACK_BUCKET, test.ack_keys[DestinationType.INF])
+                check_ack_file_content(test.name, inf_ack_content, "Success", None, test.action_sequence.actions)
+            else:
+                logger.error(f"INF ACK file not found for test: {test.name}")
+
+            if test.ack_keys[DestinationType.BUS]:
+                count += 1
+                validate_row_count(f"{test.name} - inf", test.file_name, test.ack_keys[DestinationType.BUS])
+                # check row after header
+                bus_ack_content = get_file_content_from_s3(ACK_BUCKET, test.ack_keys[DestinationType.BUS])
+                # loop through each line in the bus ack content
+
+                # sometimes OK and sometimes CREATE
+                check_inf_file_content(f"{test.name} - bus", bus_ack_content, "OK", test.operation_outcome,
+                                       test.getOperations())
+            else:
+                logger.error(f"BUS ACK file not found for test: {test.name}")
+    except Exception as e:
+        logger.error(f"Error during validation: {e}")
+    finally:
+        if count == expected_count:
+            logger.info("All responses subject to validation.")
         else:
-            logger.error(f"BUS ACK file not found for test: {test.name}")
+            logger.error(f"{count} of {expected_count} responses subject to validation.")
+        logger.info(f"Time: {time.time() - start_time:.1f} seconds")
+        assert count == expected_count, f"Only {count} of {expected_count} responses subject to validation."
+
 
-    if count == expected_count:
-        logger.info("All responses subject to validation.")
-    else:
-        logger.error(f"{count} of {expected_count} responses subject to validation.")
-    logger.info(f"Time: {time.time() - start_time:.1f} seconds")
+def create_seed_data() -> list[TestData]:
+    return [
+            # TestData("V0V8L", ActionSet.CREATE),
+            # TestData("8HK48", ActionSet.UPDATE),
+            TestData("8HA94", ActionSet.DELETE),
+            # TestData("Reinstate", "X26", [NEW, DELETE, UPDATE]),
+            # TestData("Update no Create", "YGM41", [UPDATE], expected=BusRowResult.FATAL_ERROR,
+            #          operation_outcome=OpMsgs.IMM_NOT_EXIST),
+            # TestData("Delete no Create", "YGJ", [DELETE], expected=BusRowResult.FATAL_ERROR,
+            #          operation_outcome=OpMsgs.IMM_NOT_EXIST),
+            # TestData("YGA", ActionSet.CREATE, inject_char=True, name="Create with 1252 char"),
+        ]

e2e_batch/utils.py

@@ -10,7 +10,7 @@
 from boto3.dynamodb.conditions import Key
 from io import StringIO
 from datetime import datetime, timezone
-from clients import logger, s3_client, table, audit_table
+from clients import logger, s3_client, table, audit_table, batch_fifo_queue_url, sqs_client
 from errors import AckFileNotFoundError, DynamoDBMismatchError
 from constants import (
     ACK_BUCKET,
@@ -88,7 +88,7 @@ def __init__(self,
 
     def getOperations(self) -> list[Operation]:
         operations = []
-        for action in self.actions:
+        for action in self.action_sequence.actions:
             operations.append(get_operation(action))
         return operations
 
@@ -580,7 +580,7 @@ def generate_csv_files(seed_data_list: list[TestData]) -> list[TestData]:
     """Generate CSV files based on a list of TestData instances."""
     for seed_data in seed_data_list:
         file_name = (generate_csv_file(seed_data, actions=seed_data.action_sequence.actions))
-        logger.info(f"Test file: {file_name}")
+        logger.info(f"Create Test file: {file_name}")
         seed_data.file_name = file_name
     return seed_data_list
 
@@ -590,7 +590,9 @@ def generate_csv_file(seed: TestData, actions: list[ActionFlag]) -> str:
     data = []
     for action in actions:
         unique_id = str(uuid.uuid4())
-        data.append(create_row(unique_id, seed.dose_amount, action, seed.header, seed.inject_char))
+        row = create_row(unique_id, seed.dose_amount, action, seed.header, seed.inject_char)
+        logger.info(f"> Create row ID: {unique_id}, ACTION_FLAG: {action}")
+        data.append(row)
     df = pd.DataFrame(data)
     file_name = get_file_name(seed.vax, seed.ods, seed.version)
     df.to_csv(file_name, index=False, sep="|", quoting=csv.QUOTE_MINIMAL)
@@ -629,6 +631,14 @@ def cleanup(data_list: list[TestData]):
             if ack_key:
                 if not delete_file_from_s3(ACK_BUCKET, ack_key):
                     logger.warning(f"s3 delete fail {ACK_BUCKET}: {ack_key}")
+    # purge batch_fifo_queue_url
+    try:
+        sqs_client.purge_queue(QueueUrl=batch_fifo_queue_url)
+        print("SQS purge successful.")
+    except sqs_client.exceptions.PurgeQueueInProgress:
+        print("SQS purge already in progress. Try again later.")
+    except Exception as e:
+        print(f"SQS Purge errord: {e}")
 
 
 def check_inf_file_content(desc, content, response_code, operation_outcome,