fixed lint issue along with test case

Author: amitkr-22

filenameprocessor/src/file_key_validation.py

@@ -54,7 +54,7 @@ def validate_file_key(file_key: str) -> tuple[str, str]:
         and version in Constants.VALID_VERSIONS
         and supplier  # Note that if supplier could be identified, this also implies that ODS code is valid
         and is_valid_datetime(timestamp)
-        and ((extension == "CSV") or (extension == "DAT")) # The DAT extension has been added for MESH file processing
+        and ((extension == "CSV") or (extension == "DAT"))  # The DAT extension has been added for MESH file processing
     ):
         raise InvalidFileKeyError("Initial file validation failed: invalid file key")
 

meshfileprocessor/src/converter.py

@@ -1,13 +1,11 @@
 import boto3
-import pandas as pd
-import io
 import os
-import re
 
-def lambda_handler(event, context):    
+
+def lambda_handler(event, context):
     s3 = boto3.client('s3')
-    
-    # Destination bucket name  
+
+    #  Destination bucket name
     destination_bucket = os.getenv("Destination_BUCKET_NAME")
 
     for record in event["Records"]:
@@ -16,16 +14,16 @@ def lambda_handler(event, context):
         copy_source = {
             'Bucket': record["s3"]["bucket"]["name"],
             'Key': record["s3"]["object"]["key"]
-        }    
-    
+        }
+
     # Read the .dat file from S3
     dat_obj = s3.get_object(Bucket=bucket_name, Key=file_key)
-    
+
     # Update the filename from Metadata
     file_name = ensure_dat_extension(dat_obj['Metadata'].get('mex-filename', None))
 
     s3.copy_object(CopySource=copy_source, Bucket=destination_bucket, Key=file_name)
-    
+
     return {
         'statusCode': 200,
         'body': 'Files converted and uploaded successfully!'
@@ -36,11 +34,11 @@ def ensure_dat_extension(file_name):
     if '.' in file_name:
         # Split the filename and extension
         base_name, extension = file_name.rsplit('.', 1)
-        
+
         # Check if the extension is not 'dat'
         if extension != 'dat':
             file_name = f"{base_name}.dat"
-    else:        
+    else:
         file_name += '.dat'
-    
+
     return file_name

meshfileprocessor/tests/test_converter.py

@@ -0,0 +1,61 @@
+from unittest import TestCase
+from unittest.mock import patch, MagicMock
+import os
+from src.converter import lambda_handler, ensure_dat_extension
+
+
+class TestLambdaHandler(TestCase):
+
+    @patch('boto3.client')
+    @patch('os.getenv')
+    def test_lambda_handler_success(self, mock_getenv, mock_boto_client):
+        # Mock environment variable
+        mock_getenv.return_value = "destination-bucket"
+
+        # Mock boto3 S3 client
+        mock_s3 = MagicMock()
+        mock_boto_client.return_value = mock_s3
+        mock_s3.get_object.return_value = {
+            'Metadata': {'mex-filename': '20250320121710483244_2DB240.txt'}
+        }
+
+        # Define the event
+        event = {
+            "Records": [
+                {
+                    "s3": {
+                        "bucket": {"name": "source-bucket"},
+                        "object": {"key": "20250320121710483244_2DB240.dat"}
+                    }
+                }
+            ]
+        }
+        context = {}
+
+        # Call the lambda_handler function
+        response = lambda_handler(event, context)
+
+        # Assertions
+        mock_s3.get_object.assert_called_with(Bucket="source-bucket", Key="20250320121710483244_2DB240.dat")
+        mock_s3.copy_object.assert_called_with(
+            CopySource={'Bucket': "source-bucket", 'Key': "20250320121710483244_2DB240.dat"},
+            Bucket="destination-bucket",
+            Key="20250320121710483244_2DB240.dat"
+        )
+        self.assertEqual(response['statusCode'], 200)
+        self.assertEqual(response['body'], 'Files converted and uploaded successfully!')
+
+    def test_ensure_dat_extension_with_other_extension(self):
+        # Test case where file has an extension other than 'dat'
+        result = ensure_dat_extension("COVID19_Vaccinations_v5_YGM41_20240927T13005921.txt")
+        self.assertEqual(result, "COVID19_Vaccinations_v5_YGM41_20240927T13005921.dat")
+
+    def test_ensure_dat_extension_with_dat_extension(self):
+        # Test case where file already has a 'dat' extension
+        result = ensure_dat_extension("COVID19_Vaccinations_v5_YGM41_20240927T13005921.dat")
+        self.assertEqual(result, "COVID19_Vaccinations_v5_YGM41_20240927T13005921.dat")
+
+    def test_ensure_dat_extension_without_extension(self):
+        # Test case where file has no extension
+        result = ensure_dat_extension("COVID19_Vaccinations_v5_YGM41_20240927T13005921")
+        self.assertEqual(result, "COVID19_Vaccinations_v5_YGM41_20240927T13005921.dat")

recordprocessor/src/utils_for_recordprocessor.py

@@ -17,10 +17,10 @@ def get_environment() -> str:
 
 
 def get_csv_content_dict_reader(file_key: str) -> DictReader:
-    """Returns the requested file contents from the source bucket in the form of a DictReader"""    
+    """Returns the requested file contents from the source bucket in the form of a DictReader"""
     response = s3_client.get_object(Bucket=os.getenv("SOURCE_BUCKET_NAME"), Key=file_key)
     csv_data = response["Body"].read().decode("utf-8")
-    
+
     # Verify and process the DAT file content coming from MESH
     if '.dat' in file_key:
         csv_data = extract_content(csv_data)
@@ -47,19 +47,20 @@ def invoke_filename_lambda(file_key: str, message_id: str) -> None:
         logger.error("Error invoking filename lambda: %s", error)
         raise
 
+
 def extract_content(dat_file_content):
-    
-    boundary_pattern = re.compile(r'----------------------------\d+')    
-    
+
+    boundary_pattern = re.compile(r'----------------------------\d+')
+
     parts = boundary_pattern.split(dat_file_content)
-    
+
     # Extract the content between the boundaries
     filecontent = None
     for part in parts:
-        if 'Content-Disposition' in part and 'Content-Type' in part:       
+        if 'Content-Disposition' in part and 'Content-Type' in part:
 
             content_start = part.index('Content-Type') + len('Content-Type: text/csv') + 2
-            filecontent =part[content_start:].strip()
+            filecontent = part[content_start:].strip()
             break
-    
+
     return filecontent

recordprocessor/tests/test_utils_for_recordprocessor.py

@@ -15,7 +15,7 @@
 from tests.utils_for_recordprocessor_tests.mock_environment_variables import MOCK_ENVIRONMENT_DICT, BucketNames
 
 with patch("os.environ", MOCK_ENVIRONMENT_DICT):
-    from utils_for_recordprocessor import get_environment, get_csv_content_dict_reader, create_diagnostics_dictionary
+    from utils_for_recordprocessor import get_environment, get_csv_content_dict_reader, create_diagnostics_dictionary, extract_content
 
 
 s3_client = boto3.client("s3", region_name=REGION_NAME)
@@ -76,6 +76,17 @@ def test_create_diagnostics_dictionary(self):
             },
         )
 
+    def test_extract_content_valid_input(self):
+        dat_file_content = (
+            "----------------------------1234567890\n"
+            "Content-Disposition: form-data; name=\"file\"; filename=\"COVID19_Vaccinations_v5_YGM41_20250312T113455981.csv\"\n"
+            "Content-Type: text/csv\n\n"
+            "NHS_NUMBER|PERSON_FORENAME|PERSON_SURNAME|PERSON_DOB|PERSON_GENDER_CODE|PERSON_POSTCODE\n"
+            "----------------------------1234567890\n")
+        expected_content = "NHS_NUMBER|PERSON_FORENAME|PERSON_SURNAME|PERSON_DOB|PERSON_GENDER_CODE|PERSON_POSTCODE"
+        result = extract_content(dat_file_content)
+        self.assertEqual(result, expected_content)
+
 
 if __name__ == "__main__":
     unittest.main()