test_flatten_mongo_script

import unittest from unittest.mock import Mock, patch from datetime import datetime from your_module import flatten_mongo_script # Replace 'your_module' with the actual module name class TestFlattenMongoScript(unittest.TestCase): def setUp(self): # Set up mock services and config self.mock_mongo_service = Mock() self.mock_oracle_service = Mock() self.mock_config = { 'cloud-ai': { 'EMBEDDING_URL': 'http://embedding-url.com', 'LABELLING_URL': 'http://labelling-url.com', 'GET_STRING_ID_URL': 'http://string-id-url.com', 'CLASSIFICATION_URL': 'http://classification-url.com' } } # Initialize flatten_mongo_script with mocked services self.flatten_script = flatten_mongo_script(self.mock_config, self.mock_mongo_service, self.mock_oracle_service) def test_fetch_mongo_data_success(self): # Test case: fetching data from Mongo successfully mock_query_result = [{"client_code": "test_client", "user_id": "test_user", "event_source": "EUM_TA"}] self.mock_mongo_service.execute_query.return_value = mock_query_result # Call the method result = self.flatten_script.fetch_mongo_data("LOB_TEST", "test_client", "test_user", "2024-01-01", "2024-12-31") # Assertions self.mock_mongo_service.execute_query.assert_called_once_with('events_data_ta', { "client_code": "test_client", "user_id": "test_user", "event_source": "EUM_TA" }) print("Result:", result) # Output the results if needed def test_fetch_mongo_data_no_results(self): # Test case: no results found in Mongo self.mock_mongo_service.execute_query.return_value = [] # Call the method result = self.flatten_script.fetch_mongo_data("LOB_TEST", "test_client", "test_user", "2024-01-01", "2024-12-31") # Assertions self.mock_mongo_service.execute_query.assert_called_once_with('events_data_ta', { "client_code": "test_client", "user_id": "test_user", "event_source": "EUM_TA" }) self.assertEqual(result, None) # Expect None if no results def test_fetch_mongo_data_invalid_dates(self): # Test case: Invalid date format provided with self.assertRaises(ValueError): self.flatten_script.fetch_mongo_data("LOB_TEST", "test_client", "test_user", "invalid_date", "2024-12-31") def test_fetch_mongo_data_error_handling(self): # Test case: MongoDB service throws an exception self.mock_mongo_service.execute_query.side_effect = Exception("MongoDB error") with self.assertRaises(Exception) as context: self.flatten_script.fetch_mongo_data("LOB_TEST", "test_client", "test_user", "2024-01-01", "2024-12-31") self.assertEqual(str(context.exception), "MongoDB error")

Posted By