test_task_service

import unittest from unittest.mock import MagicMock, patch from task_service import TaskService from modelclasses.profiler_request import ProfilerRequest from modelclasses.statistics_profiler_request import StatisticsProfilerRequest class TestTaskService(unittest.TestCase): def setUp(self): # Mock dependencies self.mock_config = {} self.mock_oracle_service = MagicMock() self.mock_profiling_service = MagicMock() self.mock_statistics_profiler_service = MagicMock() # Set up environment variables patch.dict('os.environ', { 'GROUP_NAME': 'TestGroup', 'RUN_TYPE': 'profiler' }).start() # Create an instance of TaskService self.task_service = TaskService( self.mock_config, self.mock_oracle_service, self.mock_profiling_service, self.mock_statistics_profiler_service ) def tearDown(self): patch.stopall() def test_run_task_with_profiler(self): # Mock Oracle service responses self.mock_oracle_service.call_stored_proc.return_value = 'test-uuid' self.mock_oracle_service.execute_query.return_value = [('test-uuid', '{"attribute": "value"}')] self.mock_profiling_service.run_profiler.return_value = 'Success' # Run the task self.task_service.run_task() # Verify Oracle service and profiling service calls self.mock_oracle_service.call_stored_proc.assert_called_once() self.mock_oracle_service.execute_query.assert_called_once() self.mock_profiling_service.run_profiler.assert_called_once_with( profiler_request=ProfilerRequest(attribute='value') ) # Verify task status update self.mock_oracle_service.execute_update_query.assert_called_with( "UPDATE JOB_TASK SET UPDATE_TIME = systimestamp, STATUS = 'Completed' where UUID ='test-uuid'" ) def test_run_task_no_uuid_returned(self): # Simulate no UUID returned self.mock_oracle_service.call_stored_proc.return_value = None # Run the task result = self.task_service.run_task() # Verify no profiling task runs self.assertFalse(result) self.mock_profiling_service.run_profiler.assert_not_called() self.mock_oracle_service.execute_query.assert_not_called() def test_update_task_status(self): # Update task status self.task_service.task_dict = {"uuid": "test-uuid"} self.task_service.update_task_status("Completed") # Verify query execution self.mock_oracle_service.execute_update_query.assert_called_once_with( "UPDATE JOB_TASK SET UPDATE_TIME = systimestamp, STATUS = 'Completed' where UUID ='test-uuid'" ) def test_run_task_with_invalid_request(self): # Mock Oracle service responses with invalid data self.mock_oracle_service.call_stored_proc.return_value = 'test-uuid' self.mock_oracle_service.execute_query.return_value = [('test-uuid', '{"invalid": "data"}')] # Mock the profiler to raise a validation error self.mock_profiling_service.run_profiler.side_effect = ValidationError # Run the task self.task_service.run_task() # Verify failure status update self.mock_oracle_service.execute_update_query.assert_called_with( "UPDATE JOB_TASK SET UPDATE_TIME = systimestamp, STATUS = 'Failed' where UUID ='test-uuid'" ) def test_enrich_ss_request_object(self): # Mock Oracle responses for table and request info self.mock_oracle_service.execute_query.side_effect = [ [('table_id_1', 'table_name_1', 'BASE')], [('2023-01-01', '2023-12-31')] ] # Mock request object request = {'datasetId': 123, 'requestId': 456} # Enrich the request enriched_request = self.task_service.enrich_ss_request_object(request) # Verify enriched values self.assertEqual(enriched_request['table_id'], 'table_id_1') self.assertEqual(enriched_request['start_time'], '2023-01-01') self.assertEqual(enriched_request['end_time'], '2023-12-31')

Posted By