]>
git.proxmox.com Git - ceph.git/blob - 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
7 from contextlib
import contextmanager
8 from unittest
import mock
12 from ..services
.ceph_service
import CephService
15 class CephServiceTest(unittest
.TestCase
):
17 'pool_name': 'good_pool',
20 'pool_name': 'bad_pool',
27 self
.list_patch
= mock
.patch('dashboard.services.ceph_service.CephService.get_pool_list')
28 self
.list = self
.list_patch
.start()
29 self
.list.return_value
= self
.pools
31 self
.mgr_patch
= mock
.patch('dashboard.mgr.get')
32 self
.mgr
= self
.mgr_patch
.start()
33 self
.mgr
.return_value
= {
35 '1': {'active+clean': 16},
36 '2': {'creating+incomplete': 16},
39 self
.service
= CephService()
42 self
.list_patch
.stop()
45 def test_get_pool_by_attribute_with_match(self
):
46 self
.assertEqual(self
.service
.get_pool_by_attribute('pool', 1), self
.pools
[0])
47 self
.assertEqual(self
.service
.get_pool_by_attribute('pool_name', 'bad_pool'), self
.pools
[1])
49 def test_get_pool_by_attribute_without_a_match(self
):
50 self
.assertEqual(self
.service
.get_pool_by_attribute('pool', 3), None)
51 self
.assertEqual(self
.service
.get_pool_by_attribute('not_there', 'sth'), None)
53 def test_get_pool_by_attribute_matching_a_not_always_set_attribute(self
):
54 self
.assertEqual(self
.service
.get_pool_by_attribute('flaky', 'option_x'), self
.pools
[1])
56 @mock.patch('dashboard.mgr.rados.pool_reverse_lookup', return_value
='good_pool')
57 def test_get_pool_name_from_id_with_match(self
, _mock
):
58 self
.assertEqual(self
.service
.get_pool_name_from_id(1), 'good_pool')
60 @mock.patch('dashboard.mgr.rados.pool_reverse_lookup', return_value
=None)
61 def test_get_pool_name_from_id_without_match(self
, _mock
):
62 self
.assertEqual(self
.service
.get_pool_name_from_id(3), None)
64 def test_get_pool_pg_status(self
):
65 self
.assertEqual(self
.service
.get_pool_pg_status('good_pool'), {'active+clean': 16})
67 def test_get_pg_status_without_match(self
):
68 self
.assertEqual(self
.service
.get_pool_pg_status('no-pool'), {})
72 def mock_smart_data(data
):
73 devices
= [{'devid': devid
} for devid
in data
]
75 def _get_smart_data(d
):
76 return {d
['devid']: data
[d
['devid']]}
78 with mock
.patch
.object(CephService
, '_get_smart_data_by_device', side_effect
=_get_smart_data
), \
79 mock
.patch
.object(CephService
, 'get_devices_by_host', return_value
=devices
), \
80 mock
.patch
.object(CephService
, 'get_devices_by_daemon', return_value
=devices
):
84 @pytest.mark
.parametrize(
87 ('host', ('osd0',), 'from host osd0'),
88 ('daemon', ('osd', '1'), 'with ID 1')
91 def test_get_smart_data(caplog
, by
, args
, log
):
92 # pylint: disable=protected-access
94 'aaa': {'device': {'name': '/dev/sda'}},
95 'bbb': {'device': {'name': '/dev/sdb'}},
97 with
mock_smart_data(expected_data
):
98 smart_data
= getattr(CephService
, 'get_smart_data_by_{}'.format(by
))(*args
)
99 getattr(CephService
, 'get_devices_by_{}'.format(by
)).assert_called_with(*args
)
100 CephService
._get
_smart
_data
_by
_device
.assert_called()
101 assert smart_data
== expected_data
103 with caplog
.at_level(logging
.DEBUG
):
104 with
mock_smart_data([]):
105 smart_data
= getattr(CephService
, 'get_smart_data_by_{}'.format(by
))(*args
)
106 getattr(CephService
, 'get_devices_by_{}'.format(by
)).assert_called_with(*args
)
107 CephService
._get
_smart
_data
_by
_device
.assert_not_called()
108 assert smart_data
== {}
109 assert log
in caplog
.text
112 @mock.patch
.object(CephService
, 'send_command')
113 def test_get_smart_data_from_appropriate_ceph_command(send_command
):
114 # pylint: disable=protected-access
115 send_command
.side_effect
= [
116 {'nodes': [{'name': 'osd.1', 'status': 'up'}, {'name': 'mon.1', 'status': 'down'}]},
117 {'fake': {'device': {'name': '/dev/sda'}}}
119 CephService
._get
_smart
_data
_by
_device
({'devid': '1', 'daemons': ['osd.1', 'mon.1']})
120 send_command
.assert_has_calls([mock
.call('mon', 'osd tree'),
121 mock
.call('osd', 'smart', '1', devid
='1')])
123 send_command
.side_effect
= [
124 {'nodes': [{'name': 'osd.1', 'status': 'down'}, {'name': 'mon.1', 'status': 'up'}]},
125 {'fake': {'device': {'name': '/dev/sda'}}}
127 CephService
._get
_smart
_data
_by
_device
({'devid': '1', 'daemons': ['osd.1', 'mon.1']})
128 send_command
.assert_has_calls([mock
.call('mon', 'osd tree'),
129 mock
.call('mon', 'device get-health-metrics', '1', devid
='1')])