added code

Author: ASubaran

e2e_batch/Makefile

@@ -0,0 +1,4 @@
+-include .env
+
+run-immunization-batch:
+	ENV=$(environment) poetry run python  -m unittest -v -c

e2e_batch/constants.py

@@ -8,4 +8,4 @@
 FORWARDEDFILE_PREFIX = "forwardedFile/"
 ACK_PREFIX = "ack/"
 PRE_VALIDATION_ERROR = "Validation errors: doseQuantity.value must be a number"
-POST_VALIDATION_ERROR = "Validation errors: contained[?(@.resourceType=='Patient')].name[0].given is a mandatory field"
+POST_VALIDATION_ERROR = "Validation errors: contained[?(@.resourceType=='Patient')].name[0].given is a mandatory field"

e2e_batch/poetry.lock

@@ -1,7 +1,7 @@
 [tool.poetry]
 name = "e2e_batch"
 version = "0.1.0"
-description = "End-to-end tests for immunization-fhir-api"
+description = "End-to-end tests for immunization-batch"
 authors = ["Your Name <you@example.com>"]
 license = "MIT"
 readme = "README.md"

e2e_batch/pyproject.toml

@@ -7,7 +7,13 @@
     wait_for_ack_file,
     check_ack_file_content,
 )
-from constants import INPUT_BUCKET, INPUT_PREFIX, ACK_BUCKET, PRE_VALIDATION_ERROR, POST_VALIDATION_ERROR
+from constants import (
+    INPUT_BUCKET,
+    INPUT_PREFIX,
+    ACK_BUCKET,
+    PRE_VALIDATION_ERROR,
+    POST_VALIDATION_ERROR,
+)
 
 
 class TestE2EBatch(unittest.TestCase):
@@ -57,5 +63,6 @@ def test_post_validation_error(self):
         ack_content = get_file_content_from_s3(ACK_BUCKET, ack_key)
         check_ack_file_content(ack_content, "Fatal Error", POST_VALIDATION_ERROR)
 
+
 if __name__ == "__main__":
     unittest.main()

e2e_batch/test_e2e_batch.py

@@ -3,7 +3,6 @@
 import pandas as pd
 import uuid
 from datetime import datetime, timezone
-import csv
 from clients import logger, s3_client, table
 from constants import ACK_BUCKET, FORWARDEDFILE_PREFIX
 
@@ -104,7 +103,9 @@ def wait_for_ack_file(ack_prefix, input_file_name, timeout=120):
                 if search_pattern in key:
                     return key
         time.sleep(5)
-    raise Exception(f"Ack file matching '{search_pattern}' not found in bucket {ACK_BUCKET} within {timeout} seconds.")
+    raise AssertionError(
+                    f"Ack file matching '{search_pattern}' not found in bucket {ACK_BUCKET} within {timeout} seconds."
+                    )
 
 
 def get_file_content_from_s3(bucket, key):
@@ -120,12 +121,12 @@ def check_ack_file_content(content, response_code, operation_outcome):
     2. If 'HEADER_RESPONSE_CODE' is 'Fatal Error', check 'OPERATION_OUTCOME'.
     3. If 'OPERATION_OUTCOME' is 'Ok', extract 'LOCAL_ID', convert it to IdentifierPK,
        fetch PK from DynamoDB, and ensure it matches 'IMMS_ID'."""
-    
+
     reader = csv.DictReader(content.splitlines(), delimiter="|")
     rows = list(reader)
     for i, row in enumerate(rows):
         if "HEADER_RESPONSE_CODE" not in row:
-            raise Exception(f"Row {i + 1} does not have a 'HEADER_RESPONSE_CODE' column.")
+            raise AssertionError(f"Row {i + 1} does not have a 'HEADER_RESPONSE_CODE' column.")
         if row["HEADER_RESPONSE_CODE"].strip() != response_code:
             raise AssertionError(
                 f"Row {i + 1}: Expected RESPONSE '{response_code}', but found '{row['HEADER_RESPONSE_CODE']}'"
@@ -138,30 +139,31 @@ def check_ack_file_content(content, response_code, operation_outcome):
 
         if row["HEADER_RESPONSE_CODE"].strip() == "OK":
             if "LOCAL_ID" not in row:
-                raise Exception(f"Row {i + 1} does not have a 'LOCAL_ID' column.")
-            
+                raise AssertionError(f"Row {i + 1} does not have a 'LOCAL_ID' column.")
+
             # Extract LOCAL_ID and convert to IdentifierPK
             try:
                 local_id, unique_id_uri = row["LOCAL_ID"].split("^")
                 identifier_pk = f"{unique_id_uri}#{local_id}"
             except ValueError:
-                raise Exception(f"Row {i + 1}: Invalid LOCAL_ID format - {row['LOCAL_ID']}")
-            
+                raise AssertionError(f"Row {i + 1}: Invalid LOCAL_ID format - {row['LOCAL_ID']}")
+
             # Fetch PK from DynamoDB
             dynamo_pk = fetch_pk_from_dynamodb(identifier_pk)
             # Compare DynamoDB PK with IMMS_ID
             if dynamo_pk != row["IMMS_ID"]:
                 raise AssertionError(
-                    f"Row {i + 1}: Mismatch - DynamoDB PK '{dynamo_pk}' does not match ACK file IMMS_ID '{row['IMMS_ID']}'"
+                 f"Row {i + 1}: Mismatch - DynamoDB PK '{dynamo_pk}' does not match ACK file IMMS_ID '{row['IMMS_ID']}'"
                 )
 
+
 def fetch_pk_from_dynamodb(identifier_pk):
     """Fetch PK with IdentifierPK as the query key."""
     try:
         response = table.query(
-            IndexName="IdentifierGSI",  
+            IndexName="IdentifierGSI",
             KeyConditionExpression="IdentifierPK = :identifier_pk",
-            ExpressionAttributeValues={":identifier_pk": identifier_pk}
+            ExpressionAttributeValues={":identifier_pk": identifier_pk},
         )
         if "Items" in response and response["Items"]:
             return response["Items"][0]["PK"]  # Return the first matched PK
@@ -170,4 +172,4 @@ def fetch_pk_from_dynamodb(identifier_pk):
 
     except Exception as e:
         logger.error(f"Error fetching from DynamoDB: {e}")
-        return "ERROR"
+        return "ERROR"