Browse Source

port_forwarding: validate args before invoking db update (cont.)

This is a follow up for https://review.opendev.org/#/c/738145/

During backporting review, it became clear that unit test had a
flaw. It assumed that order of items in dictionary that make up
the exception message did not change. That is not true, based
on the python version used.

This follow up also includes a review feedback that did not make
into the original change: rename function that raises exception
to have "raise" in its name (raise_port_forwarding_update_failed).

Change-Id: I6fcd64e205e584017e6c9022f82a5497ea1cc576
Closes-Bug: #1878299
changes/93/744993/5
Flavio Fernandes 1 year ago
committed by Flavio Fernandes
parent
commit
5f9b5dd861
  1. 6
      neutron/services/portforwarding/pf_plugin.py
  2. 5
      neutron/tests/unit/services/portforwarding/test_pf_plugin.py

6
neutron/services/portforwarding/pf_plugin.py

@ -471,7 +471,7 @@ class PortForwardingPlugin(fip_pf.PortForwardingPluginBase):
msg=message)
def _check_port_forwarding_update(self, context, pf_obj):
def _check_port_forwarding_update_failed(conflict):
def _raise_port_forwarding_update_failed(conflict):
message = _("Another port forwarding entry with the same "
"attributes already exists, conflicting "
"values are %s") % conflict
@ -499,7 +499,7 @@ class PortForwardingPlugin(fip_pf.PortForwardingPluginBase):
# Ensure there are no conflicts on the outside
if (obj.floating_ip_address == pf_obj.floating_ip_address and
obj.external_port == pf_obj.external_port):
_check_port_forwarding_update_failed(
_raise_port_forwarding_update_failed(
{'floating_ip_address': str(obj.floating_ip_address),
'external_port': obj.external_port})
# Ensure there are no conflicts in the inside
@ -507,7 +507,7 @@ class PortForwardingPlugin(fip_pf.PortForwardingPluginBase):
if (obj.internal_port_id == pf_obj.internal_port_id and
obj.internal_ip_address == pf_obj.internal_ip_address and
obj.internal_port == pf_obj.internal_port):
_check_port_forwarding_update_failed(
_raise_port_forwarding_update_failed(
{'internal_port_id': obj.internal_port_id,
'internal_ip_address': str(obj.internal_ip_address),
'internal_port': obj.internal_port})

5
neutron/tests/unit/services/portforwarding/test_pf_plugin.py

@ -246,7 +246,7 @@ class TestPortForwardingPlugin(testlib_api.SqlTestCase):
mock_pf_get_objects.return_value = [pf_obj, other_pf_obj]
self.assertRaisesRegex(
lib_exc.BadRequest,
"already exist.*same_fip_addr.*same_ext_port",
"already exist.*same_fip_addr",
self.pf_plugin._check_port_forwarding_update,
self.ctxt, pf_obj)
mock_get_port.assert_called_once_with(self.ctxt, mock.ANY)
@ -269,8 +269,7 @@ class TestPortForwardingPlugin(testlib_api.SqlTestCase):
mock_pf_get_objects.return_value = [pf_obj, other_pf_obj]
self.assertRaisesRegex(
lib_exc.BadRequest,
"already exist.*same_int_port_id.*{}.*same_int_port".format(
same_internal_ip),
"already exist.*{}".format(same_internal_ip),
self.pf_plugin._check_port_forwarding_update,
self.ctxt, pf_obj)
mock_get_port.assert_called_once_with(self.ctxt, 'same_int_port_id')

Loading…
Cancel
Save