basic tests

Author: nhsdevws

delta_backend/src/delta.py

@@ -1,25 +1,6 @@
-import boto3
-import json
-import logging
 from log_firehose import FirehoseLogger
-
-logging.basicConfig()
-logger = logging.getLogger()
-logger.setLevel("INFO")
-firehose_logger = None
+firehose_logger = FirehoseLogger()
 
 def handler(event, context):
-    global  firehose_logger
-    logger.info("Starting Delta Handler")
-    logger.info("Event: '%s'", event["text"])
-    
-    if firehose_logger is None:
-        firehose_logger = FirehoseLogger()
-        logger.info("FirehoseLogger initialized")
-
-    firehose_logger.send_log(event)
-    
-    return {
-        "statusCode": 200,
-        "body": json.dumps("Hello from Lambda!"),
-    }
+    firehose_logger.send_log(event)    
+    return { "statusCode": 200  }

delta_backend/src/log_firehose.py

@@ -1,35 +1,21 @@
 import boto3
-import logging
 import os
+import json
 from botocore.config import Config
 
-logging.basicConfig()
-logger = logging.getLogger()
-logger.setLevel("INFO")
-
-
 class FirehoseLogger:
     def __init__(
         self,
-        stream_name: str = os.getenv("SPLUNK_FIREHOSE_NAME"),
         boto_client=boto3.client("firehose", config=Config(region_name="eu-west-2")),
+        stream_name: str = os.getenv("SPLUNK_FIREHOSE_NAME")
     ):
-        logger.info(">>>>>>FirehoseLogger.init!!!")
-        if boto_client is None:
-            boto_client = boto3.client("firehose", config=Config(region_name="eu-west-2"))
         self.firehose_client = boto_client
         self.delivery_stream_name = stream_name
-        logger.info(f">>>>>> Firehose stream name: {stream_name}")
 
     def send_log(self, log_message):
-        log_to_splunk = log_message
-        logger.info(f"Log sent to Firehose for save: {log_to_splunk}")
-        try:
-            logger.info(f"Send log to Firehose")
-            response = self.firehose_client.put_record(
-                DeliveryStreamName=self.delivery_stream_name,
-                Record={"Data": log_message},
-            )
-            logger.info(f"Log sent: {response}")
-        except Exception as e:
-            logger.exception(f"Error sending log: {e}")
+        encoded_log_data = json.dumps(log_message).encode("utf-8")
+        s = self.firehose_client.put_record(
+            DeliveryStreamName=self.delivery_stream_name,
+            Record={"Data": encoded_log_data},
+        )
+        print(f"Log sent to Firehose: {s}")

delta_backend/tests/test_delta.py

@@ -1,36 +1,24 @@
 import unittest
 from unittest.mock import patch, MagicMock
-
-from src.delta import handler  # Import after setting environment variables
+from src.delta import handler
 
 
 class DeltaTestCase(unittest.TestCase):
 
-    def setUp(self):
-        # Mock firehose_logger globally
-        self.firehose_logger_patch = patch("src.delta.firehose_logger")
-        self.mock_firehose_logger = self.firehose_logger_patch.start()
-
-        # Mock the firehose_client and its put_record method
-        self.mock_firehose_client = self.mock_firehose_logger.firehose_client
-        self.mock_firehose_client.put_record.return_value = {"ResponseMetadata": {"HTTPStatusCode": 200}}
+    @patch("src.delta.firehose_logger")  # Mock the firehose_logger instance in delta.py
+    def test_handler(self, mock_firehose_logger):
+        # Arrange
+        event = {"text": "hello world"}
 
         # Mock the send_log method
-        self.mock_firehose_logger.send_log = MagicMock()
-
-        # Common setup if needed
-        self.context = {}
-
-    def tearDown(self):
-        # Stop the patch after each test
-        self.firehose_logger_patch.stop()
-
-    def test_handler(self):
-        # Arrange
-        event = {"text": "hello world!"}
+        mock_firehose_logger.send_log = MagicMock()
 
         # Act
-        result = handler(event, self.context)
+        result = handler(event, None)
 
         # Assert
+        # Check the handler's response
         self.assertEqual(result["statusCode"], 200)
+
+        # Verify that send_log was called with the correct event
+        mock_firehose_logger.send_log.assert_called_once_with(event)

delta_backend/tests/test_log_firehose.py

@@ -0,0 +1,39 @@
+import unittest
+from unittest.mock import patch, MagicMock
+import os
+import json
+from src.log_firehose import FirehoseLogger  # Assuming your FirehoseLogger class is in the log_firehose.py file.
+
+class TestFirehoseLogger(unittest.TestCase):
+
+    @patch("boto3.client")
+    def test_send_log(self, mock_boto_client):
+        # Create a mock boto3 client and mock the put_record response
+        mock_response = {
+            "RecordId": "shardId-000000000000000000000001",
+            "ResponseMetadata": {
+            "RequestId": "12345abcde67890fghijk",
+            "HTTPStatusCode": 200, 
+            "RetryAttempts": 0
+            }
+        }        
+        mock_firehose_client = MagicMock()
+        mock_boto_client.return_value = mock_firehose_client
+        mock_firehose_client.put_record.return_value = mock_response
+        
+        stream_name = "stream_name"
+        firehose_logger = FirehoseLogger(boto_client=mock_firehose_client, stream_name=stream_name)
+    
+        log_message = {"text": "Test log message"}
+
+        # Call the send_log method
+        firehose_logger.send_log(log_message)
+
+        # Assert that put_record was called exactly once with the correct parameters
+        mock_firehose_client.put_record.assert_called_once()
+
+        # Assert that the return value is as expected (optional)
+        self.assertEqual(mock_firehose_client.put_record.return_value, mock_response)
+
+if __name__ == "__main__":
+    unittest.main()