]> git.proxmox.com Git - ceph.git/blob - ceph/src/pybind/mgr/dashboard/tests/test_ceph_service.py
import 15.2.4
[ceph.git] / ceph / src / pybind / mgr / dashboard / tests / test_ceph_service.py
1 # -*- coding: utf-8 -*-
2 # pylint: disable=dangerous-default-value,too-many-public-methods
3 from __future__ import absolute_import
4
5 import unittest
6 try:
7 import mock
8 except ImportError:
9 import unittest.mock as mock
10
11 from ..services.ceph_service import CephService
12
13
14 class CephServiceTest(unittest.TestCase):
15 pools = [{
16 'pool_name': 'good_pool',
17 'pool': 1,
18 }, {
19 'pool_name': 'bad_pool',
20 'pool': 2,
21 'flaky': 'option_x'
22 }]
23
24 def setUp(self):
25 # Mock get_pool_list
26 self.list_patch = mock.patch('dashboard.services.ceph_service.CephService.get_pool_list')
27 self.list = self.list_patch.start()
28 self.list.return_value = self.pools
29 # Mock mgr.get
30 self.mgr_patch = mock.patch('dashboard.mgr.get')
31 self.mgr = self.mgr_patch.start()
32 self.mgr.return_value = {
33 'by_pool': {
34 '1': {'active+clean': 16},
35 '2': {'creating+incomplete': 16},
36 }
37 }
38 self.service = CephService()
39
40 def tearDown(self):
41 self.list_patch.stop()
42 self.mgr_patch.stop()
43
44 def test_get_pool_by_attribute_with_match(self):
45 self.assertEqual(self.service.get_pool_by_attribute('pool', 1), self.pools[0])
46 self.assertEqual(self.service.get_pool_by_attribute('pool_name', 'bad_pool'), self.pools[1])
47
48 def test_get_pool_by_attribute_without_a_match(self):
49 self.assertEqual(self.service.get_pool_by_attribute('pool', 3), None)
50 self.assertEqual(self.service.get_pool_by_attribute('not_there', 'sth'), None)
51
52 def test_get_pool_by_attribute_matching_a_not_always_set_attribute(self):
53 self.assertEqual(self.service.get_pool_by_attribute('flaky', 'option_x'), self.pools[1])
54
55 def test_get_pool_name_from_id_with_match(self):
56 self.assertEqual(self.service.get_pool_name_from_id(1), 'good_pool')
57
58 def test_get_pool_name_from_id_without_match(self):
59 self.assertEqual(self.service.get_pool_name_from_id(3), None)
60
61 def test_get_pool_pg_status(self):
62 self.assertEqual(self.service.get_pool_pg_status('good_pool'), {'active+clean': 16})
63
64 def test_get_pg_status_without_match(self):
65 self.assertEqual(self.service.get_pool_pg_status('no-pool'), {})