openstack/neutron
Code Issues Proposed changes

Merge "Add policy enforcer for QoS policy "tags" service plugin" into stable/2024.1

Browse Source
This commit is contained in:
Zuul 2024-11-29 12:51:53 +00:00 committed by Gerrit Code Review
commit 703ec0d487
2 changed files with 167 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

50
neutron/conf/policies/qos.py
View File

@ -19,6 +19,25 @@ from neutron.conf.policies import base
DEPRECATED_REASON = """
The QoS API now supports project scope and default roles.
"""
RESOURCE_PATH = '/qos/policies/{id}'
TAGS_PATH = RESOURCE_PATH + '/tags'
TAG_PATH = RESOURCE_PATH + '/tags/{tag_id}'
ACTION_GET_TAGS = [
{'method': 'GET', 'path': TAGS_PATH},
{'method': 'GET', 'path': TAG_PATH},
]
ACTION_PUT_TAGS = [
{'method': 'PUT', 'path': TAGS_PATH},
{'method': 'PUT', 'path': TAG_PATH},
]
ACTION_POST_TAGS = [
{'method': 'POST', 'path': TAGS_PATH},
]
ACTION_DELETE_TAGS = [
{'method': 'DELETE', 'path': TAGS_PATH},
{'method': 'DELETE', 'path': TAG_PATH},
]
rules = [
@ -50,6 +69,16 @@ rules = [
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='get_policies_tags',
check_str=neutron_policy.policy_or(
base.ADMIN_OR_PROJECT_READER,
'rule:shared_qos_policy'
),
scope_types=['project'],
description='Get QoS policy tags',
operations=ACTION_GET_TAGS
),
policy.DocumentedRuleDefault(
name='create_policy',
check_str=base.ADMIN,
@ -67,6 +96,13 @@ rules = [
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='create_policies_tags',
check_str=base.ADMIN,
scope_types=['project'],
description='Create the QoS policy tags',
operations=ACTION_POST_TAGS,
),
policy.DocumentedRuleDefault(
name='update_policy',
check_str=base.ADMIN,
@ -84,6 +120,13 @@ rules = [
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='update_policies_tags',
check_str=base.ADMIN,
scope_types=['project'],
description='Update the QoS policy tags',
operations=ACTION_PUT_TAGS,
),
policy.DocumentedRuleDefault(
name='delete_policy',
check_str=base.ADMIN,
@ -101,6 +144,13 @@ rules = [
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='delete_policies_tags',
check_str=base.ADMIN,
scope_types=['project'],
description='Delete the QoS policy tags',
operations=ACTION_DELETE_TAGS
),
policy.DocumentedRuleDefault(
name='get_rule_type',

117
neutron/tests/unit/conf/policies/test_qos.py
View File

@ -44,6 +44,14 @@ class SystemAdminQosPolicyTests(QosPolicyAPITestCase):
base_policy.InvalidScope,
policy.enforce, self.context, 'get_policy', self.alt_target)
def test_get_policies_tags(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce, self.context, 'get_policies_tags', self.target)
self.assertRaises(
base_policy.InvalidScope,
policy.enforce, self.context, 'get_policies_tags', self.alt_target)
def test_create_policy(self):
self.assertRaises(
base_policy.InvalidScope,
@ -52,6 +60,15 @@ class SystemAdminQosPolicyTests(QosPolicyAPITestCase):
base_policy.InvalidScope,
policy.enforce, self.context, 'create_policy', self.alt_target)
def test_create_policies_tags(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce, self.context, 'create_policies_tags', self.target)
self.assertRaises(
base_policy.InvalidScope,
policy.enforce, self.context, 'create_policies_tags',
self.alt_target)
def test_update_policy(self):
self.assertRaises(
base_policy.InvalidScope,
@ -60,6 +77,15 @@ class SystemAdminQosPolicyTests(QosPolicyAPITestCase):
base_policy.InvalidScope,
policy.enforce, self.context, 'update_policy', self.alt_target)
def test_update_policies_tags(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce, self.context, 'update_policies_tags', self.target)
self.assertRaises(
base_policy.InvalidScope,
policy.enforce, self.context, 'update_policies_tags',
self.alt_target)
def test_delete_policy(self):
self.assertRaises(
base_policy.InvalidScope,
@ -68,6 +94,15 @@ class SystemAdminQosPolicyTests(QosPolicyAPITestCase):
base_policy.InvalidScope,
policy.enforce, self.context, 'delete_policy', self.alt_target)
def test_delete_policies_tags(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce, self.context, 'delete_policies_tags', self.target)
self.assertRaises(
base_policy.InvalidScope,
policy.enforce, self.context, 'delete_policies_tags',
self.alt_target)
class SystemMemberQosPolicyTests(SystemAdminQosPolicyTests):
@ -95,24 +130,51 @@ class AdminQosPolicyTests(QosPolicyAPITestCase):
self.assertTrue(
policy.enforce(self.context, 'get_policy', self.alt_target))
def test_get_policies_tags(self):
self.assertTrue(
policy.enforce(self.context, 'get_policies_tags', self.target))
self.assertTrue(
policy.enforce(self.context, 'get_policies_tags', self.alt_target))
def test_create_policy(self):
self.assertTrue(
policy.enforce(self.context, 'create_policy', self.target))
self.assertTrue(
policy.enforce(self.context, 'create_policy', self.alt_target))
def test_create_policies_tags(self):
self.assertTrue(
policy.enforce(self.context, 'create_policies_tags', self.target))
self.assertTrue(
policy.enforce(self.context, 'create_policies_tags',
self.alt_target))
def test_update_policy(self):
self.assertTrue(
policy.enforce(self.context, 'update_policy', self.target))
self.assertTrue(
policy.enforce(self.context, 'update_policy', self.alt_target))
def test_update_policies_tags(self):
self.assertTrue(
policy.enforce(self.context, 'update_policies_tags', self.target))
self.assertTrue(
policy.enforce(self.context, 'update_policies_tags',
self.alt_target))
def test_delete_policy(self):
self.assertTrue(
policy.enforce(self.context, 'delete_policy', self.target))
self.assertTrue(
policy.enforce(self.context, 'delete_policy', self.alt_target))
def test_delete_policies_tags(self):
self.assertTrue(
policy.enforce(self.context, 'delete_policies_tags', self.target))
self.assertTrue(
policy.enforce(self.context, 'delete_policies_tags',
self.alt_target))
class ProjectMemberQosPolicyTests(AdminQosPolicyTests):
@ -127,6 +189,14 @@ class ProjectMemberQosPolicyTests(AdminQosPolicyTests):
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, 'get_policy', self.alt_target)
def test_get_policies_tags(self):
self.assertTrue(
policy.enforce(self.context, 'get_policies_tags', self.target))
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, 'get_policies_tags',
self.alt_target)
def test_create_policy(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
@ -135,6 +205,15 @@ class ProjectMemberQosPolicyTests(AdminQosPolicyTests):
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, 'create_policy', self.alt_target)
def test_create_policies_tags(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, 'create_policies_tags', self.target)
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, 'create_policies_tags',
self.alt_target)
def test_update_policy(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
@ -143,6 +222,15 @@ class ProjectMemberQosPolicyTests(AdminQosPolicyTests):
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, 'update_policy', self.alt_target)
def test_update_policies_tags(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, 'update_policies_tags', self.target)
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, 'update_policies_tags',
self.alt_target)
def test_delete_policy(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
@ -151,6 +239,15 @@ class ProjectMemberQosPolicyTests(AdminQosPolicyTests):
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, 'delete_policy', self.alt_target)
def test_delete_policies_tags(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, 'delete_policies_tags', self.target)
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, 'delete_policies_tags',
self.alt_target)
class ProjectReaderQosPolicyTests(ProjectMemberQosPolicyTests):
@ -170,21 +267,41 @@ class ServiceRoleQosPolicyTests(QosPolicyAPITestCase):
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, 'get_policy', self.target)
def test_get_policies_tags(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, 'get_policies_tags', self.target)
def test_create_policy(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, 'create_policy', self.target)
def test_create_policies_tags(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, 'create_policies_tags', self.target)
def test_update_policy(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, 'update_policy', self.target)
def test_update_policies_tags(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, 'update_policies_tags', self.target)
def test_delete_policy(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, 'delete_policy', self.target)
def test_delete_policies_tags(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, 'delete_policies_tags', self.target)
class QosRuleTypeAPITestCase(base.PolicyBaseTestCase):