openstack/neutron
Code Issues Proposed changes

Fix bulk port functioning with requested security groups

Browse Source 
When bulk ports are created with a security group supplied, the
resulting port(s) should only have that security group assigned. But the
resulting ports are getting both the requested security group as well as
the tenant default security group assigned.  This fixes that condition.

In order to ensure that bulk port creation results in the proper
assignment of security groups, add some testing.

Change-Id: I65aca7cd14447cc988e4bc4ab62bc7b9279e7522
Fixes-Bug: #1842666
This commit is contained in:
Nate Johnston 2019-09-03 15:56:59 -04:00
parent f5bcca87d1
commit 88c7be55c2
2 changed files with 53 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
neutron/plugins/ml2/plugin.py
View File

@ -1445,6 +1445,11 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
return bound_context.current
def _ensure_security_groups_on_port(self, context, port_dict):
port_compat = {'port': port_dict}
sgids = self._get_security_groups_on_port(context, port_compat)
self._process_port_create_security_group(context, port_dict, sgids)
@utils.transaction_guard
@db_api.retry_if_session_inactive()
def create_port_bulk(self, context, ports):
@ -1550,10 +1555,10 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
self._portsec_ext_port_create_processing(context, port_dict,
port_compat)
# sgids must be got after portsec checked with security group
sgids = self._get_security_groups_on_port(context, port_compat)
self._process_port_create_security_group(context, port_dict,
sgids)
# Ensure the default security group is assigned, unless one was
# specifically requested
if security_group_ids is None:
self._ensure_security_groups_on_port(context, port_dict)
# process port binding
binding = db.add_port_binding(context, port_dict['id'])
@ -1597,6 +1602,13 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
# Perform actions after the transaction is committed
completed_ports = []
for port in port_data:
# Ensure security groups are assigned to the port, if
# specifically requested
port_dict = port['port_dict']
if port_dict.get('security_group_ids') is not None:
with db_api.CONTEXT_WRITER.using(context):
self._ensure_security_groups_on_port(context, port_dict)
resource_extend.apply_funcs('ports',
port['port_dict'],
port['port_obj'].db_obj)

39
neutron/tests/functional/pecan_wsgi/test_controllers.py
View File

@ -492,10 +492,45 @@ class TestResourceController(TestRootController):
'tenant_id': 'tenid'}]
},
headers={'X-Project-Id': 'tenid'})
self.assertEqual(response.status_int, 201)
self.assertEqual(201, response.status_int)
json_body = jsonutils.loads(response.body)
self.assertIn('ports', json_body)
self.assertEqual(2, len(json_body['ports']))
ports = json_body['ports']
self.assertEqual(2, len(ports))
for port in ports:
self.assertEqual(1, len(port['security_groups']))
def test_bulk_create_with_sg(self):
sg_response = self.app.post_json(
'/v2.0/security-groups.json',
params={'security_group': {
"name": "functest",
"description": "Functional test"}},
headers={'X-Project-Id': 'tenid'})
self.assertEqual(201, sg_response.status_int)
sg_json_body = jsonutils.loads(sg_response.body)
self.assertIn('security_group', sg_json_body)
sg_id = sg_json_body['security_group']['id']
port_response = self.app.post_json(
'/v2.0/ports.json',
params={'ports': [{'network_id': self.port['network_id'],
'admin_state_up': True,
'security_groups': [sg_id],
'tenant_id': 'tenid'},
{'network_id': self.port['network_id'],
'admin_state_up': True,
'security_groups': [sg_id],
'tenant_id': 'tenid'}]
},
headers={'X-Project-Id': 'tenid'})
self.assertEqual(201, port_response.status_int)
json_body = jsonutils.loads(port_response.body)
self.assertIn('ports', json_body)
ports = json_body['ports']
self.assertEqual(2, len(ports))
for port in ports:
self.assertEqual(1, len(port['security_groups']))
def test_emulated_bulk_create(self):
self.plugin._FORCE_EMULATED_BULK = True