ceph/qa/tasks/cephadm_cases/test_cli.py

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