Merge "Add missing port_binding policies"

This commit is contained in:
Zuul 2023-07-18 18:25:23 +00:00 committed by Gerrit Code Review
commit 154ec11e69
3 changed files with 61 additions and 4 deletions

View File

@ -15,7 +15,7 @@ from oslo_policy import policy
from neutron.conf.policies import base
GET_BINDING_PATH = '/ports/{port_id}/bindings/'
BINDING_PATH = '/ports/{port_id}/bindings/'
ACTIVATE_BINDING_PATH = '/ports/{port_id}/bindings/{host}'
@ -28,7 +28,31 @@ rules = [
operations=[
{
'method': 'GET',
'path': GET_BINDING_PATH,
'path': BINDING_PATH,
},
],
),
policy.DocumentedRuleDefault(
name='create_port_binding',
check_str=base.ADMIN,
scope_types=['project'],
description='Create port binding on the host',
operations=[
{
'method': 'POST',
'path': BINDING_PATH,
},
],
),
policy.DocumentedRuleDefault(
name='delete_port_binding',
check_str=base.ADMIN,
scope_types=['project'],
description='Delete port binding on the host',
operations=[
{
'method': 'DELETE',
'path': BINDING_PATH,
},
],
),
@ -39,7 +63,7 @@ rules = [
description='Activate port binding on the host',
operations=[
{
'method': 'GET',
'method': 'PUT',
'path': ACTIVATE_BINDING_PATH,
},
],

View File

@ -38,6 +38,18 @@ class SystemAdminTests(PortBindingsAPITestCase):
policy.enforce,
self.context, "get_port_binding", self.target)
def test_create_port_binding(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, "create_port_binding", self.target)
def test_delete_port_binding(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, "delete_port_binding", self.target)
def test_activate_port_binding(self):
self.assertRaises(
base_policy.InvalidScope,
@ -69,6 +81,14 @@ class AdminTests(PortBindingsAPITestCase):
self.assertTrue(
policy.enforce(self.context, "get_port_binding", self.target))
def test_create_port_binding(self):
self.assertTrue(
policy.enforce(self.context, "create_port_binding", self.target))
def test_delete_port_binding(self):
self.assertTrue(
policy.enforce(self.context, "delete_port_binding", self.target))
def test_activate_port_binding(self):
self.assertTrue(
policy.enforce(self.context, "activate", self.target))
@ -86,6 +106,18 @@ class ProjectMemberTests(AdminTests):
policy.enforce,
self.context, "get_port_binding", self.target)
def test_create_port_binding(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, "create_port_binding", self.target)
def test_delete_port_binding(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, "delete_port_binding", self.target)
def test_activate_port_binding(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,

View File

@ -371,7 +371,8 @@ class ExtendedPortBindingTestCase(test_plugin.NeutronDbPluginV2TestCase):
if kwargs:
data['binding'].update(kwargs)
binding_resource = 'ports/%s/bindings' % port_id
binding_req = self.new_create_request(binding_resource, data, fmt)
binding_req = self.new_create_request(
binding_resource, data, fmt, as_admin=True)
return binding_req.get_response(self.api)
def _make_port_binding(self, fmt, port_id, host, **kwargs):