DynamoDb

Author: nhsdevws

lambdas/shared/src/common/clients.py

@@ -1,8 +1,6 @@
 import os
 import logging
 from boto3 import client as boto3_client
-# from botocore.config import Config
-
 
 logging.basicConfig(level="INFO")
 logger = logging.getLogger()
@@ -19,3 +17,17 @@
 # boto_config = Config(region_name=REGION_NAME)
 # secretsmanager_client = boto3_client("secretsmanager", config=boto_config)
 secrets_manager_client = boto3_client("secretsmanager", region_name=REGION_NAME)
+dynamodb_client = boto3_client("dynamodb", region_name=REGION_NAME)
+
+
+def get_delta_table(table_name):
+    """
+    Initialize the DynamoDB table resource with exception handling.
+    """
+    try:
+        logger.info("Initializing table: %s", table_name)
+        delta_table = dynamodb_client.Table(table_name)
+    except Exception as e:
+        logger.exception("Error initializing Delta Table")
+        raise e
+    return delta_table

lambdas/shared/tests/test_common/test_clients.py

@@ -1,6 +1,7 @@
 import unittest
-from unittest.mock import patch
+from unittest.mock import patch, MagicMock
 import common.clients as clients
+from common.clients import get_delta_table
 import importlib
 
 
@@ -61,3 +62,56 @@ def test_logging_initialization(self):
         self.mock_logging.assert_called_once_with()
         self.assertTrue(hasattr(clients, 'logger'))
         clients.logger.setLevel.assert_any_call("INFO")
+
+
+class TestGetDeltaTable(unittest.TestCase):
+
+    AWS_REGION = "eu-west-2"  # Add this missing constant
+
+    def setUp(self):
+        self.boto3_client_patch = patch("boto3.client")
+        self.mock_boto3_client = self.boto3_client_patch.start()
+
+        # Mock the specific logger instance used in the module
+        self.logger_patch = patch("common.clients.logger")
+        self.mock_logger = self.logger_patch.start()
+
+        self.getenv_patch = patch("os.getenv")
+        self.mock_getenv = self.getenv_patch.start()
+        self.mock_getenv.side_effect = lambda key, default=None: {
+            "AWS_REGION": self.AWS_REGION
+        }.get(key, default)
+
+        self.mock_dynamodb_client = patch("common.clients.dynamodb_client").start()
+
+    def tearDown(self):
+        patch.stopall()
+
+    def test_get_delta_table_success(self):
+        # Create a mock table object
+        table_name = "abc"
+        mock_table = MagicMock()
+        self.mock_dynamodb_client.Table.return_value = mock_table
+
+        # Call the function
+        table = get_delta_table(table_name)
+
+        self.mock_dynamodb_client.Table.assert_called_once_with(table_name)
+        self.assertEqual(table, mock_table)
+        # Verify the success logging
+        self.mock_logger.info.assert_called_once_with("Initializing table: %s", table_name)
+
+    def test_get_delta_table_failure(self):
+        # Simulate exception when accessing Table
+        msg = "DynamoDB failure"
+        self.mock_dynamodb_client.Table.side_effect = Exception(msg)
+        table_name = "abc"
+
+        with self.assertRaises(Exception) as context:
+            get_delta_table(table_name)
+
+        self.assertEqual(str(context.exception), msg)
+        # This should now work - mocking the instance method
+        self.mock_logger.exception.assert_called_once_with("Error initializing Delta Table")
+        # Also verify info logging happened before the exception
+        self.mock_logger.info.assert_called_once_with("Initializing table: %s", table_name)