openstack/neutron
Code Issues Proposed changes

Use neutron-lib constants IPv4_ANY and IPv6_ANY

Browse Source 
Use these constants wherever is possible and replace the duplicated
constants.

TrivialFix

Change-Id: I19cabbc5b0701e9e791143378fe95a5d509b6d79
This commit is contained in:
Rodolfo Alonso Hernandez 2024-09-19 07:56:33 +00:00
parent 6d38e7d975
commit 1e6ab531e2
19 changed files with 70 additions and 63 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
neutron/common/ovn/constants.py
View File

@ -464,6 +464,3 @@ OVN_SUPPORTED_VNIC_TYPES = [portbindings.VNIC_NORMAL,
portbindings.VNIC_BAREMETAL,
portbindings.VNIC_VIRTIO_FORWARDER,
]
# OVN default SNAT CIDR
OVN_DEFAULT_SNAT_CIDR = '0.0.0.0/0'

2
neutron/db/securitygroups_db.py
View File

@ -927,7 +927,7 @@ class SecurityGroupDbMixin(
none_char = '+'
if key == 'remote_ip_prefix':
all_address = ['0.0.0.0/0', '::/0', None]
all_address = [constants.IPv4_ANY, constants.IPv6_ANY, None]
if value in all_address:
return none_char
elif value is None:

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

@ -1442,7 +1442,7 @@ class OVNClient(object):
def _get_snat_cidrs_for_external_router(self, context, router_id):
if is_nested_snat():
return [ovn_const.OVN_DEFAULT_SNAT_CIDR]
return [const.IPv4_ANY]
# nat rule per attached subnet per external ip
return self._get_v4_network_of_all_router_ports(context, router_id)

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

@ -54,7 +54,6 @@ from neutron.services.ovn_l3.service_providers import driver_controller
from neutron.services.portforwarding.drivers.ovn import driver \
as port_forwarding
LOG = log.getLogger(__name__)
@ -330,9 +329,9 @@ class OVNL3RouterPlugin(service_base.ServicePluginBase,
router_ids = {port['device_id'] for port in gw_ports
if utils.is_ovn_provider_router(
l3plugin.get_router(context, port['device_id']))}
remove = [{'destination': '0.0.0.0/0', 'nexthop': orig_gw_ip}
remove = [{'destination': n_const.IPv4_ANY, 'nexthop': orig_gw_ip}
] if orig_gw_ip else []
add = [{'destination': '0.0.0.0/0', 'nexthop': current_gw_ip}
add = [{'destination': n_const.IPv4_ANY, 'nexthop': current_gw_ip}
] if current_gw_ip else []
with l3plugin._nb_ovn.transaction(check_error=True) as txn:
for router_id in router_ids:

8
neutron/tests/functional/agent/l3/test_dvr_router.py
View File

@ -312,7 +312,7 @@ class TestDvrRouter(DvrRouterTestFramework, framework.L3AgentTestFramework):
updated_route = fg_device.route.list_routes(
ip_version=lib_constants.IP_VERSION_4,
table=tbl_index)
expected_route = [{'cidr': '0.0.0.0/0',
expected_route = [{'cidr': lib_constants.IPv4_ANY,
'device': fg_port_name,
'table': tbl_index,
'via': '19.4.4.2'}]
@ -360,7 +360,7 @@ class TestDvrRouter(DvrRouterTestFramework, framework.L3AgentTestFramework):
updated_route = fg_device.route.list_routes(
ip_version=lib_constants.IP_VERSION_4,
table=tbl_index)
expected_route = [{'cidr': '0.0.0.0/0',
expected_route = [{'cidr': lib_constants.IPv4_ANY,
'device': fg_port_name,
'table': tbl_index,
'via': '19.4.4.2'}]
@ -930,7 +930,7 @@ class TestDvrRouter(DvrRouterTestFramework, framework.L3AgentTestFramework):
v6_gateway = utils.cidr_to_ip(
ip_lib.get_ipv6_lladdr(fpr_device.link.address))
expected_gateway = [{'device': device_name,
'cidr': '0.0.0.0/0',
'cidr': lib_constants.IPv4_ANY,
'via': str(fip_2_rtr.ip),
'table': dvr_fip_ns.FIP_RT_TBL},
{'device': device_name,
@ -1815,7 +1815,7 @@ class TestDvrRouter(DvrRouterTestFramework, framework.L3AgentTestFramework):
# When floatingip are associated, make sure that the
# corresponding rules and routes in route table are created
# for the router.
expected_rule = {u'from': '0.0.0.0/0',
expected_rule = {u'from': lib_constants.IPv4_ANY,
u'iif': fip_ns_int_name,
'priority': str(router_fip_table_idx),
'table': str(router_fip_table_idx),

6
neutron/tests/functional/agent/test_firewall.py
View File

@ -183,9 +183,11 @@ class FirewallTestCase(BaseFirewallTestCase):
sg_rules = [{'ethertype': 'IPv4', 'direction': 'egress'},
{'ethertype': 'IPv6', 'direction': 'egress'},
{'ethertype': 'IPv4', 'direction': 'ingress',
'source_ip_prefix': '0.0.0.0/0', 'protocol': 'icmp'},
'source_ip_prefix': constants.IPv4_ANY,
'protocol': 'icmp'},
{'ethertype': 'IPv6', 'direction': 'ingress',
'source_ip_prefix': '0::0/0', 'protocol': 'ipv6-icmp'}]
'source_ip_prefix': constants.IPv6_ANY,
'protocol': 'ipv6-icmp'}]
# make sure port ranges converge on all protocols with and without
# port ranges (prevents regression of bug 1502924)
for proto in ('tcp', 'udp', 'icmp'):

2
neutron/tests/functional/plugins/ml2/drivers/ovn/mech_driver/test_mech_driver.py
View File

@ -1433,7 +1433,7 @@ class TestRouterGWPort(_TestRouter):
nats = lr.nat
self.assertEqual(1, len(nats))
expected_logical_ip = (
ovn_const.OVN_DEFAULT_SNAT_CIDR if nested_snat else subnet_cidr
constants.IPv4_ANY if nested_snat else subnet_cidr
)
self.assertEqual(expected_logical_ip, nats[0].logical_ip)

4
neutron/tests/unit/agent/linux/openvswitch_firewall/test_rules.py
View File

@ -31,7 +31,7 @@ class TestIsValidPrefix(base.BaseTestCase):
self.assertTrue(is_valid)
def test_invalid_prefix_ipv4(self):
is_valid = rules.is_valid_prefix('0.0.0.0/0')
is_valid = rules.is_valid_prefix(constants.IPv4_ANY)
self.assertFalse(is_valid)
def test_valid_prefix_ipv6(self):
@ -104,7 +104,7 @@ class TestCreateFlowsFromRuleAndPort(base.BaseTestCase):
'ethertype': constants.IPv4,
'direction': constants.INGRESS_DIRECTION,
'source_ip_prefix': '192.168.0.0/24',
'dest_ip_prefix': '0.0.0.0/0',
'dest_ip_prefix': constants.IPv4_ANY,
}
expected_template = {
'priority': 74,

4
neutron/tests/unit/agent/linux/test_ip_lib.py
View File

@ -1259,7 +1259,7 @@ class ParseIpRuleTestCase(base.BaseTestCase):
def test_parse_from_any_ipv4(self):
parsed_rule = ip_lib._parse_ip_rule(self.rule, 4)
self.assertEqual('0.0.0.0/0', parsed_rule['from'])
self.assertEqual(constants.IPv4_ANY, parsed_rule['from'])
def test_parse_from_any_ipv6(self):
parsed_rule = ip_lib._parse_ip_rule(self.rule, 6)
@ -1338,7 +1338,7 @@ class ListIpRulesTestCase(base.BaseTestCase):
reference = [
{'type': 'unicast', 'from': '10.0.0.1/24', 'priority': '0',
'table': '100'},
{'type': 'blackhole', 'from': '0.0.0.0/0', 'priority': '0',
{'type': 'blackhole', 'from': constants.IPv4_ANY, 'priority': '0',
'table': 'local'}]
self.assertEqual(reference, retval)

6
neutron/tests/unit/agent/linux/test_ipset_manager.py
View File

@ -13,6 +13,8 @@
from unittest import mock
from neutron_lib import constants as n_const
from neutron.agent.linux import ipset_manager
from neutron.tests import base
@ -146,9 +148,9 @@ class IpsetManagerTestCase(BaseIpsetManagerTest):
self.verify_mock_calls()
def test_set_members_adding_all_zero_ipv4(self):
self.expect_set([('0.0.0.0/0', 'fa:16:3e:aa:bb:c1'), ])
self.expect_set([(n_const.IPv4_ANY, 'fa:16:3e:aa:bb:c1'), ])
self.ipset.set_members(TEST_SET_ID, ETHERTYPE,
[('0.0.0.0/0', 'fa:16:3e:aa:bb:c1'), ])
[(n_const.IPv4_ANY, 'fa:16:3e:aa:bb:c1'), ])
self.verify_mock_calls()
def test_set_members_adding_all_zero_ipv6(self):

2
neutron/tests/unit/db/test_db_base_plugin_v2.py
View File

@ -6825,7 +6825,7 @@ class DbModelMixin(object):
security_group=sg, port_range_min=1,
port_range_max=2, protocol='TCP',
ethertype='v4', direction='ingress',
remote_ip_prefix='0.0.0.0/0')
remote_ip_prefix=constants.IPv4_ANY)
ctx.session.add(sg)
ctx.session.add(rule)
return sg, rule

4
neutron/tests/unit/db/test_securitygroups_db.py
View File

@ -172,7 +172,7 @@ class SecurityGroupDbMixinTestCase(testlib_api.SqlTestCase):
'security_group_id': 'fake',
'ethertype': 'IPv4',
'direction': 'ingress',
'remote_ip_prefix': '0.0.0.0/0'}
'remote_ip_prefix': constants.IPv4_ANY}
}
self.assertRaises(securitygroup.SecurityGroupRuleExists,
self.mixin._check_for_duplicate_rules,
@ -768,7 +768,7 @@ class SecurityGroupDbMixinTestCase(testlib_api.SqlTestCase):
'id': 'fake2',
'ethertype': 'IPv4',
'direction': 'ingress',
'remote_ip_prefix': '0.0.0.0/0'}
'remote_ip_prefix': constants.IPv4_ANY}
}
self.assertRaises(
ext_sg_default_rules.DefaultSecurityGroupRuleExists,

6
neutron/tests/unit/fake_resources.py
View File

@ -608,8 +608,8 @@ class FakeSecurityGroupRule(object):
'port_range_min': 22,
'protocol': 'tcp',
'remote_group_id': None,
'remote_ip_prefix': '0.0.0.0/0',
'normalized_cidr': '0.0.0.0/0',
'remote_ip_prefix': n_const.IPv4_ANY,
'normalized_cidr': n_const.IPv4_ANY,
'security_group_id': 'security-group-id-' + fake_uuid,
'tenant_id': 'project-id-' + fake_uuid,
}
@ -924,7 +924,7 @@ class FakeOVNRouter(object):
ovn_const.OVN_ROUTER_IS_EXT_GW: 'true',
ovn_const.OVN_SUBNET_EXT_ID_KEY: _get_subnet_id(gw_info)}
routes.append(FakeStaticRoute(
ip_prefix='0.0.0.0/0', nexthop='',
ip_prefix=n_const.IPv4_ANY, nexthop='',
external_ids=external_ids))
ports.append(FakeOVNRouterPort.create_one_port(
attrs={

7
neutron/tests/unit/plugins/ml2/drivers/ovn/mech_driver/ovsdb/test_commands.py
View File

@ -14,6 +14,7 @@
from unittest import mock
from neutron_lib import constants as n_const
from ovsdbapp.backend.ovs_idl import idlutils
from neutron.common.ovn import constants as ovn_const
@ -1165,7 +1166,7 @@ class TestDeleteLRouterExtGwCommand(TestBaseCommand):
def test_delete_lrouter_extgw_routes(self):
fake_route_1 = fakes.FakeOvsdbRow.create_one_ovsdb_row(
attrs={'ip_prefix': '0.0.0.0/0', 'nexthop': '10.0.0.1',
attrs={'ip_prefix': n_const.IPv4_ANY, 'nexthop': '10.0.0.1',
'external_ids': {ovn_const.OVN_ROUTER_IS_EXT_GW: True}})
fake_route_2 = fakes.FakeOvsdbRow.create_one_ovsdb_row(
attrs={'ip_prefix': '50.0.0.0/24', 'nexthop': '40.0.0.101'})
@ -1185,11 +1186,11 @@ class TestDeleteLRouterExtGwCommand(TestBaseCommand):
def test_delete_lrouter_multiple_extgw_routes(self):
fake_route_1 = fakes.FakeOvsdbRow.create_one_ovsdb_row(
attrs={'ip_prefix': '0.0.0.0/0', 'nexthop': '10.0.0.1',
attrs={'ip_prefix': n_const.IPv4_ANY, 'nexthop': '10.0.0.1',
'external_ids': {ovn_const.OVN_ROUTER_IS_EXT_GW: True},
'output_port': '1'})
fake_route_2 = fakes.FakeOvsdbRow.create_one_ovsdb_row(
attrs={'ip_prefix': '0.0.0.0/0', 'nexthop': '10.0.0.1',
attrs={'ip_prefix': n_const.IPv4_ANY, 'nexthop': '10.0.0.1',
'external_ids': {ovn_const.OVN_ROUTER_IS_EXT_GW: True},
'output_port': '2'})
fake_route_3 = fakes.FakeOvsdbRow.create_one_ovsdb_row(

9
neutron/tests/unit/plugins/ml2/drivers/ovn/mech_driver/ovsdb/test_ovn_client.py
View File

@ -15,6 +15,7 @@
from unittest import mock
from neutron_lib import constants as n_const
from neutron_lib import context as ncontext
from oslo_config import cfg
@ -131,7 +132,7 @@ class TestOVNClient(TestOVNClientBase):
self.ovn_client._add_router_ext_gw(mock.Mock(), router, txn))
self.nb_idl.add_static_route.assert_called_once_with(
'neutron-' + router['id'],
ip_prefix='0.0.0.0/0',
ip_prefix=n_const.IPv4_ANY,
nexthop='10.42.0.1',
maintain_bfd=False,
external_ids={
@ -182,7 +183,7 @@ class TestOVNClient(TestOVNClientBase):
self.ovn_client._add_router_ext_gw(mock.Mock(), router, txn))
self.nb_idl.add_static_route.assert_has_calls([
mock.call('neutron-' + router['id'],
ip_prefix='0.0.0.0/0',
ip_prefix=n_const.IPv4_ANY,
nexthop=subnet1['gateway_ip'],
maintain_bfd=False,
external_ids={
@ -191,7 +192,7 @@ class TestOVNClient(TestOVNClientBase):
constants.OVN_LRSR_EXT_ID_KEY: 'true'},
),
mock.call('neutron-' + router['id'],
ip_prefix='0.0.0.0/0',
ip_prefix=n_const.IPv4_ANY,
nexthop=subnet2['gateway_ip'],
maintain_bfd=False,
external_ids={
@ -358,7 +359,7 @@ class TestOVNClient(TestOVNClientBase):
return_value=per_subnet_cidrs):
cidrs = self.ovn_client._get_snat_cidrs_for_external_router(
ctx, 'fake-id')
self.assertEqual([constants.OVN_DEFAULT_SNAT_CIDR], cidrs)
self.assertEqual([n_const.IPv4_ANY], cidrs)
class TestOVNClientFairMeter(TestOVNClientBase,

27
neutron/tests/unit/plugins/ml2/drivers/ovn/mech_driver/ovsdb/test_ovn_db_sync.py
View File

@ -27,7 +27,6 @@ from neutron.tests.unit import fake_resources as fakes
from neutron.tests.unit.plugins.ml2.drivers.ovn.mech_driver import \
test_mech_driver
OvnPortInfo = collections.namedtuple('OvnPortInfo', ['name'])
@ -121,7 +120,7 @@ class TestOvnNbSyncML2(test_mech_driver.OVNMechanismDriverTestCase):
{'id': 'sg1', 'tenant_id': 'tenant1',
'security_group_rules': [{'remote_group_id': None,
'direction': 'ingress',
'remote_ip_prefix': '0.0.0.0/0',
'remote_ip_prefix': const.IPv4_ANY,
'protocol': 'tcp',
'ethertype': 'IPv4',
'tenant_id': 'tenant1',
@ -133,7 +132,7 @@ class TestOvnNbSyncML2(test_mech_driver.OVNMechanismDriverTestCase):
{'id': 'sg2', 'tenant_id': 'tenant1',
'security_group_rules': [{'remote_group_id': 'sg2',
'direction': 'egress',
'remote_ip_prefix': '0.0.0.0/0',
'remote_ip_prefix': const.IPv4_ANY,
'protocol': 'tcp',
'ethertype': 'IPv4',
'tenant_id': 'tenant1',
@ -776,7 +775,7 @@ class TestOvnNbSyncML2(test_mech_driver.OVNMechanismDriverTestCase):
'destination': '12.0.0.0/24',
'external_ids': {}},
{'nexthop': '90.0.0.1',
'destination': '0.0.0.0/0',
'destination': const.IPv4_ANY,
'external_ids': {
ovn_const.OVN_ROUTER_IS_EXT_GW: 'true',
ovn_const.OVN_SUBNET_EXT_ID_KEY:
@ -785,7 +784,7 @@ class TestOvnNbSyncML2(test_mech_driver.OVNMechanismDriverTestCase):
'destination': '30.0.0.0/24',
'external_ids': {}},
{'nexthop': '100.0.0.1',
'destination': '0.0.0.0/0',
'destination': const.IPv4_ANY,
'external_ids': {
ovn_const.OVN_ROUTER_IS_EXT_GW: 'true',
ovn_const.OVN_SUBNET_EXT_ID_KEY:
@ -935,7 +934,7 @@ class TestOvnNbSyncML2(test_mech_driver.OVNMechanismDriverTestCase):
'destination': '11.0.0.0/24',
'external_ids': {}},
{'nexthop': '90.0.0.1',
'destination': '0.0.0.0/0',
'destination': const.IPv4_ANY,
'external_ids': {
ovn_const.OVN_ROUTER_IS_EXT_GW: 'true',
ovn_const.OVN_SUBNET_EXT_ID_KEY:
@ -954,7 +953,7 @@ class TestOvnNbSyncML2(test_mech_driver.OVNMechanismDriverTestCase):
'destination': '11.0.0.0/24',
'external_ids': {}},
{'nexthop': '90.0.0.1',
'destination': '0.0.0.0/0',
'destination': const.IPv4_ANY,
'external_ids': {
ovn_const.OVN_ROUTER_IS_EXT_GW: 'true',
ovn_const.OVN_SUBNET_EXT_ID_KEY:
@ -971,7 +970,7 @@ class TestOvnNbSyncML2(test_mech_driver.OVNMechanismDriverTestCase):
# remove 2 routes from ovn, add 2 routes to ovn
ovn_routes = [{'nexthop': '90.0.0.1',
'destination': '0.0.0.0/0',
'destination': const.IPv4_ANY,
'external_ids': {
ovn_const.OVN_ROUTER_IS_EXT_GW: 'true',
ovn_const.OVN_SUBNET_EXT_ID_KEY:
@ -1002,7 +1001,7 @@ class TestOvnNbSyncML2(test_mech_driver.OVNMechanismDriverTestCase):
'destination': '12.0.0.0/24',
'external_ids': {}},
{'nexthop': '90.0.0.1',
'destination': '0.0.0.0/0',
'destination': const.IPv4_ANY,
'external_ids': {
ovn_const.OVN_ROUTER_IS_EXT_GW: 'true',
ovn_const.OVN_SUBNET_EXT_ID_KEY:
@ -1018,7 +1017,7 @@ class TestOvnNbSyncML2(test_mech_driver.OVNMechanismDriverTestCase):
'external_ids': {}}]
expected_added = []
expected_deleted = [{'nexthop': '90.0.0.1',
'destination': '0.0.0.0/0',
'destination': const.IPv4_ANY,
'external_ids': {
ovn_const.OVN_ROUTER_IS_EXT_GW: 'true',
ovn_const.OVN_SUBNET_EXT_ID_KEY:
@ -1047,7 +1046,7 @@ class TestOvnNbSyncML2(test_mech_driver.OVNMechanismDriverTestCase):
'destination': '12.0.0.0/24',
'external_ids': {}},
{'nexthop': '90.0.0.1',
'destination': '0.0.0.0/0',
'destination': const.IPv4_ANY,
'external_ids': {
ovn_const.OVN_ROUTER_IS_EXT_GW: 'true',
ovn_const.OVN_SUBNET_EXT_ID_KEY:
@ -1056,7 +1055,7 @@ class TestOvnNbSyncML2(test_mech_driver.OVNMechanismDriverTestCase):
'destination': '13.0.0.0/24',
'external_ids': {}}]
expected_added = [{'nexthop': '90.0.0.1',
'destination': '0.0.0.0/0',
'destination': const.IPv4_ANY,
'external_ids': {
ovn_const.OVN_ROUTER_IS_EXT_GW: 'true',
ovn_const.OVN_SUBNET_EXT_ID_KEY:
@ -1077,7 +1076,7 @@ class TestOvnNbSyncML2(test_mech_driver.OVNMechanismDriverTestCase):
'destination': '13.0.0.0/24',
'external_ids': {}},
{'nexthop': '90.0.0.1',
'destination': '0.0.0.0/0',
'destination': const.IPv4_ANY,
'external_ids': {
ovn_const.OVN_ROUTER_IS_EXT_GW: 'true',
ovn_const.OVN_SUBNET_EXT_ID_KEY:
@ -1098,7 +1097,7 @@ class TestOvnNbSyncML2(test_mech_driver.OVNMechanismDriverTestCase):
'destination': '13.0.0.0/24',
'external_ids': {}},
{'nexthop': '90.0.0.1',
'destination': '0.0.0.0/0',
'destination': const.IPv4_ANY,
'external_ids': {
ovn_const.OVN_ROUTER_IS_EXT_GW: 'true',
ovn_const.OVN_SUBNET_EXT_ID_KEY:

13
neutron/tests/unit/services/metering/test_metering_plugin.py
View File

@ -16,6 +16,7 @@ from unittest import mock
from neutron_lib.agent import topics
from neutron_lib.api.definitions import metering as metering_apidef
from neutron_lib import constants as n_const
from neutron_lib import context
from neutron_lib.plugins import constants
from neutron_lib.plugins import directory
@ -417,7 +418,8 @@ class TestMeteringPlugin(test_db_base_plugin_v2.NeutronDbPluginV2TestCase,
'_metering_labels': [
{'rule': {
'destination_ip_prefix':
net_utils.AuthenticIPNetwork('0.0.0.0/0'),
net_utils.AuthenticIPNetwork(
n_const.IPv4_ANY),
'source_ip_prefix':
net_utils.AuthenticIPNetwork(
'10.0.0.0/24'),
@ -438,7 +440,8 @@ class TestMeteringPlugin(test_db_base_plugin_v2.NeutronDbPluginV2TestCase,
'_metering_labels': [
{'rule': {
'destination_ip_prefix':
net_utils.AuthenticIPNetwork('0.0.0.0/0'),
net_utils.AuthenticIPNetwork(
n_const.IPv4_ANY),
'source_ip_prefix':
net_utils.AuthenticIPNetwork(
'10.0.0.0/24'),
@ -451,7 +454,7 @@ class TestMeteringPlugin(test_db_base_plugin_v2.NeutronDbPluginV2TestCase,
'id': self.uuid}]
ip_prefixes = {'source_ip_prefix': '10.0.0.0/24',
'destination_ip_prefix': '0.0.0.0/0'}
'destination_ip_prefix': n_const.IPv4_ANY}
with self.router():
with self.metering_label() as label:
la = label['metering_label']
@ -472,7 +475,7 @@ class TestMeteringPlugin(test_db_base_plugin_v2.NeutronDbPluginV2TestCase,
res = self._create_metering_label_rule(
self.fmt, la['id'], 'ingress', False,
remote_ip_prefix='0.0.0.0/0',
remote_ip_prefix=n_const.IPv4_ANY,
source_ip_prefix='10.0.0.0/24')
expected_error_code = 500
@ -494,7 +497,7 @@ class TestMeteringPlugin(test_db_base_plugin_v2.NeutronDbPluginV2TestCase,
res = self._create_metering_label_rule(
self.fmt, la['id'], 'ingress', False,
remote_ip_prefix='0.0.0.0/0',
remote_ip_prefix=n_const.IPv4_ANY,
destination_ip_prefix='8.8.8.8/32')
expected_error_code = 500

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

@ -655,7 +655,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
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',
mock.call('neutron-router-id', ip_prefix=constants.IPv4_ANY,
nexthop='192.168.1.254',
maintain_bfd=False,
external_ids={
@ -838,7 +838,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
'gw-port-id', 'lrp-gw-port-id', is_gw_port=True,
lsp_address=ovn_const.DEFAULT_ADDR_FOR_LSP_WITH_PEER)
self.l3_inst._nb_ovn.add_static_route.assert_called_once_with(
'neutron-router-id', ip_prefix='0.0.0.0/0',
'neutron-router-id', ip_prefix=constants.IPv4_ANY,
maintain_bfd=False,
external_ids={ovn_const.OVN_ROUTER_IS_EXT_GW: 'true',
ovn_const.OVN_SUBNET_EXT_ID_KEY: 'ext-subnet-id',
@ -910,7 +910,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
'gw-port-id', 'lrp-gw-port-id', is_gw_port=True,
lsp_address=ovn_const.DEFAULT_ADDR_FOR_LSP_WITH_PEER)
self.l3_inst._nb_ovn.add_static_route.assert_called_once_with(
'neutron-router-id', ip_prefix='0.0.0.0/0',
'neutron-router-id', ip_prefix=constants.IPv4_ANY,
nexthop='192.168.1.254',
maintain_bfd=False,
external_ids={ovn_const.OVN_ROUTER_IS_EXT_GW: 'true',
@ -965,7 +965,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
'gw-port-id', 'lrp-gw-port-id', is_gw_port=True,
lsp_address=ovn_const.DEFAULT_ADDR_FOR_LSP_WITH_PEER)
self.l3_inst._nb_ovn.add_static_route.assert_called_once_with(
'neutron-router-id', ip_prefix='0.0.0.0/0',
'neutron-router-id', ip_prefix=constants.IPv4_ANY,
nexthop='192.168.1.254',
maintain_bfd=False,
external_ids={ovn_const.OVN_ROUTER_IS_EXT_GW: 'true',
@ -1023,7 +1023,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
# Need not check lsp and lrp here, it has been tested in other cases
self.l3_inst._nb_ovn.add_static_route.assert_called_once_with(
'neutron-router-id', ip_prefix='0.0.0.0/0',
'neutron-router-id', ip_prefix=constants.IPv4_ANY,
maintain_bfd=False,
external_ids={ovn_const.OVN_ROUTER_IS_EXT_GW: 'true',
ovn_const.OVN_SUBNET_EXT_ID_KEY: 'ext-subnet-id',
@ -2173,10 +2173,11 @@ class OVNL3ExtrarouteTests(test_l3_gw.ExtGwModeIntTestCase,
super(OVNL3ExtrarouteTests, self). \
test_update_subnet_gateway_for_external_net()
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',
'neutron-fake_device', ip_prefix=constants.IPv4_ANY,
nexthop='120.0.0.2',
external_ids={'neutron:is_static_route': 'true'})
self.l3_inst._nb_ovn.delete_static_routes.assert_called_once_with(
'neutron-fake_device', [('0.0.0.0/0', '120.0.0.1')])
'neutron-fake_device', [(constants.IPv4_ANY, '120.0.0.1')])
def test_router_update_gateway_upon_subnet_create_max_ips_ipv6(self):
super(OVNL3ExtrarouteTests, self). \
@ -2186,8 +2187,9 @@ class OVNL3ExtrarouteTests(test_l3_gw.ExtGwModeIntTestCase,
ovn_const.OVN_SUBNET_EXT_ID_KEY: mock.ANY,
ovn_const.OVN_LRSR_EXT_ID_KEY: 'true'}
add_static_route_calls = [
mock.call(mock.ANY, ip_prefix='0.0.0.0/0', nexthop='10.0.0.1',
maintain_bfd=False, external_ids=expected_ext_ids),
mock.call(mock.ANY, ip_prefix=constants.IPv4_ANY,
nexthop='10.0.0.1', maintain_bfd=False,
external_ids=expected_ext_ids),
mock.call(mock.ANY, ip_prefix='::/0', nexthop='2001:db8::',
maintain_bfd=False, external_ids=expected_ext_ids)]
self.l3_inst._nb_ovn.add_static_route.assert_has_calls(

3
neutron/tests/unit/services/revisions/test_revision_plugin.py
View File

@ -13,6 +13,7 @@
# under the License.
#
from neutron_lib import constants as n_const
from neutron_lib import context as nctx
from neutron_lib.db import api as db_api
from neutron_lib.plugins import constants
@ -216,7 +217,7 @@ class TestRevisionPlugin(test_plugin.Ml2PluginV2TestCase):
r = {'security_group_rule': {'tenant_id': 'some_tenant',
'port_range_min': 80, 'protocol': 6,
'port_range_max': 90,
'remote_ip_prefix': '0.0.0.0/0',
'remote_ip_prefix': n_const.IPv4_ANY,
'ethertype': 'IPv4',
'remote_group_id': None,
'remote_address_group_id': None,