test_callminer_service

import unittest from unittest.mock import MagicMock, patch from callminer_service import CallMinerService import requests class TestCallMinerService(unittest.TestCase): def setUp(self): self.config = { "call-miner-auth": { "USERNAME": "test_user", "PASSWORD": "test_pass", "APIKEY": "test_key", "SECURITY_URL": "https://fake-security-url", "DATA_URL": "https://fake-data-url" } } self.mongo_service = MagicMock() self.moog_service = MagicMock() self.callminer_service = CallMinerService(self.config, self.mongo_service, self.moog_service) @patch("requests.Session.post") def test_get_jwt_token_success(self, mock_post): mock_response = MagicMock() mock_response.json.return_value = {"token": "jwt_token"} mock_response.raise_for_status = MagicMock() mock_post.return_value = mock_response token = self.callminer_service.get_jwt_token() self.assertEqual(token, {"token": "jwt_token"}) mock_post.assert_called_once_with( url=self.config["call-miner-auth"]["SECURITY_URL"], data='{"Username": "test_user", "Password": "test_pass", "ApiKey": "test_key"}' ) @patch("requests.Session.post") def test_get_jwt_token_failure(self, mock_post): mock_post.side_effect = Exception("Request failed") token = self.callminer_service.get_jwt_token() self.assertIsNone(token) mock_post.assert_called_once() @patch("callminer_service.CallMinerService.get_jwt_token") def test_refresh_token(self, mock_get_jwt_token): mock_get_jwt_token.return_value = {"token": "new_jwt_token"} self.callminer_service.refresh_token() self.assertIn("Authorization", self.callminer_service.session.headers) self.assertEqual(self.callminer_service.session.headers["Authorization"], "JWT {'token': 'new_jwt_token'}") @patch("requests.Session.get") @patch("callminer_service.CallMinerService.refresh_token") def test_get_data_by_meta_search_success(self, mock_refresh_token, mock_get): mock_response = MagicMock() mock_response.json.return_value = [{"data": "record"}] mock_get.return_value = mock_response data = self.callminer_service.get_data_by_meta_search( "2023-01-01", "2023-01-31", "metaField", "metaValue", 1 ) self.assertEqual(data, [{"data": "record"}]) mock_refresh_token.assert_called_once() mock_get.assert_called_once() @patch("requests.Session.get") @patch("callminer_service.CallMinerService.refresh_token") def test_get_data_by_meta_search_failure(self, mock_refresh_token, mock_get): mock_get.side_effect = Exception("Request failed") data = self.callminer_service.get_data_by_meta_search( "2023-01-01", "2023-01-31", "metaField", "metaValue", 1 ) self.assertIsNone(data) mock_refresh_token.assert_called_once() mock_get.assert_called_once() @patch("requests.Session.get") def test_update_token_from_header_with_token(self, mock_get): mock_response = MagicMock() mock_response.headers = {"auth-token-updated": "new_token"} self.callminer_service.update_token_from_header(mock_response) self.assertEqual(self.callminer_service.session.headers["Authorization"], "JWT new_token") @patch("requests.Session.get") def test_update_token_from_header_without_token(self, mock_get): mock_response = MagicMock() mock_response.headers = {} with patch.object(self.callminer_service, "refresh_token") as mock_refresh_token: self.callminer_service.update_token_from_header(mock_response) mock_refresh_token.assert_called_once() @patch("requests.Session.get") @patch("callminer_service.CallMinerService.update_token_from_header") def test_get_call_transcript_with_metadata_success(self, mock_update_token, mock_get): mock_response = MagicMock() mock_response.text = "transcript data" mock_get.return_value = mock_response contact_id_to_metavalues_dict = { "contact_id_1": { "caller_type_vdn": "12345", "client_capture_date_utc": "2023-01-01T00:00:00Z", "client_capture_date_est": "2023-01-01T05:00:00Z", "agent_name": "Agent X", "comit_id": "CMT001", "caller_type_nuance": "Type A", "call_duration_in_secs": 300, "ani": "9876543210", "department_id": "D001", "avaya_ucid": "A001", "transferred": False, "department_name": "Sales", "create_date": "2023-01-01" } } df = self.callminer_service.get_call_transcript_with_metadata(contact_id_to_metavalues_dict) self.assertEqual(len(df), 1) self.assertIn("file_name", df.columns) self.assertEqual(df.iloc[0]["file_name"], "contact_id_1") @patch("callminer_service.CommonUtil.chunker") @patch("requests.Session.get") @patch("callminer_service.CallMinerService.get_call_transcript_with_metadata") def test_run_transcript_by_id(self, mock_get_transcripts, mock_get, mock_chunker): mock_chunker.return_value = [["id1", "id2"]] mock_get.return_value.json.return_value = [{"Contact": {"Id": "id1"}, "Others": {}, "Attributes": {}, "Measures": {}}] mock_get_transcripts.return_value = "transcripts_df" result = self.callminer_service.run_transcript_by_id(["id1", "id2"]) self.assertEqual(result, "transcripts_df") mock_chunker.assert_called_once() mock_get.assert_called_once() mock_get_transcripts.assert_called_once() @patch("requests.Session") def test_get_jwt_token_success(self, mock_session): mock_session_instance = mock_session.return_value mock_response = MagicMock() mock_response.json.return_value = "fake_token" mock_response.raise_for_status = MagicMock() mock_session_instance.post.return_value = mock_response token = self.callminer_service.get_jwt_token() self.assertEqual(token, "fake_token") mock_session_instance.post.assert_called_once_with( url=self.mock_config["call-miner-auth"]["SECURITY_URL"], data='{"Username": "test_user", "Password": "test_password", "ApiKey": "test_api_key"}' ) @patch("requests.Session") def test_get_jwt_token_exception(self, mock_session): mock_session_instance = mock_session.return_value mock_session_instance.post.side_effect = requests.exceptions.RequestException with self.assertLogs(level="INFO") as log: token = self.callminer_service.get_jwt_token() self.assertIsNone(token) self.assertIn("Exception occurred:", log.output[0]) @patch("requests.Session") def test_refresh_token(self, mock_session): mock_session_instance = mock_session.return_value self.callminer_service.get_jwt_token = MagicMock(return_value="new_fake_token") self.callminer_service.refresh_token() self.assertEqual(self.callminer_service.session.headers["Authorization"], "JWT new_fake_token") @patch("requests.Session") def test_get_data_by_meta_search_success(self, mock_session): mock_session_instance = mock_session.return_value mock_response = MagicMock() mock_response.json.return_value = [{"record": "test"}] mock_session_instance.get.return_value = mock_response result = self.callminer_service.get_data_by_meta_search( "2024-01-01T00:00:00", "2024-01-02T00:00:00", "meta_field", "meta_value", 1 ) self.assertEqual(result, [{"record": "test"}]) @patch("requests.Session") def test_get_data_by_meta_search_exception(self, mock_session): mock_session_instance = mock_session.return_value mock_session_instance.get.side_effect = requests.exceptions.RequestException with self.assertLogs(level="INFO") as log: result = self.callminer_service.get_data_by_meta_search( "2024-01-01T00:00:00", "2024-01-02T00:00:00", "meta_field", "meta_value", 1 ) self.assertIsNone(result) self.assertIn("Exception occurred:", log.output[0]) @patch("requests.Session") def test_run_transcript_for_date_range_and_dept(self, mock_session): mock_session_instance = mock_session.return_value mock_response = MagicMock() mock_response.json.return_value = [{"Contact": {"Id": "123"}, "Others": {"ClientCaptureDateUT": "2024-01-01"}}] mock_session_instance.get.return_value = mock_response self.callminer_service.get_call_transcript_with_metadata = MagicMock(return_value=pd.DataFrame()) result = self.callminer_service.run_transcript_for_date_range_and_dept( "2024-01-01T00:00:00", "2024-01-02T00:00:00", "department", None, "field" ) self.assertTrue(isinstance(result, pd.DataFrame)) @patch("your_module.CommonUtil.chunker") @patch("requests.Session") def test_run_transcript_by_id(self, mock_session, mock_chunker): mock_session_instance = mock_session.return_value mock_response = MagicMock() mock_response.json.return_value = [{"Contact": {"Id": "123"}}] mock_session_instance.get.return_value = mock_response mock_chunker.return_value = [["contact_1", "contact_2"]] self.callminer_service.get_call_transcript_with_metadata = MagicMock(return_value=pd.DataFrame()) result = self.callminer_service.run_transcript_by_id(["contact_1", "contact_2"]) self.assertTrue(isinstance(result, pd.DataFrame)) @patch('callminer_service.CallMinerService.get_number_of_pages') @patch('callminer_service.CallMinerService.get_data_by_meta_search') def test_run_transcript_for_vdn_list(self, mock_get_data, mock_get_pages): mock_get_pages.return_value = 100 mock_get_data.side_effect = [[ { "Others": {"vdn_lookup_field": "test_vdn", "ClientCaptureDateUT": "2023-01-01", "CreateDate": "2023-01-01"}, "Attributes": {"Agent": "Test Agent"}, "Measures": {"WAVLength": 300}, "Contact": {"Id": "12345"} } ]] result = self.callminer_service.run_transcript_for_vdn_list( start_day="2023-01-01", end_day="2023-01-02", vdn_list=["test_vdn"], vdn_lookup_field="vdn_lookup_field") self.assertIn("12345", result) def test_get_number_of_pages(self): with patch('callminer_service.CallMinerService.get_data_by_meta_search', return_value=[{ "RecordInfo": {"TotalRowCount": 250} }]): rows = self.callminer_service.get_number_of_pages("2023-01-01", "2023-01-02", "key", "value") self.assertEqual(rows, 250) def test_get_calls_from_mongo(self): self.mongo_service.execute_query.return_value = [ {"call_miner_id": "123", "source_payload": {"Others": {"key": "value"}}} ] result = self.callminer_service.get_calls_from_mongo("2023-01-01T00:00:00", "2023-01-02T00:00:00", ["vdn"], "collection", "vdn_field") self.assertIn("123", result) def test_get_calls_from_mongo_with_transcript(self): self.mongo_service.execute_query.return_value = [ {"call_miner_id": "123", "source_payload": {"transcript": "test_transcript", "Others": {"key": "value"}}} ] df = self.callminer_service.get_calls_from_mongo_with_transcript("2023-01-01T00:00:00", "2023-01-02T00:00:00", ["vdn"], "collection", "vdn_field") self.assertEqual(len(df), 1) self.assertIn("123", df["file_name"].values) def test_get_call_id_from_intent_action(self): self.mongo_service.execute_query.return_value = [{"call_id": "123"}] result = self.callminer_service.get_call_id_from_intent_action( "2023-01-01T00:00:00", "2023-01-02T00:00:00", ["vdn"], "provider", {"prompt_name": "test_prompt"}, "collection") self.assertEqual(result, ["123"]) @patch('callminer_service.logging.info') def test_logging_exception(self, mock_logging): self.mongo_service.execute_query.side_effect = Exception("Test exception") result = self.callminer_service.get_calls_from_mongo("2023-01-01T00:00:00", "2023-01-02T00:00:00", ["vdn"], "collection", "vdn_field") self.assertIsNone(result) self.assertTrue(mock_logging.called) if __name__ == "__main__": unittest.main()

Posted By