ceph/qa/tasks/mgr/dashboard/test_crush_rule.py

   1 # -*- coding: utf-8 -*-
   2
   3 from __future__ import absolute_import
   4
   5 from .helper import DashboardTestCase, JList, JObj
   6
   7
   8 class CrushRuleTest(DashboardTestCase):
   9
  10     AUTH_ROLES = ['pool-manager']
  11
  12     rule_schema = JObj(sub_elems={
  13         'max_size': int,
  14         'min_size': int,
  15         'rule_id': int,
  16         'rule_name': str,
  17         'ruleset': int,
  18         'steps': JList(JObj({}, allow_unknown=True))
  19     }, allow_unknown=True)
  20
  21     def create_and_delete_rule(self, data):
  22         name = data['name']
  23         # Creates rule
  24         self._post('/api/crush_rule', data)
  25         self.assertStatus(201)
  26         # Makes sure rule exists
  27         rule = self._get('/api/crush_rule/{}'.format(name))
  28         self.assertStatus(200)
  29         self.assertSchemaBody(self.rule_schema)
  30         self.assertEqual(rule['rule_name'], name)
  31         # Deletes rule
  32         self._delete('/api/crush_rule/{}'.format(name))
  33         self.assertStatus(204)
  34
  35     @DashboardTestCase.RunAs('test', 'test', ['rgw-manager'])
  36     def test_read_access_permissions(self):
  37         self._get('/api/crush_rule')
  38         self.assertStatus(403)
  39
  40     @DashboardTestCase.RunAs('test', 'test', ['read-only'])
  41     def test_write_access_permissions(self):
  42         self._get('/api/crush_rule')
  43         self.assertStatus(200)
  44         data = {'name': 'some_rule', 'root': 'default', 'failure_domain': 'osd'}
  45         self._post('/api/crush_rule', data)
  46         self.assertStatus(403)
  47         self._delete('/api/crush_rule/default')
  48         self.assertStatus(403)
  49
  50     @classmethod
  51     def tearDownClass(cls):
  52         super(CrushRuleTest, cls).tearDownClass()
  53         cls._ceph_cmd(['osd', 'crush', 'rule', 'rm', 'some_rule'])
  54         cls._ceph_cmd(['osd', 'crush', 'rule', 'rm', 'another_rule'])
  55
  56     def test_list(self):
  57         self._get('/api/crush_rule')
  58         self.assertStatus(200)
  59         self.assertSchemaBody(JList(self.rule_schema))
  60
  61     def test_create(self):
  62         self.create_and_delete_rule({
  63             'name': 'some_rule',
  64             'root': 'default',
  65             'failure_domain': 'osd'
  66         })
  67
  68     @DashboardTestCase.RunAs('test', 'test', ['pool-manager', 'cluster-manager'])
  69     def test_create_with_ssd(self):
  70         data = self._get('/api/osd/0')
  71         self.assertStatus(200)
  72         device_class = data['osd_metadata']['default_device_class']
  73         self.create_and_delete_rule({
  74             'name': 'another_rule',
  75             'root': 'default',
  76             'failure_domain': 'osd',
  77             'device_class': device_class
  78         })
  79
  80     def test_crush_rule_info(self):
  81         self._get('/ui-api/crush_rule/info')
  82         self.assertStatus(200)
  83         self.assertSchemaBody(JObj({
  84             'names': JList(str),
  85             'nodes': JList(JObj({}, allow_unknown=True)),
  86             'roots': JList(int)
  87         }))