diff --git a/simulation-system/libs/csle-rest-api/src/csle_rest_api/resources/mcmc_system_models/routes.py b/simulation-system/libs/csle-rest-api/src/csle_rest_api/resources/mcmc_system_models/routes.py index cf53ae5d9..460600c7e 100644 --- a/simulation-system/libs/csle-rest-api/src/csle_rest_api/resources/mcmc_system_models/routes.py +++ b/simulation-system/libs/csle-rest-api/src/csle_rest_api/resources/mcmc_system_models/routes.py @@ -1,10 +1,11 @@ """ Routes and sub-resources for the /mcmc-system-models resource """ -from flask import Blueprint, jsonify, request import csle_common.constants.constants as constants -import csle_rest_api.constants.constants as api_constants from csle_common.metastore.metastore_facade import MetastoreFacade +from flask import Blueprint, jsonify, request + +import csle_rest_api.constants.constants as api_constants import csle_rest_api.util.rest_api_util as rest_api_util # Creates a blueprint "sub application" of the main REST app diff --git a/simulation-system/libs/csle-rest-api/tests/test_resources_mcmc_system_models.py b/simulation-system/libs/csle-rest-api/tests/test_resources_mcmc_system_models.py new file mode 100644 index 000000000..216d4fb54 --- /dev/null +++ b/simulation-system/libs/csle-rest-api/tests/test_resources_mcmc_system_models.py @@ -0,0 +1,301 @@ +import json +import logging +from typing import Dict, List, Tuple, Union + +import csle_common.constants.constants as constants +import pytest +import pytest_mock +from csle_common.dao.system_identification.mcmc_posterior import MCMCPosterior +from csle_common.dao.system_identification.mcmc_system_model import MCMCSystemModel + +import csle_rest_api.constants.constants as api_constants +from csle_rest_api.rest_api import create_app + + +class TestResourcesMCMCSystemModelsSuite: + """ + Test suite for /system-identification-jobs resource + """ + + pytest.logger = logging.getLogger("resources_mcmc_system_models_tests") + + @pytest.fixture + def flask_app(self): + """ + :return: the flask app fixture representing the webserver + """ + return create_app(static_folder="../../../../../management-system/csle-mgmt-webapp/build") + + @pytest.fixture + def list_mcmc_m(self, mocker: pytest_mock.MockFixture): + """ + Pytest fixture for mocking the list_mcmc_system_models function + + :param mocker: the pytest mocker object + :return: a mock object with the mocked function + """ + def list_mcmc_system_models() -> List[MCMCSystemModel]: + mcmc_sys_mod = TestResourcesMCMCSystemModelsSuite.example_returner() + return [mcmc_sys_mod] + list_mcmc_system_models_mocker = mocker.MagicMock( + side_effect=list_mcmc_system_models) + return list_mcmc_system_models_mocker + + @pytest.fixture + def list_mcmc_s_m_ids(self, mocker: pytest_mock.MockFixture): + """ + Pytest fixture for mocking the list_mcmc_system_models_ids function + + :param mocker: the pytest mocker object + :return: a mock object with the mocked function + """ + def list_mcmc_system_models_ids() -> List[Tuple[str, int, int, str]]: + t = ("csle-level-10", 1, 10, + api_constants.MGMT_WEBAPP.GAUSSIAN_MIXTURE_SYSTEM_MODEL_TYPE) + l = [t] + return l + list_mcmc_system_models_ids_mocker = mocker.MagicMock( + side_effect=list_mcmc_system_models_ids) + return list_mcmc_system_models_ids_mocker + + @pytest.fixture + def get_mcmc_config(self, mocker: pytest_mock.MockFixture): + """ + pytest fixture for mocking the get_mcmc_system_model_config function + + :param mocker: the pytest mocker object + :return: a mock object with the mocked function + """ + def get_mcmc_system_model_config(id: int) -> MCMCSystemModel: + g_m_sys_mod = TestResourcesMCMCSystemModelsSuite.example_returner() + return g_m_sys_mod + + get_mcmc_system_model_config_mocker = mocker.MagicMock( + side_effect=get_mcmc_system_model_config) + + return get_mcmc_system_model_config_mocker + + @pytest.fixture + def remove(self, mocker: pytest_mock.MockFixture): + """ + pytest fixture for mocking the remove_mcmc_system_model function + + :param mocker: the pytest mocker object + :return: a mock object with the mocked function + """ + def remove_mcmc_system_model(mcmc_system_model: MCMCSystemModel) -> None: + return None + remove_mcmc_system_model_mocker = mocker.MagicMock( + side_effect=remove_mcmc_system_model) + return remove_mcmc_system_model_mocker + + @staticmethod + def example_sys_mod() -> List[Dict[str, Union[str, int]]]: + """ + Static method for returning an example list of response dicts + + :return: the example list of response dicts + """ + response_dicts = [{api_constants.MGMT_WEBAPP.EMULATION_QUERY_PARAM: 1, + api_constants.MGMT_WEBAPP.ID_PROPERTY: 'csle-level-10', + api_constants.MGMT_WEBAPP.STATISTIC_ID_PROPERTY: 10, + api_constants.MGMT_WEBAPP.SYSTEM_MODEL_TYPE: + api_constants.MGMT_WEBAPP.GAUSSIAN_MIXTURE_SYSTEM_MODEL_TYPE}] + return response_dicts + + @staticmethod + def example_returner() -> MCMCSystemModel: + """ + Static help method for returning an example MCMC System Model + + :return: the example MCMC System model + """ + + mcmc_post = MCMCPosterior(posterior_name="JohnDoePost", samples=[1.1], + densities=[0.1], + sample_space=[0.5]) + mcmc_sys_mod = MCMCSystemModel(emulation_env_name="JohnDoeEmulation", + emulation_statistic_id=10, + posteriors=[mcmc_post], descr="descr") + return mcmc_sys_mod + + def test_mcmc_s_m_get(self, flask_app, mocker: pytest_mock.MockFixture, + logged_in, not_logged_in, logged_in_as_admin, + list_mcmc_s_m_ids, list_mcmc_m) -> None: + """ + Testing for the GET HTTPS method in the /mcmc-system-models resource + :param flask_app: the pytest flask app for making requests + :param mocker: the pytest mocker object + :param logged_in_as_admin: the logged_in_as_admin fixture + :param logged_in: the logged_in fixture + :param not_logged_in: the not_logged_in fixture + :param list_mcmc_s_m_ids: the list_mcmc_s_m_ids fixture + :param list_mcmc_m: the list_mcmc_m fixture + :return: None + """ + test_obj = TestResourcesMCMCSystemModelsSuite.example_returner() + test_obj_dict = test_obj.to_dict() + mocker.patch("csle_rest_api.util.rest_api_util.check_if_user_is_authorized", + side_effect=not_logged_in) + mocker.patch("csle_common.metastore.metastore_facade.MetastoreFacade.list_mcmc_system_models", + side_effect=list_mcmc_m) + mocker.patch("csle_common.metastore.metastore_facade.MetastoreFacade.list_mcmc_system_models_ids", + side_effect=list_mcmc_s_m_ids) + response = flask_app.test_client().get(f"{api_constants.MGMT_WEBAPP.MCMC_SYSTEM_MODELS_RESOURCE}") + response_data = response.data.decode("utf-8") + response_data_list = json.loads(response_data) + assert response.status_code == constants.HTTPS.UNAUTHORIZED_STATUS_CODE + assert response_data_list == {} + mocker.patch("csle_rest_api.util.rest_api_util.check_if_user_is_authorized", side_effect=logged_in) + response = flask_app.test_client().get(f"{api_constants.MGMT_WEBAPP.MCMC_SYSTEM_MODELS_RESOURCE}" + f"?{api_constants.MGMT_WEBAPP.IDS_QUERY_PARAM}=true") + response_data = response.data.decode("utf-8") + response_data_list = json.loads(response_data) + test_dicts = TestResourcesMCMCSystemModelsSuite.example_sys_mod() + assert response.status_code == constants.HTTPS.OK_STATUS_CODE + for i in range(len(response_data_list)): + for j in response_data_list[i]: + assert response_data_list[i][j] == test_dicts[i][j] + response = flask_app.test_client().get(f"{api_constants.MGMT_WEBAPP.MCMC_SYSTEM_MODELS_RESOURCE}") + response_data = response.data.decode("utf-8") + response_data_list = json.loads(response_data) + g_m_data_dict = response_data_list[0] + assert response.status_code == constants.HTTPS.OK_STATUS_CODE + for i in (g_m_data_dict): + assert g_m_data_dict[i] == test_obj_dict[i] + mocker.patch("csle_rest_api.util.rest_api_util.check_if_user_is_authorized", side_effect=logged_in_as_admin) + response = flask_app.test_client().get(f"{api_constants.MGMT_WEBAPP.MCMC_SYSTEM_MODELS_RESOURCE}") + response_data = response.data.decode("utf-8") + response_data_list = json.loads(response_data) + g_m_data_dict = response_data_list[0] + assert response.status_code == constants.HTTPS.OK_STATUS_CODE + for i in (g_m_data_dict): + assert g_m_data_dict[i] == test_obj_dict[i] + response = flask_app.test_client().get(f"{api_constants.MGMT_WEBAPP.MCMC_SYSTEM_MODELS_RESOURCE}" + f"?{api_constants.MGMT_WEBAPP.IDS_QUERY_PARAM}=true") + response_data = response.data.decode("utf-8") + response_data_list = json.loads(response_data) + test_dicts = TestResourcesMCMCSystemModelsSuite.example_sys_mod() + assert response.status_code == constants.HTTPS.OK_STATUS_CODE + for i in range(len(response_data_list)): + for j in response_data_list[i]: + assert response_data_list[i][j] == test_dicts[i][j] + + def test_mcmc_s_m_delete(self, flask_app, mocker: pytest_mock.MockFixture, + logged_in, not_logged_in, logged_in_as_admin, + list_mcmc_m, remove) -> None: + """ + Testing for the DELETE HTTPS method in the /mcmc-system-models resource + :param flask_app: the pytest flask app for making requests + :param mocker: the pytest mocker object + :param logged_in_as_admin: the logged_in_as_admin fixture + :param logged_in: the logged_in fixture + :param not_logged_in: the not_logged_in fixture + :param list_mcmc_m: the list_mcmc_m fixture + :return: None + """ + mocker.patch("csle_rest_api.util.rest_api_util.check_if_user_is_authorized", + side_effect=not_logged_in) + mocker.patch("csle_common.metastore.metastore_facade.MetastoreFacade.list_mcmc_system_models", + side_effect=list_mcmc_m) + mocker.patch("csle_common.metastore.metastore_facade.MetastoreFacade.remove_mcmc_system_model", + side_effect=remove) + response = flask_app.test_client().delete(api_constants.MGMT_WEBAPP.MCMC_SYSTEM_MODELS_RESOURCE) + response_data = response.data.decode("utf-8") + response_data_list = json.loads(response_data) + assert response_data_list == {} + assert response.status_code == constants.HTTPS.UNAUTHORIZED_STATUS_CODE + mocker.patch("csle_rest_api.util.rest_api_util.check_if_user_is_authorized", + side_effect=logged_in) + response = flask_app.test_client().delete(api_constants.MGMT_WEBAPP.MCMC_SYSTEM_MODELS_RESOURCE) + response_data = response.data.decode("utf-8") + response_data_list = json.loads(response_data) + assert response_data_list == {} + assert response.status_code == constants.HTTPS.UNAUTHORIZED_STATUS_CODE + mocker.patch("csle_rest_api.util.rest_api_util.check_if_user_is_authorized", side_effect=logged_in_as_admin) + response = flask_app.test_client().delete(api_constants.MGMT_WEBAPP.MCMC_SYSTEM_MODELS_RESOURCE) + response_data = response.data.decode("utf-8") + response_data_list = json.loads(response_data) + assert response_data_list == {} + assert response.status_code == constants.HTTPS.OK_STATUS_CODE + + def test_gm_s_m_ids_get(self, flask_app, mocker: pytest_mock.MockFixture, logged_in, not_logged_in, + logged_in_as_admin, get_mcmc_config) -> None: + """ + Testing for the GET HTTPS method in the /mcmc-system-models/id resource + :param flask_app: the pytest flask app for making requests + :param mocker: the pytest mocker object + :param logged_in_as_admin: the logged_in_as_admin fixture + :param logged_in: the logged_in fixture + :param not_logged_in: the not_logged_in fixture + :param get_mcmc_config: the get_mcmc_config fixture + :return: None + """ + test_obj = TestResourcesMCMCSystemModelsSuite.example_returner() + test_obj_dict = test_obj.to_dict() + mocker.patch("csle_rest_api.util.rest_api_util.check_if_user_is_authorized", + side_effect=not_logged_in) + mocker.patch("csle_common.metastore.metastore_facade.MetastoreFacade.get_mcmc_system_model_config", + side_effect=get_mcmc_config) + response = flask_app.test_client().get( + f"{api_constants.MGMT_WEBAPP.MCMC_SYSTEM_MODELS_RESOURCE}/10") + response_data = response.data.decode("utf-8") + response_data_list = json.loads(response_data) + assert response_data_list == {} + assert response.status_code == constants.HTTPS.UNAUTHORIZED_STATUS_CODE + mocker.patch("csle_rest_api.util.rest_api_util.check_if_user_is_authorized", + side_effect=logged_in,) + response = flask_app.test_client().get( + f"{api_constants.MGMT_WEBAPP.MCMC_SYSTEM_MODELS_RESOURCE}/10") + response_data = response.data.decode("utf-8") + g_m_data_dict = json.loads(response_data) + assert response.status_code == constants.HTTPS.OK_STATUS_CODE + for i in (g_m_data_dict): + assert g_m_data_dict[i] == test_obj_dict[i] + mocker.patch("csle_rest_api.util.rest_api_util.check_if_user_is_authorized", + side_effect=logged_in_as_admin,) + response = flask_app.test_client().get( + f"{api_constants.MGMT_WEBAPP.MCMC_SYSTEM_MODELS_RESOURCE}/10") + response_data = response.data.decode("utf-8") + g_m_data_dict = json.loads(response_data) + assert response.status_code == constants.HTTPS.OK_STATUS_CODE + for i in (g_m_data_dict): + assert g_m_data_dict[i] == test_obj_dict[i] + + def test_mcmc_s_m_ids_delete(self, flask_app, mocker: pytest_mock.MockFixture, logged_in, not_logged_in, + logged_in_as_admin, get_mcmc_config, remove) -> None: + """ + Testing for the DELETE HTTPS method in the /mcmc-system-models/id resource + :param flask_app: the pytest flask app for making requests + :param mocker: the pytest mocker object + :param logged_in_as_admin: the logged_in_as_admin fixture + :param logged_in: the logged_in fixture + :param not_logged_in: the not_logged_in fixture + :param get_mcmc_config: the get_mcmc_config fixture + :return: None + """ + mocker.patch("csle_rest_api.util.rest_api_util.check_if_user_is_authorized", side_effect=not_logged_in) + mocker.patch("csle_common.metastore.metastore_facade.MetastoreFacade.get_mcmc_system_model_config", + side_effect=get_mcmc_config) + mocker.patch("csle_common.metastore.metastore_facade.MetastoreFacade.remove_mcmc_system_model", + side_effect=remove) + response = flask_app.test_client().delete( + f"{api_constants.MGMT_WEBAPP.MCMC_SYSTEM_MODELS_RESOURCE}/10") + response_data = response.data.decode("utf-8") + response_data_list = json.loads(response_data) + assert response_data_list == {} + assert response.status_code == constants.HTTPS.UNAUTHORIZED_STATUS_CODE + mocker.patch("csle_rest_api.util.rest_api_util.check_if_user_is_authorized", side_effect=logged_in) + response = flask_app.test_client().delete( + f"{api_constants.MGMT_WEBAPP.MCMC_SYSTEM_MODELS_RESOURCE}/10") + response_data = response.data.decode("utf-8") + response_data_list = json.loads(response_data) + assert response_data_list == {} + assert response.status_code == constants.HTTPS.UNAUTHORIZED_STATUS_CODE + mocker.patch("csle_rest_api.util.rest_api_util.check_if_user_is_authorized", side_effect=logged_in_as_admin) + response = flask_app.test_client().delete( + f"{api_constants.MGMT_WEBAPP.MCMC_SYSTEM_MODELS_RESOURCE}/10") + response_data = response.data.decode("utf-8") + response_data_list = json.loads(response_data) + assert response_data_list == {} + assert response.status_code == constants.HTTPS.OK_STATUS_CODE