ceph/src/pybind/mgr/cephadm/tests/test_ssh.py

   1 import asyncssh
   2 from asyncssh.process import SSHCompletedProcess
   3 from unittest import mock
   4 try:
   5     # AsyncMock was not added until python 3.8
   6     from unittest.mock import AsyncMock
   7 except ImportError:
   8     from asyncmock import AsyncMock
   9 except ImportError:
  10     AsyncMock = None
  11 import pytest
  12
  13
  14 try:
  15     from asyncssh.misc import ConnectionLost
  16 except ImportError:
  17     ConnectionLost = None
  18
  19 from ceph.deployment.hostspec import HostSpec
  20
  21 from cephadm import CephadmOrchestrator
  22 from cephadm.serve import CephadmServe
  23 from cephadm.tests.fixtures import with_host, wait, async_side_effect
  24 from orchestrator import OrchestratorError
  25
  26
  27 @pytest.mark.skipif(ConnectionLost is None, reason='no asyncssh')
  28 class TestWithSSH:
  29     @mock.patch("cephadm.ssh.SSHManager._execute_command")
  30     @mock.patch("cephadm.ssh.SSHManager._check_execute_command")
  31     def test_offline(self, check_execute_command, execute_command, cephadm_module):
  32         check_execute_command.side_effect = async_side_effect('')
  33         execute_command.side_effect = async_side_effect(('', '', 0))
  34
  35         if not AsyncMock:
  36             # can't run this test if we could not import AsyncMock
  37             return
  38         mock_connect = AsyncMock(return_value='')
  39         with mock.patch("asyncssh.connect", new=mock_connect) as asyncssh_connect:
  40             with with_host(cephadm_module, 'test'):
  41                 asyncssh_connect.side_effect = ConnectionLost('reason')
  42                 code, out, err = cephadm_module.check_host('test')
  43                 assert out == ''
  44                 assert "Failed to connect to test at address (1::4)" in err
  45
  46                 out = wait(cephadm_module, cephadm_module.get_hosts())[0].to_json()
  47                 assert out == HostSpec('test', '1::4', status='Offline').to_json()
  48
  49                 asyncssh_connect.return_value = mock.MagicMock()
  50                 asyncssh_connect.side_effect = None
  51                 assert CephadmServe(cephadm_module)._check_host('test') is None
  52                 out = wait(cephadm_module, cephadm_module.get_hosts())[0].to_json()
  53                 assert out == HostSpec('test', '1::4').to_json()
  54
  55     def test_ssh_remote_cmds_execution(self, cephadm_module):
  56
  57         if not AsyncMock:
  58             # can't run this test if we could not import AsyncMock
  59             return
  60
  61         class FakeConn:
  62             def __init__(self, exception=None, returncode=0):
  63                 self.exception = exception
  64                 self.returncode = returncode
  65
  66             async def run(self, *args, **kwargs):
  67                 if self.exception:
  68                     raise self.exception
  69                 else:
  70                     return SSHCompletedProcess(returncode=self.returncode, stdout="", stderr="")
  71
  72             async def close(self):
  73                 pass
  74
  75         def run_test(host, conn, expected_error):
  76             mock_connect = AsyncMock(return_value=conn)
  77             with pytest.raises(OrchestratorError, match=expected_error):
  78                 with mock.patch("asyncssh.connect", new=mock_connect):
  79                     with with_host(cephadm_module, host):
  80                         CephadmServe(cephadm_module)._check_host(host)
  81
  82         # Test case 1: command failure
  83         run_test('test1', FakeConn(returncode=1), "Command .+ failed")
  84
  85         # Test case 2: connection error
  86         run_test('test2', FakeConn(exception=asyncssh.ChannelOpenError(1, "", "")), "Unable to reach remote host test2.")
  87
  88         # Test case 3: asyncssh ProcessError
  89         stderr = "my-process-stderr"
  90         run_test('test3', FakeConn(exception=asyncssh.ProcessError(returncode=3,
  91                                                                    env="",
  92                                                                    command="",
  93                                                                    subsystem="",
  94                                                                    exit_status="",
  95                                                                    exit_signal="",
  96                                                                    stderr=stderr,
  97                                                                    stdout="")), f"Cannot execute the command.+{stderr}")
  98         # Test case 4: generic error
  99         run_test('test4', FakeConn(exception=Exception), "Generic error while executing command.+")
 100
 101
 102 @pytest.mark.skipif(ConnectionLost is not None, reason='asyncssh')
 103 class TestWithoutSSH:
 104     def test_can_run(self, cephadm_module: CephadmOrchestrator):
 105         assert cephadm_module.can_run() == (False, "loading asyncssh library:No module named 'asyncssh'")