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

   1 from unittest import mock
   2 try:
   3     # AsyncMock was not added until python 3.8
   4     from unittest.mock import AsyncMock
   5 except ImportError:
   6     from asyncmock import AsyncMock
   7 except ImportError:
   8     AsyncMock = None
   9 import pytest
  10
  11
  12 try:
  13     from asyncssh.misc import ConnectionLost
  14 except ImportError:
  15     ConnectionLost = None
  16
  17 from ceph.deployment.hostspec import HostSpec
  18
  19 from cephadm import CephadmOrchestrator
  20 from cephadm.serve import CephadmServe
  21 from cephadm.tests.fixtures import with_host, wait, async_side_effect
  22
  23
  24 @pytest.mark.skipif(ConnectionLost is None, reason='no asyncssh')
  25 class TestWithSSH:
  26     @mock.patch("cephadm.ssh.SSHManager._execute_command")
  27     @mock.patch("cephadm.ssh.SSHManager._check_execute_command")
  28     def test_offline(self, check_execute_command, execute_command, cephadm_module):
  29         check_execute_command.side_effect = async_side_effect('')
  30         execute_command.side_effect = async_side_effect(('', '', 0))
  31
  32         if not AsyncMock:
  33             # can't run this test if we could not import AsyncMock
  34             return
  35         mock_connect = AsyncMock(return_value='')
  36         with mock.patch("asyncssh.connect", new=mock_connect) as asyncssh_connect:
  37             with with_host(cephadm_module, 'test'):
  38                 asyncssh_connect.side_effect = ConnectionLost('reason')
  39                 code, out, err = cephadm_module.check_host('test')
  40                 assert out == ''
  41                 assert "Host 'test' not found" in err
  42
  43                 out = wait(cephadm_module, cephadm_module.get_hosts())[0].to_json()
  44                 assert out == HostSpec('test', '1::4', status='Offline').to_json()
  45
  46                 asyncssh_connect.return_value = mock.MagicMock()
  47                 asyncssh_connect.side_effect = None
  48                 assert CephadmServe(cephadm_module)._check_host('test') is None
  49                 out = wait(cephadm_module, cephadm_module.get_hosts())[0].to_json()
  50                 assert out == HostSpec('test', '1::4').to_json()
  51
  52
  53 @pytest.mark.skipif(ConnectionLost is not None, reason='asyncssh')
  54 class TestWithoutSSH:
  55     def test_can_run(self, cephadm_module: CephadmOrchestrator):
  56         assert cephadm_module.can_run() == (False, "loading asyncssh library:No module named 'asyncssh'")