diff options
Diffstat (limited to 'qa/tasks/mgr/dashboard/test_crush_rule.py')
-rw-r--r-- | qa/tasks/mgr/dashboard/test_crush_rule.py | 84 |
1 files changed, 84 insertions, 0 deletions
diff --git a/qa/tasks/mgr/dashboard/test_crush_rule.py b/qa/tasks/mgr/dashboard/test_crush_rule.py new file mode 100644 index 000000000..aa2250b1d --- /dev/null +++ b/qa/tasks/mgr/dashboard/test_crush_rule.py @@ -0,0 +1,84 @@ +# -*- 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={ + 'rule_id': int, + 'rule_name': str, + '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), version='2.0') + 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', version='2.0') + self.assertStatus(403) + + @DashboardTestCase.RunAs('test', 'test', ['read-only']) + def test_write_access_permissions(self): + self._get('/api/crush_rule', version='2.0') + 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', version='2.0') + 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) + })) |