WIP

Author: nhsdevws

ack_backend/poetry.lock

@@ -7,7 +7,7 @@
 from constants import ACK_HEADERS, SOURCE_BUCKET_NAME, ACK_BUCKET_NAME, FILE_NAME_PROC_LAMBDA_NAME
 from audit_table import change_audit_table_status_to_processed, get_next_queued_file_details
 from clients import s3_client, logger, lambda_client
-from utils_for_ack_lambda import get_row_count
+from utils_for_ack_lambda import get_row_count, debug_print
 
 
 def create_ack_data(
@@ -76,11 +76,16 @@ def upload_ack_file(
 ) -> None:
     """Adds the data row to the uploaded ack file"""
     for row in ack_data_rows:
+        print(f"add row {row} to ack file")
         data_row_str = [str(item) for item in row.values()]
         cleaned_row = "|".join(data_row_str).replace(" |", "|").replace("| ", "|").strip()
         accumulated_csv_content.write(cleaned_row + "\n")
+
+
     csv_file_like_object = BytesIO(accumulated_csv_content.getvalue().encode("utf-8"))
     s3_client.upload_fileobj(csv_file_like_object, ACK_BUCKET_NAME, temp_ack_file_key)
+    
+    debug_print("upload_ack_file...2")
 
     row_count_source = get_row_count(SOURCE_BUCKET_NAME, f"processing/{file_key}")
     row_count_destination = get_row_count(ACK_BUCKET_NAME, temp_ack_file_key)

ack_backend/src/update_ack_file.py

@@ -1,12 +1,34 @@
 """Utils for ack lambda"""
 
 from clients import s3_client
-
+from tests.utils.values_for_ack_backend_tests import MessageDetails
 
 def get_row_count(bucket_name: str, file_key: str) -> int:
     """
     Looks in the given bucket and returns the count of the number of lines in the given file.
     NOTE: Blank lines are not included in the count.
     """
-    response = s3_client.get_object(Bucket=bucket_name, Key=file_key)
-    return sum(1 for line in response["Body"].iter_lines() if line.strip())
+    try:
+        response = s3_client.get_object(Bucket=bucket_name, Key=file_key)
+        if "Body" not in response or response["Body"] is None:
+            print(f"Error getting row count: No Body in response for {bucket_name}/{file_key}")
+            return 0
+        ret=sum(1 for line in response["Body"].iter_lines() if line.strip())
+        print(f"Row count {bucket_name}-{file_key}: {ret}")
+    except Exception as e:
+        print(f"Row count {bucket_name}-{file_key}: 0")
+        return 0
+
+from constants import ACK_HEADERS, SOURCE_BUCKET_NAME, ACK_BUCKET_NAME, FILE_NAME_PROC_LAMBDA_NAME
+
+def debug_print(title, show_source=False):
+    rsv_ravs = MessageDetails("RSV", "RAVS", "X26")
+    print(f"Debug rowcount: {title}")
+
+    if show_source:
+        row_count_source = get_row_count(SOURCE_BUCKET_NAME, f"processing/{rsv_ravs.file_key}")
+        print(f"Row count in source bucket: {row_count_source}")    
+    
+    row_count_destination = get_row_count(ACK_BUCKET_NAME, "TempAck/RSV_Vaccinations_v5_X26_20210730T12000000_BusAck_20211120T12000000.csv")
+
+    print(f"Row count in destination bucket: {row_count_destination}")

ack_backend/src/utils_for_ack_lambda.py

@@ -51,9 +51,12 @@ def setUp(self) -> None:
             Key=f"processing/{MOCK_MESSAGE_DETAILS.file_key}",
             Body=mock_source_file_with_100_rows.getvalue(),
         )
+        self.logger_info_patcher = patch('logging_decorators.logger.info')
+        self.mock_logger_info = self.logger_info_patcher.start()
 
     def tearDown(self) -> None:
         GenericTearDown(s3_client, firehose_client)
+        self.mock_logger_info.stop()
 
     @staticmethod
     def generate_event(test_messages: list[dict]) -> dict:

ack_backend/tests/test_ack_processor.py

@@ -0,0 +1,55 @@
+import unittest
+from unittest.mock import patch, MagicMock
+import audit_table
+from errors import UnhandledAuditTableError
+
+class TestAuditTable(unittest.TestCase):
+    
+    def setUp(self):
+        self.logger_patcher = patch('audit_table.logger')
+        self.mock_logger = self.logger_patcher.start()
+        self.dynamodb_resource_patcher = patch('audit_table.dynamodb_resource')
+        self.mock_dynamodb_resource = self.dynamodb_resource_patcher.start()
+        self.dynamodb_client_patcher = patch('audit_table.dynamodb_client')
+        self.mock_dynamodb_client = self.dynamodb_client_patcher.start()
+
+    def tearDown(self):
+        self.logger_patcher.stop()
+        self.dynamodb_resource_patcher.stop()
+        self.dynamodb_client_patcher.stop()
+
+    def test_get_next_queued_file_details_returns_oldest(self):
+        # Arrange
+        mock_table = MagicMock()
+        self.mock_dynamodb_resource.Table.return_value = mock_table
+        mock_table.query.return_value = {
+            "Items": [
+                {"timestamp": 2, "foo": "bar"},
+                {"timestamp": 1, "foo": "baz"},
+            ]
+        }
+        # Act
+        result = audit_table.get_next_queued_file_details("queue1")
+        # Assert
+        self.assertEqual(result, {"timestamp": 1, "foo": "baz"})
+
+    def test_get_next_queued_file_details_returns_none_if_empty(self):
+        mock_table = MagicMock()
+        self.mock_dynamodb_resource.Table.return_value = mock_table
+        mock_table.query.return_value = {"Items": []}
+        result = audit_table.get_next_queued_file_details("queue1")
+        self.assertIsNone(result)
+
+    def test_change_audit_table_status_to_processed_success(self):
+        # Should not raise
+        self.mock_dynamodb_client.update_item.return_value = {}
+        audit_table.change_audit_table_status_to_processed("file1", "msg1")
+        self.mock_dynamodb_client.update_item.assert_called_once()
+        self.mock_logger.info.assert_called_once()
+
+    def test_change_audit_table_status_to_processed_raises(self):
+        self.mock_dynamodb_client.update_item.side_effect = Exception("fail!")
+        with self.assertRaises(UnhandledAuditTableError) as ctx:
+            audit_table.change_audit_table_status_to_processed("file1", "msg1")
+        self.assertIn("fail!", str(ctx.exception))
+        self.mock_logger.error.assert_called_once()

ack_backend/tests/test_audit_table.py

@@ -18,6 +18,8 @@
     generate_expected_ack_content,
     MOCK_MESSAGE_DETAILS,
 )
+from constants import ACK_HEADERS, SOURCE_BUCKET_NAME, ACK_BUCKET_NAME, FILE_NAME_PROC_LAMBDA_NAME
+from utils_for_ack_lambda import get_row_count, debug_print
 
 with patch.dict("os.environ", MOCK_ENVIRONMENT_DICT):
     from update_ack_file import obtain_current_ack_content, create_ack_data, update_ack_file
@@ -42,6 +44,8 @@ def setUp(self) -> None:
             Key=f"processing/{MOCK_MESSAGE_DETAILS.file_key}",
             Body=mock_source_file_with_100_rows.getvalue(),
         )
+        self.logger_patcher = patch('update_ack_file.logger')
+        self.mock_logger = self.logger_patcher.start()
 
     def tearDown(self) -> None:
         GenericTearDown(s3_client)
@@ -204,5 +208,31 @@ def test_obtain_current_ack_content_file_exists(self):
         self.assertEqual(result.getvalue(), existing_content)
 
 
+    def test_update_ack_file_moves_file_to_destination(self):
+        """Test that update_ack_file moves the ack file to the correct destination location."""
+        with patch('update_ack_file.move_file') as mock_move_file:
+            
+            
+            row_count_source = get_row_count(SOURCE_BUCKET_NAME, f"processing/{MOCK_MESSAGE_DETAILS.file_key}")
+            row_count_destination = get_row_count(ACK_BUCKET_NAME, "TempAck/RSV_Vaccinations_v5_X26_20210730T12000000_BusAck_20211120T12000000.csv")
+
+            debug_print("test...1")
+
+            update_ack_file(
+                file_key=MOCK_MESSAGE_DETAILS.file_key,
+                message_id=MOCK_MESSAGE_DETAILS.message_id,
+                supplier_queue=MOCK_MESSAGE_DETAILS.queue_name,
+                created_at_formatted_string=MOCK_MESSAGE_DETAILS.created_at_formatted_string,
+                ack_data_rows=[ValidValues.ack_data_success_dict],
+            )
+            # Assert that move_file was called with the expected arguments
+            mock_move_file.assert_called()
+            called_args_list = [call.args for call in mock_move_file.call_args_list]
+            # Check that at least one call moved the file to the DESTINATION bucket
+            self.assertTrue(
+                any(BucketNames.DESTINATION in args for args in called_args_list[0]),
+                f"File was not moved to the {BucketNames.DESTINATION} bucket"
+            )
+
 if __name__ == "__main__":
     unittest.main()