Cleanup after each test

Author: robertnovac1

delta_backend/README.md

@@ -68,74 +68,4 @@ You can now:
 - Open `output.csv` in Excel or Google Sheets to view cleanly structured records
 - Inspect `output.json` to validate the flat key-value output programmatically
 
----# 🩺 FHIR to Flat JSON Conversion Engine
-
-This project is designed to convert FHIR-compliant JSON data (e.g., Immunization records) into a flat JSON format based on a configurable schema layout. It is intended to support synchronization of Immunisation API generated data from external sources to DPS (Data Processing System) data system
-
----
-
-## 📁 File Structure Overview
-
-| File Name              | What It Does |
-|------------------------|---------------|
-| **`converter.py`**     | 🧠 The main brain — applies the schema, runs conversions, handles errors. |
-| **`FHIRParser.py`**    | 🪜 Knows how to dig into nested FHIR structures and pull out values like dates, IDs, and patient names. |
-| **`SchemaParser.py`**  | 📐 Reads your schema layout and tells the converter which FHIR fields to extract and how to rename/format them. |
-| **`ConversionLayout.py`** | ✍️ A plain Python list that defines which fields you want, and how they should be formatted (e.g. date format, renaming rules). |
-| **`ConversionChecker.py`** | 🔧 Handles transformation logic — e.g. turning a FHIR datetime into `YYYY-MM-DD`, applying lookups, gender codes, defaults, etc. |
-| **`Extractor.py`**     | 🎣 Specialized logic to pull practitioner names, site codes, addresses, and apply time-aware rules. |
-| **`ExceptionMessages.py`** | 🚨 Holds reusable error messages and codes for clean debugging and validation feedback. |
-
----
-
-
-## 🛠️ Key Features
-
-- Schema-driven field extraction and formatting
-- Support for custom date formats like `YYYYMMDD`, and CSV-safe UTC timestamps
-- Robust handling of patient, practitioner, and address data using time-aware logic
-- Extendable structure with static helper methods and modular architecture
-
----
-
-## 📦 Example Use Case
-
-- Input: FHIR `Immunization` resource (with nested fields)
-- Output: Flat JSON object with 34 standardized key-value pairs
-- Purpose: To export into CSV or push into downstream ETL systems
-
----
-
-## ✅ Getting Started with `check_conversion.py`
-
-To quickly test your conversion, use the provided `check_conversion.py` script.
-This script loads sample FHIR data, runs it through the converter, and automatically saves the output in both JSON and CSV formats.
-
-### 🔄 How to Use It
-
-1. Add your FHIR data (e.g., a dictionary or sample JSON) into the `fhir_sample` variable inside `check_conversion.py`
-2. Ensure the field mapping in `ConversionLayout.py` matches your desired output
-3. Run the script from the `tests` folder:
-
-```bash
-python check_conversion.py
-```
-
-### 📁 Output Location
-When the script runs, it will automatically:
-- Save a **flat JSON file** as `output.json`
-- Save a **CSV file** as `output.csv`
-
-These will be located one level up from the `src/` folder:
-
-```
-/mnt/c/Users/USER/desktop/shn/immunisation-fhir-api/delta_backend/output.json
-/mnt/c/Users/USER/desktop/shn/immunisation-fhir-api/delta_backend/output.csv
-```
-
-### 👀 Visualization
-You can now:
-- Open `output.csv` in Excel or Google Sheets to view cleanly structured records
-- Inspect `output.json` to validate the flat key-value output programmatically
-
 ---

e2e/utils/base_test.py

@@ -20,7 +20,7 @@
     make_apigee_product,
 )
 from utils.immunisation_api import ImmunisationApi, parse_location
-from utils.resource import generate_imms_resource
+from utils.resource import generate_imms_resource, delete_imms_records
 
 
 class ImmunizationBaseTest(unittest.TestCase):
@@ -35,7 +35,11 @@ class ImmunizationBaseTest(unittest.TestCase):
     apps: List[ApigeeApp]
     # an ImmunisationApi with default auth-type: ApplicationRestricted
     default_imms_api: ImmunisationApi
+    generated_imms_records: List[str]
 
+    # Called once before any test methods in the class are run.
+    # The purpose of setUpClass is to prepare shared resources that
+    # All tests in the class can use
     @classmethod
     def setUpClass(cls):
         cls.apps = []
@@ -87,13 +91,25 @@ def make_app_data() -> ApigeeApp:
             cls.tearDownClass()
             raise e
 
+    # Class method that runs once after all test methods in the class have finished. 
+    # It is used to clean up resources that were shared across multiple tests
     @classmethod
     def tearDownClass(cls):
         for app in cls.apps:
             cls.apigee_service.delete_application(app.name)
         if hasattr(cls, "product") and cls:
             cls.apigee_service.delete_product(cls.product.name)
 
+    # Runs after each individual test method in a test class. 
+    # It’s used to clean up resources that were initialized specifically for a single test, 
+    # such as temporary files or mock objects.
+    def tearDown(cls):
+        for api in cls.imms_apis:
+            if api.generated_test_records:
+                imms_deleted = delete_imms_records(api.generated_test_records)
+                print(imms_deleted)
+                api.generated_test_records.clear()
+
     @staticmethod
     def create_immunization_resource(imms_api: ImmunisationApi, resource: dict = None) -> str:
         """creates an Immunization resource and returns the resource url"""

e2e/utils/immunisation_api.py

@@ -3,7 +3,7 @@
 import time
 import random
 import requests
-from typing import Optional, Literal
+from typing import Optional, Literal, List
 from datetime import datetime
 
 from lib.authentication import BaseAuthentication
@@ -26,6 +26,7 @@ class ImmunisationApi:
     url: str
     headers: dict
     auth: BaseAuthentication
+    generated_test_records: List[str]
 
     def __init__(self, url, auth: BaseAuthentication):
         self.url = url
@@ -38,6 +39,7 @@ def __init__(self, url, auth: BaseAuthentication):
             "Authorization": f"Bearer {token}",
             "Content-Type": "application/fhir+json",
             "Accept": "application/fhir+json"}
+        self.generated_test_records = []
 
     def __str__(self):
         return f"ImmunizationApi: AuthType: {self.auth}"
@@ -69,7 +71,7 @@ def _make_request_with_backoff(
                 return response
 
             except Exception as e:
-                if attempt == self.MAX_RETRIES:
+                if attempt == self.MAX_RETRIES - 1:
                     raise
 
                 wait = (2 ** attempt) + random.uniform(0, 0.5)
@@ -91,14 +93,26 @@ def get_immunization_by_id(self, event_id, expected_status_code: int = 200):
         )
 
     def create_immunization(self, imms, expected_status_code: int = 201):
-        return self._make_request_with_backoff(
+        response = self._make_request_with_backoff(
             "POST",
             f"{self.url}/Immunization",
             expected_status_code,
             headers=self._update_headers(),
             json=imms
         )
 
+        if response.status_code == 201:
+            if "Location" not in response.headers:
+                raise ValueError("Missing 'Location' header in response")
+
+            imms_id = response.headers["Location"].split("Immunization/")[-1]
+            if not self._is_valid_uuid4(imms_id):
+                raise ValueError(f"Invalid UUID4: {imms_id}")
+
+            self.generated_test_records.append(imms_id)
+
+        return response
+
     def update_immunization(self, imms_id, imms, expected_status_code: int = 200):
         return self._make_request_with_backoff(
             "PUT",
@@ -154,3 +168,10 @@ def _update_headers(self, headers=None):
             "E-Tag": "1"
         }}
         return {**updated, **headers}
+
+    def _is_valid_uuid4(self, imms_id):
+        try:
+            val = uuid.UUID(imms_id, version=4)
+            return str(val) == imms_id
+        except ValueError:
+            return False

e2e/utils/resource.py

@@ -5,10 +5,10 @@
 from copy import deepcopy
 from decimal import Decimal
 from typing import Union, Literal
-from mypy_boto3_dynamodb.service_resource import DynamoDBServiceResource
+from mypy_boto3_dynamodb.service_resource import DynamoDBServiceResource, Table
 from botocore.config import Config
 from .mappings import vaccine_type_mappings, VaccineTypes
-
+from functools import lru_cache
 from .constants import valid_nhs_number1
 
 current_directory = os.path.dirname(os.path.realpath(__file__))
@@ -133,9 +133,36 @@ def get_patient_postal_code(imms: dict):
 
 
 def get_full_row_from_identifier(identifier: str) -> dict:
-    """Get the full record from the dynamodb table using the identifier"""
-    config = Config(connect_timeout=1, read_timeout=1, retries={"max_attempts": 1})
-    db: DynamoDBServiceResource = boto3.resource("dynamodb", region_name="eu-west-2", config=config)
-    table = db.Table(os.getenv("DYNAMODB_TABLE_NAME"))
+    table = get_dynamodb_table()
 
     return table.get_item(Key={"PK": f"Immunization#{identifier}"}).get("Item")
+
+
+@lru_cache()
+def get_dynamodb_table() -> Table:
+    config = Config(connect_timeout=2, read_timeout=2, retries={"max_attempts": 1})
+    db: DynamoDBServiceResource = boto3.resource("dynamodb", region_name="eu-west-2", config=config)
+    return db.Table(os.getenv("DYNAMODB_TABLE_NAME"))
+
+
+def delete_imms_records(identifiers: list[str]) -> dict:
+    """Batch delete immunization records from the DynamoDB table."""
+    table = get_dynamodb_table()
+    deleted = []
+    errors = []
+
+    try:
+        with table.batch_writer(overwrite_by_pkeys=["PK"]) as batch:
+            for identifier in identifiers:
+                key = {"PK": f"Immunization#{identifier}"}
+                try:
+                    batch.delete_item(Key=key)
+                    deleted.append(identifier)
+                except Exception as e:
+                    print(f"Failed to delete record with key {key}: {e}")
+                    errors.append({"identifier": identifier, "error": str(e)})
+    except Exception as e:
+        print(f"Batch writer failed: {e}")
+        return {"Error": str(e), "Deleted": deleted, "Failures": errors}
+
+    return {"Deleted": deleted, "Failures": errors}