test_write_to_mongo

import unittest from unittest.mock import MagicMock, patch import pandas as pd import datetime import pytz from your_module import WriteDataFromMongoToOracle # Replace with the actual module name class TestWriteDataFromMongoToOracle(unittest.TestCase): def setUp(self): # Mock the services and configuration dictionary self.config = { 'cloud-ai': { 'EMBEDDING_URL': 'http://embedding.url', 'LABELLING_URL': 'http://labelling.url', 'GET_STRING_ID_URL': 'http://string_id.url', 'CLASSIFICATION_URL': 'http://classification.url' }, 'email': { 'cc-intent-recipients': ['recipient@example.com'] } } self.mongo_service = MagicMock() self.oracle_service = MagicMock() self.moog_service = MagicMock() self.email_service = MagicMock() self.config_service = MagicMock() # Instantiate the main class with mocked dependencies self.instance = WriteDataFromMongoToOracle( self.config, self.mongo_service, self.oracle_service, self.moog_service, self.email_service, self.config_service ) @patch('your_module.datetime') # Mock datetime for consistency def test_get_model_results_from_mongo_with_valid_data(self, mock_datetime): # Set up mock data and time mock_datetime.datetime.fromtimestamp.return_value = datetime.datetime(2023, 1, 1, tzinfo=pytz.UTC) data = [{'run_time': '2023-01-01T00:00:00Z', 'call_id': '123', 'model_results': {'callSummary': 'summary'}}] self.mongo_service.execute_query.return_value = data # Call the method and verify results result = self.instance.get_model_results_from_mongo( 'client_code', None, 1672444800, 1672531200, 'trace_id', 'cc_action_collection' ) self.assertEqual(len(result), 1) self.assertIn('callID', result.columns) self.mongo_service.execute_query.assert_called_once() def test_create_vdn_caller_type_mapping(self): # Mock the oracle service response query_response = "[(1234, 'Type1'), (5678, 'Type2')]" self.oracle_service.execute_select_query.return_value = query_response # Call method vdn_mapping = self.instance.create_vdn_caller_type_mapping('client_code', 'trace_id') # Assert response matches expected dictionary expected_dict = {1234: 'Type1', 5678: 'Type2'} self.assertEqual(vdn_mapping, expected_dict) def test_split_data_on_vdn_type(self): # Mock input DataFrame data = { 'CALLER_TYPE': ['Shareholder', 'Rep'], 'callID': [1, 2] } model_tagged_data = pd.DataFrame(data) # Call method customer_data, rep_data = self.instance.split_data_on_vdn_type(model_tagged_data) # Verify customer and rep data separation self.assertEqual(len(customer_data), 1) self.assertEqual(len(rep_data), 1) self.assertEqual(customer_data.iloc[0]['CALLER_TYPE'], 'Shareholder') self.assertEqual(rep_data.iloc[0]['CALLER_TYPE'], 'Rep') def test_create_model_results_df(self): # Mock input data data = { 'callID': [1], 'model_results': [{'callSummary': 'summary'}], 'caller_type_vdn': [1234], 'ani': ['None'] } model_data = pd.DataFrame(data) vdn_caller_type_dict = {1234: 'Shareholder'} # Call method result_df = self.instance.create_model_results_df(model_data, vdn_caller_type_dict, 'client_code', '2023-01-01', 'trace_id') # Check output DataFrame structure and values self.assertIn('callID', result_df.columns) self.assertIn('CALLER_TYPE', result_df.columns) self.assertEqual(result_df.iloc[0]['CALLER_TYPE'], 'Shareholder') @patch('your_module.logger') def test_executor_updates_status_to_completed(self, mock_logger): # Mock the necessary data and methods self.mongo_service.execute_query.return_value = [] self.oracle_service.execute_query = MagicMock() # Call the method self.instance.executor('client_code', 'run_times', 'trace_id', scan_table='scan_table', scan_id='scan_id') # Verify that it updates the status to 'Completed' update_query_call_args = self.oracle_service.execute_query.call_args_list self.assertIn('Completed', update_query_call_args[1][0][0]) @patch('your_module.time') def test_timer_decorator(self, mock_time): mock_time.time.side_effect = [1, 3] # Mocking start and end times @self.instance.timer def sample_func(): return "Execution Complete" result = sample_func() self.assertEqual(result, "Execution Complete") # Assuming `logging.info` is configured, validate the logging behavior if needed if __name__ == '__main__': unittest.main()

Posted By