return names
+ def _append_mds_metadata(self, mds_versions, metadata_key):
+ metadata = mgr.get_metadata('mds', metadata_key)
+ if metadata is None:
+ return
+ mds_versions[metadata.get('ceph_version', 'unknown')].append(metadata_key)
+
# pylint: disable=too-many-statements,too-many-branches
def fs_status(self, fs_id):
mds_versions = defaultdict(list)
else:
activity = 0.0
- metadata = mgr.get_metadata('mds', info['name'])
- mds_versions[metadata.get('ceph_version', 'unknown')].append(
- info['name'])
+ self._append_mds_metadata(mds_versions, info['name'])
rank_table.append(
{
"rank": rank,
standby_table = []
for standby in fsmap['standbys']:
- metadata = mgr.get_metadata('mds', standby['name'])
- mds_versions[metadata.get('ceph_version', 'unknown')].append(
- standby['name'])
-
+ self._append_mds_metadata(mds_versions, standby['name'])
standby_table.append({
'name': standby['name']
})
@UiApiController('/cephfs', Scope.CEPHFS)
class CephFsUi(CephFS):
- RESOURCE_ID='fs_id'
+ RESOURCE_ID = 'fs_id'
@RESTController.Resource('GET')
def tabs(self, fs_id):
data['clients'] = self._clients(fs_id)
return data
-
-
--- /dev/null
+# -*- coding: utf-8 -*-
+from collections import defaultdict
+try:
+ from mock import Mock
+except ImportError:
+ from unittest.mock import Mock
+
+from .. import mgr
+from . import ControllerTestCase
+from ..controllers.cephfs import CephFS
+
+
+class MetaDataMock(object):
+ def get(self, _x, _y):
+ return 'bar'
+
+
+def get_metadata_mock(key, meta_key):
+ return {
+ 'mds': {
+ None: None, # Unknown key
+ 'foo': MetaDataMock()
+ }[meta_key]
+ }[key]
+
+
+class CephFsTest(ControllerTestCase):
+ cephFs = CephFS()
+
+ @classmethod
+ def setup_server(cls):
+ mgr.get_metadata = Mock(side_effect=get_metadata_mock)
+
+ def tearDown(self):
+ mgr.get_metadata.stop()
+
+ def test_append_of_mds_metadata_if_key_is_not_found(self):
+ mds_versions = defaultdict(list)
+ # pylint: disable=protected-access
+ self.cephFs._append_mds_metadata(mds_versions, None)
+ self.assertEqual(len(mds_versions), 0)
+
+ def test_append_of_mds_metadata_with_existing_metadata(self):
+ mds_versions = defaultdict(list)
+ # pylint: disable=protected-access
+ self.cephFs._append_mds_metadata(mds_versions, 'foo')
+ self.assertEqual(len(mds_versions), 1)
+ self.assertEqual(mds_versions['bar'], ['foo'])
@classmethod
def setup_server(cls):
mgr.list_servers.return_value = mock_list_servers
- mgr.get_metadata.return_value = mock_get_metadata
+ mgr.get_metadata = mock.Mock(return_value=mock_get_metadata)
mgr.get_daemon_status.return_value = mock_get_daemon_status
mgr.get.side_effect = lambda key: {
'osd_map': mock_osd_map,