]>
git.proxmox.com Git - ceph.git/blob - ceph/qa/tasks/mgr/dashboard/test_crush_rule.py
1 # -*- coding: utf-8 -*-
3 from __future__
import absolute_import
7 from tasks
.mgr
.dashboard
.helper
import DashboardTestCase
, JObj
, JList
10 class CrushRuleTest(DashboardTestCase
):
12 AUTH_ROLES
= ['pool-manager']
14 rule_schema
= JObj(sub_elems
={
18 'rule_name': six
.string_types
,
20 'steps': JList(JObj({}, allow_unknown
=True))
21 }, allow_unknown
=True)
23 def create_and_delete_rule(self
, data
):
26 self
._post
('/api/crush_rule', data
)
27 self
.assertStatus(201)
28 # Makes sure rule exists
29 rule
= self
._get
('/api/crush_rule/{}'.format(name
))
30 self
.assertStatus(200)
31 self
.assertSchemaBody(self
.rule_schema
)
32 self
.assertEqual(rule
['rule_name'], name
)
34 self
._delete
('/api/crush_rule/{}'.format(name
))
35 self
.assertStatus(204)
37 @DashboardTestCase.RunAs('test', 'test', ['rgw-manager'])
38 def test_read_access_permissions(self
):
39 self
._get
('/api/crush_rule')
40 self
.assertStatus(403)
42 @DashboardTestCase.RunAs('test', 'test', ['read-only'])
43 def test_write_access_permissions(self
):
44 self
._get
('/api/crush_rule')
45 self
.assertStatus(200)
46 data
= {'name': 'some_rule', 'root': 'default', 'failure_domain': 'osd'}
47 self
._post
('/api/crush_rule', data
)
48 self
.assertStatus(403)
49 self
._delete
('/api/crush_rule/default')
50 self
.assertStatus(403)
53 def tearDownClass(cls
):
54 super(CrushRuleTest
, cls
).tearDownClass()
55 cls
._ceph
_cmd
(['osd', 'crush', 'rule', 'rm', 'some_rule'])
56 cls
._ceph
_cmd
(['osd', 'crush', 'rule', 'rm', 'another_rule'])
59 self
._get
('/api/crush_rule')
60 self
.assertStatus(200)
61 self
.assertSchemaBody(JList(self
.rule_schema
))
63 def test_create(self
):
64 self
.create_and_delete_rule({
67 'failure_domain': 'osd'
70 @DashboardTestCase.RunAs('test', 'test', ['pool-manager', 'cluster-manager'])
71 def test_create_with_ssd(self
):
72 data
= self
._get
('/api/osd/0')
73 self
.assertStatus(200)
74 device_class
= data
['osd_metadata']['default_device_class']
75 self
.create_and_delete_rule({
76 'name': 'another_rule',
78 'failure_domain': 'osd',
79 'device_class': device_class
82 def test_crush_rule_info(self
):
83 self
._get
('/ui-api/crush_rule/info')
84 self
.assertStatus(200)
85 self
.assertSchemaBody(JObj({
86 'names': JList(six
.string_types
),
87 'nodes': JList(JObj({}, allow_unknown
=True))