]>
git.proxmox.com Git - ceph.git/blob - ceph/src/pybind/mgr/cephadm/tests/test_ssh.py
1 from unittest
import mock
3 # AsyncMock was not added until python 3.8
4 from unittest
.mock
import AsyncMock
6 from asyncmock
import AsyncMock
13 from asyncssh
.misc
import ConnectionLost
17 from ceph
.deployment
.hostspec
import HostSpec
19 from cephadm
import CephadmOrchestrator
20 from cephadm
.serve
import CephadmServe
21 from cephadm
.tests
.fixtures
import with_host
, wait
, async_side_effect
24 @pytest.mark
.skipif(ConnectionLost
is None, reason
='no asyncssh')
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))
33 # can't run this test if we could not import AsyncMock
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')
41 assert "Host 'test' not found" in err
43 out
= wait(cephadm_module
, cephadm_module
.get_hosts())[0].to_json()
44 assert out
== HostSpec('test', '1::4', status
='Offline').to_json()
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()
53 @pytest.mark
.skipif(ConnectionLost
is not None, reason
='asyncssh')
55 def test_can_run(self
, cephadm_module
: CephadmOrchestrator
):
56 assert cephadm_module
.can_run() == (False, "loading asyncssh library:No module named 'asyncssh'")