mirror of
https://github.com/ceph/ceph
synced 2025-02-09 03:49:38 +00:00
96c8a48010
allow a user of the orchestrator interface to express that the inventory query should not read from any cached inventory state. Signed-off-by: Noah Watkins <noahwatkins@gmail.com>
111 lines
3.4 KiB
Python
111 lines
3.4 KiB
Python
import json
|
|
import logging
|
|
from tempfile import NamedTemporaryFile
|
|
|
|
from teuthology.exceptions import CommandFailedError
|
|
|
|
from mgr_test_case import MgrTestCase
|
|
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
class TestOrchestratorCli(MgrTestCase):
|
|
MGRS_REQUIRED = 1
|
|
|
|
def _orch_cmd(self, *args):
|
|
return self.mgr_cluster.mon_manager.raw_cluster_cmd("orchestrator", *args)
|
|
|
|
def _orch_cmd_result(self, *args, **kwargs):
|
|
"""
|
|
superfluous, but raw_cluster_cmd doesn't support kwargs.
|
|
"""
|
|
res = self.mgr_cluster.mon_manager.raw_cluster_cmd_result("orchestrator", *args, **kwargs)
|
|
self.assertEqual(res, 0)
|
|
|
|
|
|
def setUp(self):
|
|
super(TestOrchestratorCli, self).setUp()
|
|
|
|
self._load_module("orchestrator_cli")
|
|
self._load_module("test_orchestrator")
|
|
self._orch_cmd("set", "backend", "test_orchestrator")
|
|
|
|
def test_status(self):
|
|
ret = self._orch_cmd("status")
|
|
self.assertIn("test_orchestrator", ret)
|
|
|
|
def test_device_ls(self):
|
|
ret = self._orch_cmd("device", "ls")
|
|
self.assertIn("localhost:", ret)
|
|
|
|
def test_device_ls_refresh(self):
|
|
ret = self._orch_cmd("device", "ls", "--refresh")
|
|
self.assertIn("localhost:", ret)
|
|
|
|
def test_device_ls_hoshs(self):
|
|
ret = self._orch_cmd("device", "ls", "localhost", "host1")
|
|
self.assertIn("localhost:", ret)
|
|
|
|
|
|
def test_device_ls_json(self):
|
|
ret = self._orch_cmd("device", "ls", "--format", "json")
|
|
self.assertIn("localhost", ret)
|
|
self.assertIsInstance(json.loads(ret), list)
|
|
|
|
def test_service_ls(self):
|
|
ret = self._orch_cmd("service", "ls")
|
|
self.assertIn("ceph-mgr", ret)
|
|
|
|
def test_service_ls_json(self):
|
|
ret = self._orch_cmd("service", "ls", "--format", "json")
|
|
self.assertIsInstance(json.loads(ret), list)
|
|
self.assertIn("ceph-mgr", ret)
|
|
|
|
|
|
def test_service_action(self):
|
|
self._orch_cmd("service", "reload", "mds", "cephfs")
|
|
self._orch_cmd("service", "stop", "mds", "cephfs")
|
|
self._orch_cmd("service", "start", "mds", "cephfs")
|
|
|
|
def test_service_instance_action(self):
|
|
self._orch_cmd("service-instance", "reload", "mds", "a")
|
|
self._orch_cmd("service-instance", "stop", "mds", "a")
|
|
self._orch_cmd("service-instance", "start", "mds", "a")
|
|
|
|
def test_osd_create(self):
|
|
self._orch_cmd("osd", "create", "*:device")
|
|
self._orch_cmd("osd", "create", "*:device,device2")
|
|
|
|
drive_group = {
|
|
"host_pattern": "*",
|
|
"data_devices": {"paths": ["/dev/sda"]}
|
|
}
|
|
|
|
self._orch_cmd_result("osd", "create", "-i", "-", stdin=json.dumps(drive_group))
|
|
|
|
with self.assertRaises(CommandFailedError):
|
|
self._orch_cmd("osd", "create", "notfound:device")
|
|
|
|
def test_mds_add(self):
|
|
self._orch_cmd("mds", "add", "service_name")
|
|
|
|
def test_rgw_add(self):
|
|
self._orch_cmd("rgw", "add", "service_name")
|
|
|
|
def test_nfs_add(self):
|
|
self._orch_cmd("nfs", "add", "service_name", "pool", "--namespace", "ns")
|
|
self._orch_cmd("nfs", "add", "service_name", "pool")
|
|
|
|
def test_osd_rm(self):
|
|
self._orch_cmd("osd", "rm", "osd.0")
|
|
|
|
def test_mds_rm(self):
|
|
self._orch_cmd("mds", "rm", "foo")
|
|
|
|
def test_rgw_rm(self):
|
|
self._orch_cmd("rgw", "rm", "foo")
|
|
|
|
def test_nfs_rm(self):
|
|
self._orch_cmd("nfs", "rm", "service_name")
|