NSX|P: Update slaac config on router

In neutron, slaac is enabled per subnet. However on policy
slaac is router configuration.
In order to avoid another passthrough API, router will be
configured with slaac profile if at least one of its subnet
has slaac enabled.

Change-Id: I7a055aa4d73425011c22248c8d7d9d2e0a383dc8
This commit is contained in:
Anna Khmelnitsky 2019-03-28 16:45:35 -07:00
parent 59bf4b8a4c
commit ac06b1b985
7 changed files with 423 additions and 281 deletions

View File

@ -383,7 +383,7 @@ nsx_v3_and_p = [
"routers to connect other that the one connected to"
" the Tier0 router")),
cfg.ListOpt('transit_networks',
default=['100.64.0.0/16'],
default=['100.64.0.0/16', 'fc3d:e3c3:7b93::/48'],
help=_("List of transit networks used by NSX tier0 routers. "
"Neutron subnets will not be allowed to use those "
"cidrs")),

View File

@ -208,7 +208,8 @@ class NsxPluginBase(db_base_plugin_v2.NeutronDbPluginV2,
'subnetpool_id': subnet.subnetpool_id,
'ip_version': subnet.ip_version,
'network_id': subnet.network_id,
'gateway_ip': subnet.gateway_ip})
'gateway_ip': subnet.gateway_ip,
'ipv6_address_mode': subnet.ipv6_address_mode})
return subnets
def _find_router_gw_subnets(self, context, router):

View File

@ -1907,17 +1907,43 @@ class NsxPluginV3Base(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
LOG.error(msg)
raise n_exc.InvalidInput(error_message=msg)
def _subnet_with_native_dhcp(self, subnet):
native_metadata = self._has_native_dhcp_metadata()
# DHCPv6 is not yet supported, but slaac is
# When configuring slaac, neutron requires the user
# to enable dhcp, however plugin code does not consider
# slaac as dhcp.
return (native_metadata and
subnet.get('enable_dhcp', False) and
subnet.get('ipv6_address_mode') != 'slaac')
def _validate_subnet_ip_version(self, subnet):
# This validation only needs to be called at create,
# since ip version and ipv6 mode attributes are read only
if subnet.get('ip_version') == 4:
# No dhcp restrictions for V4
return
enable_dhcp = subnet.get('enable_dhcp', False)
is_slaac = (subnet.get('ipv6_address_mode') == 'slaac')
if enable_dhcp and not is_slaac:
# No DHCPv6 support yet
msg = _("DHCPv6 is not supported")
LOG.error(msg)
raise n_exc.InvalidInput(error_message=msg)
def _create_subnet(self, context, subnet):
self._validate_number_of_subnet_static_routes(subnet)
self._validate_host_routes_input(subnet)
self._validate_subnet_ip_version(subnet['subnet'])
network = self._get_network(
context, subnet['subnet']['network_id'])
self._validate_single_ipv6_subnet(context, network, subnet['subnet'])
# TODO(berlin): public external subnet announcement
native_metadata = self._has_native_dhcp_metadata()
if (native_metadata and subnet['subnet'].get('enable_dhcp', False)):
if self._subnet_with_native_dhcp(subnet['subnet']):
self._validate_external_subnet(context,
subnet['subnet']['network_id'])
self._ensure_native_dhcp()
@ -2126,7 +2152,7 @@ class NsxPluginV3Base(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
self._subnet_check_ip_allocations_internal_router_ports(
context, subnet_id)
subnet = self.get_subnet(context, subnet_id)
if subnet['enable_dhcp']:
if self._subnet_with_native_dhcp(subnet):
lock = 'nsxv3_network_' + subnet['network_id']
with locking.LockManager.get_lock(lock):
# Check if it is the last DHCP-enabled subnet to delete.
@ -2159,9 +2185,10 @@ class NsxPluginV3Base(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
context, network, subnet['subnet'])
if self._has_native_dhcp_metadata():
enable_dhcp = subnet['subnet'].get('enable_dhcp')
enable_dhcp = self._subnet_with_native_dhcp(subnet['subnet'])
orig_enable_dhcp = self._subnet_with_native_dhcp(orig_subnet)
if (enable_dhcp is not None and
enable_dhcp != orig_subnet['enable_dhcp']):
enable_dhcp != orig_enable_dhcp):
self._ensure_native_dhcp()
lock = 'nsxv3_network_' + orig_subnet['network_id']
with locking.LockManager.get_lock(lock):
@ -2208,8 +2235,7 @@ class NsxPluginV3Base(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
context, subnet['subnet'], updated_subnet)
# Check if needs to update logical DHCP server for native DHCP.
if (self._has_native_dhcp_metadata() and
updated_subnet['enable_dhcp']):
if self._subnet_with_native_dhcp(updated_subnet):
self._ensure_native_dhcp()
kwargs = {}
for key in ('dns_nameservers', 'gateway_ip', 'host_routes'):
@ -2438,7 +2464,7 @@ class NsxPluginV3Base(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
def _has_no_dhcp_enabled_subnet(self, context, network):
# Check if there is no DHCP-enabled subnet in the network.
for subnet in network.subnets:
if subnet.enable_dhcp:
if subnet.enable_dhcp and subnet.ipv6_address_mode != 'slaac':
return False
return True
@ -2456,9 +2482,6 @@ class NsxPluginV3Base(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
return cidr0.first <= cidr1.last and cidr1.first <= cidr0.last
def _validate_address_space(self, context, subnet):
# Only working for IPv4 at the moment
if (subnet['ip_version'] != 4):
return
# get the subnet IPs
if ('allocation_pools' in subnet and

View File

@ -729,6 +729,38 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
return updated_net
def _update_slaac_on_router(self, context, router_id,
subnet, delete=False):
# TODO(annak): for vlan networks it should be possible
# to enable slaac on interface basis
# This code is optimised to deal with concurrency challenges
# (which can not be always solved by lock because the plugin
# can run on different hosts).
# We prefer to make another backend call for attaching the
# profile even if it is already attached, than rely on DB
# to have an accurate picture of existing subnets.
profile_id = None
rtr_subnets = self._find_router_subnets(context.elevated(),
router_id)
slaac_subnets = [s for s in rtr_subnets
if s['id'] != subnet['id'] and
s['ipv6_address_mode'] == 'slaac']
slaac_subnet = (subnet['ipv6_address_mode'] == 'slaac')
if slaac_subnet and not delete:
# slaac subnet connected - verify slaac is set on router
profile_id = SLAAC_NDRA_PROFILE_ID
if not slaac_subnets and slaac_subnet and delete:
# this was the last slaac subnet connected -
# need to disable slaac on router
profile_id = DEFAULT_NDRA_PROFILE_ID
if profile_id:
self.nsxpolicy.tier1.update(router_id,
ipv6_ndra_profile_id=profile_id)
def create_subnet(self, context, subnet):
return self._create_subnet(context, subnet)
@ -1665,6 +1697,10 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
name=net_name,
tier1_id=router_id,
subnets=pol_subnets)
# will update the router only if needed
self._update_slaac_on_router(context, router_id,
subnet)
else:
# Vlan interface
pol_subnets = []
@ -1766,6 +1802,9 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
self.nsxpolicy.tier1.remove_segment_interface(
router_id, segment_id)
# will update the router only if needed
self._update_slaac_on_router(context, router_id,
subnet, delete=True)
# try to delete the SNAT/NO_DNAT rules of this subnet
router_db = self._get_router(context, router_id)
if (subnet and router_db.gw_port and router_db.enable_snat and

View File

@ -19,6 +19,8 @@ import decorator
from neutron.tests.unit.db import test_db_base_plugin_v2 as test_plugin
from neutron_lib import constants
class FixExternalNetBaseTest(object):
"""Base class providing utilities for handling tests which require updating
@ -143,6 +145,26 @@ def with_external_subnet(f, *args, **kwargs):
return result
@decorator.decorator
def with_disable_dhcp(f, *args, **kwargs):
obj = args[0]
obj.force_disable_dhcp = True
result = f(*args, **kwargs)
obj.force_disable_dhcp = False
return result
@decorator.decorator
def with_disable_dhcp_once(f, *args, **kwargs):
obj = args[0]
obj.force_disable_dhcp = True
obj.force_disable_dhcp_once = True
result = f(*args, **kwargs)
obj.force_disable_dhcp = False
obj.force_disable_dhcp_once = False
return result
def init_subnet_calls(self, n):
self.subnet_calls = []
for i in range(0, n - 1):
@ -200,3 +222,215 @@ def with_no_dhcp_subnet(f, *args, **kwargs):
result = f(*args, **kwargs)
obj.subnet = obj.original_subnet
return result
# TODO(annak): remove this when DHCPv6 is supported
@decorator.decorator
def with_force_slaac(f, *args, **kwargs):
obj = args[0]
obj.force_slaac = True
result = f(*args, **kwargs)
obj.force_slaac = False
return result
class NsxV3SubnetMixin(object):
def setUp(self, *args, **kwargs):
super(NsxV3SubnetMixin, self).setUp(*args, **kwargs)
self.force_slaac = False
self.force_disable_dhcp = False
self.force_disable_dhcp_once = False
def _test_create_subnet(self, network=None, expected=None, **kwargs):
# Until DHCPv6 is supported, switch all test to slaac-only
if (self.force_slaac and
'ipv6_ra_mode' in kwargs and 'ipv6_address_mode' in kwargs):
kwargs['ipv6_ra_mode'] = constants.IPV6_SLAAC
kwargs['ipv6_address_mode'] = constants.IPV6_SLAAC
return super(NsxV3SubnetMixin,
self)._test_create_subnet(network, expected, **kwargs)
def _create_subnet(self, fmt, net_id, cidr,
expected_res_status=None, **kwargs):
if self.force_disable_dhcp:
kwargs['enable_dhcp'] = False
if self.force_disable_dhcp_once:
self.force_disable_dhcp = False
return super(NsxV3SubnetMixin, self)._create_subnet(
fmt, net_id, cidr, expected_res_status, **kwargs)
class NsxV3TestSubnets(NsxV3SubnetMixin,
test_plugin.TestSubnetsV2):
@with_disable_dhcp
def test_list_subnets_filtering_by_project_id(self):
super(NsxV3TestSubnets,
self).test_list_subnets_filtering_by_project_id()
@with_disable_dhcp
def test_list_subnets(self):
super(NsxV3TestSubnets, self).test_list_subnets()
@with_disable_dhcp
def test_list_subnets_with_parameter(self):
super(NsxV3TestSubnets, self).test_list_subnets_with_parameter()
def test_create_subnet_ipv6_gw_is_nw_end_addr_returns_201(self):
self.skipTest('No DHCP v6 Support yet')
def test_create_subnet_with_v6_pd_allocation_pool(self):
self.skipTest('No DHCP v6 Support yet')
def test_create_subnet_with_v6_allocation_pool(self):
self.skipTest('No DHCP v6 Support yet')
def test_create_subnet_ipv6_pd_gw_values(self):
self.skipTest('No DHCP v6 Support yet')
def test_create_subnet_ipv6_slaac_with_dhcp_port_on_network(self):
self.skipTest('No DHCP v6 Support yet')
def test_create_subnet_ipv6_slaac_with_port_not_found(self):
self.skipTest('No DHCP v6 Support yet')
@with_disable_dhcp
def test_update_subnet_inconsistent_ipv6_hostroute_dst_v4(self):
super(NsxV3TestSubnets,
self).test_update_subnet_inconsistent_ipv6_hostroute_dst_v4()
@with_disable_dhcp
def test_create_two_subnets(self):
super(NsxV3TestSubnets, self).test_create_two_subnets()
@with_disable_dhcp
def test_create_subnets_bulk_emulated(self):
super(NsxV3TestSubnets, self).test_create_subnets_bulk_emulated()
@with_disable_dhcp
def test_create_subnets_bulk_native(self):
super(NsxV3TestSubnets, self).test_create_subnets_bulk_native()
@with_disable_dhcp
def test_get_subnets_count(self):
super(NsxV3TestSubnets, self).test_get_subnets_count()
@with_disable_dhcp
def test_get_subnets_count_filter_by_project_id(self):
super(NsxV3TestSubnets,
self).test_get_subnets_count_filter_by_project_id()
@with_disable_dhcp
def test_get_subnets_count_filter_by_unknown_filter(self):
super(NsxV3TestSubnets,
self).test_get_subnets_count_filter_by_unknown_filter()
@with_disable_dhcp
def test_delete_subnet_dhcp_port_associated_with_other_subnets(self):
super(NsxV3TestSubnets,
self).test_get_subnets_count_filter_by_unknown_filter()
@with_disable_dhcp
def test_delete_subnet_with_other_subnet_on_network_still_in_use(self):
super(NsxV3TestSubnets, self).\
test_delete_subnet_with_other_subnet_on_network_still_in_use()
@with_force_slaac
def test_create_subnet_ipv6_gw_values(self):
super(NsxV3TestSubnets, self).test_create_subnet_ipv6_gw_values()
@with_force_slaac
def test_create_subnet_ipv6_out_of_cidr_global(self):
super(NsxV3TestSubnets,
self).test_create_subnet_ipv6_out_of_cidr_global()
@with_disable_dhcp
def test_update_subnet_inconsistent_ipv6_gatewayv4(self):
super(NsxV3TestSubnets,
self).test_update_subnet_inconsistent_ipv6_gatewayv4()
@with_disable_dhcp
def test_update_subnet_inconsistent_ipv6_hostroute_np_v4(self):
super(NsxV3TestSubnets,
self).test_update_subnet_inconsistent_ipv6_hostroute_np_v4()
def test_update_subnet_ipv6_ra_mode_fails(self):
self.skipTest('No DHCP v6 Support yet')
def test_create_subnet_dhcpv6_stateless_with_port_on_network(self):
self.skipTest('No DHCP v6 Support yet')
def test_delete_subnet_port_exists_owned_by_network(self):
self.skipTest('No support for multiple ips')
def test_create_subnet_dhcpv6_stateless_with_ip_already_allocated(self):
self.skipTest('No DHCP v6 Support yet')
def test_create_subnet_ipv6_slaac_with_db_reference_error(self):
self.skipTest('No DHCP v6 Support yet')
def test_create_subnet_ipv6_slaac_with_ip_already_allocated(self):
self.skipTest('No DHCP v6 Support yet')
def test_subnet_update_ipv4_and_ipv6_pd_v6stateless_subnets(self):
self.skipTest('Multiple fixed ips on a port are not supported')
def test_subnet_update_ipv4_and_ipv6_pd_slaac_subnets(self):
self.skipTest('Multiple fixed ips on a port are not supported')
class NsxV3TestPorts(test_plugin.TestPortsV2):
def test_create_port_with_ipv6_dhcp_stateful_subnet_in_fixed_ips(self):
self.skipTest('No DHCP v6 Support yet')
def test_update_port_update_ip_address_only(self):
self.skipTest('Multiple fixed ips on a port are not supported')
def test_update_port_with_new_ipv6_slaac_subnet_in_fixed_ips(self):
self.skipTest('Multiple fixed ips on a port are not supported')
def test_update_port_mac_v6_slaac(self):
self.skipTest('Multiple fixed ips on a port are not supported')
def test_requested_invalid_fixed_ips(self):
self.skipTest('Multiple fixed ips on a port are not supported')
def test_requested_subnet_id_v4_and_v6_slaac(self):
self.skipTest('Multiple fixed ips on a port are not supported')
def test_range_allocation(self):
self.skipTest('Multiple fixed ips on a port are not supported')
def test_create_port_anticipating_allocation(self):
self.skipTest('Multiple fixed ips on a port are not supported')
def test_update_port_add_additional_ip(self):
self.skipTest('Multiple fixed ips on a port are not supported')
def test_delete_network_port_exists_owned_by_network_race(self):
self.skipTest('Skip need to address in future')
def test_delete_network_port_exists_owned_by_network_port_not_found(self):
self.skipTest('Skip need to address in future')
def test_delete_network_port_exists_owned_by_network(self):
self.skipTest('Skip need to address in future')
def test_duplicate_mac_generation(self):
self.skipTest('No DHCP v6 Support yet')
@with_disable_dhcp
def test_update_port_update_ip(self):
return super(NsxV3TestPorts, self).test_update_port_update_ip()
def test_create_router_port_ipv4_and_ipv6_slaac_no_fixed_ips(self):
self.skipTest('No DHCP v6 Support yet')
def test_ip_allocation_for_ipv6_2_subnet_slaac_mode(self):
self.skipTest('Only one ipv6 subnet per network is supported')
def test_create_port_with_multiple_ipv4_and_ipv6_subnets(self):
self.skipTest('Only one ipv6 subnet per network is supported')

View File

@ -15,8 +15,6 @@
import mock
import decorator
from oslo_config import cfg
from oslo_utils import uuidutils
from webob import exc
@ -594,86 +592,13 @@ class NsxPTestNetworks(test_db_base_plugin_v2.TestNetworksV2,
network['id'], data)
class NsxPTestPorts(test_db_base_plugin_v2.TestPortsV2,
class NsxPTestPorts(common_v3.NsxV3TestPorts,
common_v3.NsxV3SubnetMixin,
NsxPPluginTestCaseMixin):
def setUp(self, **kwargs):
super(NsxPTestPorts, self).setUp(**kwargs)
self.disable_dhcp = False
def _make_subnet(self, *args, **kwargs):
"""Override the original make_subnet to control the DHCP status"""
if self.disable_dhcp:
if 'enable_dhcp' in kwargs:
kwargs['enable_dhcp'] = False
else:
if len(args) > 7:
arg_list = list(args)
arg_list[7] = False
args = tuple(arg_list)
return super(NsxPTestPorts, self)._make_subnet(*args, **kwargs)
@decorator.decorator
def with_disable_dhcp(f, *args, **kwargs):
"""Change the default subnet DHCP status to disable.
This is used to allow tests with 2 subnets on the same net
"""
obj = args[0]
obj.disable_dhcp = True
result = f(*args, **kwargs)
obj.disable_dhcp = False
return result
def test_update_port_update_ip_address_only(self):
self.skipTest('Multiple fixed ips on a port are not supported')
def test_update_port_with_new_ipv6_slaac_subnet_in_fixed_ips(self):
self.skipTest('Multiple fixed ips on a port are not supported')
def test_update_port_mac_v6_slaac(self):
self.skipTest('Multiple fixed ips on a port are not supported')
@with_disable_dhcp
def test_requested_subnet_id_v4_and_v6(self):
return super(NsxPTestPorts, self).test_requested_subnet_id_v4_and_v6()
def test_requested_invalid_fixed_ips(self):
self.skipTest('Multiple fixed ips on a port are not supported')
def test_requested_subnet_id_v4_and_v6_slaac(self):
self.skipTest('Multiple fixed ips on a port are not supported')
def test_range_allocation(self):
self.skipTest('Multiple fixed ips on a port are not supported')
def test_create_port_anticipating_allocation(self):
self.skipTest('Multiple fixed ips on a port are not supported')
def test_update_port_add_additional_ip(self):
self.skipTest('Multiple fixed ips on a port are not supported')
def test_delete_network_port_exists_owned_by_network_race(self):
self.skipTest('Skip need to address in future')
def test_delete_network_port_exists_owned_by_network_port_not_found(self):
self.skipTest('Skip need to address in future')
def test_delete_network_port_exists_owned_by_network(self):
self.skipTest('Skip need to address in future')
@with_disable_dhcp
def test_duplicate_mac_generation(self):
self.skipTest('No DHCP v6 Support yet')
return super(NsxPTestPorts, self).test_duplicate_mac_generation()
@with_disable_dhcp
def test_update_port_update_ip(self):
return super(NsxPTestPorts, self).test_update_port_update_ip()
def test_create_router_port_ipv4_and_ipv6_slaac_no_fixed_ips(self):
self.skipTest('No DHCP v6 Support yet')
@with_disable_dhcp
@common_v3.with_disable_dhcp
def test_create_port_with_multiple_ipv4_and_ipv6_subnets(self):
return super(
NsxPTestPorts,
@ -688,41 +613,45 @@ class NsxPTestPorts(test_db_base_plugin_v2.TestPortsV2,
def test_update_port_excluding_ipv6_slaac_subnet_from_fixed_ips(self):
self.skipTest('No DHCP v6 Support yet')
@with_disable_dhcp
@common_v3.with_disable_dhcp
def test_requested_subnet_id_v4_and_v6(self):
return super(NsxPTestPorts, self).test_requested_subnet_id_v4_and_v6()
@common_v3.with_disable_dhcp
def test_requested_ips_only(self):
return super(NsxPTestPorts, self).test_requested_ips_only()
@with_disable_dhcp
@common_v3.with_disable_dhcp
def test_list_ports_with_sort_emulated(self):
return super(NsxPTestPorts,
self).test_list_ports_with_sort_emulated()
@with_disable_dhcp
@common_v3.with_disable_dhcp
def test_list_ports_with_pagination_native(self):
return super(NsxPTestPorts,
self).test_list_ports_with_pagination_native()
@with_disable_dhcp
@common_v3.with_disable_dhcp
def test_list_ports_for_network_owner(self):
return super(NsxPTestPorts, self).test_list_ports_for_network_owner()
@with_disable_dhcp
@common_v3.with_disable_dhcp
def test_list_ports_public_network(self):
return super(NsxPTestPorts, self).test_list_ports_public_network()
@with_disable_dhcp
@common_v3.with_disable_dhcp
def test_list_ports(self):
return super(NsxPTestPorts, self).test_list_ports()
@with_disable_dhcp
@common_v3.with_disable_dhcp
def test_get_ports_count(self):
return super(NsxPTestPorts, self).test_get_ports_count()
@with_disable_dhcp
@common_v3.with_disable_dhcp
def test_list_ports_with_sort_native(self):
return super(NsxPTestPorts, self).test_list_ports_with_sort_native()
@with_disable_dhcp
@common_v3.with_disable_dhcp
def test_list_ports_with_pagination_emulated(self):
return super(NsxPTestPorts,
self).test_list_ports_with_pagination_emulated()
@ -1115,11 +1044,11 @@ class NsxPTestPorts(test_db_base_plugin_v2.TestPortsV2,
self.assertEqual(res.status_int, exc.HTTPBadRequest.code)
class NsxPTestSubnets(test_db_base_plugin_v2.TestSubnetsV2,
class NsxPTestSubnets(common_v3.NsxV3TestSubnets,
NsxPPluginTestCaseMixin):
def setUp(self, plugin=PLUGIN_NAME, ext_mgr=None):
super(NsxPTestSubnets, self).setUp(plugin=plugin, ext_mgr=ext_mgr)
self.disable_dhcp = False
self.force_slaac = False
def _create_subnet_bulk(self, fmt, number, net_id, name,
ip_version=4, **kwargs):
@ -1135,103 +1064,15 @@ class NsxPTestSubnets(test_db_base_plugin_v2.TestSubnetsV2,
kwargs.update({'override': overrides})
return self._create_bulk(fmt, number, 'subnet', base_data, **kwargs)
def _make_subnet(self, *args, **kwargs):
"""Override the original make_subnet to control the DHCP status"""
if self.disable_dhcp:
if 'enable_dhcp' in kwargs:
kwargs['enable_dhcp'] = False
else:
if len(args) > 7:
arg_list = list(args)
arg_list[7] = False
args = tuple(arg_list)
return super(NsxPTestSubnets, self)._make_subnet(*args, **kwargs)
def _test_create_subnet(self, network=None, expected=None, **kwargs):
# Until DHCPv6 is supported, switch all test to slaac-only
if (self.force_slaac and
'ipv6_ra_mode' in kwargs and 'ipv6_address_mode' in kwargs):
kwargs['ipv6_ra_mode'] = constants.IPV6_SLAAC
kwargs['ipv6_address_mode'] = constants.IPV6_SLAAC
@decorator.decorator
def with_disable_dhcp(f, *args, **kwargs):
"""Change the default subnet DHCP status to disable.
This is used to allow tests with 2 subnets on the same net
"""
obj = args[0]
obj.disable_dhcp = True
result = f(*args, **kwargs)
obj.disable_dhcp = False
return result
@with_disable_dhcp
def test_list_subnets_filtering_by_project_id(self):
super(NsxPTestSubnets,
self).test_list_subnets_filtering_by_project_id()
@with_disable_dhcp
def test_list_subnets(self):
super(NsxPTestSubnets, self).test_list_subnets()
@with_disable_dhcp
def test_list_subnets_with_parameter(self):
super(NsxPTestSubnets, self).test_list_subnets_with_parameter()
@with_disable_dhcp
def test_create_two_subnets(self):
super(NsxPTestSubnets, self).test_create_two_subnets()
@with_disable_dhcp
def test_create_subnets_bulk_emulated(self):
super(NsxPTestSubnets, self).test_create_subnets_bulk_emulated()
@with_disable_dhcp
def test_create_subnets_bulk_native(self):
super(NsxPTestSubnets, self).test_create_subnets_bulk_native()
@with_disable_dhcp
def test_get_subnets_count(self):
super(NsxPTestSubnets, self).test_get_subnets_count()
@with_disable_dhcp
def test_get_subnets_count_filter_by_project_id(self):
super(NsxPTestSubnets,
self).test_get_subnets_count_filter_by_project_id()
@with_disable_dhcp
def test_get_subnets_count_filter_by_unknown_filter(self):
super(NsxPTestSubnets,
self).test_get_subnets_count_filter_by_unknown_filter()
@with_disable_dhcp
def test_delete_subnet_dhcp_port_associated_with_other_subnets(self):
super(NsxPTestSubnets,
self).test_get_subnets_count_filter_by_unknown_filter()
@with_disable_dhcp
def _test_create_subnet_ipv6_auto_addr_with_port_on_network(
self, *args, **kwargs):
super(NsxPTestSubnets,
self)._test_create_subnet_ipv6_auto_addr_with_port_on_network(
*args, **kwargs)
@with_disable_dhcp
def test_delete_subnet_with_other_subnet_on_network_still_in_use(self):
super(NsxPTestSubnets, self).\
test_delete_subnet_with_other_subnet_on_network_still_in_use()
def test_delete_subnet_port_exists_owned_by_network(self):
self.skipTest('No support for multiple ips')
def test_create_subnet_dhcpv6_stateless_with_ip_already_allocated(self):
self.skipTest('No DHCP v6 Support yet')
def test_create_subnet_ipv6_slaac_with_db_reference_error(self):
self.skipTest('No DHCP v6 Support yet')
def test_create_subnet_ipv6_slaac_with_ip_already_allocated(self):
self.skipTest('No DHCP v6 Support yet')
def test_subnet_update_ipv4_and_ipv6_pd_v6stateless_subnets(self):
self.skipTest('Multiple fixed ips on a port are not supported')
def test_subnet_update_ipv4_and_ipv6_pd_slaac_subnets(self):
self.skipTest('Multiple fixed ips on a port are not supported')
return super(NsxPTestSubnets,
self)._test_create_subnet(network, expected, **kwargs)
def test_create_external_subnet_with_conflicting_t0_address(self):
with self._create_l3_ext_network() as network:
@ -1268,6 +1109,11 @@ class NsxPTestSubnets(test_db_base_plugin_v2.TestSubnetsV2,
self.plugin.create_subnet(
context.get_admin_context(), data)
@common_v3.with_disable_dhcp_once
def test_create_subnet_ipv6_slaac_with_port_on_network(self):
super(NsxPTestSubnets,
self).test_create_subnet_ipv6_slaac_with_port_on_network()
class NsxPTestSecurityGroup(common_v3.FixExternalNetBaseTest,
NsxPPluginTestCaseMixin,
@ -1406,6 +1252,7 @@ class NsxPTestL3ExtensionManager(object):
class NsxPTestL3NatTest(common_v3.FixExternalNetBaseTest,
common_v3.NsxV3SubnetMixin,
NsxPPluginTestCaseMixin,
test_l3_plugin.L3BaseForIntTests,
test_address_scope.AddressScopeTestCase):
@ -1465,31 +1312,6 @@ class NsxPTestL3NatTestCase(NsxPTestL3NatTest,
def setUp(self, *args, **kwargs):
super(NsxPTestL3NatTestCase, self).setUp(*args, **kwargs)
self.disable_dhcp = False
def _make_subnet(self, *args, **kwargs):
"""Override the original make_subnet to control the DHCP status"""
if self.disable_dhcp:
if 'enable_dhcp' in kwargs:
kwargs['enable_dhcp'] = False
else:
if len(args) > 7:
arg_list = list(args)
arg_list[7] = False
args = tuple(arg_list)
return super(NsxPTestL3NatTestCase, self)._make_subnet(*args, **kwargs)
@decorator.decorator
def with_disable_dhcp(f, *args, **kwargs):
"""Change the default subnet DHCP status to disable.
This is used to allow tests with 2 subnets on the same net
"""
obj = args[0]
obj.disable_dhcp = True
result = f(*args, **kwargs)
obj.disable_dhcp = False
return result
def test__notify_gateway_port_ip_changed(self):
self.skipTest('not supported')
@ -1572,7 +1394,7 @@ class NsxPTestL3NatTestCase(NsxPTestL3NatTest,
def test_router_delete_dhcpv6_stateless_subnet_inuse_returns_409(self):
self.skipTest('not supported')
@with_disable_dhcp
@common_v3.with_disable_dhcp
@common_v3.with_external_network
def test_router_update_gateway_upon_subnet_create_ipv6(self):
super(NsxPTestL3NatTestCase,
@ -1585,7 +1407,7 @@ class NsxPTestL3NatTestCase(NsxPTestL3NatTest,
self.skipTest('not supported')
def test_router_add_interface_ipv6_subnet(self):
self.skipTest('slaac not supported')
self.skipTest('DHCPv6 not supported')
def test_router_add_dual_stack_subnets(self):
"""Add dual stack subnets to router interface"""
@ -1624,35 +1446,36 @@ class NsxPTestL3NatTestCase(NsxPTestL3NatTest,
def test_router_add_interface_ipv6_single_subnet(self):
with self.router() as r, self.network() as n:
with self.subnet(network=n, cidr='fd00::1/64',
gateway_ip='fd00::1', ip_version=6) as s:
gateway_ip='fd00::1', ip_version=6,
enable_dhcp=False) as s:
self._test_router_add_interface_subnet(r, s)
@with_disable_dhcp
@common_v3.with_disable_dhcp
def test_route_clear_routes_with_None(self):
super(NsxPTestL3NatTestCase,
self).test_route_clear_routes_with_None()
@with_disable_dhcp
@common_v3.with_disable_dhcp
def test_route_update_with_multi_routes(self):
super(NsxPTestL3NatTestCase,
self).test_route_update_with_multi_routes()
@with_disable_dhcp
@common_v3.with_disable_dhcp
def test_route_update_with_one_route(self):
super(NsxPTestL3NatTestCase,
self).test_route_update_with_one_route()
@with_disable_dhcp
@common_v3.with_disable_dhcp
def test_router_update_delete_routes(self):
super(NsxPTestL3NatTestCase,
self).test_router_update_delete_routes()
@with_disable_dhcp
@common_v3.with_disable_dhcp
def test_router_interface_in_use_by_route(self):
super(NsxPTestL3NatTestCase,
self).test_router_interface_in_use_by_route()
@with_disable_dhcp
@common_v3.with_disable_dhcp
def test_create_floatingip_with_assoc_to_ipv4_and_ipv6_port(self):
super(NsxPTestL3NatTestCase,
self).test_create_floatingip_with_assoc_to_ipv4_and_ipv6_port()
@ -1993,3 +1816,26 @@ class NsxPTestL3NatTestCase(NsxPTestL3NatTest,
p['port']['id'],
expected_code=exc.HTTPBadRequest.code)
self.assertIn('NeutronError', res)
@common_v3.with_disable_dhcp
def test_create_floatingip_with_assoc_to_ipv6_subnet(self):
super(NsxPTestL3NatTestCase,
self).test_create_floatingip_with_assoc_to_ipv6_subnet()
@common_v3.with_disable_dhcp
def test_router_add_interface_ipv6_subnet_without_gateway_ip(self):
super(NsxPTestL3NatTestCase,
self).test_router_add_interface_ipv6_subnet_without_gateway_ip()
@common_v3.with_disable_dhcp
def test_router_add_interface_multiple_ipv6_subnets_different_net(self):
super(NsxPTestL3NatTestCase, self).\
test_router_add_interface_multiple_ipv6_subnets_different_net()
@common_v3.with_disable_dhcp
def test_create_floatingip_ipv6_only_network_returns_400(self):
super(NsxPTestL3NatTestCase,
self).test_create_floatingip_ipv6_only_network_returns_400()
def test_router_add_iface_ipv6_ext_ra_subnet_returns_400(self):
self.skipTest('DHCPv6 not supported')

View File

@ -13,8 +13,6 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import decorator
import mock
import netaddr
from neutron.db import models_v2
@ -799,7 +797,10 @@ class TestNetworksV2(test_plugin.TestNetworksV2, NsxV3PluginTestCaseMixin):
res['NeutronError']['type'])
class TestSubnetsV2(test_plugin.TestSubnetsV2, NsxV3PluginTestCaseMixin):
class TestSubnetsV2(common_v3.NsxV3TestSubnets, NsxV3PluginTestCaseMixin):
def setUp(self, plugin=PLUGIN_NAME, ext_mgr=None):
super(TestSubnetsV2, self).setUp(plugin=plugin, ext_mgr=ext_mgr)
def test_create_subnet_with_shared_address_space(self):
with self.network() as network:
@ -847,12 +848,6 @@ class TestSubnetsV2(test_plugin.TestSubnetsV2, NsxV3PluginTestCaseMixin):
self.plugin.create_subnet,
context.get_admin_context(), data)
def test_subnet_update_ipv4_and_ipv6_pd_v6stateless_subnets(self):
self.skipTest('Multiple fixed ips on a port are not supported')
def test_subnet_update_ipv4_and_ipv6_pd_slaac_subnets(self):
self.skipTest('Multiple fixed ips on a port are not supported')
def test_subnet_native_dhcp_subnet_enabled(self):
self._enable_native_dhcp_md()
with self.network() as network:
@ -982,7 +977,8 @@ class TestSubnetsV2(test_plugin.TestSubnetsV2, NsxV3PluginTestCaseMixin):
context.get_admin_context(), subnet['id'], data)
class TestPortsV2(test_plugin.TestPortsV2, NsxV3PluginTestCaseMixin,
class TestPortsV2(common_v3.NsxV3SubnetMixin,
common_v3.NsxV3TestPorts, NsxV3PluginTestCaseMixin,
test_bindings.PortBindingsTestCase,
test_bindings.PortBindingsHostTestCaseMixin,
test_bindings.PortBindingsVnicTestCaseMixin):
@ -1714,33 +1710,6 @@ class TestPortsV2(test_plugin.TestPortsV2, NsxV3PluginTestCaseMixin,
self.assertEqual('InvalidInput',
res['NeutronError']['type'])
def test_update_port_update_ip_address_only(self):
self.skipTest('Multiple fixed ips on a port are not supported')
def test_update_port_with_new_ipv6_slaac_subnet_in_fixed_ips(self):
self.skipTest('Multiple fixed ips on a port are not supported')
def test_update_port_mac_v6_slaac(self):
self.skipTest('Multiple fixed ips on a port are not supported')
def test_requested_invalid_fixed_ips(self):
self.skipTest('Multiple fixed ips on a port are not supported')
def test_requested_subnet_id_v4_and_v6_slaac(self):
self.skipTest('Multiple fixed ips on a port are not supported')
def test_range_allocation(self):
self.skipTest('Multiple fixed ips on a port are not supported')
def test_create_port_anticipating_allocation(self):
self.skipTest('Multiple fixed ips on a port are not supported')
def test_ip_allocation_for_ipv6_2_subnet_slaac_mode(self):
self.skipTest('Only one ipv6 subnet per network is supported')
def test_create_port_with_multiple_ipv4_and_ipv6_subnets(self):
self.skipTest('Only one ipv6 subnet per network is supported')
def test_create_compute_port_with_relay_no_router(self):
"""Compute port creation should fail
@ -1880,6 +1849,10 @@ class TestPortsV2(test_plugin.TestPortsV2, NsxV3PluginTestCaseMixin,
**kwargs)
self.assertEqual(res.status_int, exc.HTTPBadRequest.code)
@common_v3.with_disable_dhcp
def test_requested_subnet_id_v4_and_v6(self):
return super(TestPortsV2, self).test_requested_subnet_id_v4_and_v6()
class DHCPOptsTestCase(test_dhcpopts.TestExtraDhcpOpt,
NsxV3PluginTestCaseMixin):
@ -1922,8 +1895,10 @@ class TestL3ExtensionManager(object):
return []
class L3NatTest(test_l3_plugin.L3BaseForIntTests, NsxV3PluginTestCaseMixin,
class L3NatTest(test_l3_plugin.L3BaseForIntTests,
NsxV3PluginTestCaseMixin,
common_v3.FixExternalNetBaseTest,
common_v3.NsxV3SubnetMixin,
test_address_scope.AddressScopeTestCase):
def setUp(self, plugin=PLUGIN_NAME, ext_mgr=None,
@ -2042,6 +2017,16 @@ class TestL3NatTestCase(L3NatTest,
super(TestL3NatTestCase,
self).test_router_update_gateway_with_different_external_subnet()
@common_v3.with_disable_dhcp
def test_create_floatingip_ipv6_only_network_returns_400(self):
super(TestL3NatTestCase,
self).test_create_floatingip_ipv6_only_network_returns_400()
@common_v3.with_disable_dhcp
def test_create_floatingip_with_assoc_to_ipv4_and_ipv6_port(self):
super(L3NatTest,
self).test_create_floatingip_with_assoc_to_ipv4_and_ipv6_port()
@common_v3.with_external_subnet_once
def test_router_update_gateway_with_existed_floatingip(self):
with self.subnet(cidr='20.0.0.0/24') as subnet:
@ -2130,6 +2115,31 @@ class TestL3NatTestCase(L3NatTest,
def test_floatingip_via_router_interface_returns_404(self):
self.skipTest('not supported')
def test_router_delete_dhcpv6_stateless_subnet_inuse_returns_409(self):
self.skipTest('DHCPv6 not supported')
@common_v3.with_disable_dhcp
def test_router_add_interface_ipv6_subnet(self):
self.skipTest('DHCPv6 not supported')
@common_v3.with_disable_dhcp
def test_router_add_interface_ipv6_subnet_without_gateway_ip(self):
super(TestL3NatTestCase,
self).test_router_add_interface_ipv6_subnet_without_gateway_ip()
@common_v3.with_disable_dhcp
def test_router_add_interface_multiple_ipv6_subnets_different_net(self):
super(TestL3NatTestCase, self).\
test_router_add_interface_multiple_ipv6_subnets_different_net()
@common_v3.with_disable_dhcp
def test_create_floatingip_with_assoc_to_ipv6_subnet(self):
super(TestL3NatTestCase,
self).test_create_floatingip_with_assoc_to_ipv6_subnet()
def test_router_add_iface_ipv6_ext_ra_subnet_returns_400(self):
self.skipTest('DHCPv6 not supported')
@common_v3.with_external_subnet
def test_floatingip_list_with_sort(self):
super(TestL3NatTestCase,
@ -3113,17 +3123,6 @@ class ExtGwModeTestCase(test_ext_gw_mode.ExtGwModeIntTestCase,
def test_router_gateway_set_fail_after_port_create(self):
self.skipTest("TBD")
# Override subnet/network creation in some tests to create external
# networks immediately instead of updating it post creation, which the
# v3 plugin does not support
@decorator.decorator
def with_external_subnet(f, *args, **kwargs):
obj = args[0]
obj.subnet = obj.external_subnet
result = f(*args, **kwargs)
obj.subnet = obj.original_subnet
return result
@common_v3.with_external_subnet
def _test_router_update_ext_gwinfo(self, snat_input_value,
snat_expected_value=False,