Skip to content

Commit

Permalink
add type hints
Browse files Browse the repository at this point in the history
  • Loading branch information
@Limmen
Limmen committed Aug 9, 2023
1 parent 11bccd9 commit 8377a44
Showing 1 changed file with 40 additions and 67 deletions.
107 changes: 40 additions & 67 deletions simulation-system/libs/csle-rest-api/tests/test_resources_emulation_traces.py
View file
Original file line number Diff line number Diff line change
@@ -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

Expand Down Expand Up @@ -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"],
Expand Down Expand Up @@ -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
Expand All @@ -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")
Expand Down Expand Up @@ -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
Expand All @@ -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)
Expand All @@ -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
Expand All @@ -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)
Expand All @@ -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)
Expand All @@ -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
Expand All @@ -335,24 +311,21 @@ 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")
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().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)
Expand Down

0 comments on commit 8377a44

Please sign in to comment.