diff --git a/simulation-system/libs/csle-rest-api/tests/test_resources_emulation_traces.py b/simulation-system/libs/csle-rest-api/tests/test_resources_emulation_traces.py index ea364fc06..183f2c8ba 100644 --- a/simulation-system/libs/csle-rest-api/tests/test_resources_emulation_traces.py +++ b/simulation-system/libs/csle-rest-api/tests/test_resources_emulation_traces.py @@ -1,46 +1,27 @@ -import json from typing import List, Tuple - -import csle_common.constants.constants as constants +import json import pytest import pytest_mock -from csle_collector.client_manager.client_population_metrics import ( - ClientPopulationMetrics, -) +import csle_common.constants.constants as constants +from csle_collector.client_manager.client_population_metrics import ClientPopulationMetrics from csle_collector.docker_stats_manager.docker_stats import DockerStats from csle_collector.host_manager.host_metrics import HostMetrics -from csle_collector.ossec_ids_manager.dao.ossec_ids_alert_counters import ( - OSSECIdsAlertCounters, -) -from csle_collector.snort_ids_manager.snort_ids_alert_counters import ( - SnortIdsAlertCounters, -) -from csle_collector.snort_ids_manager.snort_ids_rule_counters import ( - SnortIdsRuleCounters, -) -from csle_common.dao.emulation_action.attacker.emulation_attacker_action import ( - EmulationAttackerAction, -) -from csle_common.dao.emulation_action.defender.emulation_defender_action import ( - EmulationDefenderAction, -) -from csle_common.dao.emulation_action.defender.emulation_defender_action_id import ( - EmulationDefenderActionId, -) -from csle_common.dao.emulation_action.defender.emulation_defender_action_outcome import ( - EmulationDefenderActionOutcome, -) -from csle_common.dao.emulation_action.defender.emulation_defender_action_type import ( - EmulationDefenderActionType, -) +from csle_collector.ossec_ids_manager.dao.ossec_ids_alert_counters import OSSECIdsAlertCounters +from csle_collector.snort_ids_manager.snort_ids_alert_counters import SnortIdsAlertCounters +from csle_collector.snort_ids_manager.snort_ids_rule_counters import SnortIdsRuleCounters +from csle_common.dao.emulation_action.attacker.emulation_attacker_action import EmulationAttackerAction +from csle_common.dao.emulation_action.attacker.emulation_attacker_action_id import EmulationAttackerActionId +from csle_common.dao.emulation_action.attacker.emulation_attacker_action_outcome import EmulationAttackerActionOutcome +from csle_common.dao.emulation_action.attacker.emulation_attacker_action_type import EmulationAttackerActionType +from csle_common.dao.emulation_action.defender.emulation_defender_action import EmulationDefenderAction +from csle_common.dao.emulation_action.defender.emulation_defender_action_id import EmulationDefenderActionId +from csle_common.dao.emulation_action.defender.emulation_defender_action_outcome import EmulationDefenderActionOutcome +from csle_common.dao.emulation_action.defender.emulation_defender_action_type import EmulationDefenderActionType from csle_common.dao.emulation_config.emulation_trace import EmulationTrace -from csle_common.dao.emulation_observation.attacker.emulation_attacker_observation_state import ( - EmulationAttackerObservationState, -) -from csle_common.dao.emulation_observation.defender.emulation_defender_observation_state import ( - EmulationDefenderObservationState, -) - +from csle_common.dao.emulation_observation.attacker.emulation_attacker_observation_state \ + import EmulationAttackerObservationState +from csle_common.dao.emulation_observation.defender.emulation_defender_observation_state \ + import EmulationDefenderObservationState import csle_rest_api.constants.constants as api_constants from csle_rest_api.rest_api import create_app @@ -127,16 +108,17 @@ def get_emulation_trace() -> None: return get_emulation_trace_mocker @staticmethod - def get_ex_em_tr(): + def get_ex_em_tr() -> EmulationTrace: """ Static help method for obtaining an EmulationTrace object + :return: an EmulationTrace object of dummy-values instances """ - attacker_action = EmulationAttackerAction(id=EmulationDefenderActionId.CONTINUE, name="JohnDoe", + attacker_action = EmulationAttackerAction(id=EmulationAttackerActionId.CONTINUE, name="JohnDoe", cmds=["123.456.78.99"], - type=EmulationDefenderActionType.CONTINUE, descr="null", + type=EmulationAttackerActionType.CONTINUE, descr="null", ips=["123.456.78.99"], index=2, - action_outcome=EmulationDefenderActionOutcome.GAME_END, + action_outcome=EmulationAttackerActionOutcome.GAME_END, alt_cmds=["null"], execution_time=10.0, ts=10.0) defender_action = EmulationDefenderAction(id=EmulationDefenderActionId.STOP, name="JohnDoe", cmds=["null"], @@ -174,8 +156,8 @@ def get_ex_em_tr(): emulation_name="JohnDoe") return ex_em_tr - def test_emulation_traces_get(self, mocker, flask_app, not_logged_in, logged_in, logged_in_as_admin, - list_em_trac_ids, list_em_trac): + def test_emulation_traces_get(self, mocker: pytest_mock.MockFixture, flask_app, not_logged_in, logged_in, + logged_in_as_admin, list_em_trac_ids, list_em_trac) -> None: """ Testing the GET HTTPS method for the /emulation-traces resource @@ -188,8 +170,7 @@ def test_emulation_traces_get(self, mocker, flask_app, not_logged_in, logged_in, side_effect=list_em_trac_ids) mocker.patch("csle_common.metastore.metastore_facade.MetastoreFacade.list_emulation_traces", side_effect=list_em_trac) - mocker.patch("csle_rest_api.util.rest_api_util.check_if_user_is_authorized", - side_effect=not_logged_in) + 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.EMULATION_TRACES_RESOURCE}" f"?{api_constants.MGMT_WEBAPP.IDS_QUERY_PARAM}=true") response_data = response.data.decode("utf-8") @@ -235,8 +216,8 @@ def test_emulation_traces_get(self, mocker, flask_app, not_logged_in, logged_in, response = flask_app.test_client().post(f"{api_constants.MGMT_WEBAPP.EMULATION_TRACES_RESOURCE}") assert response.status_code == constants.HTTPS.METHOD_NOT_ALLOWED_CODE - def test_emulation_traces_delete(self, mocker, flask_app, not_logged_in, logged_in, logged_in_as_admin, - list_em_trac, remove): + def test_emulation_traces_delete(self, mocker: pytest_mock.MockFixture, flask_app, not_logged_in, logged_in, + logged_in_as_admin, list_em_trac, remove) -> None: """ Testing the DELETE HTTPS method for the /emulation-traces resource @@ -250,8 +231,7 @@ def test_emulation_traces_delete(self, mocker, flask_app, not_logged_in, logged_ side_effect=list_em_trac) mocker.patch("csle_common.metastore.metastore_facade.MetastoreFacade.remove_emulation_trace", side_effect=remove) - mocker.patch("csle_rest_api.util.rest_api_util.check_if_user_is_authorized", - side_effect=not_logged_in) + 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.EMULATION_TRACES_RESOURCE}") response_data = response.data.decode("utf-8") response_data_dict = json.loads(response_data) @@ -264,16 +244,15 @@ def test_emulation_traces_delete(self, mocker, flask_app, not_logged_in, logged_ response_data_dict = json.loads(response_data) assert response.status_code == constants.HTTPS.UNAUTHORIZED_STATUS_CODE assert response_data_dict == {} - mocker.patch("csle_rest_api.util.rest_api_util.check_if_user_is_authorized", - side_effect=logged_in_as_admin) + 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.EMULATION_TRACES_RESOURCE}") response_data = response.data.decode("utf-8") response_data_dict = json.loads(response_data) assert response.status_code == constants.HTTPS.OK_STATUS_CODE assert response_data_dict == {} - def test_emulation_traces_ids_get(self, mocker, flask_app, not_logged_in, logged_in, logged_in_as_admin, - get_em_tr, get_em_tr_none): + def test_emulation_traces_ids_get(self, mocker: pytest_mock.MockFixture, flask_app, not_logged_in, logged_in, + logged_in_as_admin, get_em_tr, get_em_tr_none) -> None: """ Testing the GET HTTPS method for the /emulation-traces/id @@ -284,15 +263,13 @@ def test_emulation_traces_ids_get(self, mocker, flask_app, not_logged_in, logged """ mocker.patch("csle_common.metastore.metastore_facade.MetastoreFacade.list_emulation_traces_ids", side_effect=get_em_tr) - mocker.patch("csle_rest_api.util.rest_api_util.check_if_user_is_authorized", - side_effect=not_logged_in) + 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.EMULATION_TRACES_RESOURCE}/10") response_data = response.data.decode("utf-8") response_data_dict = json.loads(response_data) assert response.status_code == constants.HTTPS.UNAUTHORIZED_STATUS_CODE assert response_data_dict == {} - mocker.patch("csle_rest_api.util.rest_api_util.check_if_user_is_authorized", - side_effect=logged_in) + 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.EMULATION_TRACES_RESOURCE}/10") response_data = response.data.decode("utf-8") response_data_dict = json.loads(response_data) @@ -307,8 +284,7 @@ def test_emulation_traces_ids_get(self, mocker, flask_app, not_logged_in, logged response_data_dict = json.loads(response_data) assert response_data_dict == {} assert response.status_code == constants.HTTPS.OK_STATUS_CODE - mocker.patch("csle_rest_api.util.rest_api_util.check_if_user_is_authorized", - side_effect=logged_in_as_admin) + 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.EMULATION_TRACES_RESOURCE}/10") response_data = response.data.decode("utf-8") response_data_dict = json.loads(response_data) @@ -322,8 +298,8 @@ def test_emulation_traces_ids_get(self, mocker, flask_app, not_logged_in, logged assert response_data_dict == {} assert response.status_code == constants.HTTPS.OK_STATUS_CODE - def test_emulation_traces_ids_delete(self, mocker, flask_app, not_logged_in, logged_in, logged_in_as_admin, - get_em_tr, get_em_tr_none, remove): + def test_emulation_traces_ids_delete(self, mocker: pytest_mock.MockFixture, flask_app, not_logged_in, + logged_in, logged_in_as_admin, get_em_tr, get_em_tr_none, remove) -> None: """ Testing the DELETE HTTPS method for the /emulation-traces/id @@ -335,8 +311,7 @@ def test_emulation_traces_ids_delete(self, mocker, flask_app, not_logged_in, log """ mocker.patch("csle_common.metastore.metastore_facade.MetastoreFacade.list_emulation_traces_ids", side_effect=get_em_tr) - mocker.patch("csle_rest_api.util.rest_api_util.check_if_user_is_authorized", - side_effect=not_logged_in) + 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.remove_emulation_trace", side_effect=remove) response = flask_app.test_client().delete(f"{api_constants.MGMT_WEBAPP.EMULATION_TRACES_RESOURCE}/10") @@ -344,15 +319,13 @@ def test_emulation_traces_ids_delete(self, mocker, flask_app, not_logged_in, log response_data_dict = json.loads(response_data) assert response.status_code == constants.HTTPS.UNAUTHORIZED_STATUS_CODE assert response_data_dict == {} - mocker.patch("csle_rest_api.util.rest_api_util.check_if_user_is_authorized", - side_effect=logged_in) + 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.EMULATION_TRACES_RESOURCE}/10") response_data = response.data.decode("utf-8") response_data_dict = json.loads(response_data) assert response.status_code == constants.HTTPS.UNAUTHORIZED_STATUS_CODE assert response_data_dict == {} - mocker.patch("csle_rest_api.util.rest_api_util.check_if_user_is_authorized", - side_effect=logged_in_as_admin) + 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.EMULATION_TRACES_RESOURCE}/10") response_data = response.data.decode("utf-8") response_data_dict = json.loads(response_data)