openstack
/
neutron
Code Issues Proposed changes

Merge "Change TestOVNL3RouterPlugin to support unittest.mock"

This commit is contained in:
Zuul 2020-03-24 16:20:15 +00:00 committed by Gerrit Code Review
parent a0e1b5804e 8c7aac4f3c
commit 9531e065ce
1 changed files with 109 additions and 204 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -219,13 +219,13 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
self._start_mock(
'neutron.plugins.ml2.plugin.Ml2Plugin.get_network',
return_value=self.fake_network)
self._start_mock(
self.get_port = self._start_mock(
'neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_port',
return_value=self.fake_router_port)
self._start_mock(
self.get_subnet = self._start_mock(
'neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_subnet',
return_value=self.fake_subnet)
self._start_mock(
self.get_router = self._start_mock(
'neutron.db.l3_db.L3_NAT_dbonly_mixin.get_router',
return_value=self.fake_router)
self._start_mock(
@ -255,12 +255,12 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
self._start_mock(
'neutron.db.l3_db.L3_NAT_dbonly_mixin.create_floatingip',
return_value=self.fake_floating_ip)
self._start_mock(
'neutron.db.ovn_revision_numbers_db.bump_revision',
return_value=None)
self._start_mock(
self._get_floatingip = self._start_mock(
'neutron.db.l3_db.L3_NAT_dbonly_mixin._get_floatingip',
return_value=self.fake_floating_ip)
self._start_mock(
'neutron.db.l3_db.L3_NAT_dbonly_mixin.update_floatingip_status',
return_value=None)
self._start_mock(
'neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb.ovn_client.'
'OVNClient.update_floatingip_status',
@ -275,11 +275,9 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
'neutron.common.ovn.utils.get_revision_number',
return_value=1)
self.admin_context = mock.Mock()
self.get_a_ctx_mock_p = mock.patch(
self._start_mock(
'neutron_lib.context.get_admin_context',
return_value=self.admin_context)
self.addCleanup(self.get_a_ctx_mock_p.stop)
self.get_a_ctx_mock_p.start()
self.mock_is_lb_member_fip = mock.patch(
'neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb.ovn_client'
'.OVNClient._is_lb_member_fip',
@ -304,8 +302,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
ovn_const.TYPE_ROUTER_PORTS)
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin.add_router_interface')
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_port')
def test_add_router_interface_update_lrouter_port(self, getp, func):
def test_add_router_interface_update_lrouter_port(self, func):
router_id = 'router-id'
interface_info = {'port_id': 'router-port-id'}
func.return_value = {'id': router_id,
@ -319,7 +316,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
'subnet_id': 'subnet-id2'}],
'mac_address': 'aa:aa:aa:aa:aa:aa'
}
getp.return_value = {
self.get_port.return_value = {
'id': 'router-port-id',
'fixed_ips': [
{'ip_address': '2001:db8::1', 'subnet_id': 'subnet-id1'},
@ -340,11 +337,11 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
'router-port-id', 'lrp-router-port-id', is_gw_port=False,
lsp_address=ovn_const.DEFAULT_ADDR_FOR_LSP_WITH_PEER)
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_port')
def test_remove_router_interface(self, getp):
def test_remove_router_interface(self):
router_id = 'router-id'
interface_info = {'port_id': 'router-port-id'}
getp.side_effect = n_exc.PortNotFound(port_id='router-port-id')
self.get_port.side_effect = n_exc.PortNotFound(
port_id='router-port-id')
self.l3_inst.remove_router_interface(
self.context, router_id, interface_info)
@ -373,12 +370,10 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
@mock.patch('neutron.db.extraroute_db.ExtraRoute_dbonly_mixin.'
'update_router')
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin.get_router')
@mock.patch('neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb'
'.ovn_client.OVNClient._get_v4_network_of_all_router_ports')
def test_update_router_admin_state_change(self, get_rps, get_r, func):
def test_update_router_admin_state_change(self, get_rps, func):
router_id = 'router-id'
get_r.return_value = self.fake_router
new_router = self.fake_router.copy()
updated_data = {'admin_state_up': True}
new_router.update(updated_data)
@ -393,12 +388,10 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
@mock.patch('neutron.db.extraroute_db.ExtraRoute_dbonly_mixin.'
'update_router')
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin.get_router')
@mock.patch('neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb.'
'ovn_client.OVNClient._get_v4_network_of_all_router_ports')
def test_update_router_name_change(self, get_rps, get_r, func):
def test_update_router_name_change(self, get_rps, func):
router_id = 'router-id'
get_r.return_value = self.fake_router
new_router = self.fake_router.copy()
updated_data = {'name': 'test'}
new_router.update(updated_data)
@ -435,10 +428,9 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
@mock.patch.object(utils, 'get_lrouter_non_gw_routes')
@mock.patch('neutron.db.extraroute_db.ExtraRoute_dbonly_mixin.'
'update_router')
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin.get_router')
@mock.patch('neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb.'
'ovn_client.OVNClient._get_v4_network_of_all_router_ports')
def test_update_router_static_route_change(self, get_rps, get_r, func,
def test_update_router_static_route_change(self, get_rps, func,
mock_routes):
router_id = 'router-id'
get_rps.return_value = [{'device_id': '',
@ -449,7 +441,6 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
'id': 'router-port-id'}]
mock_routes.return_value = self.fake_router['routes']
get_r.return_value = self.fake_router
new_router = self.fake_router.copy()
updated_data = {'routes': [{'destination': '2.2.2.0/24',
'nexthop': '10.0.0.3'}]}
@ -467,10 +458,9 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
@mock.patch.object(utils, 'get_lrouter_non_gw_routes')
@mock.patch('neutron.db.extraroute_db.ExtraRoute_dbonly_mixin.'
'update_router')
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin.get_router')
@mock.patch('neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb.'
'ovn_client.OVNClient._get_v4_network_of_all_router_ports')
def test_update_router_static_route_clear(self, get_rps, get_r, func,
def test_update_router_static_route_clear(self, get_rps, func,
mock_routes):
router_id = 'router-id'
get_rps.return_value = [{'device_id': '',
@ -481,7 +471,6 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
'id': 'router-port-id'}]
mock_routes.return_value = self.fake_router['routes']
get_r.return_value = self.fake_router
new_router = self.fake_router.copy()
updated_data = {'routes': []}
new_router.update(updated_data)
@ -493,15 +482,13 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
'neutron-router-id',
ip_prefix='1.1.1.0/24', nexthop='10.0.0.2')
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_port')
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_subnet')
@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, get_subnet, get_port):
def test_create_router_with_ext_gw(self, get_rps):
self.l3_inst._ovn.is_col_present.return_value = True
router = {'router': {'name': 'router'}}
get_subnet.return_value = self.fake_ext_subnet
get_port.return_value = self.fake_ext_gw_port
self.get_subnet.return_value = self.fake_ext_subnet
self.get_port.return_value = self.fake_ext_gw_port
get_rps.return_value = self.fake_ext_subnet['cidr']
self.l3_inst.create_router(self.context, router)
@ -538,31 +525,23 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
@mock.patch('neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb.ovn_client'
'.OVNClient._get_router_ports')
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin.get_router')
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_subnet')
def test_delete_router_with_ext_gw(self, gs, gr, gprs):
gr.return_value = self.fake_router_with_ext_gw
gs.return_value = self.fake_ext_subnet
def test_delete_router_with_ext_gw(self, gprs):
self.get_router.return_value = self.fake_router_with_ext_gw
self.get_subnet.return_value = self.fake_ext_subnet
self.l3_inst.delete_router(self.context, 'router-id')
self.l3_inst._ovn.delete_lrouter.assert_called_once_with(
'neutron-router-id')
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_port')
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_subnet')
@mock.patch('neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb.ovn_client'
'.OVNClient._get_router_ports')
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin.get_router')
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin.add_router_interface')
def test_add_router_interface_with_gateway_set(self, ari, gr, grps,
gs, gp):
def test_add_router_interface_with_gateway_set(self, ari, grps):
router_id = 'router-id'
interface_info = {'port_id': 'router-port-id'}
ari.return_value = self.fake_router_interface_info
gr.return_value = self.fake_router_with_ext_gw
gs.return_value = self.fake_subnet
gp.return_value = self.fake_router_port
self.get_router.return_value = self.fake_router_with_ext_gw
self.l3_inst.add_router_interface(self.context, router_id,
interface_info)
@ -580,21 +559,17 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
mock.ANY, self.fake_router_port,
ovn_const.TYPE_ROUTER_PORTS)
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_port')
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_subnet')
@mock.patch('neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb.ovn_client'
'.OVNClient._get_router_ports')
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin.get_router')
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin.add_router_interface')
def test_add_router_interface_with_gateway_set_and_snat_disabled(
self, ari, gr, grps, gs, gp):
self, ari, grps):
router_id = 'router-id'
interface_info = {'port_id': 'router-port-id'}
ari.return_value = self.fake_router_interface_info
gr.return_value = self.fake_router_with_ext_gw
gr.return_value['external_gateway_info']['enable_snat'] = False
gs.return_value = self.fake_subnet
gp.return_value = self.fake_router_port
get_router = self.fake_router_with_ext_gw
get_router['external_gateway_info']['enable_snat'] = False
self.get_router.return_value = get_router
self.l3_inst.add_router_interface(self.context, router_id,
interface_info)
@ -608,20 +583,14 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
self.l3_inst._ovn.add_nat_rule_in_lrouter.assert_not_called()
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_network')
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_port')
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_subnet')
@mock.patch('neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb.ovn_client'
'.OVNClient._get_router_ports')
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin.get_router')
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin.add_router_interface')
def test_add_router_interface_vlan_network(self, ari, gr, grps, gs,
gp, gn):
def test_add_router_interface_vlan_network(self, ari, grps, gn):
router_id = 'router-id'
interface_info = {'port_id': 'router-port-id'}
ari.return_value = self.fake_router_interface_info
gr.return_value = self.fake_router_with_ext_gw
gs.return_value = self.fake_subnet
gp.return_value = self.fake_router_port
self.get_router.return_value = self.fake_router_with_ext_gw
# Set the type to be VLAN
fake_network_vlan = self.fake_network
@ -651,16 +620,13 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
mock.ANY, self.fake_router_port,
ovn_const.TYPE_ROUTER_PORTS)
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_port')
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_subnet')
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin.get_router')
def test_remove_router_interface_with_gateway_set(self, gr, gs, gp):
def test_remove_router_interface_with_gateway_set(self):
router_id = 'router-id'
interface_info = {'port_id': 'router-port-id',
'subnet_id': 'subnet-id'}
gr.return_value = self.fake_router_with_ext_gw
gs.return_value = self.fake_subnet
gp.side_effect = n_exc.PortNotFound(port_id='router-port-id')
self.get_router.return_value = self.fake_router_with_ext_gw
self.get_port.side_effect = n_exc.PortNotFound(
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(
@ -671,21 +637,18 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
self.del_rev_p.assert_called_with(
self.context, 'router-port-id', ovn_const.TYPE_ROUTER_PORTS)
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_port')
@mock.patch('neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb.ovn_client'
'.OVNClient._get_router_ports')
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_subnet')
@mock.patch('neutron.db.extraroute_db.ExtraRoute_dbonly_mixin.'
'update_router')
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin.get_router')
def test_update_router_with_ext_gw(self, gr, ur, gs, grps, gp):
def test_update_router_with_ext_gw(self, ur, grps):
self.l3_inst._ovn.is_col_present.return_value = True
router = {'router': {'name': 'router'}}
gr.return_value = self.fake_router_without_ext_gw
self.get_router.return_value = self.fake_router_without_ext_gw
ur.return_value = self.fake_router_with_ext_gw
gs.side_effect = lambda ctx, sid: {
self.get_subnet.side_effect = lambda ctx, sid: {
'ext-subnet-id': self.fake_ext_subnet}.get(sid, self.fake_subnet)
gp.return_value = self.fake_ext_gw_port
self.get_port.return_value = self.fake_ext_gw_port
grps.return_value = self.fake_router_ports
self.l3_inst.update_router(self.context, 'router-id', router)
@ -709,15 +672,12 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
ovn_const.TYPE_ROUTER_PORTS)
@mock.patch.object(utils, 'get_lrouter_ext_gw_static_route')
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_port')
@mock.patch('neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb.ovn_client'
'.OVNClient._get_router_ports')
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_subnet')
@mock.patch('neutron.db.extraroute_db.ExtraRoute_dbonly_mixin.'
'update_router')
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin.get_router')
def test_update_router_ext_gw_change_subnet(self, gr, ur, gs,
grps, gp, mock_get_gw):
def test_update_router_ext_gw_change_subnet(self, ur,
grps, mock_get_gw):
self.l3_inst._ovn.is_col_present.return_value = True
mock_get_gw.return_value = [mock.sentinel.GwRoute]
router = {'router': {'name': 'router'}}
@ -726,18 +686,18 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
'cidr': '192.168.2.0/24',
'gateway_ip': '192.168.2.254'}
# Old gateway info with same network and different subnet
gr.return_value = copy.copy(self.fake_router_with_ext_gw)
gr.return_value['external_gateway_info'] = {
self.get_router.return_value = copy.copy(self.fake_router_with_ext_gw)
self.get_router.return_value['external_gateway_info'] = {
'network_id': 'ext-network-id',
'external_fixed_ips': [{'ip_address': '192.168.2.1',
'subnet_id': 'old-ext-subnet-id'}]}
gr.return_value['gw_port_id'] = 'old-gw-port-id'
self.get_router.return_value['gw_port_id'] = 'old-gw-port-id'
ur.return_value = self.fake_router_with_ext_gw
gs.side_effect = lambda ctx, sid: {
self.get_subnet.side_effect = lambda ctx, sid: {
'ext-subnet-id': self.fake_ext_subnet,
'old-ext-subnet-id': fake_old_ext_subnet}.get(sid,
self.fake_subnet)
gp.return_value = self.fake_ext_gw_port
self.get_port.return_value = self.fake_ext_gw_port
grps.return_value = self.fake_router_ports
self.l3_inst.update_router(self.context, 'router-id', router)
@ -769,15 +729,12 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
mock.ANY, 'old-gw-port-id', ovn_const.TYPE_ROUTER_PORTS)
@mock.patch.object(utils, 'get_lrouter_ext_gw_static_route')
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_port')
@mock.patch('neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb.ovn_client.'
'OVNClient._get_router_ports')
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_subnet')
@mock.patch('neutron.db.extraroute_db.ExtraRoute_dbonly_mixin.'
'update_router')
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin.get_router')
def test_update_router_ext_gw_change_ip_address(self, gr, ur, gs,
grps, gp, mock_get_gw):
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
mock_get_gw.return_value = [mock.sentinel.GwRoute]
router = {'router': {'name': 'router'}}
@ -786,11 +743,11 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
gr_value['external_gateway_info'][
'external_fixed_ips'][0]['ip_address'] = '192.168.1.2'
gr_value['gw_port_id'] = 'old-gw-port-id'
gr.return_value = gr_value
self.get_router.return_value = gr_value
ur.return_value = self.fake_router_with_ext_gw
gs.side_effect = lambda ctx, sid: {
self.get_subnet.side_effect = lambda ctx, sid: {
'ext-subnet-id': self.fake_ext_subnet}.get(sid, self.fake_subnet)
gp.return_value = self.fake_ext_gw_port
self.get_port.return_value = self.fake_ext_gw_port
grps.return_value = self.fake_router_ports
self.l3_inst.update_router(self.context, 'router-id', router)
@ -818,10 +775,9 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
'ovn_client.OVNClient._get_v4_network_of_all_router_ports')
@mock.patch('neutron.db.extraroute_db.ExtraRoute_dbonly_mixin.'
'update_router')
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin.get_router')
def test_update_router_ext_gw_no_change(self, gr, ur, get_rps):
def test_update_router_ext_gw_no_change(self, ur, get_rps):
router = {'router': {'name': 'router'}}
gr.return_value = self.fake_router_with_ext_gw
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 = (
fake_resources.FakeOVNRouter.from_neutron_router(
@ -837,23 +793,19 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
self.l3_inst._ovn.add_static_route.assert_not_called()
self.l3_inst._ovn.add_nat_rule_in_lrouter.assert_not_called()
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_port')
@mock.patch('neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb.ovn_client'
'.OVNClient._get_v4_network_of_all_router_ports')
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_subnet')
@mock.patch('neutron.db.extraroute_db.ExtraRoute_dbonly_mixin.'
'update_router')
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin.get_router')
def test_update_router_with_ext_gw_and_disabled_snat(self, gr, ur,
gs, grps, gp):
def test_update_router_with_ext_gw_and_disabled_snat(self, ur, grps):
self.l3_inst._ovn.is_col_present.return_value = True
router = {'router': {'name': 'router'}}
gr.return_value = self.fake_router_without_ext_gw
self.get_router.return_value = self.fake_router_without_ext_gw
ur.return_value = self.fake_router_with_ext_gw
ur.return_value['external_gateway_info']['enable_snat'] = False
gs.side_effect = lambda ctx, sid: {
self.get_subnet.side_effect = lambda ctx, sid: {
'ext-subnet-id': self.fake_ext_subnet}.get(sid, self.fake_subnet)
gp.return_value = self.fake_ext_gw_port
self.get_port.return_value = self.fake_ext_gw_port
grps.return_value = self.fake_router_ports
self.l3_inst.update_router(self.context, 'router-id', router)
@ -866,24 +818,22 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
nexthop='192.168.1.254')
self.l3_inst._ovn.add_nat_rule_in_lrouter.assert_not_called()
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_port')
@mock.patch('neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb.ovn_client'
'.OVNClient._get_router_ports')
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_subnet')
@mock.patch('neutron.db.extraroute_db.ExtraRoute_dbonly_mixin.'
'update_router')
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin.get_router')
def test_enable_snat(self, gr, ur, gs, grps, gp):
def test_enable_snat(self, ur, grps):
router = {'router': {'name': 'router'}}
gr.return_value = copy.deepcopy(self.fake_router_with_ext_gw)
gr.return_value['external_gateway_info']['enable_snat'] = False
gr_value = copy.deepcopy(self.fake_router_with_ext_gw)
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 = (
fake_resources.FakeOVNRouter.from_neutron_router(
self.fake_router_with_ext_gw))
gs.side_effect = lambda ctx, sid: {
self.get_subnet.side_effect = lambda ctx, sid: {
'ext-subnet-id': self.fake_ext_subnet}.get(sid, self.fake_subnet)
gp.return_value = self.fake_ext_gw_port
self.get_port.return_value = self.fake_ext_gw_port
grps.return_value = self.fake_router_ports
self.l3_inst.update_router(self.context, 'router-id', router)
@ -901,25 +851,22 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
@mock.patch.object(utils, 'get_lrouter_ext_gw_static_route')
@mock.patch('neutron.common.ovn.utils.is_snat_enabled',
mock.Mock(return_value=True))
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_port')
@mock.patch('neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb.'
'ovn_client.OVNClient._get_router_ports')
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_subnet')
@mock.patch('neutron.db.extraroute_db.ExtraRoute_dbonly_mixin.'
'update_router')
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin.get_router')
def test_disable_snat(self, gr, ur, gs, grps, gp, mock_get_gw, mock_snats,
def test_disable_snat(self, ur, grps, mock_get_gw, mock_snats,
mock_ext_ips):
mock_get_gw.return_value = [mock.sentinel.GwRoute]
mock_snats.return_value = [mock.sentinel.NAT]
mock_ext_ips.return_value = False
router = {'router': {'name': 'router'}}
gr.return_value = self.fake_router_with_ext_gw
self.get_router.return_value = self.fake_router_with_ext_gw
ur.return_value = copy.deepcopy(self.fake_router_with_ext_gw)
ur.return_value['external_gateway_info']['enable_snat'] = False
gs.side_effect = lambda ctx, sid: {
self.get_subnet.side_effect = lambda ctx, sid: {
'ext-subnet-id': self.fake_ext_subnet}.get(sid, self.fake_subnet)
gp.return_value = self.fake_ext_gw_port
self.get_port.return_value = self.fake_ext_gw_port
grps.return_value = self.fake_router_ports
self.l3_inst.update_router(self.context, 'router-id', router)
@ -931,11 +878,9 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
self.l3_inst._ovn.add_static_route.assert_not_called()
self.l3_inst._ovn.add_nat_rule_in_lrouter.assert_not_called()
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin._get_floatingip')
def test_create_floatingip(self, gf):
self.get_a_ctx_mock_p.stop()
def test_create_floatingip(self):
self.l3_inst._ovn.is_col_present.return_value = True
gf.return_value = {'floating_port_id': 'fip-port-id'}
self._get_floatingip.return_value = {'floating_port_id': 'fip-port-id'}
self.l3_inst.create_floatingip(self.context, 'floatingip')
expected_ext_ids = {
ovn_const.OVN_FIP_EXT_ID_KEY: self.fake_floating_ip['id'],
@ -953,14 +898,11 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
self.l3_inst._ovn.delete_lswitch_port.assert_called_once_with(
'fip-port-id', 'neutron-fip-net-id')
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_port')
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin._get_floatingip')
def test_create_floatingip_distributed(self, gf, gp):
self.get_a_ctx_mock_p.stop()
def test_create_floatingip_distributed(self):
self.l3_inst._ovn.is_col_present.return_value = True
gp.return_value = {'mac_address': '00:01:02:03:04:05',
self.get_port.return_value = {'mac_address': '00:01:02:03:04:05',
'network_id': 'port-network-id'}
gf.return_value = {'floating_port_id': 'fip-port-id'}
self._get_floatingip.return_value = {'floating_port_id': 'fip-port-id'}
config.cfg.CONF.set_override(
'enable_distributed_floating_ip', True, group='ovn')
self.l3_inst.create_floatingip(self.context, 'floatingip')
@ -978,18 +920,15 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
logical_port='port_id',
external_ids=expected_ext_ids)
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_port')
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin._get_floatingip')
def test_create_floatingip_distributed_logical_port_down(self, gf, gp):
self.get_a_ctx_mock_p.stop()
def test_create_floatingip_distributed_logical_port_down(self):
# 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 = (
False)
gp.return_value = {'mac_address': '00:01:02:03:04:05'}
gf.return_value = {'floating_port_id': 'fip-port-id'}
self.get_port.return_value = {'mac_address': '00:01:02:03:04:05'}
self._get_floatingip.return_value = {'floating_port_id': 'fip-port-id'}
config.cfg.CONF.set_override(
'enable_distributed_floating_ip', True, group='ovn')
self.l3_inst.create_floatingip(self.context, 'floatingip')
@ -1007,11 +946,9 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
logical_port='port_id',
external_ids=expected_ext_ids)
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin._get_floatingip')
def test_create_floatingip_external_ip_present_in_nat_rule(self, gf):
self.get_a_ctx_mock_p.stop()
def test_create_floatingip_external_ip_present_in_nat_rule(self):
self.l3_inst._ovn.is_col_present.return_value = True
gf.return_value = {'floating_port_id': 'fip-port-id'}
self._get_floatingip.return_value = {'floating_port_id': 'fip-port-id'}
self.l3_inst._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'}]
@ -1032,11 +969,9 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
self.l3_inst._ovn.delete_lswitch_port.assert_called_once_with(
'fip-port-id', 'neutron-fip-net-id')
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin._get_floatingip')
def test_create_floatingip_external_ip_present_type_snat(self, gf):
self.get_a_ctx_mock_p.stop()
def test_create_floatingip_external_ip_present_type_snat(self):
self.l3_inst._ovn.is_col_present.return_value = True
gf.return_value = {'floating_port_id': 'fip-port-id'}
self._get_floatingip.return_value = {'floating_port_id': 'fip-port-id'}
self.l3_inst._ovn.get_lrouter_nat_rules.return_value = [
{'external_ip': '192.168.0.10', 'logical_ip': '10.0.0.0/24',
'type': 'snat', 'uuid': 'uuid1'}]
@ -1058,9 +993,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
self.l3_inst._ovn.delete_lswitch_port.assert_called_once_with(
'fip-port-id', 'neutron-fip-net-id')
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin._get_floatingip')
def test_create_floatingip_lsp_external_id(self, gf):
self.get_a_ctx_mock_p.stop()
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
@ -1072,16 +1005,12 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
'192.168.0.10'}))]
self.l3_inst._ovn.db_set.assert_has_calls(calls)
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin._get_floatingip')
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_port')
def test_create_floatingip_lb_member_fip(self, gp, gf):
self.get_a_ctx_mock_p.stop()
def test_create_floatingip_lb_member_fip(self):
config.cfg.CONF.set_override(
'enable_distributed_floating_ip', True, group='ovn')
# Stop this mock.
self.mock_is_lb_member_fip.stop()
gp.return_value = self.member_port
gf.return_value = self.fake_floating_ip
self.get_port.return_value = self.member_port
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.create_floatingip(self.context, 'floatingip')
@ -1093,12 +1022,10 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
logical_ip='10.0.0.10',
type='dnat_and_snat')
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_subnet')
def test_create_floatingip_lb_vip_fip(self, gs):
self.get_a_ctx_mock_p.stop()
def test_create_floatingip_lb_vip_fip(self):
config.cfg.CONF.set_override(
'enable_distributed_floating_ip', True, group='ovn')
gs.return_value = self.member_subnet
self.get_subnet.return_value = self.member_subnet
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.ovn_lb],
@ -1134,14 +1061,11 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
logical_ip='10.0.0.10',
external_ip='192.168.0.10')
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_subnet')
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin._get_floatingip')
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin.delete_floatingip')
def test_delete_floatingip_lb_vip_fip(self, df, gf, gs):
def test_delete_floatingip_lb_vip_fip(self, df):
config.cfg.CONF.set_override(
'enable_distributed_floating_ip', True, group='ovn')
gs.return_value = self.member_subnet
gf.return_value = self.fake_floating_ip
self.get_subnet.return_value = self.member_subnet
self.l3_inst._ovn.get_floatingip.return_value = (
self.fake_ovn_nat_rule)
self.l3_inst._ovn.get_lswitch_port.return_value = self.lb_vip_lsp
@ -1170,10 +1094,8 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
mock.call('NAT', self.fake_ovn_nat_rule.uuid,
('external_mac', 'aa:aa:aa:aa:aa:aa'))])
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin._get_floatingip')
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin.delete_floatingip')
def test_delete_floatingip_lsp_external_id(self, df, gf):
gf.return_value = self.fake_floating_ip
def test_delete_floatingip_lsp_external_id(self, df):
self.l3_inst._ovn.get_floatingip.return_value = (
self.fake_ovn_nat_rule)
@ -1189,23 +1111,18 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
'external_ids', ovn_const.OVN_PORT_FIP_EXT_ID_KEY)]
self.l3_inst._ovn.db_remove.assert_has_calls(calls)
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin._get_floatingip')
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin.delete_floatingip')
def test_delete_floatingip_no_lsp_external_id(self, df, gf):
gf.return_value = self.fake_floating_ip
def test_delete_floatingip_no_lsp_external_id(self, df):
self.l3_inst._ovn.get_floatingip.return_value = (
self.fake_ovn_nat_rule)
self.l3_inst._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()
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin._get_floatingip')
@mock.patch('neutron.db.extraroute_db.ExtraRoute_dbonly_mixin.'
'update_floatingip')
def test_update_floatingip(self, uf, gf):
self.get_a_ctx_mock_p.stop()
def test_update_floatingip(self, uf):
self.l3_inst._ovn.is_col_present.return_value = True
gf.return_value = self.fake_floating_ip
uf.return_value = self.fake_floating_ip_new
self.l3_inst._ovn.get_floatingip.return_value = (
self.fake_ovn_nat_rule)
@ -1229,14 +1146,11 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
external_ip='192.168.0.10',
external_ids=expected_ext_ids)
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin._get_floatingip')
@mock.patch('neutron.db.extraroute_db.ExtraRoute_dbonly_mixin.'
'update_floatingip')
def test_update_floatingip_associate(self, uf, gf):
self.get_a_ctx_mock_p.stop()
def test_update_floatingip_associate(self, uf):
self.l3_inst._ovn.is_col_present.return_value = True
self.fake_floating_ip.update({'fixed_port_id': None})
gf.return_value = self.fake_floating_ip
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()
@ -1255,17 +1169,13 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
external_ids=expected_ext_ids)
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_network')
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_port')
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin._get_floatingip')
@mock.patch('neutron.db.extraroute_db.ExtraRoute_dbonly_mixin.'
'update_floatingip')
def test_update_floatingip_associate_distributed(self, uf, gf, gp, gn):
self.get_a_ctx_mock_p.stop()
def test_update_floatingip_associate_distributed(self, uf, gn):
self.l3_inst._ovn.is_col_present.return_value = True
self.fake_floating_ip.update({'fixed_port_id': None})
gp.return_value = {'mac_address': '00:01:02:03:04:05',
self.get_port.return_value = {'mac_address': '00:01:02:03:04:05',
'network_id': 'port-network-id'}
gf.return_value = self.fake_floating_ip
uf.return_value = self.fake_floating_ip_new
fake_network_vlan = self.fake_network
@ -1290,17 +1200,14 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
external_mac='00:01:02:03:04:05', logical_port='new-port_id',
external_ids=expected_ext_ids)
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin._get_floatingip')
@mock.patch('neutron.db.extraroute_db.ExtraRoute_dbonly_mixin.'
'update_floatingip')
def test_update_floatingip_association_empty_update(self, uf, gf):
self.get_a_ctx_mock_p.stop()
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 = (
self.fake_ovn_nat_rule)
self.fake_floating_ip.update({'fixed_port_id': 'foo'})
self.fake_floating_ip_new.update({'port_id': 'foo'})
gf.return_value = self.fake_floating_ip
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(
@ -1322,18 +1229,15 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
external_ip='192.168.0.10',
external_ids=expected_ext_ids)
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin._get_floatingip')
@mock.patch('neutron.db.extraroute_db.ExtraRoute_dbonly_mixin.'
'update_floatingip')
def test_update_floatingip_reassociate_to_same_port_diff_fixed_ip(
self, uf, gf):
self.get_a_ctx_mock_p.stop()
self, uf):
self.l3_inst._ovn.is_col_present.return_value = True
self.l3_inst._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'})
gf.return_value = self.fake_floating_ip
uf.return_value = self.fake_floating_ip_new
self.l3_inst.update_floatingip(self.context, 'id', 'floatingip')
@ -1439,20 +1343,27 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
mock_updt_status.assert_called_once_with(
mock.ANY, fake_port_id, constants.PORT_STATUS_DOWN)
def test_schedule_unhosted_gateways_no_gateways(self):
self.get_a_ctx_mock_p.stop()
@mock.patch('neutron.services.ovn_l3.plugin.OVNL3RouterPlugin.'
'_get_gateway_port_physnet_mapping')
def test_schedule_unhosted_gateways_no_gateways(self, get_gppm):
get_gppm.return_value = {}
self.nb_idl().get_unhosted_gateways.return_value = []
self.l3_inst.schedule_unhosted_gateways()
self.nb_idl().update_lrouter_port.assert_not_called()
def test_schedule_unhosted_gateways(self):
self.get_a_ctx_mock_p.stop()
@mock.patch('neutron.services.ovn_l3.plugin.OVNL3RouterPlugin.'
'_get_gateway_port_physnet_mapping')
def test_schedule_unhosted_gateways(self, get_gppm):
physnet_dict = {'foo-1': 'physnet1',
'foo-2': 'physnet1',
'foo-3': 'physnet1'}
unhosted_gws = ['lrp-foo-1', 'lrp-foo-2', 'lrp-foo-3']
chassis_mappings = {
'chassis1': ['physnet1'],
'chassis2': ['physnet1'],
'chassis3': ['physnet1']}
chassis = ['chassis1', 'chassis2', 'chassis3']
get_gppm.return_value = physnet_dict
self.sb_idl().get_chassis_and_physnets.return_value = (
chassis_mappings)
self.sb_idl().get_gateway_chassis_from_cms_options.return_value = (
@ -1498,25 +1409,19 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
gateway_chassis=['chassis3', 'chassis2', 'chassis1'])])
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_network')
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_port')
@mock.patch('neutron.db.db_base_plugin_v2.NeutronDbPluginV2.get_subnet')
@mock.patch('neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb.'
'ovn_client.OVNClient._get_router_ports')
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin.get_router')
@mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin.add_router_interface')
def test_add_router_interface_need_to_frag_enabled(self, ari, gr, grps,
gs, gp, gn):
def test_add_router_interface_need_to_frag_enabled(self, ari, grps, gn):
config.cfg.CONF.set_override(
'ovn_emit_need_to_frag', True, group='ovn')
router_id = 'router-id'
interface_info = {'port_id': 'router-port-id'}
ari.return_value = self.fake_router_interface_info
gr.return_value = self.fake_router_with_ext_gw
gs.return_value = self.fake_subnet
self.get_router.return_value = self.fake_router_with_ext_gw
gn.return_value = self.fake_network
self.fake_router_port['device_owner'] = (
constants.DEVICE_OWNER_ROUTER_GW)
gp.return_value = self.fake_router_port
self.l3_inst.add_router_interface(self.context, router_id,
interface_info)