ceph/src/pybind/mgr/dashboard/tests/test_cephfs.py

   1 # -*- coding: utf-8 -*-
   2 from collections import defaultdict
   3 try:
   4     from mock import Mock
   5 except ImportError:
   6     from unittest.mock import Mock
   7
   8 from .. import mgr
   9 from . import ControllerTestCase
  10 from ..controllers.cephfs import CephFS
  11
  12
  13 class MetaDataMock(object):
  14     def get(self, _x, _y):
  15         return 'bar'
  16
  17
  18 def get_metadata_mock(key, meta_key):
  19     return {
  20         'mds': {
  21             None: None,  # Unknown key
  22             'foo': MetaDataMock()
  23         }[meta_key]
  24     }[key]
  25
  26
  27 class CephFsTest(ControllerTestCase):
  28     cephFs = CephFS()
  29
  30     @classmethod
  31     def setup_server(cls):
  32         mgr.get_metadata = Mock(side_effect=get_metadata_mock)
  33
  34     def tearDown(self):
  35         mgr.get_metadata.stop()
  36
  37     def test_append_of_mds_metadata_if_key_is_not_found(self):
  38         mds_versions = defaultdict(list)
  39         # pylint: disable=protected-access
  40         self.cephFs._append_mds_metadata(mds_versions, None)
  41         self.assertEqual(len(mds_versions), 0)
  42
  43     def test_append_of_mds_metadata_with_existing_metadata(self):
  44         mds_versions = defaultdict(list)
  45         # pylint: disable=protected-access
  46         self.cephFs._append_mds_metadata(mds_versions, 'foo')
  47         self.assertEqual(len(mds_versions), 1)
  48         self.assertEqual(mds_versions['bar'], ['foo'])