From 698357ef11bb8216b8b2bc5c57e9252c6b98dbf6 Mon Sep 17 00:00:00 2001 From: Kim Hammar Date: Mon, 15 Jul 2024 17:13:22 +0200 Subject: [PATCH 01/13] add dataset link --- README.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/README.md b/README.md index a7a858f12..97dd322b0 100644 --- a/README.md +++ b/README.md @@ -131,6 +131,10 @@ features. We currently support each release for a window of 6 months.

+## Datasets + +A dataset of 6400 intrusion traces can be found [here](https://zenodo.org/records/10234379). + ## Maintainer From 5be311974e6de14fcca3fbcc9f637f164a603b84 Mon Sep 17 00:00:00 2001 From: Yuhu-kth Date: Tue, 16 Jul 2024 19:35:40 +0200 Subject: [PATCH 02/13] add test_grpc_util --- .../libs/csle-common/tests/test_grpc_util.py | 38 +++++++++++++++++++ 1 file changed, 38 insertions(+) create mode 100644 simulation-system/libs/csle-common/tests/test_grpc_util.py diff --git a/simulation-system/libs/csle-common/tests/test_grpc_util.py b/simulation-system/libs/csle-common/tests/test_grpc_util.py new file mode 100644 index 000000000..23564b1eb --- /dev/null +++ b/simulation-system/libs/csle-common/tests/test_grpc_util.py @@ -0,0 +1,38 @@ +import grpc +import pytest +from unittest.mock import patch, MagicMock +from csle_common.util.grpc_util import GrpcUtil + +class TestGrpcUtilSuite: + """ + Test suite for grpc util + """ + @patch("grpc.channel_ready_future") + def test_grpc_server_on(self, mock_channel_ready_future): + """ + Test + + :param mock_channel_ready_future: _description_ + :type mock_channel_ready_future: _type_ + """ + mock_future = MagicMock() + mock_channel_ready_future.return_value = mock_future + result = GrpcUtil.grpc_server_on(mock_channel_ready_future) + mock_future.result.assert_called() + assert result + + @patch("grpc.channel_ready_future") + def test_grpc_server_on_timeout(self, mock_channel_ready_future): + """ + Test + + :param mock_channel_ready_future: _description_ + :type mock_channel_ready_future: _type_ + """ + mock_future = MagicMock() + mock_future.result.side_effect = grpc.FutureTimeoutError() + mock_channel_ready_future.return_value = mock_future + result = GrpcUtil.grpc_server_on(mock_channel_ready_future) + mock_future.result.assert_called() + assert not result + \ No newline at end of file From e12c148d6c673563c70cd90ac122fc89e1e6f041 Mon Sep 17 00:00:00 2001 From: Yuhu-kth Date: Tue, 16 Jul 2024 19:51:14 +0200 Subject: [PATCH 03/13] add test_grpc_util --- .../libs/csle-common/tests/test_grpc_util.py | 18 ++++++++++-------- 1 file changed, 10 insertions(+), 8 deletions(-) diff --git a/simulation-system/libs/csle-common/tests/test_grpc_util.py b/simulation-system/libs/csle-common/tests/test_grpc_util.py index 23564b1eb..051dca870 100644 --- a/simulation-system/libs/csle-common/tests/test_grpc_util.py +++ b/simulation-system/libs/csle-common/tests/test_grpc_util.py @@ -8,12 +8,13 @@ class TestGrpcUtilSuite: Test suite for grpc util """ @patch("grpc.channel_ready_future") - def test_grpc_server_on(self, mock_channel_ready_future): + def test_grpc_server_on(self, mock_channel_ready_future) -> None: """ - Test + Test utility function to test if a given gRPC channel is working or not - :param mock_channel_ready_future: _description_ - :type mock_channel_ready_future: _type_ + :param mock_channel_ready_future: mock_channel_ready_future + + :return: None """ mock_future = MagicMock() mock_channel_ready_future.return_value = mock_future @@ -22,12 +23,13 @@ def test_grpc_server_on(self, mock_channel_ready_future): assert result @patch("grpc.channel_ready_future") - def test_grpc_server_on_timeout(self, mock_channel_ready_future): + def test_grpc_server_on_timeout(self, mock_channel_ready_future) -> None: """ - Test + Test utility function to test if a given gRPC channel is working or not - :param mock_channel_ready_future: _description_ - :type mock_channel_ready_future: _type_ + :param mock_channel_ready_future: mock_channel_ready_future + + :return: None """ mock_future = MagicMock() mock_future.result.side_effect = grpc.FutureTimeoutError() From c0921d03718a79487f78ac1150bbfaa4a3296971 Mon Sep 17 00:00:00 2001 From: Yuhu-kth Date: Tue, 16 Jul 2024 21:03:02 +0200 Subject: [PATCH 04/13] add test_grpc_util --- .../libs/csle-common/tests/test_grpc_util.py | 40 +++++++++++++++++++ 1 file changed, 40 insertions(+) create mode 100644 simulation-system/libs/csle-common/tests/test_grpc_util.py diff --git a/simulation-system/libs/csle-common/tests/test_grpc_util.py b/simulation-system/libs/csle-common/tests/test_grpc_util.py new file mode 100644 index 000000000..645f56fb2 --- /dev/null +++ b/simulation-system/libs/csle-common/tests/test_grpc_util.py @@ -0,0 +1,40 @@ +import grpc +from unittest.mock import patch, MagicMock +from csle_common.util.grpc_util import GrpcUtil + + +class TestGrpcUtilSuite: + """ + Test suite for grpc util + """ + + @patch("grpc.channel_ready_future") + def test_grpc_server_on(self, mock_channel_ready_future) -> None: + """ + Test utility function to test if a given gRPC channel is working or not + + :param mock_channel_ready_future: mock_channel_ready_future + + :return: None + """ + mock_future = MagicMock() + mock_channel_ready_future.return_value = mock_future + result = GrpcUtil.grpc_server_on(mock_channel_ready_future) + mock_future.result.assert_called() + assert result + + @patch("grpc.channel_ready_future") + def test_grpc_server_on_timeout(self, mock_channel_ready_future) -> None: + """ + Test utility function to test if a given gRPC channel is working or not + + :param mock_channel_ready_future: mock_channel_ready_future + + :return: None + """ + mock_future = MagicMock() + mock_future.result.side_effect = grpc.FutureTimeoutError() + mock_channel_ready_future.return_value = mock_future + result = GrpcUtil.grpc_server_on(mock_channel_ready_future) + mock_future.result.assert_called() + assert not result From cf8a3e34e4cdc829ad1bee1a997a53898adb3507 Mon Sep 17 00:00:00 2001 From: Yuhu-kth Date: Tue, 16 Jul 2024 21:04:10 +0200 Subject: [PATCH 05/13] add test_grpc_util --- .../libs/csle-common/tests/test_grpc_util.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/simulation-system/libs/csle-common/tests/test_grpc_util.py b/simulation-system/libs/csle-common/tests/test_grpc_util.py index 051dca870..645f56fb2 100644 --- a/simulation-system/libs/csle-common/tests/test_grpc_util.py +++ b/simulation-system/libs/csle-common/tests/test_grpc_util.py @@ -1,19 +1,20 @@ import grpc -import pytest from unittest.mock import patch, MagicMock from csle_common.util.grpc_util import GrpcUtil + class TestGrpcUtilSuite: """ Test suite for grpc util """ + @patch("grpc.channel_ready_future") def test_grpc_server_on(self, mock_channel_ready_future) -> None: """ Test utility function to test if a given gRPC channel is working or not :param mock_channel_ready_future: mock_channel_ready_future - + :return: None """ mock_future = MagicMock() @@ -21,14 +22,14 @@ def test_grpc_server_on(self, mock_channel_ready_future) -> None: result = GrpcUtil.grpc_server_on(mock_channel_ready_future) mock_future.result.assert_called() assert result - + @patch("grpc.channel_ready_future") def test_grpc_server_on_timeout(self, mock_channel_ready_future) -> None: """ Test utility function to test if a given gRPC channel is working or not :param mock_channel_ready_future: mock_channel_ready_future - + :return: None """ mock_future = MagicMock() @@ -37,4 +38,3 @@ def test_grpc_server_on_timeout(self, mock_channel_ready_future) -> None: result = GrpcUtil.grpc_server_on(mock_channel_ready_future) mock_future.result.assert_called() assert not result - \ No newline at end of file From d1190474281e65fe35e1f6bac2d8108976684efb Mon Sep 17 00:00:00 2001 From: Yuhu-kth Date: Tue, 16 Jul 2024 22:41:41 +0200 Subject: [PATCH 06/13] add test_import_util --- .../csle-common/tests/test_import_util.py | 64 +++++++++++++++++++ 1 file changed, 64 insertions(+) create mode 100644 simulation-system/libs/csle-common/tests/test_import_util.py diff --git a/simulation-system/libs/csle-common/tests/test_import_util.py b/simulation-system/libs/csle-common/tests/test_import_util.py new file mode 100644 index 000000000..2cb3374a3 --- /dev/null +++ b/simulation-system/libs/csle-common/tests/test_import_util.py @@ -0,0 +1,64 @@ +from unittest.mock import patch, MagicMock +from csle_common.util.import_util import ImportUtil + + +class TestImportUtilSuite: + """ + Test suite for import_util + """ + + @patch("os.path.exists") + @patch("csle_common.dao.system_identification.emulation_statistics.EmulationStatistics.from_json_file") + @patch("csle_common.metastore.metastore_facade.MetastoreFacade.save_emulation_statistic") + def test_import_emulation_statistics_from_disk_json( + self, mock_save_emulation_statistic, mock_from_json_file, mock_path_exists + ) -> None: + """ + Test the method that imports emulation statistics from disk to the metastore + + :param mock_save_emulation_statistic: mock_save_emulation_statistic + :param mock_from_json_file: mock_from_json_file + :param mock_path_exists: mock_path_exists + + :return: None + """ + mock_path_exists.return_value = True + mock_statistics = MagicMock() + mock_from_json_file.return_value = mock_statistics + input_file = "file.json" + emulation_name = "test_emulation" + ImportUtil.import_emulation_statistics_from_disk_json(input_file=input_file, emulation_name=emulation_name) + + mock_path_exists.assert_called() + mock_from_json_file.assert_called_once_with(input_file) + assert mock_statistics.emulation_name == emulation_name + mock_save_emulation_statistic.assert_called() + + @patch("os.path.exists") + @patch("csle_common.dao.emulation_config.emulation_trace.EmulationTrace.load_traces_from_disk") + @patch("csle_common.metastore.metastore_facade.MetastoreFacade.save_emulation_trace") + def test_import_emulation_traces_from_disk_json( + self, mock_save_emulation_trace, mock_load_traces_from_disk, mock_path_exists + ) -> None: + """ + Test the method that imports emulation traces from disk to the metastore + + :param mock_save_emulation_trace: mock_save_emulation_trac + :param mock_load_traces_from_disk: mock_load_traces_from_disk + :param mock_path_exists: mock_path_exists + + :return: None + """ + mock_path_exists.return_value = True + mock_trace_1 = MagicMock() + mock_trace_2 = MagicMock() + mock_load_traces_from_disk.return_value = [mock_trace_1, mock_trace_2] + input_file = "file.json" + emulation_name = "test_emulation" + ImportUtil.import_emulation_traces_from_disk_json(input_file=input_file, emulation_name=emulation_name) + + mock_path_exists.assert_called() + mock_load_traces_from_disk.assert_called() + assert mock_trace_1.emulation_name == emulation_name + assert mock_trace_2.emulation_name == emulation_name + assert mock_save_emulation_trace.call_count == 2 From ee29b234cf6712b9e4df4389aa532cb2a068cd28 Mon Sep 17 00:00:00 2001 From: Yuhu-kth Date: Tue, 16 Jul 2024 22:44:34 +0200 Subject: [PATCH 07/13] add test_grpc_util --- simulation-system/libs/csle-common/tests/test_grpc_util.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/simulation-system/libs/csle-common/tests/test_grpc_util.py b/simulation-system/libs/csle-common/tests/test_grpc_util.py index 645f56fb2..067139d78 100644 --- a/simulation-system/libs/csle-common/tests/test_grpc_util.py +++ b/simulation-system/libs/csle-common/tests/test_grpc_util.py @@ -26,7 +26,7 @@ def test_grpc_server_on(self, mock_channel_ready_future) -> None: @patch("grpc.channel_ready_future") def test_grpc_server_on_timeout(self, mock_channel_ready_future) -> None: """ - Test utility function to test if a given gRPC channel is working or not + Test utility function to test if a given gRPC channel is not working :param mock_channel_ready_future: mock_channel_ready_future From 5a010d403d0a8356e2ca338ffbba41f0d60ac80f Mon Sep 17 00:00:00 2001 From: Yuhu-kth Date: Wed, 17 Jul 2024 11:03:27 +0200 Subject: [PATCH 08/13] add test_management_util --- .../csle-common/tests/test_management_util.py | 80 +++++++++++++++++++ 1 file changed, 80 insertions(+) create mode 100644 simulation-system/libs/csle-common/tests/test_management_util.py diff --git a/simulation-system/libs/csle-common/tests/test_management_util.py b/simulation-system/libs/csle-common/tests/test_management_util.py new file mode 100644 index 000000000..1815e3b62 --- /dev/null +++ b/simulation-system/libs/csle-common/tests/test_management_util.py @@ -0,0 +1,80 @@ +import csle_common.constants.constants as constants +from csle_common.util.management_util import ManagementUtil +from unittest.mock import patch + +class TestManagementUtilSuite: + """ + Test suite for management util + """ + + @patch("csle_common.metastore.metastore_facade.MetastoreFacade.list_management_users") + @patch("bcrypt.gensalt") + @patch("bcrypt.hashpw") + @patch("csle_common.metastore.metastore_facade.MetastoreFacade.save_management_user") + def test_create_default_management_admin_account( + self, mock_save_management_user, mock_hashpw, mock_gensalt, mock_list_management_users + ) -> None: + """ + Test the method that creates the default management admin account + + :param mock_save_management_user: mock_save_management_user + :param mock_hashpw: mock_hashpw + :param mock_gensalt: mock_gensalt + :param mock_list_management_users: mock_list_management_users + + :return: None + """ + mock_list_management_users.return_value = [] + mock_salt = b"salt" + mock_gensalt.return_value = mock_salt + mock_hash = b"hashed_password" + mock_hashpw.return_value = mock_hash + + constants.CSLE_ADMIN.MANAGEMENT_USER = "admin" + constants.CSLE_ADMIN.MANAGEMENT_PW = "password" + constants.CSLE_ADMIN.MANAGEMENT_FIRST_NAME = "first" + constants.CSLE_ADMIN.MANAGEMENT_LAST_NAME = "last" + constants.CSLE_ADMIN.MANAGEMENT_ORGANIZATION = "organization" + constants.CSLE_ADMIN.MANAGEMENT_EMAIL = "admin@email.com" + + ManagementUtil.create_default_management_admin_account() + mock_list_management_users.assert_called_once() + mock_gensalt.assert_called_once() + mock_hashpw.assert_called_once_with(constants.CSLE_ADMIN.MANAGEMENT_PW.encode("utf-8"), mock_salt) + mock_save_management_user.assert_called_once() + + @patch("csle_common.metastore.metastore_facade.MetastoreFacade.list_management_users") + @patch("bcrypt.gensalt") + @patch("bcrypt.hashpw") + @patch("csle_common.metastore.metastore_facade.MetastoreFacade.save_management_user") + def test_create_default_management_guest_account( + self, mock_save_management_user, mock_hashpw, mock_gensalt, mock_list_management_users + ) -> None: + """ + Test the method that creates the default management guest account + + :param mock_save_management_user: mock_save_management_user + :param mock_hashpw: mock_hashpw + :param mock_gensalt: mock_gensalt + :param mock_list_management_users: mock_list_management_users + + :return: None + """ + mock_list_management_users.return_value = [] + mock_salt = b"salt" + mock_gensalt.return_value = mock_salt + mock_hash = b"hashed_password" + mock_hashpw.return_value = mock_hash + + constants.CSLE_GUEST.MANAGEMENT_USER = "user" + constants.CSLE_GUEST.MANAGEMENT_PW = "password" + constants.CSLE_GUEST.MANAGEMENT_FIRST_NAME = "guest_first" + constants.CSLE_GUEST.MANAGEMENT_LAST_NAME = "guest_last" + constants.CSLE_GUEST.MANAGEMENT_ORGANIZATION = "guest_organization" + constants.CSLE_GUEST.MANAGEMENT_EMAIL = "guest@email.com" + + ManagementUtil.create_default_management_guest_account() + mock_list_management_users.assert_called_once() + mock_gensalt.assert_called_once() + mock_hashpw.assert_called_once_with(constants.CSLE_GUEST.MANAGEMENT_PW.encode("utf-8"), mock_salt) + mock_save_management_user.assert_called_once() From 9b77ab37c3f0fc7d0577fb008ac6d50e67501eb3 Mon Sep 17 00:00:00 2001 From: Yuhu-kth Date: Wed, 17 Jul 2024 11:05:29 +0200 Subject: [PATCH 09/13] add test_grpc_util --- simulation-system/libs/csle-common/tests/test_grpc_util.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/simulation-system/libs/csle-common/tests/test_grpc_util.py b/simulation-system/libs/csle-common/tests/test_grpc_util.py index 067139d78..22d052e2e 100644 --- a/simulation-system/libs/csle-common/tests/test_grpc_util.py +++ b/simulation-system/libs/csle-common/tests/test_grpc_util.py @@ -5,7 +5,7 @@ class TestGrpcUtilSuite: """ - Test suite for grpc util + Test suite for grpc_util """ @patch("grpc.channel_ready_future") From 971d097cbfd301a695ffcb1599f227c2526b899e Mon Sep 17 00:00:00 2001 From: Yuhu-kth Date: Wed, 17 Jul 2024 11:08:12 +0200 Subject: [PATCH 10/13] add test_import_util --- simulation-system/libs/csle-common/tests/test_import_util.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/simulation-system/libs/csle-common/tests/test_import_util.py b/simulation-system/libs/csle-common/tests/test_import_util.py index 2cb3374a3..0cb158cf9 100644 --- a/simulation-system/libs/csle-common/tests/test_import_util.py +++ b/simulation-system/libs/csle-common/tests/test_import_util.py @@ -6,7 +6,7 @@ class TestImportUtilSuite: """ Test suite for import_util """ - + @patch("os.path.exists") @patch("csle_common.dao.system_identification.emulation_statistics.EmulationStatistics.from_json_file") @patch("csle_common.metastore.metastore_facade.MetastoreFacade.save_emulation_statistic") From 03db65386726dc6afec662d20a854c52aeb62ba7 Mon Sep 17 00:00:00 2001 From: Yuhu-kth Date: Wed, 17 Jul 2024 14:00:56 +0200 Subject: [PATCH 11/13] add test_multiprocessing_util --- .../tests/test_multiprocessing_util.py | 43 +++++++++++++++++++ 1 file changed, 43 insertions(+) create mode 100644 simulation-system/libs/csle-common/tests/test_multiprocessing_util.py diff --git a/simulation-system/libs/csle-common/tests/test_multiprocessing_util.py b/simulation-system/libs/csle-common/tests/test_multiprocessing_util.py new file mode 100644 index 000000000..09664c081 --- /dev/null +++ b/simulation-system/libs/csle-common/tests/test_multiprocessing_util.py @@ -0,0 +1,43 @@ +from unittest.mock import patch +from csle_common.util.multiprocessing_util import NoDaemonProcess +from csle_common.util.multiprocessing_util import NoDaemonContext +from csle_common.util.multiprocessing_util import NestablePool + + +class TestMultiprocessingUtilSuite: + """ + Test suite for multiprocessing util + """ + + def test_daemon(self) -> None: + """ + Test the process with daemon property set to false + + :return: None + """ + result = NoDaemonProcess(target=lambda: None).daemon + assert not result + + def test_no_daemon_context(self) -> None: + """ + Test the NoDaemonContext method + + :return: None + """ + context = NoDaemonContext() + process = context.Process(target=lambda: None) + assert isinstance(process, NoDaemonProcess) + assert not process.daemon + + @patch("multiprocessing.get_context") + def test_nestable_pool_initialization(self, mock_get_context) -> None: + """ + Test the method that initializes the pool + + :param mock_get_context: mock_get_context + + :return: None + """ + mock_get_context.return_value = NoDaemonContext() + pool = NestablePool() + assert pool From b5584f5426a834c020360ae3e52a3b625031f5c9 Mon Sep 17 00:00:00 2001 From: Yuhu-kth Date: Wed, 17 Jul 2024 22:05:02 +0200 Subject: [PATCH 12/13] test_plotting_util w/mypy checked --- .../csle-common/tests/test_plotting_util.py | 48 +++++++++++++++++++ 1 file changed, 48 insertions(+) create mode 100644 simulation-system/libs/csle-common/tests/test_plotting_util.py diff --git a/simulation-system/libs/csle-common/tests/test_plotting_util.py b/simulation-system/libs/csle-common/tests/test_plotting_util.py new file mode 100644 index 000000000..d21f0988a --- /dev/null +++ b/simulation-system/libs/csle-common/tests/test_plotting_util.py @@ -0,0 +1,48 @@ +from csle_common.util.plotting_util import PlottingUtil +from scipy import stats +import numpy as np + + +class TestPlottingUtilSuite: + """ + Test suite for plotting util + """ + + def test_running_average(self) -> None: + """ + Test the function used to compute the running average of the last N elements of a vector x + + :return: None + """ + x = np.array([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]) + N = 3 + expected = np.array([1, 2, 3, 3, 4, 5, 6, 7, 8, 9]) + result = PlottingUtil.running_average(x, N) + assert result.any() == expected.any() + + def test_mean_confidence_interval(self) -> None: + """ + Test function that computes confidence intervals + + :return: None + """ + data = np.array([1, 2, 3, 4, 5]) + mean, h = PlottingUtil.mean_confidence_interval(data=data, confidence=0.95) + expected_mean = np.mean(data) + expected_se = stats.sem(data) + expected_h = expected_se * stats.t.ppf((1 + 0.95) / 2.0, len(data) - 1) + assert expected_mean == mean + assert expected_h == h + + def test_min_max_norm(self) -> None: + """ + Test function that computes min-max normalization of a vector + + :return: None + """ + vec = np.array([1, 2, 3, 4, 5]) + min_val = 1 + max_val = 5 + expected = np.array([0.0, 0.25, 0.5, 0.75, 1.0]) + result = PlottingUtil.min_max_norm(vec, max_val, min_val) + assert result.any() == expected.any() From 415183bc1b96351cbdaf96515947ae87544d9e06 Mon Sep 17 00:00:00 2001 From: Yuhu-kth Date: Thu, 18 Jul 2024 15:37:09 +0200 Subject: [PATCH 13/13] add ssh_util test w/mypy test --- .../libs/csle-common/tests/test_ssh_util.py | 53 +++++++++++++++++++ 1 file changed, 53 insertions(+) create mode 100644 simulation-system/libs/csle-common/tests/test_ssh_util.py diff --git a/simulation-system/libs/csle-common/tests/test_ssh_util.py b/simulation-system/libs/csle-common/tests/test_ssh_util.py new file mode 100644 index 000000000..2fa9acb91 --- /dev/null +++ b/simulation-system/libs/csle-common/tests/test_ssh_util.py @@ -0,0 +1,53 @@ +from csle_common.util.ssh_util import SSHUtil +from unittest.mock import patch, MagicMock +import pytest + + +class TestSSHUtilSuite: + """ + Test suite for ssh_util + """ + + @pytest.fixture(autouse=True) + def mock_sleep(self): + """ + Mock time.sleep to avoid delays + """ + with patch("time.sleep", return_value=None): + yield + + @patch("csle_common.util.ssh_util.SSHUtil.execute_ssh_cmd") + def test_execute_ssh_cmds(self, mock_execute_ssh_cmd) -> None: + """ + Test the method that executes a list of commands over an ssh connection to the emulation + + :param mock_execute_ssh_cmd: mock_execute_ssh_cmd + + :return: None + """ + mock_execute_ssh_cmd.return_value = (b"output", b"error", 1.0) + cmds = ["ls", "pwd", "whoami"] + conn = MagicMock() + results = SSHUtil.execute_ssh_cmds(cmds, conn) + mock_execute_ssh_cmd.assert_called() + assert results == [(b"output", b"error", 1.0)] * len(cmds) + + def test_execute_ssh_cmd(self) -> None: + """ + Test the method that executes an action on the emulation over a ssh connection + + :return: None + """ + conn = MagicMock() + mock_transport = MagicMock() + mock_session = MagicMock() + mock_session.exit_status_ready.return_value = True + mock_session.recv_ready.return_value = True + mock_session.recv_stderr_ready.return_value = True + mock_session.recv.side_effect = [b"output", b""] + mock_session.recv_stderr.side_effect = [b"error", b""] + conn.get_transport.return_value = mock_transport + mock_transport.open_session.return_value = mock_session + + with pytest.raises(ConnectionError, match="Connection failed"): + SSHUtil.execute_ssh_cmd("ls", conn)