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

   1 # -*- coding: utf-8 -*-
   2 # pylint: disable=dangerous-default-value,too-many-public-methods
   3
   4 import unittest
   5 from datetime import datetime
   6 from unittest.mock import MagicMock
   7
   8 try:
   9     import mock
  10 except ImportError:
  11     import unittest.mock as mock
  12
  13 from .. import mgr
  14 from ..services.rbd import RbdConfiguration, RBDSchedulerInterval, RbdService, \
  15     get_image_spec, parse_image_spec
  16
  17
  18 class ImageNotFoundStub(Exception):
  19     def __init__(self, message, errno=None):
  20         super(ImageNotFoundStub, self).__init__(
  21             'RBD image not found (%s)' % message, errno)
  22
  23
  24 class RbdServiceTest(unittest.TestCase):
  25
  26     def setUp(self):
  27         # pylint: disable=protected-access
  28         RbdService._rbd_inst = mock.Mock()
  29         self.rbd_inst_mock = RbdService._rbd_inst
  30
  31     def test_compose_image_spec(self):
  32         self.assertEqual(get_image_spec('mypool', 'myns', 'myimage'), 'mypool/myns/myimage')
  33         self.assertEqual(get_image_spec('mypool', None, 'myimage'), 'mypool/myimage')
  34
  35     def test_parse_image_spec(self):
  36         self.assertEqual(parse_image_spec('mypool/myns/myimage'), ('mypool', 'myns', 'myimage'))
  37         self.assertEqual(parse_image_spec('mypool/myimage'), ('mypool', None, 'myimage'))
  38
  39     @mock.patch('dashboard.services.rbd.RbdConfiguration._rbd.config_list')
  40     @mock.patch('dashboard.mgr.get')
  41     @mock.patch('dashboard.services.ceph_service.CephService.get_pool_list')
  42     def test_pool_rbd_configuration_with_different_pg_states(self, get_pool_list, get, config_list):
  43         get_pool_list.return_value = [{
  44             'pool_name': 'good-pool',
  45             'pool': 1,
  46         }, {
  47             'pool_name': 'bad-pool',
  48             'pool': 2,
  49         }]
  50         get.return_value = {
  51             'by_pool': {
  52                 '1': {'active+clean': 32},
  53                 '2': {'creating+incomplete': 32},
  54             }
  55         }
  56         config_list.return_value = [1, 2, 3]
  57         config = RbdConfiguration('bad-pool')
  58         self.assertEqual(config.list(), [])
  59         config = RbdConfiguration('good-pool')
  60         self.assertEqual(config.list(), [1, 2, 3])
  61
  62     def test_rbd_image_stat_removing(self):
  63         time = datetime.utcnow()
  64         self.rbd_inst_mock.trash_get.return_value = {
  65             'id': '3c1a5ee60a88',
  66             'name': 'test_rbd',
  67             'source': 'REMOVING',
  68             'deletion_time': time,
  69             'deferment_end_time': time
  70         }
  71
  72         ioctx_mock = MagicMock()
  73
  74         # pylint: disable=protected-access
  75         rbd = RbdService._rbd_image_stat_removing(ioctx_mock, 'test_pool', '', '3c1a5ee60a88')
  76         self.assertEqual(rbd, {
  77             'id': '3c1a5ee60a88',
  78             'unique_id': 'test_pool/3c1a5ee60a88',
  79             'name': 'test_rbd',
  80             'source': 'REMOVING',
  81             'deletion_time': '{}Z'.format(time.isoformat()),
  82             'deferment_end_time': '{}Z'.format(time.isoformat()),
  83             'pool_name': 'test_pool',
  84             'namespace': ''
  85         })
  86
  87     @mock.patch('dashboard.services.rbd.rbd.ImageNotFound', new_callable=lambda: ImageNotFoundStub)
  88     def test_rbd_image_stat_filter_source_user(self, _):
  89         self.rbd_inst_mock.trash_get.return_value = {
  90             'id': '3c1a5ee60a88',
  91             'name': 'test_rbd',
  92             'source': 'USER'
  93         }
  94
  95         ioctx_mock = MagicMock()
  96         with self.assertRaises(ImageNotFoundStub) as ctx:
  97             # pylint: disable=protected-access
  98             RbdService._rbd_image_stat_removing(ioctx_mock, 'test_pool', '', '3c1a5ee60a88')
  99         self.assertIn('No image test_pool/3c1a5ee60a88 in status `REMOVING` found.',
 100                       str(ctx.exception))
 101
 102     @mock.patch('dashboard.services.rbd.rbd.ImageNotFound', new_callable=lambda: ImageNotFoundStub)
 103     @mock.patch('dashboard.services.rbd.RbdService._pool_namespaces')
 104     @mock.patch('dashboard.services.rbd.RbdService._rbd_image_stat_removing')
 105     @mock.patch('dashboard.services.rbd.RbdService._rbd_image_stat')
 106     @mock.patch('dashboard.services.rbd.RbdService._rbd_image_refs')
 107     def test_rbd_pool_list(self, rbd_image_ref_mock, rbd_image_stat_mock,
 108                            rbd_image_stat_removing_mock, pool_namespaces, _):
 109         time = datetime.utcnow()
 110
 111         ioctx_mock = MagicMock()
 112         mgr.rados = MagicMock()
 113         mgr.rados.open_ioctx.return_value = ioctx_mock
 114
 115         self.rbd_inst_mock.namespace_list.return_value = []
 116         rbd_image_ref_mock.return_value = [{'name': 'test_rbd', 'id': '3c1a5ee60a88'}]
 117         pool_namespaces.return_value = ['']
 118
 119         rbd_image_stat_mock.side_effect = mock.Mock(side_effect=ImageNotFoundStub(
 120             'RBD image not found test_pool/3c1a5ee60a88'))
 121
 122         rbd_image_stat_removing_mock.return_value = {
 123             'id': '3c1a5ee60a88',
 124             'unique_id': 'test_pool/3c1a5ee60a88',
 125             'name': 'test_rbd',
 126             'source': 'REMOVING',
 127             'deletion_time': '{}Z'.format(time.isoformat()),
 128             'deferment_end_time': '{}Z'.format(time.isoformat()),
 129             'pool_name': 'test_pool',
 130             'namespace': ''
 131         }
 132
 133         # test with limit 0, it should return a list of pools with an empty list, but
 134         rbd_pool_list = RbdService.rbd_pool_list(['test_pool'], offset=0, limit=0)
 135         self.assertEqual(rbd_pool_list, ([], 1))
 136
 137         self.rbd_inst_mock.namespace_list.return_value = []
 138
 139         rbd_pool_list = RbdService.rbd_pool_list(['test_pool'], offset=0, limit=5)
 140         self.assertEqual(rbd_pool_list, ([{
 141             'id': '3c1a5ee60a88',
 142             'unique_id': 'test_pool/3c1a5ee60a88',
 143             'name': 'test_rbd',
 144             'source': 'REMOVING',
 145             'deletion_time': '{}Z'.format(time.isoformat()),
 146             'deferment_end_time': '{}Z'.format(time.isoformat()),
 147             'pool_name': 'test_pool',
 148             'namespace': ''
 149         }], 1))
 150
 151     def test_valid_interval(self):
 152         test_cases = [
 153             ('15m', False),
 154             ('1h', False),
 155             ('5d', False),
 156             ('m', True),
 157             ('d', True),
 158             ('1s', True),
 159             ('11', True),
 160             ('1m1', True),
 161         ]
 162         for interval, error in test_cases:
 163             if error:
 164                 with self.assertRaises(ValueError):
 165                     RBDSchedulerInterval(interval)
 166             else:
 167                 self.assertEqual(str(RBDSchedulerInterval(interval)), interval)
 168
 169     def test_rbd_image_refs_cache(self):
 170         ioctx_mock = MagicMock()
 171         mgr.rados = MagicMock()
 172         mgr.rados.open_ioctx.return_value = ioctx_mock
 173         images = [{'image': str(i), 'id': str(i)} for i in range(10)]
 174         for i in range(5):
 175             self.rbd_inst_mock.list2.return_value = images[i*2:(i*2)+2]
 176             ioctx_mock = MagicMock()
 177             # pylint: disable=protected-access
 178             res = RbdService._rbd_image_refs(ioctx_mock, str(i))
 179             self.assertEqual(res, images[i*2:(i*2)+2])