]>
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
5 from .helper
import DashboardTestCase
, JList
, JObj
8 class CrushRuleTest(DashboardTestCase
):
10 AUTH_ROLES
= ['pool-manager']
12 rule_schema
= JObj(sub_elems
={
15 'steps': JList(JObj({}, allow_unknown
=True))
16 }, allow_unknown
=True)
18 def create_and_delete_rule(self
, data
):
21 self
._post
('/api/crush_rule', data
)
22 self
.assertStatus(201)
23 # Makes sure rule exists
24 rule
= self
._get
('/api/crush_rule/{}'.format(name
), version
='2.0')
25 self
.assertStatus(200)
26 self
.assertSchemaBody(self
.rule_schema
)
27 self
.assertEqual(rule
['rule_name'], name
)
29 self
._delete
('/api/crush_rule/{}'.format(name
))
30 self
.assertStatus(204)
32 @DashboardTestCase.RunAs('test', 'test', ['rgw-manager'])
33 def test_read_access_permissions(self
):
34 self
._get
('/api/crush_rule', version
='2.0')
35 self
.assertStatus(403)
37 @DashboardTestCase.RunAs('test', 'test', ['read-only'])
38 def test_write_access_permissions(self
):
39 self
._get
('/api/crush_rule', version
='2.0')
40 self
.assertStatus(200)
41 data
= {'name': 'some_rule', 'root': 'default', 'failure_domain': 'osd'}
42 self
._post
('/api/crush_rule', data
)
43 self
.assertStatus(403)
44 self
._delete
('/api/crush_rule/default')
45 self
.assertStatus(403)
48 def tearDownClass(cls
):
49 super(CrushRuleTest
, cls
).tearDownClass()
50 cls
._ceph
_cmd
(['osd', 'crush', 'rule', 'rm', 'some_rule'])
51 cls
._ceph
_cmd
(['osd', 'crush', 'rule', 'rm', 'another_rule'])
54 self
._get
('/api/crush_rule', version
='2.0')
55 self
.assertStatus(200)
56 self
.assertSchemaBody(JList(self
.rule_schema
))
58 def test_create(self
):
59 self
.create_and_delete_rule({
62 'failure_domain': 'osd'
65 @DashboardTestCase.RunAs('test', 'test', ['pool-manager', 'cluster-manager'])
66 def test_create_with_ssd(self
):
67 data
= self
._get
('/api/osd/0')
68 self
.assertStatus(200)
69 device_class
= data
['osd_metadata']['default_device_class']
70 self
.create_and_delete_rule({
71 'name': 'another_rule',
73 'failure_domain': 'osd',
74 'device_class': device_class
77 def test_crush_rule_info(self
):
78 self
._get
('/ui-api/crush_rule/info')
79 self
.assertStatus(200)
80 self
.assertSchemaBody(JObj({
82 'nodes': JList(JObj({}, allow_unknown
=True)),