mirror of
https://github.com/ceph/ceph
synced 2025-01-24 03:53:54 +00:00
f26b4b23fd
Now a crush rule can be created and deleted through the pool form, similar to the ECP profile. The creation form is somewhat more intelligent as it checks the crush map to help create a usable rule, with only a few clicks through preselections. Fixes: https://tracker.ceph.com/issues/43260 Signed-off-by: Stephan Müller <smueller@suse.com>
90 lines
2.8 KiB
Python
90 lines
2.8 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
from __future__ import absolute_import
|
|
|
|
import six
|
|
|
|
from .helper import DashboardTestCase, JObj, JList
|
|
|
|
|
|
class CrushRuleTest(DashboardTestCase):
|
|
|
|
AUTH_ROLES = ['pool-manager']
|
|
|
|
rule_schema = JObj(sub_elems={
|
|
'max_size': int,
|
|
'min_size': int,
|
|
'rule_id': int,
|
|
'rule_name': six.string_types,
|
|
'ruleset': int,
|
|
'steps': JList(JObj({}, allow_unknown=True))
|
|
}, allow_unknown=True)
|
|
|
|
def create_and_delete_rule(self, data):
|
|
name = data['name']
|
|
# Creates rule
|
|
self._post('/api/crush_rule', data)
|
|
self.assertStatus(201)
|
|
# Makes sure rule exists
|
|
rule = self._get('/api/crush_rule/{}'.format(name))
|
|
self.assertStatus(200)
|
|
self.assertSchemaBody(self.rule_schema)
|
|
self.assertEqual(rule['rule_name'], name)
|
|
# Deletes rule
|
|
self._delete('/api/crush_rule/{}'.format(name))
|
|
self.assertStatus(204)
|
|
|
|
@DashboardTestCase.RunAs('test', 'test', ['rgw-manager'])
|
|
def test_read_access_permissions(self):
|
|
self._get('/api/crush_rule')
|
|
self.assertStatus(403)
|
|
|
|
@DashboardTestCase.RunAs('test', 'test', ['read-only'])
|
|
def test_write_access_permissions(self):
|
|
self._get('/api/crush_rule')
|
|
self.assertStatus(200)
|
|
data = {'name': 'some_rule', 'root': 'default', 'failure_domain': 'osd'}
|
|
self._post('/api/crush_rule', data)
|
|
self.assertStatus(403)
|
|
self._delete('/api/crush_rule/default')
|
|
self.assertStatus(403)
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
super(CrushRuleTest, cls).tearDownClass()
|
|
cls._ceph_cmd(['osd', 'crush', 'rule', 'rm', 'some_rule'])
|
|
cls._ceph_cmd(['osd', 'crush', 'rule', 'rm', 'another_rule'])
|
|
|
|
def test_list(self):
|
|
self._get('/api/crush_rule')
|
|
self.assertStatus(200)
|
|
self.assertSchemaBody(JList(self.rule_schema))
|
|
|
|
def test_create(self):
|
|
self.create_and_delete_rule({
|
|
'name': 'some_rule',
|
|
'root': 'default',
|
|
'failure_domain': 'osd'
|
|
})
|
|
|
|
@DashboardTestCase.RunAs('test', 'test', ['pool-manager', 'cluster-manager'])
|
|
def test_create_with_ssd(self):
|
|
data = self._get('/api/osd/0')
|
|
self.assertStatus(200)
|
|
device_class = data['osd_metadata']['default_device_class']
|
|
self.create_and_delete_rule({
|
|
'name': 'another_rule',
|
|
'root': 'default',
|
|
'failure_domain': 'osd',
|
|
'device_class': device_class
|
|
})
|
|
|
|
def test_crush_rule_info(self):
|
|
self._get('/ui-api/crush_rule/info')
|
|
self.assertStatus(200)
|
|
self.assertSchemaBody(JObj({
|
|
'names': JList(six.string_types),
|
|
'nodes': JList(JObj({}, allow_unknown=True))
|
|
}))
|
|
|