Merge "Get rid of DVR override of _port_has_ipv6_address"

This commit is contained in:
Jenkins 2016-12-08 15:11:06 +00:00 committed by Gerrit Code Review
commit 1031e137d4
3 changed files with 7 additions and 36 deletions

View File

@ -707,9 +707,13 @@ class L3_NAT_dbonly_mixin(l3.RouterPluginBase,
if netaddr.IPNetwork(fixed_ip['ip_address']).version == 6:
return True
def _find_ipv6_router_port_by_network(self, router, net_id):
def _find_ipv6_router_port_by_network(self, context, router, net_id):
router_dev_owner = self._get_device_owner(context, router)
for port in router.attached_ports:
p = port['port']
if p['device_owner'] != router_dev_owner:
# we don't want any special purpose internal ports
continue
if p['network_id'] == net_id and self._port_has_ipv6_address(p):
return port
@ -735,7 +739,7 @@ class L3_NAT_dbonly_mixin(l3.RouterPluginBase,
ipv6_utils.is_ipv6_pd_enabled(subnet)):
# Add new prefix to an existing ipv6 port with the same network id
# if one exists
port = self._find_ipv6_router_port_by_network(router,
port = self._find_ipv6_router_port_by_network(context, router,
subnet['network_id'])
if port:
fixed_ips = list(map(dict, port['port']['fixed_ips']))

View File

@ -431,21 +431,13 @@ class L3_NAT_with_dvr_db_mixin(l3_db.L3_NAT_db_mixin,
return router_interface_info
def _port_has_ipv6_address(self, port, csnat_port_check=True):
"""Overridden to return False if DVR SNAT port."""
if csnat_port_check:
if port['device_owner'] == const.DEVICE_OWNER_ROUTER_SNAT:
return False
return super(L3_NAT_with_dvr_db_mixin,
self)._port_has_ipv6_address(port)
def _find_v6_router_port_by_network_and_device_owner(
self, router, net_id, device_owner):
for port in router.attached_ports:
p = port['port']
if (p['network_id'] == net_id and
p['device_owner'] == device_owner and
self._port_has_ipv6_address(p, csnat_port_check=False)):
self._port_has_ipv6_address(p)):
return self._core_plugin._make_port_dict(p)
def _check_for_multiprefix_csnat_port_and_update(

View File

@ -235,31 +235,6 @@ class L3DvrTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
routers = self.mixin._build_routers_list(self.ctx, routers, gw_ports)
self.assertIsNone(routers[0].get('gw_port'))
def setup_port_has_ipv6_address(self, port):
with mock.patch.object(l3_dvr_db.l3_db.L3_NAT_db_mixin,
'_port_has_ipv6_address') as pv6:
pv6.return_value = True
result = self.mixin._port_has_ipv6_address(port)
return result, pv6
def test__port_has_ipv6_address_for_dvr_snat_port(self):
port = {
'id': 'my_port_id',
'device_owner': const.DEVICE_OWNER_ROUTER_SNAT,
}
result, pv6 = self.setup_port_has_ipv6_address(port)
self.assertFalse(result)
self.assertFalse(pv6.called)
def test__port_has_ipv6_address_for_non_snat_ports(self):
port = {
'id': 'my_port_id',
'device_owner': const.DEVICE_OWNER_DVR_INTERFACE,
}
result, pv6 = self.setup_port_has_ipv6_address(port)
self.assertTrue(result)
self.assertTrue(pv6.called)
def _helper_delete_floatingip_agent_gateway_port(self, port_host):
ports = [{
'id': 'my_port_id',