openstack
/
neutron
Code Issues Proposed changes

Cosmetic: Change ._ovn to ._nb_ovn to match ._sb_ovn 

Change-Id: I15dc080129582a0303dda499322b7aae2f4d6565
This commit is contained in:
Terry Wilson 2020-04-09 13:25:18 -05:00 committed by Slawek Kaplonski
parent 0a89986932
commit 92b081145e
6 changed files with 195 additions and 185 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
neutron/plugins/ml2/drivers/ovn/mech_driver/ovsdb/impl_idl_ovn.py
View File

@ -404,7 +404,7 @@ class OvsdbNbOvnIdl(nb_impl_idl.OvnNbApiIdlImpl, Backend):
# Iterate over each acl in a lswitch and store the acl in
# a key:value representation for e.g. acl_string. This
# key:value representation can invoke the code -
# self._ovn.add_acl(**acl_string)
# self.add_acl(**acl_string)
for acl in acls:
ext_ids = getattr(acl, 'external_ids', {})
port_id = ext_ids.get('neutron:lport')

2
neutron/plugins/ml2/drivers/ovn/mech_driver/ovsdb/ovn_db_sync.py
View File

@ -86,7 +86,7 @@ class OvnNbSynchronizer(OvnDbSynchronizer):
def stop(self):
if utils.is_ovn_l3(self.l3_plugin):
self.l3_plugin._ovn.ovsdb_connection.stop()
self.l3_plugin._nb_ovn.ovsdb_connection.stop()
self.l3_plugin._sb_ovn.ovsdb_connection.stop()
super(OvnNbSynchronizer, self).stop()

24
neutron/services/ovn_l3/plugin.py
View File

@ -114,12 +114,12 @@ class OVNL3RouterPlugin(service_base.ServicePluginBase,
@property
def _ovn_client(self):
if self._ovn_client_inst is None:
self._ovn_client_inst = ovn_client.OVNClient(self._ovn,
self._ovn_client_inst = ovn_client.OVNClient(self._nb_ovn,
self._sb_ovn)
return self._ovn_client_inst
@property
def _ovn(self):
def _nb_ovn(self):
return self._plugin_driver.nb_ovn
@property
@ -362,7 +362,7 @@ class OVNL3RouterPlugin(service_base.ServicePluginBase,
if not self._plugin_driver.list_availability_zones(context):
return []
lrp = self._ovn.get_lrouter_port(lrp_name)
lrp = self._nb_ovn.get_lrouter_port(lrp_name)
router = self.get_router(
context, lrp.external_ids[ovn_const.OVN_ROUTER_NAME_EXT_ID_KEY])
az_hints = common_utils.get_az_hints(router)
@ -373,7 +373,7 @@ class OVNL3RouterPlugin(service_base.ServicePluginBase,
port_physnet_dict = self._get_gateway_port_physnet_mapping()
# Filter out unwanted ports in case of event.
if event_from_chassis:
gw_chassis = self._ovn.get_chassis_gateways(
gw_chassis = self._nb_ovn.get_chassis_gateways(
chassis_name=event_from_chassis)
if not gw_chassis:
return
@ -394,17 +394,17 @@ class OVNL3RouterPlugin(service_base.ServicePluginBase,
chassis_with_physnets = self._sb_ovn.get_chassis_and_physnets()
# All chassis with enable_as_gw_chassis set
all_gw_chassis = self._sb_ovn.get_gateway_chassis_from_cms_options()
unhosted_gateways = self._ovn.get_unhosted_gateways(
unhosted_gateways = self._nb_ovn.get_unhosted_gateways(
port_physnet_dict, chassis_with_physnets,
all_gw_chassis)
for g_name in unhosted_gateways:
physnet = port_physnet_dict.get(g_name[len(ovn_const.LRP_PREFIX):])
# Remove any invalid gateway chassis from the list, otherwise
# we can have a situation where all existing_chassis are invalid
existing_chassis = self._ovn.get_gateway_chassis_binding(g_name)
existing_chassis = self._nb_ovn.get_gateway_chassis_binding(g_name)
primary = existing_chassis[0] if existing_chassis else None
existing_chassis = self.scheduler.filter_existing_chassis(
nb_idl=self._ovn, gw_chassis=all_gw_chassis,
nb_idl=self._nb_ovn, gw_chassis=all_gw_chassis,
physnet=physnet, chassis_physnets=chassis_with_physnets,
existing_chassis=existing_chassis)
az_hints = self._get_availability_zones_from_router_port(g_name)
@ -413,7 +413,7 @@ class OVNL3RouterPlugin(service_base.ServicePluginBase,
chassis_physnets=chassis_with_physnets,
availability_zone_hints=az_hints)
chassis = self.scheduler.select(
self._ovn, self._sb_ovn, g_name, candidates=candidates,
self._nb_ovn, self._sb_ovn, g_name, candidates=candidates,
existing_chassis=existing_chassis)
if primary and primary != chassis[0]:
if primary not in chassis:
@ -433,8 +433,8 @@ class OVNL3RouterPlugin(service_base.ServicePluginBase,
chassis[0], chassis[index] = chassis[index], chassis[0]
# NOTE(dalvarez): Let's commit the changes in separate transactions
# as we will rely on those for scheduling subsequent gateways.
with self._ovn.transaction(check_error=True) as txn:
txn.add(self._ovn.update_lrouter_port(
with self._nb_ovn.transaction(check_error=True) as txn:
txn.add(self._nb_ovn.update_lrouter_port(
g_name, gateway_chassis=chassis))
@staticmethod
@ -460,7 +460,7 @@ class OVNL3RouterPlugin(service_base.ServicePluginBase,
] if orig_gw_ip else []
add = [{'destination': '0.0.0.0/0', 'nexthop': current_gw_ip}
] if current_gw_ip else []
with l3plugin._ovn.transaction(check_error=True) as txn:
with l3plugin._nb_ovn.transaction(check_error=True) as txn:
for router_id in router_ids:
l3plugin._ovn_client.update_router_routes(
context, router_id, add, remove, txn=txn)
@ -502,7 +502,7 @@ class OVNL3RouterPlugin(service_base.ServicePluginBase,
current, if_exists=True)
def get_router_availability_zones(self, router):
lr = self._ovn.get_lrouter(router['id'])
lr = self._nb_ovn.get_lrouter(router['id'])
if not lr:
return []

10
neutron/services/portforwarding/drivers/ovn/driver.py
View File

@ -170,7 +170,7 @@ class OVNPortForwarding(object):
if not payload:
return
context = payload.context
ovn_nb = self._l3_plugin._ovn
ovn_nb = self._l3_plugin._nb_ovn
with ovn_nb.transaction(check_error=True) as ovn_txn:
if event_type == events.AFTER_CREATE:
self._handler.port_forwarding_created(ovn_txn, ovn_nb,
@ -213,7 +213,7 @@ class OVNPortForwarding(object):
pf_objs = self._get_pf_objs(context, fip_id)
LOG.debug("Maintenance port forwarding under fip %s : %s",
fip_id, pf_objs)
ovn_nb = self._l3_plugin._ovn
ovn_nb = self._l3_plugin._nb_ovn
with ovn_nb.transaction(check_error=True) as ovn_txn:
for lb_name in self._handler.lb_names(fip_id):
ovn_txn.add(ovn_nb.lb_del(lb_name, vip=None, if_exists=True))
@ -240,7 +240,7 @@ class OVNPortForwarding(object):
def maintenance_delete(self, _context, fip_id):
LOG.info("Maintenance DELETE port-forwarding entries under fip %s",
fip_id)
ovn_nb = self._l3_plugin._ovn
ovn_nb = self._l3_plugin._nb_ovn
with ovn_nb.transaction(check_error=True) as ovn_txn:
for lb_name in self._handler.lb_names(fip_id):
ovn_txn.add(ovn_nb.lb_del(lb_name, vip=None, if_exists=True))
@ -250,7 +250,7 @@ class OVNPortForwarding(object):
# NOTE: Since the db_sync callback is not granular to the level
# of the affected pfs AND the fact that pfs are all vips
# in a load balancer entry, it is cheap enough to simply rebuild.
ovn_nb = self._l3_plugin._ovn
ovn_nb = self._l3_plugin._nb_ovn
pf_objs = self._get_pf_objs(context, fip_id)
LOG.debug("Db sync port forwarding under fip %s : %s", fip_id, pf_objs)
for lb_name in self._handler.lb_names(fip_id):
@ -262,7 +262,7 @@ class OVNPortForwarding(object):
def db_sync_delete(self, context, fip_id, ovn_txn):
LOG.info("db_sync DELETE entries under fip %s", fip_id)
ovn_nb = self._l3_plugin._ovn
ovn_nb = self._l3_plugin._nb_ovn
for lb_name in self._handler.lb_names(fip_id):
ovn_txn.add(ovn_nb.lb_del(lb_name, vip=None, if_exists=True))

291
neutron/tests/unit/services/ovn_l3/test_plugin.py
View File

@ -219,7 +219,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
'ports': [self.lb_vip_lsp, self.member_lsp],
'uuid': self.fake_network['id']})
self.nb_idl = self._start_mock(
'neutron.services.ovn_l3.plugin.OVNL3RouterPlugin._ovn',
'neutron.services.ovn_l3.plugin.OVNL3RouterPlugin._nb_ovn',
new_callable=mock.PropertyMock,
return_value=fake_resources.FakeOvsdbNbOvnIdl())
self.sb_idl = self._start_mock(
@ -322,9 +322,9 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
func.return_value = self.fake_router_interface_info
self.l3_inst.add_router_interface(self.context, router_id,
interface_info)
self.l3_inst._ovn.add_lrouter_port.assert_called_once_with(
self.l3_inst._nb_ovn.add_lrouter_port.assert_called_once_with(
**self.fake_router_port_assert)
self.l3_inst._ovn.set_lrouter_port_in_lswitch_port.\
self.l3_inst._nb_ovn.set_lrouter_port_in_lswitch_port.\
assert_called_once_with(
'router-port-id', 'lrp-router-port-id', is_gw_port=False,
lsp_address=ovn_const.DEFAULT_ADDR_FOR_LSP_WITH_PEER)
@ -358,12 +358,13 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
self.l3_inst.add_router_interface(self.context, router_id,
interface_info)
called_args_dict = (
self.l3_inst._ovn.update_lrouter_port.call_args_list[0][1])
self.l3_inst._nb_ovn.update_lrouter_port.call_args_list[0][1])
self.assertEqual(1, self.l3_inst._ovn.update_lrouter_port.call_count)
self.assertEqual(1,
self.l3_inst._nb_ovn.update_lrouter_port.call_count)
self.assertCountEqual(fake_rtr_intf_networks,
called_args_dict.get('networks', []))
self.l3_inst._ovn.set_lrouter_port_in_lswitch_port.\
self.l3_inst._nb_ovn.set_lrouter_port_in_lswitch_port.\
assert_called_once_with(
'router-port-id', 'lrp-router-port-id', is_gw_port=False,
lsp_address=ovn_const.DEFAULT_ADDR_FOR_LSP_WITH_PEER)
@ -377,7 +378,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
self.l3_inst.remove_router_interface(
self.context, router_id, interface_info)
self.l3_inst._ovn.lrp_del.assert_called_once_with(
self.l3_inst._nb_ovn.lrp_del.assert_called_once_with(
'lrp-router-port-id', 'neutron-router-id', if_exists=True)
self.del_rev_p.assert_called_once_with(
self.context, 'router-port-id', ovn_const.TYPE_ROUTER_PORTS)
@ -388,7 +389,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
self.l3_inst.remove_router_interface(
self.context, router_id, interface_info)
self.l3_inst._ovn.update_lrouter_port.assert_called_once_with(
self.l3_inst._nb_ovn.update_lrouter_port.assert_called_once_with(
if_exists=False, name='lrp-router-port-id',
ipv6_ra_configs={},
networks=['10.0.0.100/24'],
@ -411,7 +412,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
self.context, router_id, interface_info)
self.get_router.assert_called_once_with(self.context, 'router-id')
self.l3_inst._ovn.lrp_del.assert_called_once_with(
self.l3_inst._nb_ovn.lrp_del.assert_called_once_with(
'lrp-router-port-id', 'neutron-router-id', if_exists=True)
self.del_rev_p.assert_called_once_with(
self.context, 'router-port-id', ovn_const.TYPE_ROUTER_PORTS)
@ -428,7 +429,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
func.return_value = new_router
self.l3_inst.update_router(self.context, router_id,
{'router': updated_data})
self.l3_inst._ovn.update_lrouter.assert_called_once_with(
self.l3_inst._nb_ovn.update_lrouter.assert_called_once_with(
'neutron-router-id', enabled=True, external_ids={
ovn_const.OVN_GW_PORT_EXT_ID_KEY: '',
ovn_const.OVN_REV_NUM_EXT_ID_KEY: '1',
@ -447,7 +448,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
func.return_value = new_router
self.l3_inst.update_router(self.context, router_id,
{'router': updated_data})
self.l3_inst._ovn.update_lrouter.assert_called_once_with(
self.l3_inst._nb_ovn.update_lrouter.assert_called_once_with(
'neutron-router-id', enabled=False,
external_ids={ovn_const.OVN_ROUTER_NAME_EXT_ID_KEY: 'test',
ovn_const.OVN_REV_NUM_EXT_ID_KEY: '1',
@ -472,8 +473,8 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
update_data = {'router': {'routes': [{'destination': '1.1.1.0/24',
'nexthop': '10.0.0.2'}]}}
self.l3_inst.update_router(self.context, router_id, update_data)
self.assertFalse(self.l3_inst._ovn.add_static_route.called)
self.assertFalse(self.l3_inst._ovn.delete_static_route.called)
self.assertFalse(self.l3_inst._nb_ovn.add_static_route.called)
self.assertFalse(self.l3_inst._nb_ovn.delete_static_route.called)
@mock.patch.object(utils, 'get_lrouter_non_gw_routes')
@mock.patch('neutron.db.extraroute_db.ExtraRoute_dbonly_mixin.'
@ -498,10 +499,10 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
func.return_value = new_router
self.l3_inst.update_router(self.context, router_id,
{'router': updated_data})
self.l3_inst._ovn.add_static_route.assert_called_once_with(
self.l3_inst._nb_ovn.add_static_route.assert_called_once_with(
'neutron-router-id',
ip_prefix='2.2.2.0/24', nexthop='10.0.0.3')
self.l3_inst._ovn.delete_static_route.assert_called_once_with(
self.l3_inst._nb_ovn.delete_static_route.assert_called_once_with(
'neutron-router-id',
ip_prefix='1.1.1.0/24', nexthop='10.0.0.2')
@ -527,15 +528,15 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
func.return_value = new_router
self.l3_inst.update_router(self.context, router_id,
{'router': updated_data})
self.l3_inst._ovn.add_static_route.assert_not_called()
self.l3_inst._ovn.delete_static_route.assert_called_once_with(
self.l3_inst._nb_ovn.add_static_route.assert_not_called()
self.l3_inst._nb_ovn.delete_static_route.assert_called_once_with(
'neutron-router-id',
ip_prefix='1.1.1.0/24', nexthop='10.0.0.2')
@mock.patch('neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb.'
'ovn_client.OVNClient._get_v4_network_of_all_router_ports')
def test_create_router_with_ext_gw(self, get_rps):
self.l3_inst._ovn.is_col_present.return_value = True
self.l3_inst._nb_ovn.is_col_present.return_value = True
router = {'router': {'name': 'router'}}
self.get_subnet.return_value = self.fake_ext_subnet
self.get_port.return_value = self.fake_ext_gw_port
@ -547,11 +548,11 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
ovn_const.OVN_REV_NUM_EXT_ID_KEY: '1',
ovn_const.OVN_GW_PORT_EXT_ID_KEY: 'gw-port-id',
ovn_const.OVN_AZ_HINTS_EXT_ID_KEY: ''}
self.l3_inst._ovn.create_lrouter.assert_called_once_with(
self.l3_inst._nb_ovn.create_lrouter.assert_called_once_with(
'neutron-router-id', external_ids=external_ids,
enabled=True, options={'always_learn_from_arp_request': 'false',
'dynamic_neigh_routers': 'true'})
self.l3_inst._ovn.add_lrouter_port.assert_called_once_with(
self.l3_inst._nb_ovn.add_lrouter_port.assert_called_once_with(
**self.fake_ext_gw_port_assert)
expected_calls = [
mock.call('neutron-router-id', ip_prefix='0.0.0.0/0',
@ -559,11 +560,11 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
external_ids={
ovn_const.OVN_ROUTER_IS_EXT_GW: 'true',
ovn_const.OVN_SUBNET_EXT_ID_KEY: 'ext-subnet-id'})]
self.l3_inst._ovn.set_lrouter_port_in_lswitch_port.\
self.l3_inst._nb_ovn.set_lrouter_port_in_lswitch_port.\
assert_called_once_with(
'gw-port-id', 'lrp-gw-port-id', is_gw_port=True,
lsp_address=ovn_const.DEFAULT_ADDR_FOR_LSP_WITH_PEER)
self.l3_inst._ovn.add_static_route.assert_has_calls(expected_calls)
self.l3_inst._nb_ovn.add_static_route.assert_has_calls(expected_calls)
bump_rev_calls = [mock.call(mock.ANY, self.fake_ext_gw_port,
ovn_const.TYPE_ROUTER_PORTS),
@ -583,7 +584,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
self.l3_inst.delete_router(self.context, 'router-id')
self.l3_inst._ovn.delete_lrouter.assert_called_once_with(
self.l3_inst._nb_ovn.delete_lrouter.assert_called_once_with(
'neutron-router-id')
@mock.patch('neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb.ovn_client'
@ -598,13 +599,13 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
self.l3_inst.add_router_interface(self.context, router_id,
interface_info)
self.l3_inst._ovn.add_lrouter_port.assert_called_once_with(
self.l3_inst._nb_ovn.add_lrouter_port.assert_called_once_with(
**self.fake_router_port_assert)
self.l3_inst._ovn.set_lrouter_port_in_lswitch_port.\
self.l3_inst._nb_ovn.set_lrouter_port_in_lswitch_port.\
assert_called_once_with(
'router-port-id', 'lrp-router-port-id', is_gw_port=False,
lsp_address=ovn_const.DEFAULT_ADDR_FOR_LSP_WITH_PEER)
self.l3_inst._ovn.add_nat_rule_in_lrouter.assert_called_once_with(
self.l3_inst._nb_ovn.add_nat_rule_in_lrouter.assert_called_once_with(
'neutron-router-id', logical_ip='10.0.0.0/24',
external_ip='192.168.1.1', type='snat')
self.bump_rev_p.assert_called_with(
@ -626,13 +627,13 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
self.l3_inst.add_router_interface(self.context, router_id,
interface_info)
self.l3_inst._ovn.add_lrouter_port.assert_called_once_with(
self.l3_inst._nb_ovn.add_lrouter_port.assert_called_once_with(
**self.fake_router_port_assert)
self.l3_inst._ovn.set_lrouter_port_in_lswitch_port.\
self.l3_inst._nb_ovn.set_lrouter_port_in_lswitch_port.\
assert_called_once_with(
'router-port-id', 'lrp-router-port-id', is_gw_port=False,
lsp_address=ovn_const.DEFAULT_ADDR_FOR_LSP_WITH_PEER)
self.l3_inst._ovn.add_nat_rule_in_lrouter.assert_not_called()
self.l3_inst._nb_ovn.add_nat_rule_in_lrouter.assert_not_called()
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_network')
@mock.patch('neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb.ovn_client'
@ -658,13 +659,13 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
fake_router_port_assert['options'] = {
'reside-on-redirect-chassis': 'true'}
self.l3_inst._ovn.add_lrouter_port.assert_called_once_with(
self.l3_inst._nb_ovn.add_lrouter_port.assert_called_once_with(
**fake_router_port_assert)
self.l3_inst._ovn.set_lrouter_port_in_lswitch_port.\
self.l3_inst._nb_ovn.set_lrouter_port_in_lswitch_port.\
assert_called_once_with(
'router-port-id', 'lrp-router-port-id', is_gw_port=False,
lsp_address=ovn_const.DEFAULT_ADDR_FOR_LSP_WITH_PEER)
self.l3_inst._ovn.add_nat_rule_in_lrouter.assert_called_once_with(
self.l3_inst._nb_ovn.add_nat_rule_in_lrouter.assert_called_once_with(
'neutron-router-id', logical_ip='10.0.0.0/24',
external_ip='192.168.1.1', type='snat')
@ -681,9 +682,10 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
port_id='router-port-id')
self.l3_inst.remove_router_interface(
self.context, router_id, interface_info)
self.l3_inst._ovn.lrp_del.assert_called_once_with(
nb_ovn = self.l3_inst._nb_ovn
nb_ovn.lrp_del.assert_called_once_with(
'lrp-router-port-id', 'neutron-router-id', if_exists=True)
self.l3_inst._ovn.delete_nat_rule_in_lrouter.assert_called_once_with(
nb_ovn.delete_nat_rule_in_lrouter.assert_called_once_with(
'neutron-router-id', logical_ip='10.0.0.0/24',
external_ip='192.168.1.1', type='snat')
self.del_rev_p.assert_called_with(
@ -694,7 +696,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
@mock.patch('neutron.db.extraroute_db.ExtraRoute_dbonly_mixin.'
'update_router')
def test_update_router_with_ext_gw(self, ur, grps):
self.l3_inst._ovn.is_col_present.return_value = True
self.l3_inst._nb_ovn.is_col_present.return_value = True
router = {'router': {'name': 'router'}}
self.get_router.return_value = self.fake_router_without_ext_gw
ur.return_value = self.fake_router_with_ext_gw
@ -705,18 +707,18 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
self.l3_inst.update_router(self.context, 'router-id', router)
self.l3_inst._ovn.add_lrouter_port.assert_called_once_with(
self.l3_inst._nb_ovn.add_lrouter_port.assert_called_once_with(
**self.fake_ext_gw_port_assert)
self.l3_inst._ovn.set_lrouter_port_in_lswitch_port.\
self.l3_inst._nb_ovn.set_lrouter_port_in_lswitch_port.\
assert_called_once_with(
'gw-port-id', 'lrp-gw-port-id', is_gw_port=True,
lsp_address=ovn_const.DEFAULT_ADDR_FOR_LSP_WITH_PEER)
self.l3_inst._ovn.add_static_route.assert_called_once_with(
self.l3_inst._nb_ovn.add_static_route.assert_called_once_with(
'neutron-router-id', ip_prefix='0.0.0.0/0',
external_ids={ovn_const.OVN_ROUTER_IS_EXT_GW: 'true',
ovn_const.OVN_SUBNET_EXT_ID_KEY: 'ext-subnet-id'},
nexthop='192.168.1.254')
self.l3_inst._ovn.add_nat_rule_in_lrouter.assert_called_once_with(
self.l3_inst._nb_ovn.add_nat_rule_in_lrouter.assert_called_once_with(
'neutron-router-id', type='snat',
logical_ip='10.0.0.0/24', external_ip='192.168.1.1')
self.bump_rev_p.assert_called_with(
@ -730,7 +732,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
'update_router')
def test_update_router_ext_gw_change_subnet(self, ur,
grps, mock_get_gw):
self.l3_inst._ovn.is_col_present.return_value = True
self.l3_inst._nb_ovn.is_col_present.return_value = True
mock_get_gw.return_value = [mock.sentinel.GwRoute]
router = {'router': {'name': 'router'}}
fake_old_ext_subnet = {'id': 'old-ext-subnet-id',
@ -755,22 +757,22 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
self.l3_inst.update_router(self.context, 'router-id', router)
# Check deleting old router gateway
self.l3_inst._ovn.delete_lrouter_ext_gw.assert_called_once_with(
self.l3_inst._nb_ovn.delete_lrouter_ext_gw.assert_called_once_with(
'neutron-router-id')
# Check adding new router gateway
self.l3_inst._ovn.add_lrouter_port.assert_called_once_with(
self.l3_inst._nb_ovn.add_lrouter_port.assert_called_once_with(
**self.fake_ext_gw_port_assert)
self.l3_inst._ovn.set_lrouter_port_in_lswitch_port.\
self.l3_inst._nb_ovn.set_lrouter_port_in_lswitch_port.\
assert_called_once_with(
'gw-port-id', 'lrp-gw-port-id', is_gw_port=True,
lsp_address=ovn_const.DEFAULT_ADDR_FOR_LSP_WITH_PEER)
self.l3_inst._ovn.add_static_route.assert_called_once_with(
self.l3_inst._nb_ovn.add_static_route.assert_called_once_with(
'neutron-router-id', ip_prefix='0.0.0.0/0',
nexthop='192.168.1.254',
external_ids={ovn_const.OVN_ROUTER_IS_EXT_GW: 'true',
ovn_const.OVN_SUBNET_EXT_ID_KEY: 'ext-subnet-id'})
self.l3_inst._ovn.add_nat_rule_in_lrouter.assert_called_once_with(
self.l3_inst._nb_ovn.add_nat_rule_in_lrouter.assert_called_once_with(
'neutron-router-id', type='snat', logical_ip='10.0.0.0/24',
external_ip='192.168.1.1')
@ -787,7 +789,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
'update_router')
def test_update_router_ext_gw_change_ip_address(self, ur,
grps, mock_get_gw):
self.l3_inst._ovn.is_col_present.return_value = True
self.l3_inst._nb_ovn.is_col_present.return_value = True
mock_get_gw.return_value = [mock.sentinel.GwRoute]
router = {'router': {'name': 'router'}}
# Old gateway info with same subnet and different ip address
@ -805,21 +807,21 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
self.l3_inst.update_router(self.context, 'router-id', router)
# Check deleting old router gateway
self.l3_inst._ovn.delete_lrouter_ext_gw.assert_called_once_with(
self.l3_inst._nb_ovn.delete_lrouter_ext_gw.assert_called_once_with(
'neutron-router-id')
# Check adding new router gateway
self.l3_inst._ovn.add_lrouter_port.assert_called_once_with(
self.l3_inst._nb_ovn.add_lrouter_port.assert_called_once_with(
**self.fake_ext_gw_port_assert)
self.l3_inst._ovn.set_lrouter_port_in_lswitch_port.\
self.l3_inst._nb_ovn.set_lrouter_port_in_lswitch_port.\
assert_called_once_with(
'gw-port-id', 'lrp-gw-port-id', is_gw_port=True,
lsp_address=ovn_const.DEFAULT_ADDR_FOR_LSP_WITH_PEER)
self.l3_inst._ovn.add_static_route.assert_called_once_with(
self.l3_inst._nb_ovn.add_static_route.assert_called_once_with(
'neutron-router-id', ip_prefix='0.0.0.0/0',
nexthop='192.168.1.254',
external_ids={ovn_const.OVN_ROUTER_IS_EXT_GW: 'true',
ovn_const.OVN_SUBNET_EXT_ID_KEY: 'ext-subnet-id'})
self.l3_inst._ovn.add_nat_rule_in_lrouter.assert_called_once_with(
self.l3_inst._nb_ovn.add_nat_rule_in_lrouter.assert_called_once_with(
'neutron-router-id', type='snat', logical_ip='10.0.0.0/24',
external_ip='192.168.1.1')
@ -829,28 +831,29 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
'update_router')
def test_update_router_ext_gw_no_change(self, ur, get_rps):
router = {'router': {'name': 'router'}}
nb_ovn = self.l3_inst._nb_ovn
self.get_router.return_value = self.fake_router_with_ext_gw
ur.return_value = self.fake_router_with_ext_gw
self.l3_inst._ovn.get_lrouter.return_value = (
nb_ovn.get_lrouter.return_value = (
fake_resources.FakeOVNRouter.from_neutron_router(
self.fake_router_with_ext_gw))
self.l3_inst.update_router(self.context, 'router-id', router)
self.l3_inst._ovn.lrp_del.assert_not_called()
self.l3_inst._ovn.delete_static_route.assert_not_called()
self.l3_inst._ovn.delete_nat_rule_in_lrouter.assert_not_called()
self.l3_inst._ovn.add_lrouter_port.assert_not_called()
self.l3_inst._ovn.set_lrouter_port_in_lswitch_port.assert_not_called()
self.l3_inst._ovn.add_static_route.assert_not_called()
self.l3_inst._ovn.add_nat_rule_in_lrouter.assert_not_called()
nb_ovn.lrp_del.assert_not_called()
nb_ovn.delete_static_route.assert_not_called()
nb_ovn.delete_nat_rule_in_lrouter.assert_not_called()
nb_ovn.add_lrouter_port.assert_not_called()
nb_ovn.set_lrouter_port_in_lswitch_port.assert_not_called()
nb_ovn.add_static_route.assert_not_called()
nb_ovn.add_nat_rule_in_lrouter.assert_not_called()
@mock.patch('neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb.ovn_client'
'.OVNClient._get_v4_network_of_all_router_ports')
@mock.patch('neutron.db.extraroute_db.ExtraRoute_dbonly_mixin.'
'update_router')
def test_update_router_with_ext_gw_and_disabled_snat(self, ur, grps):
self.l3_inst._ovn.is_col_present.return_value = True
self.l3_inst._nb_ovn.is_col_present.return_value = True
router = {'router': {'name': 'router'}}
self.get_router.return_value = self.fake_router_without_ext_gw
ur.return_value = self.fake_router_with_ext_gw
@ -863,12 +866,12 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
self.l3_inst.update_router(self.context, 'router-id', router)
# Need not check lsp and lrp here, it has been tested in other cases
self.l3_inst._ovn.add_static_route.assert_called_once_with(
self.l3_inst._nb_ovn.add_static_route.assert_called_once_with(
'neutron-router-id', ip_prefix='0.0.0.0/0',
external_ids={ovn_const.OVN_ROUTER_IS_EXT_GW: 'true',
ovn_const.OVN_SUBNET_EXT_ID_KEY: 'ext-subnet-id'},
nexthop='192.168.1.254')
self.l3_inst._ovn.add_nat_rule_in_lrouter.assert_not_called()
self.l3_inst._nb_ovn.add_nat_rule_in_lrouter.assert_not_called()
@mock.patch('neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb.ovn_client'
'.OVNClient._get_router_ports')
@ -880,7 +883,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
gr_value['external_gateway_info']['enable_snat'] = False
self.get_router.return_value = gr_value
ur.return_value = self.fake_router_with_ext_gw
self.l3_inst._ovn.get_lrouter.return_value = (
self.l3_inst._nb_ovn.get_lrouter.return_value = (
fake_resources.FakeOVNRouter.from_neutron_router(
self.fake_router_with_ext_gw))
self.get_subnet.side_effect = lambda ctx, sid: {
@ -890,10 +893,10 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
self.l3_inst.update_router(self.context, 'router-id', router)
self.l3_inst._ovn.delete_static_route.assert_not_called()
self.l3_inst._ovn.delete_nat_rule_in_lrouter.assert_not_called()
self.l3_inst._ovn.add_static_route.assert_not_called()
self.l3_inst._ovn.add_nat_rule_in_lrouter.assert_called_once_with(
self.l3_inst._nb_ovn.delete_static_route.assert_not_called()
self.l3_inst._nb_ovn.delete_nat_rule_in_lrouter.assert_not_called()
self.l3_inst._nb_ovn.add_static_route.assert_not_called()
self.l3_inst._nb_ovn.add_nat_rule_in_lrouter.assert_called_once_with(
'neutron-router-id', type='snat', logical_ip='10.0.0.0/24',
external_ip='192.168.1.1')
@ -921,15 +924,16 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
self.l3_inst.update_router(self.context, 'router-id', router)
self.l3_inst._ovn.delete_static_route.assert_not_called()
self.l3_inst._ovn.delete_nat_rule_in_lrouter.assert_called_once_with(
nb_ovn = self.l3_inst._nb_ovn
nb_ovn.delete_static_route.assert_not_called()
nb_ovn.delete_nat_rule_in_lrouter.assert_called_once_with(
'neutron-router-id', type='snat', logical_ip='10.0.0.0/24',
external_ip='192.168.1.1')
self.l3_inst._ovn.add_static_route.assert_not_called()
self.l3_inst._ovn.add_nat_rule_in_lrouter.assert_not_called()
nb_ovn.add_static_route.assert_not_called()
nb_ovn.add_nat_rule_in_lrouter.assert_not_called()
def test_create_floatingip(self):
self.l3_inst._ovn.is_col_present.return_value = True
self.l3_inst._nb_ovn.is_col_present.return_value = True
self._get_floatingip.return_value = {'floating_port_id': 'fip-port-id'}
self.l3_inst.create_floatingip(self.context, 'floatingip')
expected_ext_ids = {
@ -942,7 +946,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
ovn_const.OVN_FIP_EXT_MAC_KEY: 'aa:aa:aa:aa:aa:aa',
ovn_const.OVN_FIP_NET_ID:
self.fake_floating_ip['floating_network_id']}
self.l3_inst._ovn.add_nat_rule_in_lrouter.assert_called_once_with(
self.l3_inst._nb_ovn.add_nat_rule_in_lrouter.assert_called_once_with(
'neutron-router-id',
type='dnat_and_snat',
logical_ip='10.0.0.10',
@ -952,7 +956,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
options={'stateless': 'true'})
def test_create_floatingip_distributed(self):
self.l3_inst._ovn.is_col_present.return_value = True
self.l3_inst._nb_ovn.is_col_present.return_value = True
self.get_port.return_value = {'mac_address': '00:01:02:03:04:05',
'network_id': 'port-network-id'}
self._get_floatingip.return_value = {'floating_port_id': 'fip-port-id'}
@ -969,7 +973,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
ovn_const.OVN_FIP_EXT_MAC_KEY: '00:01:02:03:04:05',
ovn_const.OVN_FIP_NET_ID:
self.fake_floating_ip['floating_network_id']}
self.l3_inst._ovn.add_nat_rule_in_lrouter.assert_called_once_with(
self.l3_inst._nb_ovn.add_nat_rule_in_lrouter.assert_called_once_with(
'neutron-router-id', type='dnat_and_snat', logical_ip='10.0.0.10',
external_ip='192.168.0.10', external_mac='00:01:02:03:04:05',
logical_port='port_id',
@ -980,8 +984,8 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
# Check that when the port is down, the external_mac field is not
# populated. This falls back to centralized routing for ports that
# are not bound to a chassis.
self.l3_inst._ovn.is_col_present.return_value = True
self.l3_inst._ovn.lsp_get_up.return_value.execute.return_value = (
self.l3_inst._nb_ovn.is_col_present.return_value = True
self.l3_inst._nb_ovn.lsp_get_up.return_value.execute.return_value = (
False)
self.get_port.return_value = {'mac_address': '00:01:02:03:04:05'}
self._get_floatingip.return_value = {'floating_port_id': 'fip-port-id'}
@ -998,7 +1002,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
ovn_const.OVN_FIP_EXT_MAC_KEY: '00:01:02:03:04:05',
ovn_const.OVN_FIP_NET_ID:
self.fake_floating_ip['floating_network_id']}
self.l3_inst._ovn.add_nat_rule_in_lrouter.assert_called_once_with(
self.l3_inst._nb_ovn.add_nat_rule_in_lrouter.assert_called_once_with(
'neutron-router-id', type='dnat_and_snat', logical_ip='10.0.0.10',
external_ip='192.168.0.10',
logical_port='port_id',
@ -1006,9 +1010,9 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
options={'stateless': 'true'})
def test_create_floatingip_external_ip_present_in_nat_rule(self):
self.l3_inst._ovn.is_col_present.return_value = True
self.l3_inst._nb_ovn.is_col_present.return_value = True
self._get_floatingip.return_value = {'floating_port_id': 'fip-port-id'}
self.l3_inst._ovn.get_lrouter_nat_rules.return_value = [
self.l3_inst._nb_ovn.get_lrouter_nat_rules.return_value = [
{'external_ip': '192.168.0.10', 'logical_ip': '10.0.0.6',
'type': 'dnat_and_snat', 'uuid': 'uuid1'}]
self.l3_inst.create_floatingip(self.context, 'floatingip')
@ -1022,7 +1026,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
ovn_const.OVN_FIP_EXT_MAC_KEY: 'aa:aa:aa:aa:aa:aa',
ovn_const.OVN_FIP_NET_ID:
self.fake_floating_ip['floating_network_id']}
self.l3_inst._ovn.add_nat_rule_in_lrouter.assert_called_once_with(
self.l3_inst._nb_ovn.add_nat_rule_in_lrouter.assert_called_once_with(
'neutron-router-id',
type='dnat_and_snat',
logical_ip='10.0.0.10',
@ -1032,13 +1036,13 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
options={'stateless': 'true'})
def test_create_floatingip_external_ip_present_type_snat(self):
self.l3_inst._ovn.is_col_present.return_value = True
self.l3_inst._nb_ovn.is_col_present.return_value = True
self._get_floatingip.return_value = {'floating_port_id': 'fip-port-id'}
self.l3_inst._ovn.get_lrouter_nat_rules.return_value = [
self.l3_inst._nb_ovn.get_lrouter_nat_rules.return_value = [
{'external_ip': '192.168.0.10', 'logical_ip': '10.0.0.0/24',
'type': 'snat', 'uuid': 'uuid1'}]
self.l3_inst.create_floatingip(self.context, 'floatingip')
self.l3_inst._ovn.set_nat_rule_in_lrouter.assert_not_called()
self.l3_inst._nb_ovn.set_nat_rule_in_lrouter.assert_not_called()
expected_ext_ids = {
ovn_const.OVN_FIP_EXT_ID_KEY: self.fake_floating_ip['id'],
ovn_const.OVN_REV_NUM_EXT_ID_KEY: '1',
@ -1049,7 +1053,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
ovn_const.OVN_FIP_EXT_MAC_KEY: 'aa:aa:aa:aa:aa:aa',
ovn_const.OVN_FIP_NET_ID:
self.fake_floating_ip['floating_network_id']}
self.l3_inst._ovn.add_nat_rule_in_lrouter.assert_called_once_with(
self.l3_inst._nb_ovn.add_nat_rule_in_lrouter.assert_called_once_with(
'neutron-router-id',
type='dnat_and_snat',
logical_ip='10.0.0.10',
@ -1061,14 +1065,14 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
def test_create_floatingip_lsp_external_id(self):
foo_lport = fake_resources.FakeOvsdbRow.create_one_ovsdb_row()
foo_lport.uuid = 'foo-port'
self.l3_inst._ovn.get_lswitch_port.return_value = foo_lport
self.l3_inst._nb_ovn.get_lswitch_port.return_value = foo_lport
self.l3_inst.create_floatingip(self.context, 'floatingip')
calls = [mock.call(
'Logical_Switch_Port',
'foo-port',
('external_ids', {ovn_const.OVN_PORT_FIP_EXT_ID_KEY:
'192.168.0.10'}))]
self.l3_inst._ovn.db_set.assert_has_calls(calls)
self.l3_inst._nb_ovn.db_set.assert_has_calls(calls)
def test_create_floatingip_lb_member_fip(self):
config.cfg.CONF.set_override(
@ -1076,9 +1080,9 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
# Stop this mock.
self.mock_is_lb_member_fip.stop()
self.get_port.return_value = self.member_port
self.l3_inst._ovn.is_col_present.return_value = True
self.l3_inst._ovn.lookup.return_value = self.lb_network
self.l3_inst._ovn.get_lswitch_port.return_value = self.member_lsp
self.l3_inst._nb_ovn.is_col_present.return_value = True
self.l3_inst._nb_ovn.lookup.return_value = self.lb_network
self.l3_inst._nb_ovn.get_lswitch_port.return_value = self.member_lsp
fip = self.l3_inst.create_floatingip(self.context, 'floatingip')
# Validate that there is no external_mac and logical_port while
# setting the NAT entry.
@ -1089,7 +1093,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
ovn_const.OVN_ROUTER_NAME_EXT_ID_KEY: 'neutron-router-id',
ovn_const.OVN_FIP_EXT_MAC_KEY: self.member_port['mac_address'],
ovn_const.OVN_FIP_NET_ID: fip['floating_network_id']}
self.l3_inst._ovn.add_nat_rule_in_lrouter.assert_called_once_with(
self.l3_inst._nb_ovn.add_nat_rule_in_lrouter.assert_called_once_with(
'neutron-router-id',
external_ip='192.168.0.10',
logical_ip='10.0.0.10',
@ -1101,14 +1105,14 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
config.cfg.CONF.set_override(
'enable_distributed_floating_ip', True, group='ovn')
self.get_subnet.return_value = self.member_subnet
self.l3_inst._ovn.is_col_present.return_value = True
self.l3_inst._ovn.get_lswitch_port.return_value = self.lb_vip_lsp
self.l3_inst._ovn.db_find_rows.return_value.execute.side_effect = [
self.l3_inst._nb_ovn.is_col_present.return_value = True
self.l3_inst._nb_ovn.get_lswitch_port.return_value = self.lb_vip_lsp
self.l3_inst._nb_ovn.db_find_rows.return_value.execute.side_effect = [
[self.ovn_lb],
[self.lb_network],
[self.fake_ovn_nat_rule],
]
self.l3_inst._ovn.lookup.return_value = self.lb_network
self.l3_inst._nb_ovn.lookup.return_value = self.lb_network
fip = self.l3_inst.create_floatingip(self.context, 'floatingip')
expected_ext_ids = {
ovn_const.OVN_FIP_EXT_ID_KEY: fip['id'],
@ -1118,7 +1122,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
ovn_const.OVN_FIP_EXT_MAC_KEY: self.member_port['mac_address'],
ovn_const.OVN_FIP_NET_ID: fip['floating_network_id']}
self.l3_inst._ovn.add_nat_rule_in_lrouter.assert_called_once_with(
self.l3_inst._nb_ovn.add_nat_rule_in_lrouter.assert_called_once_with(
'neutron-router-id',
external_ip='192.168.0.10',
external_mac='aa:aa:aa:aa:aa:aa',
@ -1127,20 +1131,21 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
type='dnat_and_snat',
external_ids=expected_ext_ids,
options={'stateless': 'true'})
self.l3_inst._ovn.db_find_rows.assert_called_with(
self.l3_inst._nb_ovn.db_find_rows.assert_called_with(
'NAT', ('external_ids', '=', {ovn_const.OVN_FIP_PORT_EXT_ID_KEY:
self.member_lsp.name}))
# Validate that it clears external_mac/logical_port for member NAT.
self.l3_inst._ovn.db_clear.assert_has_calls([
self.l3_inst._nb_ovn.db_clear.assert_has_calls([
mock.call('NAT', self.fake_ovn_nat_rule.uuid, 'external_mac'),
mock.call('NAT', self.fake_ovn_nat_rule.uuid, 'logical_port')])
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin.delete_floatingip')
def test_delete_floatingip(self, df):
self.l3_inst._ovn.get_floatingip.return_value = (
nb_ovn = self.l3_inst._nb_ovn
nb_ovn.get_floatingip.return_value = (
self.fake_ovn_nat_rule)
self.l3_inst.delete_floatingip(self.context, 'floatingip-id')
self.l3_inst._ovn.delete_nat_rule_in_lrouter.assert_called_once_with(
nb_ovn.delete_nat_rule_in_lrouter.assert_called_once_with(
'neutron-router-id',
type='dnat_and_snat',
logical_ip='10.0.0.10',
@ -1151,29 +1156,30 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
config.cfg.CONF.set_override(
'enable_distributed_floating_ip', True, group='ovn')
self.get_subnet.return_value = self.member_subnet
self.l3_inst._ovn.get_floatingip.return_value = (
nb_ovn = self.l3_inst._nb_ovn
nb_ovn.get_floatingip.return_value = (
self.fake_ovn_nat_rule)
self.l3_inst._ovn.get_lswitch_port.return_value = self.lb_vip_lsp
self.l3_inst._ovn.db_find_rows.return_value.execute.side_effect = [
nb_ovn.get_lswitch_port.return_value = self.lb_vip_lsp
nb_ovn.db_find_rows.return_value.execute.side_effect = [
[self.ovn_lb],
[self.lb_network],
[self.fake_ovn_nat_rule],
]
self.l3_inst._ovn.lookup.return_value = self.lb_network
nb_ovn.lookup.return_value = self.lb_network
self.l3_inst.delete_floatingip(self.context, 'floatingip-id')
self.l3_inst._ovn.delete_nat_rule_in_lrouter.assert_called_once_with(
nb_ovn.delete_nat_rule_in_lrouter.assert_called_once_with(
'neutron-router-id',
type='dnat_and_snat',
logical_ip='10.0.0.10',
external_ip='192.168.0.10')
self.l3_inst._ovn.db_find_rows.assert_called_with(
nb_ovn.db_find_rows.assert_called_with(
'NAT', ('external_ids', '=',
{ovn_const.OVN_FIP_PORT_EXT_ID_KEY: self.member_lsp.name}))
self.l3_inst._plugin.get_port.assert_called_once_with(
mock.ANY, self.member_lsp.name)
# Validate that it adds external_mac/logical_port back.
self.l3_inst._ovn.db_set.assert_has_calls([
nb_ovn.db_set.assert_has_calls([
mock.call('NAT', self.fake_ovn_nat_rule.uuid,
('logical_port', self.member_lsp.name)),
mock.call('NAT', self.fake_ovn_nat_rule.uuid,
@ -1181,38 +1187,39 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin.delete_floatingip')
def test_delete_floatingip_lsp_external_id(self, df):
self.l3_inst._ovn.get_floatingip.return_value = (
self.l3_inst._nb_ovn.get_floatingip.return_value = (
self.fake_ovn_nat_rule)
foo_lport = fake_resources.FakeOvsdbRow.create_one_ovsdb_row()
foo_lport.uuid = 'foo-port'
foo_lport.external_ids = {
ovn_const.OVN_PORT_FIP_EXT_ID_KEY: 'foo-port'}
self.l3_inst._ovn.get_lswitch_port.return_value = foo_lport
self.l3_inst._nb_ovn.get_lswitch_port.return_value = foo_lport
self.l3_inst.delete_floatingip(self.context, 'floatingip-id')
calls = [mock.call(
'Logical_Switch_Port', 'foo-port',
'external_ids', ovn_const.OVN_PORT_FIP_EXT_ID_KEY)]
self.l3_inst._ovn.db_remove.assert_has_calls(calls)
self.l3_inst._nb_ovn.db_remove.assert_has_calls(calls)
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin.delete_floatingip')
def test_delete_floatingip_no_lsp_external_id(self, df):
self.l3_inst._ovn.get_floatingip.return_value = (
self.l3_inst._nb_ovn.get_floatingip.return_value = (
self.fake_ovn_nat_rule)
self.l3_inst._ovn.get_lswitch_port.return_value = None
self.l3_inst._nb_ovn.get_lswitch_port.return_value = None
self.l3_inst.delete_floatingip(self.context, 'floatingip-id')
self.l3_inst._ovn.db_remove.assert_not_called()
self.l3_inst._nb_ovn.db_remove.assert_not_called()
@mock.patch('neutron.db.extraroute_db.ExtraRoute_dbonly_mixin.'
'update_floatingip')
def test_update_floatingip(self, uf):
self.l3_inst._ovn.is_col_present.return_value = True
nb_ovn = self.l3_inst._nb_ovn
nb_ovn.is_col_present.return_value = True
uf.return_value = self.fake_floating_ip_new
self.l3_inst._ovn.get_floatingip.return_value = (
nb_ovn.get_floatingip.return_value = (
self.fake_ovn_nat_rule)
self.l3_inst.update_floatingip(self.context, 'id', 'floatingip')
self.l3_inst._ovn.delete_nat_rule_in_lrouter.assert_called_once_with(
nb_ovn.delete_nat_rule_in_lrouter.assert_called_once_with(
'neutron-router-id',
type='dnat_and_snat',
logical_ip='10.0.0.10',
@ -1227,7 +1234,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
ovn_const.OVN_FIP_EXT_MAC_KEY: 'aa:aa:aa:aa:aa:aa',
ovn_const.OVN_FIP_NET_ID:
self.fake_floating_ip['floating_network_id']}
self.l3_inst._ovn.add_nat_rule_in_lrouter.assert_called_once_with(
self.l3_inst._nb_ovn.add_nat_rule_in_lrouter.assert_called_once_with(
'neutron-new-router-id',
type='dnat_and_snat',
logical_ip='10.10.10.10',
@ -1239,11 +1246,11 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
@mock.patch('neutron.db.extraroute_db.ExtraRoute_dbonly_mixin.'
'update_floatingip')
def test_update_floatingip_associate(self, uf):
self.l3_inst._ovn.is_col_present.return_value = True
self.l3_inst._nb_ovn.is_col_present.return_value = True
self.fake_floating_ip.update({'fixed_port_id': None})
uf.return_value = self.fake_floating_ip_new
self.l3_inst.update_floatingip(self.context, 'id', 'floatingip')
self.l3_inst._ovn.delete_nat_rule_in_lrouter.assert_not_called()
self.l3_inst._nb_ovn.delete_nat_rule_in_lrouter.assert_not_called()
expected_ext_ids = {
ovn_const.OVN_FIP_EXT_ID_KEY: self.fake_floating_ip_new['id'],
ovn_const.OVN_REV_NUM_EXT_ID_KEY: '1',
@ -1254,7 +1261,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
ovn_const.OVN_FIP_EXT_MAC_KEY: 'aa:aa:aa:aa:aa:aa',
ovn_const.OVN_FIP_NET_ID:
self.fake_floating_ip['floating_network_id']}
self.l3_inst._ovn.add_nat_rule_in_lrouter.assert_called_once_with(
self.l3_inst._nb_ovn.add_nat_rule_in_lrouter.assert_called_once_with(
'neutron-new-router-id',
type='dnat_and_snat',
logical_ip='10.10.10.10',
@ -1267,7 +1274,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
@mock.patch('neutron.db.extraroute_db.ExtraRoute_dbonly_mixin.'
'update_floatingip')
def test_update_floatingip_associate_distributed(self, uf, gn):
self.l3_inst._ovn.is_col_present.return_value = True
self.l3_inst._nb_ovn.is_col_present.return_value = True
self.fake_floating_ip.update({'fixed_port_id': None})
self.get_port.return_value = {'mac_address': '00:01:02:03:04:05',
'network_id': 'port-network-id'}
@ -1280,7 +1287,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
config.cfg.CONF.set_override(
'enable_distributed_floating_ip', True, group='ovn')
self.l3_inst.update_floatingip(self.context, 'id', 'floatingip')
self.l3_inst._ovn.delete_nat_rule_in_lrouter.assert_not_called()
self.l3_inst._nb_ovn.delete_nat_rule_in_lrouter.assert_not_called()
expected_ext_ids = {
ovn_const.OVN_FIP_EXT_ID_KEY: self.fake_floating_ip_new['id'],
ovn_const.OVN_REV_NUM_EXT_ID_KEY: '1',
@ -1291,7 +1298,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
ovn_const.OVN_FIP_EXT_MAC_KEY: '00:01:02:03:04:05',
ovn_const.OVN_FIP_NET_ID:
self.fake_floating_ip['floating_network_id']}
self.l3_inst._ovn.add_nat_rule_in_lrouter.assert_called_once_with(
self.l3_inst._nb_ovn.add_nat_rule_in_lrouter.assert_called_once_with(
'neutron-new-router-id', type='dnat_and_snat',
logical_ip='10.10.10.10', external_ip='192.168.0.10',
external_mac='00:01:02:03:04:05', logical_port='new-port_id',
@ -1300,14 +1307,15 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
@mock.patch('neutron.db.extraroute_db.ExtraRoute_dbonly_mixin.'
'update_floatingip')
def test_update_floatingip_association_empty_update(self, uf):
self.l3_inst._ovn.is_col_present.return_value = True
self.l3_inst._ovn.get_floatingip.return_value = (
nb_ovn = self.l3_inst._nb_ovn
nb_ovn.is_col_present.return_value = True
nb_ovn.get_floatingip.return_value = (
self.fake_ovn_nat_rule)
self.fake_floating_ip.update({'fixed_port_id': 'foo'})
self.fake_floating_ip_new.update({'port_id': 'foo'})
uf.return_value = self.fake_floating_ip_new
self.l3_inst.update_floatingip(self.context, 'id', 'floatingip')
self.l3_inst._ovn.delete_nat_rule_in_lrouter.assert_called_once_with(
nb_ovn.delete_nat_rule_in_lrouter.assert_called_once_with(
'neutron-router-id',
type='dnat_and_snat',
logical_ip='10.0.0.10',
@ -1322,7 +1330,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
ovn_const.OVN_FIP_EXT_MAC_KEY: 'aa:aa:aa:aa:aa:aa',
ovn_const.OVN_FIP_NET_ID:
self.fake_floating_ip['floating_network_id']}
self.l3_inst._ovn.add_nat_rule_in_lrouter.assert_called_once_with(
self.l3_inst._nb_ovn.add_nat_rule_in_lrouter.assert_called_once_with(
'neutron-new-router-id',
type='dnat_and_snat',
logical_ip='10.10.10.10',
@ -1335,15 +1343,16 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
'update_floatingip')
def test_update_floatingip_reassociate_to_same_port_diff_fixed_ip(
self, uf):
self.l3_inst._ovn.is_col_present.return_value = True
self.l3_inst._ovn.get_floatingip.return_value = (
nb_ovn = self.l3_inst._nb_ovn
nb_ovn.is_col_present.return_value = True
nb_ovn.get_floatingip.return_value = (
self.fake_ovn_nat_rule)
self.fake_floating_ip_new.update({'port_id': 'port_id',
'fixed_port_id': 'port_id'})
uf.return_value = self.fake_floating_ip_new
self.l3_inst.update_floatingip(self.context, 'id', 'floatingip')
self.l3_inst._ovn.delete_nat_rule_in_lrouter.assert_called_once_with(
nb_ovn.delete_nat_rule_in_lrouter.assert_called_once_with(
'neutron-router-id',
type='dnat_and_snat',
logical_ip='10.0.0.10',
@ -1358,7 +1367,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
ovn_const.OVN_FIP_EXT_MAC_KEY: 'aa:aa:aa:aa:aa:aa',
ovn_const.OVN_FIP_NET_ID:
self.fake_floating_ip['floating_network_id']}
self.l3_inst._ovn.add_nat_rule_in_lrouter.assert_called_once_with(
self.l3_inst._nb_ovn.add_nat_rule_in_lrouter.assert_called_once_with(
'neutron-new-router-id',
type='dnat_and_snat',
logical_ip='10.10.10.10',
@ -1393,8 +1402,8 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
for fip in gfs.return_value]
self.assertEqual(
len(delete_nat_calls),
self.l3_inst._ovn.delete_nat_rule_in_lrouter.call_count)
self.l3_inst._ovn.delete_nat_rule_in_lrouter.assert_has_calls(
self.l3_inst._nb_ovn.delete_nat_rule_in_lrouter.call_count)
self.l3_inst._nb_ovn.delete_nat_rule_in_lrouter.assert_has_calls(
delete_nat_calls, any_order=True)
@mock.patch('neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb.'
@ -1597,13 +1606,13 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
ovn_const.OVN_ROUTER_PORT_GW_MTU_OPTION:
str(prov_net['mtu'])}
self.l3_inst._ovn.add_lrouter_port.assert_called_once_with(
self.l3_inst._nb_ovn.add_lrouter_port.assert_called_once_with(
**fake_router_port_assert)
self.l3_inst._ovn.set_lrouter_port_in_lswitch_port.\
self.l3_inst._nb_ovn.set_lrouter_port_in_lswitch_port.\
assert_called_once_with(
'router-port-id', 'lrp-router-port-id', is_gw_port=True,
lsp_address=ovn_const.DEFAULT_ADDR_FOR_LSP_WITH_PEER)
self.l3_inst._ovn.add_nat_rule_in_lrouter.assert_called_once_with(
self.l3_inst._nb_ovn.add_nat_rule_in_lrouter.assert_called_once_with(
'neutron-router-id', logical_ip='10.0.0.0/24',
external_ip='192.168.1.1', type='snat')
@ -1615,7 +1624,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
lr = fake_resources.FakeOvsdbRow.create_one_ovsdb_row(
attrs={'id': 'fake-router', 'external_ids': {
ovn_const.OVN_AZ_HINTS_EXT_ID_KEY: azs}})
self.l3_inst._ovn.get_lrouter.return_value = lr
self.l3_inst._nb_ovn.get_lrouter.return_value = lr
azs_list = self.l3_inst.get_router_availability_zones(lr)
self.assertEqual(sorted(expected), sorted(azs_list))
@ -1690,7 +1699,7 @@ class OVNL3ExtrarouteTests(test_l3_gw.ExtGwModeIntTestCase,
test_extensions.setup_extensions_middleware(l3_gw_mgr)
self.l3_inst = directory.get_plugin(plugin_constants.L3)
self._start_mock(
'neutron.services.ovn_l3.plugin.OVNL3RouterPlugin._ovn',
'neutron.services.ovn_l3.plugin.OVNL3RouterPlugin._nb_ovn',
new_callable=mock.PropertyMock,
return_value=fake_resources.FakeOvsdbNbOvnIdl())
self._start_mock(
@ -1744,9 +1753,9 @@ class OVNL3ExtrarouteTests(test_l3_gw.ExtGwModeIntTestCase,
def test_update_subnet_gateway_for_external_net(self):
super(OVNL3ExtrarouteTests, self). \
test_update_subnet_gateway_for_external_net()
self.l3_inst._ovn.add_static_route.assert_called_once_with(
self.l3_inst._nb_ovn.add_static_route.assert_called_once_with(
'neutron-fake_device', ip_prefix='0.0.0.0/0', nexthop='120.0.0.2')
self.l3_inst._ovn.delete_static_route.assert_called_once_with(
self.l3_inst._nb_ovn.delete_static_route.assert_called_once_with(
'neutron-fake_device', ip_prefix='0.0.0.0/0', nexthop='120.0.0.1')
def test_router_update_gateway_upon_subnet_create_max_ips_ipv6(self):
@ -1760,5 +1769,5 @@ class OVNL3ExtrarouteTests(test_l3_gw.ExtGwModeIntTestCase,
external_ids=expected_ext_ids),
mock.call(mock.ANY, ip_prefix='::/0', nexthop='2001:db8::',
external_ids=expected_ext_ids)]
self.l3_inst._ovn.add_static_route.assert_has_calls(
self.l3_inst._nb_ovn.add_static_route.assert_has_calls(
add_static_route_calls, any_order=True)

51
neutron/tests/unit/services/portforwarding/drivers/ovn/test_driver.py
View File

@ -34,8 +34,8 @@ class TestOVNPortForwardingBase(base.BaseTestCase):
super(TestOVNPortForwardingBase, self).setUp()
self.context = mock.Mock()
self.l3_plugin = mock.Mock()
self.l3_plugin._ovn = fake_resources.FakeOvsdbNbOvnIdl()
self.txn = self.l3_plugin._ovn.transaction
self.l3_plugin._nb_ovn = fake_resources.FakeOvsdbNbOvnIdl()
self.txn = self.l3_plugin._nb_ovn.transaction
def _fake_pf_obj(self, **kwargs):
pf_obj_defaults_dict = {
@ -108,7 +108,7 @@ class TestOVNPortForwardingHandler(TestOVNPortForwardingBase):
_get_lb_attributes(fake_pf_obj))
exp_protocol = self.handler._get_lb_protocol(fake_pf_obj)
self.handler.port_forwarding_created(
self.txn, self.l3_plugin._ovn, fake_pf_obj)
self.txn, self.l3_plugin._nb_ovn, fake_pf_obj)
info_args, _info_kwargs = m_info.call_args_list[0]
self.assertIn('CREATE for port-forwarding', info_args[0])
self.assertEqual(2, len(self.txn.add.call_args_list))
@ -117,10 +117,10 @@ class TestOVNPortForwardingHandler(TestOVNPortForwardingBase):
ovn_const.OVN_FIP_EXT_ID_KEY: fake_pf_obj.floatingip_id,
ovn_const.OVN_ROUTER_NAME_EXT_ID_KEY: exp_rtr_name,
}
self.l3_plugin._ovn.lb_add.assert_called_once_with(
self.l3_plugin._nb_ovn.lb_add.assert_called_once_with(
exp_lb_name, exp_vip, exp_internal_ips, exp_protocol,
may_exist=True, external_ids=exp_external_ids)
self.l3_plugin._ovn.lr_lb_add.assert_called_once_with(
self.l3_plugin._nb_ovn.lr_lb_add.assert_called_once_with(
exp_rtr_name, exp_lb_name, may_exist=True)
@mock.patch.object(port_forwarding.LOG, 'info')
@ -132,12 +132,12 @@ class TestOVNPortForwardingHandler(TestOVNPortForwardingBase):
fake_pf_obj = self._fake_pf_obj(protocol='udp')
fake_orig_pf_obj = self._fake_pf_obj(protocol='tcp')
self.handler.port_forwarding_updated(
self.txn, self.l3_plugin._ovn, fake_pf_obj, fake_orig_pf_obj)
self.txn, self.l3_plugin._nb_ovn, fake_pf_obj, fake_orig_pf_obj)
info_args, _info_kwargs = m_info.call_args_list[0]
self.assertIn('UPDATE for port-forwarding', info_args[0])
m_deleted.assert_called_once_with(self.txn, self.l3_plugin._ovn,
m_deleted.assert_called_once_with(self.txn, self.l3_plugin._nb_ovn,
fake_orig_pf_obj)
m_created.assert_called_once_with(self.txn, self.l3_plugin._ovn,
m_created.assert_called_once_with(self.txn, self.l3_plugin._nb_ovn,
fake_pf_obj)
@mock.patch.object(port_forwarding.LOG, 'info')
@ -146,11 +146,11 @@ class TestOVNPortForwardingHandler(TestOVNPortForwardingBase):
exp_lb_name, exp_vip, _, _ = self.handler._get_lb_attributes(
fake_pf_obj)
self.handler.port_forwarding_deleted(
self.txn, self.l3_plugin._ovn, fake_pf_obj)
self.txn, self.l3_plugin._nb_ovn, fake_pf_obj)
info_args, _info_kwargs = m_info.call_args_list[0]
self.assertIn('DELETE for port-forwarding', info_args[0])
self.assertEqual(1, len(self.txn.add.call_args_list))
self.l3_plugin._ovn.lb_del.assert_called_once_with(
self.l3_plugin._nb_ovn.lb_del.assert_called_once_with(
exp_lb_name, exp_vip, if_exists=mock.ANY)
@ -248,7 +248,8 @@ class TestOVNPortForwarding(TestOVNPortForwardingBase):
mock_get_fip_objs.assert_called_once_with(self.context, payload)
if fip_objs:
calls = [
mock.call(mock.ANY, self.l3_plugin._ovn, fip_id, fip_obj)
mock.call(mock.ANY, self.l3_plugin._nb_ovn, fip_id,
fip_obj)
for fip_id, fip_obj in fip_objs.items()]
self.fake_check_rev.assert_has_calls(calls)
self.fake_db_rev.assert_called_once_with(
@ -265,7 +266,7 @@ class TestOVNPortForwarding(TestOVNPortForwardingBase):
self._handle_notification_common(events.AFTER_CREATE,
fake_payload_entry)
self.handler.port_forwarding_created.assert_called_once_with(
mock.ANY, self.l3_plugin._ovn, fake_payload_entry.latest_state)
mock.ANY, self.l3_plugin._nb_ovn, fake_payload_entry.latest_state)
def test_handle_notification_create(self):
fip_objs = {1: {'description': 'one'},
@ -275,7 +276,7 @@ class TestOVNPortForwarding(TestOVNPortForwardingBase):
self._handle_notification_common(events.AFTER_CREATE,
fake_payload,
fip_objs)
calls = [mock.call(mock.ANY, self.l3_plugin._ovn,
calls = [mock.call(mock.ANY, self.l3_plugin._nb_ovn,
fake_payload.latest_state)]
self.handler.port_forwarding_created.assert_has_calls(calls)
update_calls = [mock.call(
@ -289,7 +290,7 @@ class TestOVNPortForwarding(TestOVNPortForwardingBase):
fake_payload = self._fake_pf_payload_entry(100, 100)
self._handle_notification_common(events.AFTER_UPDATE, fake_payload,
fip_objs)
calls = [mock.call(mock.ANY, self.l3_plugin._ovn,
calls = [mock.call(mock.ANY, self.l3_plugin._nb_ovn,
fake_payload.latest_state,
fake_payload.states[0])]
self.handler.port_forwarding_updated.assert_has_calls(calls)
@ -297,7 +298,7 @@ class TestOVNPortForwarding(TestOVNPortForwardingBase):
fake_payload = self._fake_pf_payload_entry(101, 101)
self._handle_notification_common(events.AFTER_UPDATE, fake_payload,
fip_objs)
calls = [mock.call(mock.ANY, self.l3_plugin._ovn,
calls = [mock.call(mock.ANY, self.l3_plugin._nb_ovn,
fake_payload.latest_state,
fake_payload.states[0])]
self.handler.port_forwarding_updated.assert_has_calls(calls)
@ -314,7 +315,7 @@ class TestOVNPortForwarding(TestOVNPortForwardingBase):
self._handle_notification_common(events.AFTER_DELETE,
fake_payload, fip_objs)
calls = [mock.call(
mock.ANY, self.l3_plugin._ovn, fake_payload.states[0])]
mock.ANY, self.l3_plugin._nb_ovn, fake_payload.states[0])]
self.handler.port_forwarding_deleted.assert_has_calls(calls)
update_calls = [mock.call(
self.context, fake_payload.states[0].floatingip_id,
@ -333,12 +334,12 @@ class TestOVNPortForwarding(TestOVNPortForwardingBase):
with mock.patch.object(self._ovn_pf, '_get_pf_objs',
return_value=pf_objs) as mock_get_pf_objs:
self._ovn_pf._maintenance_create_update(self.context, fip_id)
self.l3_plugin._ovn.transaction.assert_called_once_with(
self.l3_plugin._nb_ovn.transaction.assert_called_once_with(
check_error=True)
calls = [mock.call(lb_name, vip=None, if_exists=True)
for lb_name in fake_lb_names]
self.l3_plugin._ovn.lb_del.assert_has_calls(calls)
calls = [mock.call(mock.ANY, self.l3_plugin._ovn, pf_obj)
self.l3_plugin._nb_ovn.lb_del.assert_has_calls(calls)
calls = [mock.call(mock.ANY, self.l3_plugin._nb_ovn, pf_obj)
for pf_obj in pf_objs]
self.handler.port_forwarding_created.assert_has_calls(calls)
mock_get_pf_objs.assert_called_once_with(self.context, fip_id)
@ -358,11 +359,11 @@ class TestOVNPortForwarding(TestOVNPortForwardingBase):
with mock.patch.object(self._ovn_pf, '_get_pf_objs',
return_value=pf_objs) as mock_get_pf_objs:
self._ovn_pf.maintenance_delete(self.context, fip_id)
self.l3_plugin._ovn.transaction.assert_called_once_with(
self.l3_plugin._nb_ovn.transaction.assert_called_once_with(
check_error=True)
calls = [mock.call(lb_name, vip=None, if_exists=True)
for lb_name in fake_lb_names]
self.l3_plugin._ovn.lb_del.assert_has_calls(calls)
self.l3_plugin._nb_ovn.lb_del.assert_has_calls(calls)
self.handler.port_forwarding_created.assert_not_called()
mock_get_pf_objs.assert_not_called()
self.l3_plugin.get_floatingip.assert_not_called()
@ -386,14 +387,14 @@ class TestOVNPortForwarding(TestOVNPortForwardingBase):
mock_get_pf_objs.assert_called_once_with(self.context, fip_id)
calls = [mock.call(lb_name, vip=None, if_exists=True)
for lb_name in fake_lb_names]
self.l3_plugin._ovn.lb_del.assert_has_calls(calls)
calls = [mock.call(mock.ANY, self.l3_plugin._ovn, pf_obj)
self.l3_plugin._nb_ovn.lb_del.assert_has_calls(calls)
calls = [mock.call(mock.ANY, self.l3_plugin._nb_ovn, pf_obj)
for pf_obj in pf_objs]
self.handler.port_forwarding_created.assert_has_calls(calls)
self.l3_plugin.get_floatingip.assert_called_once_with(
self.context, fip_id)
self.fake_check_rev.assert_called_once_with(
self.txn, self.l3_plugin._ovn, fip_id, fake_fip_obj)
self.txn, self.l3_plugin._nb_ovn, fip_id, fake_fip_obj)
@mock.patch.object(port_forwarding.LOG, 'info')
def test_db_sync_delete(self, m_info):
@ -405,4 +406,4 @@ class TestOVNPortForwarding(TestOVNPortForwardingBase):
self.assertIn('db_sync DELETE entries', info_args[0])
calls = [mock.call(lb_name, vip=None, if_exists=True)
for lb_name in fake_lb_names]
self.l3_plugin._ovn.lb_del.assert_has_calls(calls)
self.l3_plugin._nb_ovn.lb_del.assert_has_calls(calls)