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
   4 import logging
   5 import unittest
   6 from contextlib import contextmanager
   7 from unittest import mock
   8
   9 import pytest
  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     @mock.patch('dashboard.mgr.rados.pool_reverse_lookup', return_value='good_pool')
  56     def test_get_pool_name_from_id_with_match(self, _mock):
  57         self.assertEqual(self.service.get_pool_name_from_id(1), 'good_pool')
  58
  59     @mock.patch('dashboard.mgr.rados.pool_reverse_lookup', return_value=None)
  60     def test_get_pool_name_from_id_without_match(self, _mock):
  61         self.assertEqual(self.service.get_pool_name_from_id(3), None)
  62
  63     def test_get_pool_pg_status(self):
  64         self.assertEqual(self.service.get_pool_pg_status('good_pool'), {'active+clean': 16})
  65
  66     def test_get_pg_status_without_match(self):
  67         self.assertEqual(self.service.get_pool_pg_status('no-pool'), {})
  68
  69
  70 @contextmanager
  71 def mock_smart_data(data):
  72     devices = [{'devid': devid} for devid in data]
  73
  74     def _get_smart_data(d):
  75         return {d['devid']: data[d['devid']]}
  76
  77     with mock.patch.object(CephService, '_get_smart_data_by_device', side_effect=_get_smart_data), \
  78             mock.patch.object(CephService, 'get_devices_by_host', return_value=devices), \
  79             mock.patch.object(CephService, 'get_devices_by_daemon', return_value=devices):
  80         yield
  81
  82
  83 @pytest.mark.parametrize(
  84     "by,args,log",
  85     [
  86         ('host', ('osd0',), 'from host osd0'),
  87         ('daemon', ('osd', '1'), 'with ID 1')
  88     ]
  89 )
  90 def test_get_smart_data(caplog, by, args, log):
  91     # pylint: disable=protected-access
  92     expected_data = {
  93         'aaa': {'device': {'name': '/dev/sda'}},
  94         'bbb': {'device': {'name': '/dev/sdb'}},
  95     }
  96     with mock_smart_data(expected_data):
  97         smart_data = getattr(CephService, 'get_smart_data_by_{}'.format(by))(*args)
  98         getattr(CephService, 'get_devices_by_{}'.format(by)).assert_called_with(*args)
  99         CephService._get_smart_data_by_device.assert_called()
 100         assert smart_data == expected_data
 101
 102     with caplog.at_level(logging.DEBUG):
 103         with mock_smart_data([]):
 104             smart_data = getattr(CephService, 'get_smart_data_by_{}'.format(by))(*args)
 105             getattr(CephService, 'get_devices_by_{}'.format(by)).assert_called_with(*args)
 106             CephService._get_smart_data_by_device.assert_not_called()
 107             assert smart_data == {}
 108             assert log in caplog.text
 109
 110
 111 @mock.patch.object(CephService, 'send_command')
 112 def test_get_smart_data_by_device(send_command):
 113     # pylint: disable=protected-access
 114     device_id = 'Hitachi_HUA72201_JPW9K0N20D22SE'
 115     osd_tree_payload = {'nodes':
 116                         [
 117                             {'name': 'osd.1', 'status': 'down'},
 118                             {'name': 'osd.2', 'status': 'up'},
 119                             {'name': 'osd.3', 'status': 'up'}
 120                         ]}
 121     health_metrics_payload = {device_id: {'ata_apm': {'enabled': False}}}
 122     side_effect = [osd_tree_payload, health_metrics_payload]
 123
 124     # Daemons associated: 1 osd down, 2 osd up.
 125     send_command.side_effect = side_effect
 126     smart_data = CephService._get_smart_data_by_device(
 127         {'devid': device_id, 'daemons': ['osd.1', 'osd.2', 'osd.3']})
 128     assert smart_data == health_metrics_payload
 129     send_command.assert_has_calls([mock.call('mon', 'osd tree'),
 130                                    mock.call('osd', 'smart', '2', devid=device_id)])
 131
 132     # Daemons associated: 1 osd down.
 133     send_command.reset_mock()
 134     send_command.side_effect = [osd_tree_payload]
 135     smart_data = CephService._get_smart_data_by_device({'devid': device_id, 'daemons': ['osd.1']})
 136     assert smart_data == {}
 137     send_command.assert_has_calls([mock.call('mon', 'osd tree')])
 138
 139     # Daemons associated: 1 osd down, 1 mon.
 140     send_command.reset_mock()
 141     send_command.side_effect = side_effect
 142     smart_data = CephService._get_smart_data_by_device(
 143         {'devid': device_id, 'daemons': ['osd.1', 'mon.1']})
 144     assert smart_data == health_metrics_payload
 145     send_command.assert_has_calls([mock.call('mon', 'osd tree'),
 146                                    mock.call('mon', 'device query-daemon-health-metrics',
 147                                              who='mon.1')])
 148
 149     # Daemons associated: 1 mon.
 150     send_command.reset_mock()
 151     send_command.side_effect = side_effect
 152     smart_data = CephService._get_smart_data_by_device({'devid': device_id, 'daemons': ['mon.1']})
 153     assert smart_data == health_metrics_payload
 154     send_command.assert_has_calls([mock.call('mon', 'osd tree'),
 155                                    mock.call('mon', 'device query-daemon-health-metrics',
 156                                              who='mon.1')])
 157
 158     # Daemons associated: 1 other (non-osd, non-mon).
 159     send_command.reset_mock()
 160     send_command.side_effect = [osd_tree_payload]
 161     smart_data = CephService._get_smart_data_by_device({'devid': device_id, 'daemons': ['rgw.1']})
 162     assert smart_data == {}
 163     send_command.assert_has_calls([mock.call('mon', 'osd tree')])
 164
 165     # Daemons associated: no daemons.
 166     send_command.reset_mock()
 167     smart_data = CephService._get_smart_data_by_device({'devid': device_id, 'daemons': []})
 168     assert smart_data == {}
 169     send_command.assert_has_calls([])