test_discovery_service

import unittest from unittest.mock import patch, MagicMock import pandas as pd from DiscoveryService import DiscoveryService class TestDiscoveryService(unittest.TestCase): @classmethod def setUpClass(cls): cls.config = { 'cloud-ai': { 'CLUSTERING_URL': 'http://clustering.url', 'GET_LABEL_URL': 'http://label2.url', 'GET_MULTIPLE_LABELS_URL': 'http://label1.url' } } cls.oracle_service = MagicMock() cls.audit_service = MagicMock() cls.moog_service = MagicMock() cls.string_id_util = MagicMock() cls.discovery_service = DiscoveryService(cls.config, cls.oracle_service, cls.audit_service, cls.moog_service, cls.string_id_util) @patch('DiscoveryService.requests.Session') def test_call_cloud_ai(self, mock_session): # Mocking requests response mock_response = MagicMock() mock_response.json.return_value = {"Data": ["cluster1", "cluster2"]} mock_session.return_value.post.return_value = mock_response mock_session.return_value.get.return_value = mock_response # Test with POST method result = self.discovery_service.call_cloud_ai("http://dummy.url", "post", {"data": "test"}) self.assertEqual(result.json(), {"Data": ["cluster1", "cluster2"]}) # Test with GET method result = self.discovery_service.call_cloud_ai("http://dummy.url", "get", {"data": "test"}) self.assertEqual(result.json(), {"Data": ["cluster1", "cluster2"]}) def test_list_to_string(self): # Test for joining list into string result = self.discovery_service.list_to_string(['item1', 'item2', 'item3']) self.assertEqual(result, 'item1, item2, item3') def test_split_data_by_type(self): # Create a mock dataframe data = { 'AGENT_ACTION': ['action1', 'action2'], 'CALL_INTENT': ['intent1', 'intent2'], 'REASON_FOR_CALLING': ['reason1', 'reason2'], 'CALL_ID': [1, 2], 'CALLER_TYPE': ['type1', 'type2'], 'CLIENT_CODE': ['code1', 'code2'], 'LOB': ['lob1', 'lob2'] } df = pd.DataFrame(data) agent_action_data, call_intent_data, reason_for_calling_data = self.discovery_service.split_data_by_type(df) # Verify split data self.assertEqual(agent_action_data['Data'].tolist(), ['action1', 'action2']) self.assertEqual(call_intent_data['Data'].tolist(), ['intent1', 'intent2']) self.assertEqual(reason_for_calling_data['Data'].tolist(), ['reason1', 'reason2']) @patch('DiscoveryService.DiscoveryService.call_cloud_ai') def test_get_category_2_labels(self, mock_call_cloud_ai): # Mock API response mock_call_cloud_ai.return_value.json.return_value = {'categories': ['cat1', 'cat2']} result = self.discovery_service.get_category_2_labels("sample list") self.assertEqual(result, {'categories': ['cat1', 'cat2']}) @patch('DiscoveryService.DiscoveryService.call_cloud_ai') def test_get_category_1_labels(self, mock_call_cloud_ai): # Mock API response mock_call_cloud_ai.return_value.json.return_value = {'categories': ['cat1', 'cat2']} result = self.discovery_service.get_category_1_labels("sample list", 10) self.assertEqual(result, {'categories': ['cat1', 'cat2']}) @patch('DiscoveryService.DiscoveryService.get_data') @patch('DiscoveryService.DiscoveryService.split_data_by_type') @patch('DiscoveryService.DiscoveryService.call_cloud_ai') def test_executor(self, mock_call_cloud_ai, mock_split_data, mock_get_data): # Setup mock returns mock_get_data.return_value = pd.DataFrame({ 'CALL_INTENT': ['intent1', 'intent2'], 'REASON_FOR_CALLING': ['reason1', 'reason2'], 'AGENT_ACTION': ['action1', 'action2'], 'CALL_ID': [1, 2], 'CALLER_TYPE': ['type1', 'type2'], 'CLIENT_CODE': ['code1', 'code2'], 'LOB': ['lob1', 'lob2'], 'INSERTION_TIME': [datetime.now(), datetime.now()] }) mock_split_data.return_value = ( pd.DataFrame({'Data': ['action1', 'action2']}), pd.DataFrame({'Data': ['intent1', 'intent2']}), pd.DataFrame({'Data': ['reason1', 'reason2']}) ) # Mock responses for clustering mock_call_cloud_ai.return_value.json.side_effect = [ {'Data': {'0': ['action1'], '1': ['action2']}}, {'Data': {'0': ['intent1'], '1': ['intent2']}}, {'Data': {'0': ['reason1'], '1': ['reason2']}} ] # Run executor self.discovery_service.executor('metric_scan_id', 'client_code', 'insertion_time', 'start_date', 'end_date', 'lob', 'caller_type') # Check that inserts were called self.assertTrue(self.oracle_service.execute_temp_insert_query.called) @patch('DiscoveryService.DiscoveryService.get_data') def test_get_data_exception(self, mock_get_data): # Simulate an exception and check if audit service is called mock_get_data.side_effect = Exception("Database Error") with self.assertLogs('DiscoveryService', level='INFO') as log: self.discovery_service.get_data('insertion_time', 'min_date', 'max_date', 'metric_scan_id', 'client_code', 'lob', 'caller_type') self.assertIn("Database Error", log.output[-1]) self.audit_service.update_audit_entry.assert_called_with('metric_scan_id', ['STATUS'], ["'Failed'"]) if __name__ == '__main__': unittest.main()

Posted By