test_statistics_profiler_service

import unittest from unittest.mock import MagicMock, patch from statistics_profiler_service import StatisticsProfilerService from modelclasses.statistics_profiler_request import StatisticsProfilerRequest class TestStatisticsProfilerService(unittest.TestCase): def setUp(self): self.oracle_service = MagicMock() self.profiler_service = StatisticsProfilerService(self.oracle_service) self.request = StatisticsProfilerRequest( request_type="regular", table_id=1, table_name="TEST_TABLE", start_time="2023-01-01 00:00:00", end_time="2023-12-31 23:59:59", scan_period="daily" ) def test_get_table_detail(self): self.oracle_service.execute_query.return_value = [(1, 'TEST_TABLE')] result = self.profiler_service.get_table_detail(self.request) self.assertEqual(result, {1: 'TEST_TABLE'}) self.oracle_service.execute_query.assert_called_once() def test_call_statistics_function_single_partition(self): self.oracle_service.execute_query.return_value = [(45000,)] self.profiler_service.table_statistics = MagicMock(return_value='Success') self.profiler_service.update_std_dev_min = MagicMock() result = self.profiler_service.call_statistics_function(self.request) self.assertEqual(result, 'Success') self.profiler_service.table_statistics.assert_called_once() self.profiler_service.update_std_dev_min.assert_called_once() def test_call_statistics_function_multiple_partitions(self): self.oracle_service.execute_query.return_value = [(120000,)] self.profiler_service.table_statistics_by_partition = MagicMock(return_value='Success') self.profiler_service.update_std_dev_min = MagicMock() result = self.profiler_service.call_statistics_function(self.request) self.assertEqual(result, 'Success') self.profiler_service.table_statistics_by_partition.assert_called_once() self.profiler_service.update_std_dev_min.assert_called_once() def test_call_statistics_function_no_dictionary_id(self): self.oracle_service.execute_query.return_value = [(None,)] result = self.profiler_service.call_statistics_function(self.request) self.assertEqual(result, 'Success') def test_run_statistics_profiler_self_service(self): self.request.request_type = "self-service" self.profiler_service.call_statistics_function = MagicMock(return_value="Success") result = self.profiler_service.run_statistics_profiler(self.request) self.assertEqual(result, "Success") def test_run_statistics_profiler_regular(self): self.oracle_service.execute_query.return_value = [(1, "TEST_TABLE")] self.profiler_service.call_statistics_function = MagicMock(return_value="Success") result = self.profiler_service.run_statistics_profiler(self.request) self.assertEqual(result, "Success") def test_table_statistics_by_partition(self): self.oracle_service.execute_query.return_value = [(1, 2, 3.0, 4.0, 0.0, 1.0, 5.0, 25.0, 50.0, 75.0, 95.0, 99.0, 100.0)] self.profiler_service.upsert_statistics_to_db = MagicMock(return_value="Success") result = self.profiler_service.table_statistics_by_partition(self.request, 2) self.assertEqual(result, "Success") def test_table_statistics(self): self.oracle_service.execute_query.return_value = [(1, 2, 3.0, 4.0, 0.0, 1.0, 5.0, 25.0, 50.0, 75.0, 95.0, 99.0, 100.0)] self.profiler_service.upsert_statistics_to_db = MagicMock(return_value="Success") result = self.profiler_service.table_statistics(self.request) self.assertEqual(result, "Success") def test_upsert_statistics_to_db_success(self): self.oracle_service.upsert_df_to_db = MagicMock() result = self.profiler_service.upsert_statistics_to_db(self.request, MagicMock()) self.assertEqual(result, "Success") def test_upsert_statistics_to_db_failure(self): self.oracle_service.upsert_df_to_db.side_effect = Exception("DB Error") result = self.profiler_service.upsert_statistics_to_db(self.request, MagicMock()) self.assertEqual(result, "Failed") def test_update_std_dev_min(self): self.oracle_service.execute_query.return_value = [(1,), (2,)] self.profiler_service.update_std_dev_min(self.request) self.assertEqual(self.oracle_service.execute_update_query.call_count, 2) if __name__ == '__main__': unittest.main()

Posted By