Add policy for packet rate limit rules

This is going to add policy rules for packet rate limit
rules of https://review.opendev.org/c/openstack/neutron/+/796363

Partially-Implements: bp/packet-rate-limit
Related-Bug: #1938966
Related-Bug: #1912460
Change-Id: I20e45f73869d23f93acf4d7bc4cd378d1fa9a986
This commit is contained in:
LIU Yulong 2022-02-15 14:16:31 +08:00
parent ca9b8ec0be
commit b80f152edf
2 changed files with 214 additions and 0 deletions

View File

@ -192,6 +192,62 @@ rules = [
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='get_policy_packet_rate_limit_rule',
check_str=base.PROJECT_READER,
scope_types=['project'],
description='Get a QoS packet rate limit rule',
operations=[
{
'method': 'GET',
'path': '/qos/policies/{policy_id}/packet_rate_limit_rules',
},
{
'method': 'GET',
'path': ('/qos/policies/{policy_id}/'
'packet_rate_limit_rules/{rule_id}'),
},
]
),
policy.DocumentedRuleDefault(
name='create_policy_packet_rate_limit_rule',
check_str=base.PROJECT_ADMIN,
scope_types=['project'],
description='Create a QoS packet rate limit rule',
operations=[
{
'method': 'POST',
'path': '/qos/policies/{policy_id}/packet_rate_limit_rules',
},
]
),
policy.DocumentedRuleDefault(
name='update_policy_packet_rate_limit_rule',
check_str=base.PROJECT_ADMIN,
scope_types=['project'],
description='Update a QoS packet rate limit rule',
operations=[
{
'method': 'PUT',
'path': ('/qos/policies/{policy_id}/'
'packet_rate_limit_rules/{rule_id}'),
},
]
),
policy.DocumentedRuleDefault(
name='delete_policy_packet_rate_limit_rule',
check_str=base.PROJECT_ADMIN,
scope_types=['project'],
description='Delete a QoS packet rate limit rule',
operations=[
{
'method': 'DELETE',
'path': ('/qos/policies/{policy_id}/'
'packet_rate_limit_rules/{rule_id}'),
},
]
),
policy.DocumentedRuleDefault(
name='get_policy_dscp_marking_rule',
check_str=base.PROJECT_READER,

View File

@ -521,6 +521,164 @@ class ProjectReaderQosBandwidthLimitRuleTests(
self.context = self.project_reader_ctx
class SystemAdminQosPacketRateLimitRuleTests(QosRulesAPITestCase):
def setUp(self):
super(SystemAdminQosPacketRateLimitRuleTests, self).setUp()
self.context = self.system_admin_ctx
def test_get_policy_packet_rate_limit_rule(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'get_policy_packet_rate_limit_rule',
self.target)
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'get_policy_packet_rate_limit_rule',
self.alt_target)
def test_create_policy_packet_rate_limit_rule(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'create_policy_packet_rate_limit_rule',
self.target)
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'create_policy_packet_rate_limit_rule',
self.alt_target)
def test_update_policy_packet_rate_limit_rule(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'update_policy_packet_rate_limit_rule',
self.target)
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'update_policy_packet_rate_limit_rule',
self.alt_target)
def test_delete_policy_packet_rate_limit_rule(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'delete_policy_packet_rate_limit_rule',
self.target)
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'delete_policy_packet_rate_limit_rule',
self.alt_target)
class ProjectAdminQosPacketRateLimitRuleTests(QosRulesAPITestCase):
def setUp(self):
super(ProjectAdminQosPacketRateLimitRuleTests, self).setUp()
self.context = self.project_admin_ctx
def test_get_policy_packet_rate_limit_rule(self):
self.assertTrue(
policy.enforce(self.context,
'get_policy_packet_rate_limit_rule',
self.target))
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'get_policy_packet_rate_limit_rule',
self.alt_target)
def test_create_policy_packet_rate_limit_rule(self):
self.assertTrue(
policy.enforce(self.context,
'create_policy_packet_rate_limit_rule',
self.target))
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'create_policy_packet_rate_limit_rule',
self.alt_target)
def test_update_policy_packet_rate_limit_rule(self):
self.assertTrue(
policy.enforce(self.context,
'update_policy_packet_rate_limit_rule',
self.target))
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'update_policy_packet_rate_limit_rule',
self.alt_target)
def test_delete_policy_packet_rate_limit_rule(self):
self.assertTrue(
policy.enforce(self.context,
'delete_policy_packet_rate_limit_rule',
self.target))
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'delete_policy_packet_rate_limit_rule',
self.alt_target)
class ProjectMemberQosPacketRateLimitRuleTests(
ProjectAdminQosPacketRateLimitRuleTests):
def setUp(self):
super(ProjectMemberQosPacketRateLimitRuleTests, self).setUp()
self.context = self.project_member_ctx
def test_create_policy_packet_rate_limit_rule(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'create_policy_packet_rate_limit_rule',
self.target)
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'create_policy_packet_rate_limit_rule',
self.alt_target)
def test_update_policy_packet_rate_limit_rule(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'update_policy_packet_rate_limit_rule',
self.target)
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'update_policy_packet_rate_limit_rule',
self.alt_target)
def test_delete_policy_packet_rate_limit_rule(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'delete_policy_packet_rate_limit_rule',
self.target)
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'delete_policy_packet_rate_limit_rule',
self.alt_target)
class ProjectReaderQosPacketRateLimitRuleTests(
ProjectMemberQosPacketRateLimitRuleTests):
def setUp(self):
super(ProjectReaderQosPacketRateLimitRuleTests, self).setUp()
self.context = self.project_reader_ctx
class SystemAdminQosDSCPMarkingRuleTests(QosRulesAPITestCase):
def setUp(self):