test_profile_service_3

import unittest from unittest.mock import MagicMock, patch from collections import defaultdict from fastapi.responses import JSONResponse from fastapi import BackgroundTasks import pandas as pd import pymongo from modelclasses.profiler_request import ProfilerRequest from modelclasses.get_profiler_request import GetProfilerRequest class TestProfilerService(unittest.TestCase): def setUp(self): self.config = {"db": "test_db"} self.mongo_service = MagicMock() self.oracle_service = MagicMock() self.profiler_service = ProfilerService(self.config, mongo_service=self.mongo_service) @patch("module_name.convert_df_obj_columns") # Replace module_name with the actual module @patch("module_name.create_mongo_dict") @patch("module_name.create_base_custom_dict") @patch("module_name.fill_analysis") @patch("module_name.replace_period_with_hyphen") @patch("module_name.reduce_word_counts_size") @patch("module_name.fill_use_for_anomaly") @patch("module_name.convert_epoch_to_dt") @patch("module_name.convert_percent_to_p") def test_run_profiler( self, mock_convert_percent_to_p, mock_convert_epoch_to_dt, mock_fill_use_for_anomaly, mock_reduce_word_counts_size, mock_replace_period_with_hyphen, mock_fill_analysis, mock_create_base_custom_dict, mock_create_mongo_dict, mock_convert_df_obj_columns, ): profiler_request = ProfilerRequest( datasetId="test_dataset", requestId="test_request", UUID="test_uuid", ) # Mocking dependencies self.oracle_service.execute_query.side_effect = [ [{"DATASET_NAME": "test_collection"}], [{"DATASET_TIMESTAMP_COLUMN": "test_timestamp"}], ] self.mongo_service.execute_aggregate_pipeline.side_effect = [ [{"numFields": 10}], [{"METRIC_TIME": 1629936000}, {"METRIC_TIME": 1629939600}], ] mock_convert_df_obj_columns.return_value = ( pd.DataFrame({"col1": [1, 2, 3]}), "mock_resolution", [{"col_name": "col1", "mean_len": 10, "dtype": "str"}], ) mock_create_mongo_dict.return_value = {} mock_create_base_custom_dict.return_value = defaultdict(dict) result = self.profiler_service.run_profiler(profiler_request) # Assertions self.assertEqual(result, "Success") self.mongo_service.upsert_query.assert_called_once() mock_convert_df_obj_columns.assert_called_once() def test_profiler_request_success(self): profiler_request = ProfilerRequest( datasetId="test_dataset", requestId="test_request", UUID="test_uuid", ) background_task = MagicMock() response = self.profiler_service.profiler_request(profiler_request, background_task) self.assertIsInstance(response, JSONResponse) self.assertEqual(response.status_code, 200) background_task.add_task.assert_called_once() def test_profiler_request_failure(self): profiler_request = ProfilerRequest( datasetId="test_dataset", requestId="test_request", UUID="test_uuid", ) background_task = MagicMock() self.profiler_service.run_profiler = MagicMock(side_effect=Exception("Test exception")) response = self.profiler_service.profiler_request(profiler_request, background_task) self.assertIsInstance(response, JSONResponse) self.assertEqual(response.status_code, 500) def test_get_profiler_no_results(self): self.mongo_service.execute_query.return_value = [] profiler_request = GetProfilerRequest( datasetId="test_dataset", requestId=None, UUID=None, ) response = self.profiler_service.get_profiler(profiler_request) self.assertIsInstance(response, JSONResponse) self.assertEqual(response.status_code, 404) self.assertIn("Error", response.body.decode()) def test_get_profiler_with_results(self): query_results = [ { "requestId": "test_request", "datasetId": "test_dataset", "uuid": "test_uuid", "data": {}, } ] self.mongo_service.execute_query.return_value = query_results profiler_request = GetProfilerRequest( datasetId="test_dataset", requestId="test_request", UUID=None, ) response = self.profiler_service.get_profiler(profiler_request) self.assertIsInstance(response, JSONResponse) self.assertEqual(response.status_code, 200) self.assertIn("test_request", response.body.decode()) if __name__ == "__main__": unittest.main()

Posted By