test_convert.py

import json
import os
import unittest
from copy import deepcopy
from datetime import datetime
from unittest.mock import patch

from boto3 import resource as boto3_resource
from moto import mock_aws

from mappings import ActionFlag, EventName, Operation
from utils_for_converter_tests import ErrorValuesForTests, ValuesForTests, make_mock_logger

MOCK_ENV_VARS = {
    "AWS_SQS_QUEUE_URL": "https://sqs.eu-west-2.amazonaws.com/123456789012/test-queue",
    "DELTA_TABLE_NAME": "immunisation-batch-internal-dev-audit-test-table",
    "DELTA_TTL_DAYS": "14",
    "SOURCE": "test-source",
}
request_json_data = ValuesForTests.json_data
with patch.dict("os.environ", MOCK_ENV_VARS):
    from delta import Converter, handler


@patch.dict("os.environ", MOCK_ENV_VARS, clear=True)
class TestConvertToFlatJson(unittest.TestCase):
    maxDiff = None

    def setUp(self):
        # Start moto AWS mocks
        self.mock = mock_aws()
        self.mock.start()

        """Set up mock DynamoDB table."""
        self.dynamodb_resource = boto3_resource("dynamodb", "eu-west-2")

        self.table = self.dynamodb_resource.create_table(
            TableName="immunisation-batch-internal-dev-audit-test-table",
            KeySchema=[
                {"AttributeName": "PK", "KeyType": "HASH"},
            ],
            AttributeDefinitions=[
                {"AttributeName": "PK", "AttributeType": "S"},
                {"AttributeName": "Operation", "AttributeType": "S"},
                {"AttributeName": "IdentifierPK", "AttributeType": "S"},
                {"AttributeName": "SupplierSystem", "AttributeType": "S"},
            ],
            ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
            GlobalSecondaryIndexes=[
                {
                    "IndexName": "IdentifierGSI",
                    "KeySchema": [{"AttributeName": "IdentifierPK", "KeyType": "HASH"}],
                    "Projection": {"ProjectionType": "ALL"},
                    "ProvisionedThroughput": {
                        "ReadCapacityUnits": 5,
                        "WriteCapacityUnits": 5,
                    },
                },
                {
                    "IndexName": "PatientGSI",
                    "KeySchema": [
                        {"AttributeName": "Operation", "KeyType": "HASH"},
                        {"AttributeName": "SupplierSystem", "KeyType": "RANGE"},
                    ],
                    "Projection": {"ProjectionType": "ALL"},
                    "ProvisionedThroughput": {
                        "ReadCapacityUnits": 5,
                        "WriteCapacityUnits": 5,
                    },
                },
            ],
        )
        self.logger_patcher = patch("delta.logger", make_mock_logger())
        self.logger_patcher.start()

        self.send_log_to_firehose_patcher = patch("delta.send_log_to_firehose")
        self.mock_send_log_to_firehose = self.send_log_to_firehose_patcher.start()

        self.sqs_client_patcher = patch("common.clients.global_sqs_client")
        self.mock_sqs_client = self.sqs_client_patcher.start()

    def tearDown(self):
        patch.stopall()

        self.mock.stop()

    @staticmethod
    def get_event(event_name=EventName.CREATE, operation="operation", supplier="EMIS"):
        """Returns test event data."""
        return ValuesForTests.get_event(event_name, operation, supplier)

    def assert_structured_error_records(self, error_records):
        self.assertTrue(error_records)

        for error_record in error_records:
            self.assertEqual(set(error_record), {"code", "field", "value", "message"})
            self.assertTrue(error_record["message"])

    def assert_dynamodb_record(
        self,
        operation_flag,
        action_flag,
        items,
        expected_values,
        expected_imms,
        response,
    ):
        """
        Asserts that a record with the expected structure exists in DynamoDB.
        Ignores dynamic fields: PK, DateTimeStamp, ExpiresAt.
        Validates exact Imms payload and TTL offset.
        """
        self.assertTrue(response)

        unfiltered_items = [
            {k: v for k, v in item.items()}
            for item in items
            if item.get("Operation") == operation_flag and item.get("Imms", {}).get("ACTION_FLAG") == action_flag
        ]

        filtered_items = [
            {k: v for k, v in item.items() if k not in ["PK", "DateTimeStamp", "ExpiresAt"]} for item in unfiltered_items
        ]

        self.assertGreater(len(filtered_items), 0, f"No matching item found for {operation_flag}")

        imms_data = filtered_items[0]["Imms"]
        if isinstance(imms_data, str):
            imms_data = json.loads(imms_data)

        self.assertIsInstance(imms_data, dict)
        self.assertGreater(len(imms_data), 0)
        self.assertEqual(imms_data, expected_imms, "Imms data does not match expected JSON structure")

        for key, expected_value in expected_values.items():
            self.assertIn(key, filtered_items[0], f"{key} is missing")
            self.assertEqual(filtered_items[0][key], expected_value, f"{key} mismatch")

        expected_seconds = int(os.environ["DELTA_TTL_DAYS"]) * 24 * 60 * 60
        date_time = int(datetime.fromisoformat(unfiltered_items[0]["DateTimeStamp"]).timestamp())
        expires_at = unfiltered_items[0]["ExpiresAt"]
        self.assertEqual(expires_at - date_time, expected_seconds)

    def test_fhir_converter_json_direct_data(self):
        """it should convert fhir json data to flat json"""
        json_data = json.dumps(ValuesForTests.json_data)

        fhir_converter = Converter(json_data)
        result = fhir_converter.run_conversion()

        expected_imms_value = deepcopy(ValuesForTests.expected_imms2)  # UPDATE is currently the default action-flag
        self.assertEqual(result, expected_imms_value)
        self.assertFalse(fhir_converter.get_error_records())

    def test_fhir_converter_json_error_scenario(self):
        """it should convert fhir json data to flat json - error scenarios"""
        error_test_cases = {
            "missing_json": ErrorValuesForTests.missing_json,
            "json_dob_error": ErrorValuesForTests.json_dob_error,
        }

        for test_name, test_case in error_test_cases.items():
            with self.subTest(test_name=test_name):
                fhir_converter = Converter(json.dumps(test_case))
                fhir_converter.run_conversion()

                self.assert_structured_error_records(fhir_converter.get_error_records())

    def test_fhir_converter_json_incorrect_data_scenario(self):
        """it should convert fhir json data to flat json - error scenarios"""

        with self.assertRaisesRegex(ValueError, "FHIR data is required for initialization."):
            Converter(None)

    def test_handler_persists_structured_conversion_errors(self):
        event = self.get_event(operation=Operation.UPDATE)
        event["Records"][0]["dynamodb"]["NewImage"]["Resource"]["S"] = json.dumps(ErrorValuesForTests.json_dob_error)

        response = handler(event, None)

        result = self.table.scan()
        items = result.get("Items", [])
        self.assertTrue(response)
        self.assertEqual(len(items), 1)

        error_records = items[0]["Imms"]["CONVERSION_ERRORS"]
        self.assert_structured_error_records(error_records)
        self.assertEqual(error_records[0]["field"], "PERSON_DOB")
        self.assertEqual(error_records[0]["value"], "196513-28")

    def test_handler_imms_convert_to_flat_json(self):
        """Test that the Imms field contains the correct flat JSON data for CREATE, UPDATE, and DELETE operations."""
        expected_action_flags = [
            {"Operation": Operation.CREATE, "EXPECTED_ACTION_FLAG": ActionFlag.CREATE},
            {"Operation": Operation.UPDATE, "EXPECTED_ACTION_FLAG": ActionFlag.UPDATE},
            {
                "Operation": Operation.DELETE_LOGICAL,
                "EXPECTED_ACTION_FLAG": ActionFlag.DELETE_LOGICAL,
            },
        ]

        for test_case in expected_action_flags:
            with self.subTest(test_case["Operation"]):
                event = self.get_event(operation=test_case["Operation"])

                response = handler(event, None)

                # Retrieve items from DynamoDB
                result = self.table.scan()
                items = result.get("Items", [])

                expected_values = ValuesForTests.expected_static_values
                expected_imms = ValuesForTests.get_expected_imms(test_case["EXPECTED_ACTION_FLAG"])

                self.assert_dynamodb_record(
                    test_case["Operation"],
                    test_case["EXPECTED_ACTION_FLAG"],
                    items,
                    expected_values,
                    expected_imms,
                    response,
                )

                self.clear_table()

    def test_handler_imms_convert_to_flat_json_legacy_patientsk_compatibility(self):
        """
        Ensures legacy PatientSK input is still accepted (backward compatibility).
        """
        event = self.get_event(operation=Operation.CREATE)

        new_image = event["Records"][0]["dynamodb"]["NewImage"]

        # Some fixtures already provide PatientSK and no SK
        if "SK" in new_image:
            new_image["PatientSK"] = deepcopy(new_image["SK"])
            del new_image["SK"]
        elif "PatientSK" not in new_image:
            self.fail("Fixture must contain either SK or PatientSK")

        response = handler(event, None)

        result = self.table.scan()
        items = result.get("Items", [])
        self.assertGreater(len(items), 0)
        self.assertTrue(response)

    def clear_table(self):
        scan = self.table.scan()
        with self.table.batch_writer() as batch:
            for item in scan.get("Items", []):
                batch.delete_item(Key={"PK": item["PK"]})