From 7a33a76e32b69c224034c625d20199777ce3b5b3 Mon Sep 17 00:00:00 2001 From: nforsg Date: Thu, 10 Aug 2023 09:44:03 +0200 Subject: [PATCH] testing finished --- .../resources/statistics_datasets/routes.py | 25 +- .../test_resources_statistics_datasets.py | 280 ++++++++++++++++++ 2 files changed, 294 insertions(+), 11 deletions(-) create mode 100644 simulation-system/libs/csle-rest-api/tests/test_resources_statistics_datasets.py diff --git a/simulation-system/libs/csle-rest-api/src/csle_rest_api/resources/statistics_datasets/routes.py b/simulation-system/libs/csle-rest-api/src/csle_rest_api/resources/statistics_datasets/routes.py index 622e46a74..8ac0bb0d1 100644 --- a/simulation-system/libs/csle-rest-api/src/csle_rest_api/resources/statistics_datasets/routes.py +++ b/simulation-system/libs/csle-rest-api/src/csle_rest_api/resources/statistics_datasets/routes.py @@ -1,15 +1,16 @@ """ Routes and sub-resources for the /statistics-datasets resource """ -from typing import Tuple import os -from flask import Blueprint, jsonify, request, send_from_directory, Response +from typing import Tuple + 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 csle_common.dao.datasets.statistics_dataset import StatisticsDataset -import csle_rest_api.util.rest_api_util as rest_api_util +from csle_common.metastore.metastore_facade import MetastoreFacade +from flask import Blueprint, Response, jsonify, request, send_from_directory +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 statistics_datasets_bp = Blueprint( @@ -25,11 +26,12 @@ def statistics_datasets() -> Tuple[Response, int]: :return: A list of statistics datasets or a list of ids of the statistics datasets or deletes the datasets """ + requires_admin = False if request.method == api_constants.MGMT_WEBAPP.HTTP_REST_DELETE: requires_admin = True - authorized = rest_api_util.check_if_user_is_authorized(request=request, requires_admin=requires_admin) - if authorized is not None: - return authorized + authorized = rest_api_util.check_if_user_is_authorized(request=request, requires_admin=requires_admin) + if authorized is not None: + return authorized if request.method == api_constants.MGMT_WEBAPP.HTTP_REST_GET: # Check if ids query parameter is True, then only return the ids and not the whole list of statistics datasets @@ -78,11 +80,12 @@ def statistics_dataset(statistics_dataset_id: int) -> Tuple[Response, int]: :param statistics_dataset_id: the id of the statistics dataset :return: The given statistics dataset or deletes the statistics dataset """ + requires_admin = False if request.method == api_constants.MGMT_WEBAPP.HTTP_REST_DELETE: requires_admin = True - authorized = rest_api_util.check_if_user_is_authorized(request=request, requires_admin=requires_admin) - if authorized is not None: - return authorized + authorized = rest_api_util.check_if_user_is_authorized(request=request, requires_admin=requires_admin) + if authorized is not None: + return authorized statistics_dataset = MetastoreFacade.get_statistics_dataset_metadata(id=statistics_dataset_id) response = jsonify({}) diff --git a/simulation-system/libs/csle-rest-api/tests/test_resources_statistics_datasets.py b/simulation-system/libs/csle-rest-api/tests/test_resources_statistics_datasets.py new file mode 100644 index 000000000..cc5c9b0ce --- /dev/null +++ b/simulation-system/libs/csle-rest-api/tests/test_resources_statistics_datasets.py @@ -0,0 +1,280 @@ +import json +from typing import List, Tuple + +import csle_common.constants.constants as constants +import pytest +import pytest_mock +from csle_common.dao.datasets.statistics_dataset import StatisticsDataset + +import csle_rest_api.constants.constants as api_constants +from csle_rest_api.rest_api import create_app + + +class TestResourcesStatisticsDataSetsSuite: + """ + Test suite for /statistics-datasets resource + """ + + @pytest.fixture + def flask_app(self): + """ + Gets the Flask app + + :return: the flask app fixture representing the webserver + """ + return create_app(static_folder="../../../../../management-system/csle-mgmt-webapp/build") + + @pytest.fixture + def list_stat_ds(self, mocker): + """ + Pytest fixture for mocking the list_statistics_datasets method + + :param mocker: the pytest mocker object + :return: the mocked function + """ + def list_statistics_datasets() -> List[StatisticsDataset]: + stat_ds = TestResourcesStatisticsDataSetsSuite.get_ex_stat_ds() + return [stat_ds] + list_statistics_datasets_mocker = mocker.MagicMock(side_effect=list_statistics_datasets) + return list_statistics_datasets_mocker + + @pytest.fixture + def list_stat_ds_ids(self, mocker): + """ + Pytest fixture for mocking the list_statistics_datasets_ids method + + :param mocker the pytest mocker object + :return: the mocked function + """ + def list_statistics_datasets_ids() -> List[Tuple[int, str]]: + return [(1, "JDoeStatistics")] + list_statistics_datasets_ids_mocker = mocker.MagicMock(side_effect=list_statistics_datasets_ids) + return list_statistics_datasets_ids_mocker + + @pytest.fixture + def remove(self, mocker: pytest_mock.MockFixture): + """ + Pytest fixture for mocking the remove_statistics_dataset method + :param mocker: the pytest mocker object + :return: the mocked function + """ + def remove_statistics_dataset(statistics_dataset: StatisticsDataset) -> None: + return None + remove_statistics_dataset_mocker = mocker.MagicMock(side_effect=remove_statistics_dataset) + return remove_statistics_dataset_mocker + + @pytest.fixture + def get_stat_ds(self, mocker): + """ + Pytest fixture for mocking the get_statistics_dataset_metadata + + :param mocker: the pytest mocker object + :return: the mocked function + """ + def get_statistics_dataset_metadata(id: int) -> StatisticsDataset: + sim_tr = TestResourcesStatisticsDataSetsSuite.get_ex_stat_ds() + return sim_tr + get_statistics_dataset_metadata_mocker = mocker.MagicMock(side_effect=get_statistics_dataset_metadata) + return get_statistics_dataset_metadata_mocker + + @pytest.fixture + def update(self, mocker): + """ + Pytest fixture for mocking the update_statistics_dataset method + + :param mocker: the pytest mocker object + :return: the mocked function + """ + def update_statistics_dataset(statistics_dataset: StatisticsDataset, id: int) -> None: + return None + update_statistics_dataset_mocker = mocker.MagicMock(side_effect=update_statistics_dataset) + return update_statistics_dataset_mocker + + @staticmethod + def get_ex_stat_ds(): + """ + Static help method for obtainng a SimulationTrace object + + :return: A SimulationTrace object + """ + sim_trace = StatisticsDataset(name="JohnDoe", description="null", download_count=1, file_path="null", + url="null", date_added="null", num_measurements=1, num_metrics=1, + size_in_gb=1.0, compressed_size_in_gb=0.5, citation="null", + num_files=4, file_format="null", added_by="JDoe", conditions="null", + metrics="null", num_conditions=1) + return sim_trace + + def test_statistics_datasets_get(self, mocker, flask_app, not_logged_in, logged_in, + logged_in_as_admin, list_stat_ds, list_stat_ds_ids): + """ + Testing the GET HTTPS for the /statistics-datasets resource + :param mocker: the pytest mocker object + :param flask_app: the flask_app fixture + :param not_logged_in: the not_logged_in fixture + :param logged_in: the logged_in fixture + :param logged_in_as_admin: the logged_in_as_admin fixture + :param list_stat_ds: the list_stat_ds fixture + :param list_stat_ds_ids: the list_stat_ds_ids fixture + """ + mocker.patch("csle_common.metastore.metastore_facade.MetastoreFacade.list_statistics_datasets", + side_effect=list_stat_ds) + mocker.patch("csle_common.metastore.metastore_facade.MetastoreFacade.list_statistics_datasets_ids", + side_effect=list_stat_ds_ids) + mocker.patch("csle_rest_api.util.rest_api_util.check_if_user_is_authorized", + side_effect=not_logged_in) + response = flask_app.test_client().get(f"{api_constants.MGMT_WEBAPP.STATISTICS_DATASETS_RESOURCE}") + response_data = response.data.decode("utf-8") + response_data_dict = json.loads(response_data) + assert response_data_dict == {} + 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.STATISTICS_DATASETS_RESOURCE}" + f"?{api_constants.MGMT_WEBAPP.IDS_QUERY_PARAM}=true") + response_data = response.data.decode("utf-8") + response_data_list = json.loads(response_data) + response_data_dict = response_data_list[0] + assert response.status_code == constants.HTTPS.OK_STATUS_CODE + assert response_data_dict[api_constants.MGMT_WEBAPP.ID_PROPERTY] == 1 + assert response_data_dict[api_constants.MGMT_WEBAPP.STATISTICS_DATASET_PROPERTY] == "JDoeStatistics" + response = flask_app.test_client().get(f"{api_constants.MGMT_WEBAPP.STATISTICS_DATASETS_RESOURCE}") + response_data = response.data.decode("utf-8") + response_data_list = json.loads(response_data) + response_data_dict = response_data_list[0] + ex_st_ds = TestResourcesStatisticsDataSetsSuite.get_ex_stat_ds() + ex_st_ds_dict = ex_st_ds.to_dict() + for k in response_data_dict: + assert response_data_dict[k] == ex_st_ds_dict[k] + 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.STATISTICS_DATASETS_RESOURCE}" + f"?{api_constants.MGMT_WEBAPP.IDS_QUERY_PARAM}=true") + response_data = response.data.decode("utf-8") + response_data_list = json.loads(response_data) + response_data_dict = response_data_list[0] + assert response.status_code == constants.HTTPS.OK_STATUS_CODE + assert response_data_dict[api_constants.MGMT_WEBAPP.ID_PROPERTY] == 1 + assert response_data_dict[api_constants.MGMT_WEBAPP.STATISTICS_DATASET_PROPERTY] == "JDoeStatistics" + response = flask_app.test_client().get(f"{api_constants.MGMT_WEBAPP.STATISTICS_DATASETS_RESOURCE}") + response_data = response.data.decode("utf-8") + response_data_list = json.loads(response_data) + response_data_dict = response_data_list[0] + ex_st_ds = TestResourcesStatisticsDataSetsSuite.get_ex_stat_ds() + ex_st_ds_dict = ex_st_ds.to_dict() + for k in response_data_dict: + assert response_data_dict[k] == ex_st_ds_dict[k] + + def test_statistics_datasets_delete(self, mocker, flask_app, not_logged_in, logged_in, + logged_in_as_admin, list_stat_ds, remove): + """ + Testing the DELETE HTTPS for the /statistics-datasets resource + :param mocker: the pytest mocker object + :param flask_app: the flask_app fixture + :param not_logged_in: the not_logged_in fixture + :param logged_in: the logged_in fixture + :param logged_in_as_admin: the logged_in_as_admin fixture + :param list_stat_ds: the list_stat_ds fixture + """ + mocker.patch("csle_common.metastore.metastore_facade.MetastoreFacade.list_statistics_datasets", + side_effect=list_stat_ds) + mocker.patch("csle_common.metastore.metastore_facade.MetastoreFacade.remove_statistics_dataset", + side_effect=remove) + mocker.patch("csle_rest_api.util.rest_api_util.check_if_user_is_authorized", + side_effect=not_logged_in) + response = flask_app.test_client().delete(f"{api_constants.MGMT_WEBAPP.STATISTICS_DATASETS_RESOURCE}") + response_data = response.data.decode("utf-8") + response_data_dict = json.loads(response_data) + assert response_data_dict == {} + 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.STATISTICS_DATASETS_RESOURCE}") + response_data = response.data.decode("utf-8") + response_data_dict = json.loads(response_data) + assert response_data_dict == {} + 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.STATISTICS_DATASETS_RESOURCE}") + response_data = response.data.decode("utf-8") + response_data_dict = json.loads(response_data) + assert response_data_dict == {} + assert response.status_code == constants.HTTPS.OK_STATUS_CODE + + def test_satistics_datasets_ids_get(self, mocker, flask_app, not_logged_in, logged_in, + logged_in_as_admin, get_stat_ds, update): + """ + Testing the GET HTTPS for the /statistics-datasets/id resource + :param mocker: the pytest mocker object + :param flask_app: the flask_app fixture + :param not_logged_in: the not_logged_in fixture + :param logged_in: the logged_in fixture + :param logged_in_as_admin: the logged_in_as_admin fixture + :param get_stat_ds: the get_stat_ds fixture + """ + mocker.patch("csle_common.metastore.metastore_facade.MetastoreFacade.get_statistics_dataset_metadata", + side_effect=get_stat_ds) + mocker.patch("csle_common.metastore.metastore_facade.MetastoreFacade.update_statistics_dataset", + side_effect=update) + mocker.patch("csle_rest_api.util.rest_api_util.check_if_user_is_authorized", + side_effect=not_logged_in) + response = flask_app.test_client().get(f"{api_constants.MGMT_WEBAPP.STATISTICS_DATASETS_RESOURCE}/10") + response_data = response.data.decode("utf-8") + response_data_dict = json.loads(response_data) + assert response_data_dict == {} + 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.STATISTICS_DATASETS_RESOURCE}/10") + response_data = response.data.decode("utf-8") + response_data_dict = json.loads(response_data) + ex_st_ds = TestResourcesStatisticsDataSetsSuite.get_ex_stat_ds() + ex_st_ds_dict = ex_st_ds.to_dict() + for k in response_data_dict: + assert response_data_dict[k] == ex_st_ds_dict[k] + 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.STATISTICS_DATASETS_RESOURCE}/10") + response_data = response.data.decode("utf-8") + response_data_dict = json.loads(response_data) + ex_st_ds = TestResourcesStatisticsDataSetsSuite.get_ex_stat_ds() + ex_st_ds_dict = ex_st_ds.to_dict() + for k in response_data_dict: + assert response_data_dict[k] == ex_st_ds_dict[k] + + def test_statistics_datasets_ids_delete(self, mocker, flask_app, not_logged_in, logged_in, + logged_in_as_admin, get_stat_ds, remove): + """ + Testing the DELETE HTTPS for the /statistics-datasets/id resource + :param mocker: the pytest mocker object + :param flask_app: the flask_app fixture + :param not_logged_in: the not_logged_in fixture + :param logged_in: the logged_in fixture + :param logged_in_as_admin: the logged_in_as_admin fixture + :param get_stat_ds: the get_stat_ds fixture + """ + mocker.patch("csle_common.metastore.metastore_facade.MetastoreFacade.get_statistics_dataset_metadata", + side_effect=get_stat_ds) + mocker.patch("csle_common.metastore.metastore_facade.MetastoreFacade.remove_statistics_dataset", + side_effect=remove) + mocker.patch("csle_rest_api.util.rest_api_util.check_if_user_is_authorized", + side_effect=not_logged_in) + response = flask_app.test_client().delete(f"{api_constants.MGMT_WEBAPP.STATISTICS_DATASETS_RESOURCE}/10") + response_data = response.data.decode("utf-8") + response_data_dict = json.loads(response_data) + assert response_data_dict == {} + 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.STATISTICS_DATASETS_RESOURCE}/10") + response_data = response.data.decode("utf-8") + response_data_dict = json.loads(response_data) + assert response_data_dict == {} + 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.STATISTICS_DATASETS_RESOURCE}/10") + response_data = response.data.decode("utf-8") + response_data_dict = json.loads(response_data) + assert response_data_dict == {} + assert response.status_code == constants.HTTPS.OK_STATUS_CODE