Merge pull request #394 from Limmen/ssh

unit test ssh_util

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)