From: Taewan Kim Date: Mon, 5 Jun 2023 09:04:21 +0000 (+0900) Subject: Add and update unit tests of TM operations X-Git-Tag: 1.1.0~1 X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=commitdiff_plain;h=refs%2Fchanges%2F61%2F11261%2F1;p=aiml-fw%2Fawmf%2Ftm.git Add and update unit tests of TM operations - `delete_dme_filtered_data_job` added - `training_start` and `create_dme_filtered_data_job` updated - Unnecessary `class Test_upload_pipeline` deleted Issue-ID: AIMLFW-13 Change-Id: I1f2cbc4d25c5fff3ccbdcef4c5d3845ffdf0e1f1 Signed-off-by: Taewan Kim --- diff --git a/tests/test_trainingmgr_operations.py b/tests/test_trainingmgr_operations.py index 9bc1ab9..c8ae330 100644 --- a/tests/test_trainingmgr_operations.py +++ b/tests/test_trainingmgr_operations.py @@ -32,66 +32,66 @@ from threading import Lock from trainingmgr import trainingmgr_main from trainingmgr.common import trainingmgr_operations from trainingmgr.common.tmgr_logger import TMLogger +from trainingmgr.common.exceptions_utls import TMException +from trainingmgr.common.trainingmgr_util import MIMETYPE_JSON from trainingmgr.common.trainingmgr_config import TrainingMgrConfig trainingmgr_main.LOGGER = pytest.logger trainingmgr_main.LOCK = Lock() trainingmgr_main.DATAEXTRACTION_JOBS_CACHE = {} class DummyVariable: - kf_adapter_ip = "localhost" - kf_adapter_port = 5001 - logger = trainingmgr_main.LOGGER + kf_adapter_ip = "localhost" + kf_adapter_port = 5001 + logger = trainingmgr_main.LOGGER class Test_training_start: def setup_method(self): self.client = trainingmgr_main.APP.test_client(self) self.logger = trainingmgr_main.LOGGER - - def test_negative_training_start(self): - training_config_obj = DummyVariable() + + ts_result = Response() + ts_result.status_code = status.HTTP_200_OK + ts_result.headers={'content-type': MIMETYPE_JSON} + @patch('trainingmgr.common.trainingmgr_operations.requests.post', return_value = ts_result) + def test_success(self, mock1): + trainingjob_name = "usecase12" dict_data = { - "brand": "Ford", - "model": "Mustang", - "year": 1964 - } + "pipeline_name": "qoe", + "experiment_name": "default", + "arguments": "{epoches : 1}", + "pipeline_version": 1 + } + training_config_obj = DummyVariable() + try: + response = trainingmgr_operations.training_start(training_config_obj,dict_data,trainingjob_name) + assert response.headers['content-type'] == MIMETYPE_JSON + assert response.status_code == status.HTTP_200_OK + except Exception: + assert False + + def test_fail(self): trainingjob_name = "usecase12" - expected_data = { - "brand": "Ford", - "model": "Mustang", - "year": 1964 - } + dict_data = { + "pipeline_name": "qoe", + "experiment_name": "default", + "arguments": "{epoches : 1}", + "pipeline_version": 1 + } + training_config_obj = DummyVariable() try: response = trainingmgr_operations.training_start(training_config_obj,dict_data,trainingjob_name) - assert response == expected_data,"data not equal" assert False - except Exception: + except requests.exceptions.ConnectionError: assert True - -class Test_upload_pipeline: - def setup_method(self): - self.client = trainingmgr_main.APP.test_client(self) - self.logger = trainingmgr_main.LOGGER - - mocked_TRAININGMGR_CONFIG_OBJ=mock.Mock(name="TRAININGMGR_CONFIG_OBJ") - attrs_TRAININGMGR_CONFIG_OBJ = {'kf_adapter_ip.return_value': '123', 'kf_adapter_port.return_value' : '100'} - mocked_TRAININGMGR_CONFIG_OBJ.configure_mock(**attrs_TRAININGMGR_CONFIG_OBJ) - @patch('trainingmgr.trainingmgr_main.TRAININGMGR_CONFIG_OBJ', return_value = mocked_TRAININGMGR_CONFIG_OBJ) - def test_upload_pipeline_negative(self, mock1): - expected_data = "result" - trainingjob_req = { - "pipe_name":"usecase1", - } - response = self.client.post("/pipelines//upload".format("usecase1"), data=json.dumps(trainingjob_req), - content_type="application/json") - assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR - assert expected_data in response.json.keys() + except Exception: + assert False class Test_create_dme_filtered_data_job: the_response=Response() the_response.status_code=status.HTTP_201_CREATED @patch('trainingmgr.common.trainingmgr_operations.requests.put', return_value=the_response) - def test_create_dme_filtered_data_job(self, mock1): + def test_success(self, mock1): mocked_TRAININGMGR_CONFIG_OBJ=mock.Mock(name="TRAININGMGR_CONFIG_OBJ") attrs_TRAININGMGR_CONFIG_OBJ = {'kf_adapter_ip.return_value': '123', 'kf_adapter_port.return_value' : '100'} mocked_TRAININGMGR_CONFIG_OBJ.configure_mock(**attrs_TRAININGMGR_CONFIG_OBJ) @@ -104,10 +104,9 @@ class Test_create_dme_filtered_data_job: host="10.0.0.50" port="31840" response=trainingmgr_operations.create_dme_filtered_data_job(mocked_TRAININGMGR_CONFIG_OBJ, source_name, db_org, bucket_name, token, features, feature_group_name, host, port) - assert response.status_code==201, "create_dme_filtered_data_job failed" + assert response.status_code==status.HTTP_201_CREATED, "create_dme_filtered_data_job failed" - @patch('trainingmgr.common.trainingmgr_operations.create_url_host_port', side_effect=Exception("Mocked Error")) - def test_negative_create_dme_filtered_data_job(self, mock1): + def test_create_url_host_port_fail(self): mocked_TRAININGMGR_CONFIG_OBJ=mock.Mock(name="TRAININGMGR_CONFIG_OBJ") attrs_TRAININGMGR_CONFIG_OBJ = {'kf_adapter_ip.return_value': '123', 'kf_adapter_port.return_value' : '100'} mocked_TRAININGMGR_CONFIG_OBJ.configure_mock(**attrs_TRAININGMGR_CONFIG_OBJ) @@ -117,10 +116,42 @@ class Test_create_dme_filtered_data_job: token="" features=[] feature_group_name="test" - host="10.0.0.50" + host="url error" port="31840" try: response=trainingmgr_operations.create_dme_filtered_data_job(mocked_TRAININGMGR_CONFIG_OBJ, source_name, db_org, bucket_name, token, features, feature_group_name, host, port) assert False + except TMException as err: + assert "URL validation error: " in err.message except Exception: - assert True + assert False + +class Test_delete_dme_filtered_data_job: + + the_response=Response() + the_response.status_code=status.HTTP_204_NO_CONTENT + @patch('trainingmgr.common.trainingmgr_operations.requests.delete', return_value=the_response) + def test_success(self, mock1): + mocked_TRAININGMGR_CONFIG_OBJ=mock.Mock(name="TRAININGMGR_CONFIG_OBJ") + attrs_TRAININGMGR_CONFIG_OBJ = {'kf_adapter_ip.return_value': '123', 'kf_adapter_port.return_value' : '100'} + mocked_TRAININGMGR_CONFIG_OBJ.configure_mock(**attrs_TRAININGMGR_CONFIG_OBJ) + feature_group_name="test" + host="10.0.0.50" + port="31840" + response=trainingmgr_operations.delete_dme_filtered_data_job(mocked_TRAININGMGR_CONFIG_OBJ, feature_group_name, host, port) + assert response.status_code==status.HTTP_204_NO_CONTENT, "delete_dme_filtered_data_job failed" + + def test_create_url_host_port_fail(self): + mocked_TRAININGMGR_CONFIG_OBJ=mock.Mock(name="TRAININGMGR_CONFIG_OBJ") + attrs_TRAININGMGR_CONFIG_OBJ = {'kf_adapter_ip.return_value': '123', 'kf_adapter_port.return_value' : '100'} + mocked_TRAININGMGR_CONFIG_OBJ.configure_mock(**attrs_TRAININGMGR_CONFIG_OBJ) + feature_group_name="test" + host="url error" + port="31840" + try: + response=trainingmgr_operations.delete_dme_filtered_data_job(mocked_TRAININGMGR_CONFIG_OBJ, feature_group_name, host, port) + assert False + except TMException as err: + assert "URL validation error: " in err.message + except Exception: + assert False