1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
|
import logging
from tasks.mgr.mgr_test_case import MgrTestCase
log = logging.getLogger(__name__)
class TestCephadmCLI(MgrTestCase):
def _cmd(self, *args) -> str:
assert self.mgr_cluster is not None
return self.mgr_cluster.mon_manager.raw_cluster_cmd(*args)
def _orch_cmd(self, *args) -> str:
return self._cmd("orch", *args)
def setUp(self):
super(TestCephadmCLI, self).setUp()
def test_yaml(self):
"""
to prevent oddities like
>>> import yaml
... from collections import OrderedDict
... assert yaml.dump(OrderedDict()) == '!!python/object/apply:collections.OrderedDict\\n- []\\n'
"""
out = self._orch_cmd('device', 'ls', '--format', 'yaml')
self.assertNotIn('!!python', out)
out = self._orch_cmd('host', 'ls', '--format', 'yaml')
self.assertNotIn('!!python', out)
out = self._orch_cmd('ls', '--format', 'yaml')
self.assertNotIn('!!python', out)
out = self._orch_cmd('ps', '--format', 'yaml')
self.assertNotIn('!!python', out)
out = self._orch_cmd('status', '--format', 'yaml')
self.assertNotIn('!!python', out)
def test_pause(self):
self._orch_cmd('pause')
self.wait_for_health('CEPHADM_PAUSED', 60)
self._orch_cmd('resume')
self.wait_for_health_clear(60)
def test_daemon_restart(self):
self._orch_cmd('daemon', 'stop', 'osd.0')
self.wait_for_health('OSD_DOWN', 60)
self._orch_cmd('daemon', 'start', 'osd.0')
self.wait_for_health_clear(120)
self._orch_cmd('daemon', 'restart', 'osd.0')
def test_device_ls_wide(self):
self._orch_cmd('device', 'ls', '--wide')
def test_cephfs_mirror(self):
self._orch_cmd('apply', 'cephfs-mirror')
self.wait_until_true(lambda: 'cephfs-mirror' in self._orch_cmd('ps'), 60)
self.wait_for_health_clear(60)
self._orch_cmd('rm', 'cephfs-mirror')
self.wait_until_true(lambda: 'cephfs-mirror' not in self._orch_cmd('ps'), 60)
|
