# test_audio_deletion_service.py

# test_audio_deletion_service.py import unittest from unittest.mock import Mock, patch from audio_deletion_service import AudioDeletionService from datetime import datetime import pandas as pd class TestAudioDeletionService(unittest.TestCase): def setUp(self): # Mock configuration and external dependencies config = Mock() self.call_miner_util = Mock() self.oracle_service = Mock() self.mongo_service = Mock() self.s3_service = Mock() self.auth = Mock() self.config_service = Mock() self.cert_service = Mock() self.workflow_service = Mock() self.audit_service = Mock() # Initialize AudioDeletionService with mocked dependencies self.audio_service = AudioDeletionService( config, self.call_miner_util, self.oracle_service, self.mongo_service, self.s3_service, self.auth, self.config_service, self.cert_service, self.workflow_service, self.audit_service ) @patch("audio_deletion_service.logger") def test_delete_audio_from_s3_no_files(self, mock_logger): # Test when no files are found for deletion trace_id = "test_trace" start_time = int(datetime.now().timestamp()) end_time = start_time + 1000 collection = "test_collection" # Mock MongoDB query to return an empty DataFrame self.audio_service.get_audio_paths_from_mongo = Mock(return_value=pd.DataFrame()) # Execute the method self.audio_service.delete_audio_from_s3(collection, start_time, end_time, trace_id) # Verify that no deletion occurred and appropriate log was generated mock_logger.assert_any_call( 'info', trace_id, 'delete_audio_from_s3', f'No files in collection "{collection}" between {datetime.utcfromtimestamp(start_time)} and {datetime.utcfromtimestamp(end_time)}' ) self.s3_service.delete_from_bucket.assert_not_called() self.audio_service.check_and_update_scheduler.assert_called_once_with(None, None, trace_id, "Processing") @patch("audio_deletion_service.logger") def test_delete_audio_from_s3_successful_deletion(self, mock_logger): # Test successful deletion case trace_id = "test_trace" start_time = int(datetime.now().timestamp()) end_time = start_time + 1000 collection = "test_collection" # Mock MongoDB query to return audio paths audio_paths_df = pd.DataFrame({"audio_path": ["path1/audio1.mp3", "path2/audio2.mp3"]}) self.audio_service.get_audio_paths_from_mongo = Mock(return_value=audio_paths_df) # Mock S3 delete_from_bucket to return the number of deleted files self.s3_service.delete_from_bucket = Mock(return_value=2) # Execute the method self.audio_service.delete_audio_from_s3(collection, start_time, end_time, trace_id) # Verify logs and scheduler updates mock_logger.assert_any_call( 'info', trace_id, 'delete_audio_from_s3', f'Beginning deletion of 2 audio files in S3 between {datetime.utcfromtimestamp(start_time)} and {datetime.utcfromtimestamp(end_time)}' ) mock_logger.assert_any_call( 'info', trace_id, 'delete_audio_from_s3', 'Successfully deleted 2 audio files from S3' ) self.audio_service.check_and_update_scheduler.assert_any_call( None, None, trace_id, "Processing" ) self.audio_service.check_and_update_scheduler.assert_any_call( None, None, trace_id, "Completed", 2, 2, 0 ) @patch("audio_deletion_service.logger") def test_delete_audio_from_s3_exception_handling(self, mock_logger): # Test exception handling during audio deletion trace_id = "test_trace" start_time = int(datetime.now().timestamp()) end_time = start_time + 1000 collection = "test_collection" # Simulate an exception in delete_from_bucket self.s3_service.delete_from_bucket.side_effect = Exception("S3 Deletion Failed") # Mock MongoDB to return a non-empty DataFrame audio_paths_df = pd.DataFrame({"audio_path": ["path1/audio1.mp3", "path2/audio2.mp3"]}) self.audio_service.get_audio_paths_from_mongo = Mock(return_value=audio_paths_df) # Execute the method and verify error handling with self.assertRaises(Exception): self.audio_service.delete_audio_from_s3(collection, start_time, end_time, trace_id) mock_logger.assert_any_call("error", trace_id, "executor", "Exception occurred in delete_audio_from_s3") self.audio_service.check_and_update_scheduler.assert_any_call(None, None, trace_id, "Failed") def test_get_audio_paths_from_mongo(self): # Test get_audio_paths_from_mongo to ensure MongoDB query is executed and formatted correctly collection = "test_collection" start_time = datetime(2023, 1, 1) end_time = datetime(2023, 1, 2) # Mock the MongoDB response mock_data = [ {"audio_path": "path/to/audio1.mp3"}, {"audio_path": "path/to/audio2.mp3"} ] self.mongo_service.execute_query = Mock(return_value=mock_data) # Execute the method df = self.audio_service.get_audio_paths_from_mongo(collection, start_time, end_time) # Verify the result is a DataFrame with expected data expected_df = pd.DataFrame(mock_data) pd.testing.assert_frame_equal(df, expected_df) self.mongo_service.execute_query.assert_called_once_with( collection, {'client_capture_date': {"$gte": start_time, "$lte": end_time}}, {'audio_path': 1} )

Posted By