update ceph source to reef 18.2.1

[ceph.git] / ceph / src / pybind / mgr / cephadm / tests / test_ssh.py

diff --git a/ceph/src/pybind/mgr/cephadm/tests/test_ssh.py b/ceph/src/pybind/mgr/cephadm/tests/test_ssh.py
index 4197d8d7ef0ede87bbdf529aa62d25cc06e9b3e4..29f01b6c79724001bb86ad6768827e57b96a7184 100644 (file)
--- a/ceph/src/pybind/mgr/cephadm/tests/test_ssh.py
+++ b/ceph/src/pybind/mgr/cephadm/tests/test_ssh.py
@@ -1,3 +1,5 @@
+import asyncssh
+from asyncssh.process import SSHCompletedProcess
 from unittest import mock
 try:
     # AsyncMock was not added until python 3.8
@@ -19,6 +21,7 @@ from ceph.deployment.hostspec import HostSpec
 from cephadm import CephadmOrchestrator
 from cephadm.serve import CephadmServe
 from cephadm.tests.fixtures import with_host, wait, async_side_effect
+from orchestrator import OrchestratorError
 
 
 @pytest.mark.skipif(ConnectionLost is None, reason='no asyncssh')
@@ -49,6 +52,52 @@ class TestWithSSH:
                 out = wait(cephadm_module, cephadm_module.get_hosts())[0].to_json()
                 assert out == HostSpec('test', '1::4').to_json()
 
+    def test_ssh_remote_cmds_execution(self, cephadm_module):
+
+        if not AsyncMock:
+            # can't run this test if we could not import AsyncMock
+            return
+
+        class FakeConn:
+            def __init__(self, exception=None, returncode=0):
+                self.exception = exception
+                self.returncode = returncode
+
+            async def run(self, *args, **kwargs):
+                if self.exception:
+                    raise self.exception
+                else:
+                    return SSHCompletedProcess(returncode=self.returncode, stdout="", stderr="")
+
+            async def close(self):
+                pass
+
+        def run_test(host, conn, expected_error):
+            mock_connect = AsyncMock(return_value=conn)
+            with pytest.raises(OrchestratorError, match=expected_error):
+                with mock.patch("asyncssh.connect", new=mock_connect):
+                    with with_host(cephadm_module, host):
+                        CephadmServe(cephadm_module)._check_host(host)
+
+        # Test case 1: command failure
+        run_test('test1', FakeConn(returncode=1), "Command .+ failed")
+
+        # Test case 2: connection error
+        run_test('test2', FakeConn(exception=asyncssh.ChannelOpenError(1, "", "")), "Unable to reach remote host test2.")
+
+        # Test case 3: asyncssh ProcessError
+        stderr = "my-process-stderr"
+        run_test('test3', FakeConn(exception=asyncssh.ProcessError(returncode=3,
+                                                                   env="",
+                                                                   command="",
+                                                                   subsystem="",
+                                                                   exit_status="",
+                                                                   exit_signal="",
+                                                                   stderr=stderr,
+                                                                   stdout="")), f"Cannot execute the command.+{stderr}")
+        # Test case 4: generic error
+        run_test('test4', FakeConn(exception=Exception), "Generic error while executing command.+")
+
 
 @pytest.mark.skipif(ConnectionLost is not None, reason='asyncssh')
 class TestWithoutSSH: