1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
|
# -*- coding: utf-8 -*-
from __future__ import absolute_import
from .helper import DashboardTestCase, JList, JObj
class CrushRuleTest(DashboardTestCase):
AUTH_ROLES = ['pool-manager']
rule_schema = JObj(sub_elems={
'max_size': int,
'min_size': int,
'rule_id': int,
'rule_name': str,
'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(str),
'nodes': JList(JObj({}, allow_unknown=True)),
'roots': JList(int)
}))
|
