x
/
vmware-nsx
Code Issues Proposed changes

NSX|P: Support policy DHCP v6 

Change-Id: Ibe4936b6f0b64e67cb3c7838d0f1a16304eb1180
This commit is contained in:
asarfaty 2020-02-09 09:37:09 +02:00 committed by Adit Sarfaty
parent ea07bbc2fb
commit fd1ad40b6a
7 changed files with 972 additions and 253 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
vmware_nsx/plugins/common/plugin.py
View File

@ -231,6 +231,7 @@ class NsxPluginBase(db_base_plugin_v2.NeutronDbPluginV2,
'ip_version': subnet.ip_version,
'network_id': subnet.network_id,
'gateway_ip': subnet.gateway_ip,
'enable_dhcp': subnet.enable_dhcp,
'ipv6_address_mode': subnet.ipv6_address_mode}
for subnet in db_subnets]
return subnets

42
vmware_nsx/plugins/common_v3/plugin.py
View File

@ -1489,15 +1489,7 @@ class NsxPluginV3Base(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
LOG.warning(err_msg)
raise n_exc.InvalidInput(error_message=err_msg)
def _build_dhcp_server_config(self, context, network, subnet, port, az):
name = self.nsxlib.native_dhcp.build_server_name(
network['name'], network['id'])
net_tags = self.nsxlib.build_v3_tags_payload(
network, resource_type='os-neutron-net-id',
project_name=context.tenant_name)
def _get_network_dns_domain(self, az, network):
dns_domain = None
if network.get('dns_domain'):
net_dns = network['dns_domain']
@ -1507,6 +1499,18 @@ class NsxPluginV3Base(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
dns_domain = net_dns.dns_domain
if not dns_domain or not validators.is_attr_set(dns_domain):
dns_domain = az.dns_domain
return dns_domain
def _build_dhcp_server_config(self, context, network, subnet, port, az):
name = self.nsxlib.native_dhcp.build_server_name(
network['name'], network['id'])
net_tags = self.nsxlib.build_v3_tags_payload(
network, resource_type='os-neutron-net-id',
project_name=context.tenant_name)
dns_domain = self._get_network_dns_domain(az, network)
dns_nameservers = subnet['dns_nameservers']
if not dns_nameservers or not validators.is_attr_set(dns_nameservers):
@ -2032,15 +2036,15 @@ class NsxPluginV3Base(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
native_metadata = self._has_native_dhcp_metadata()
default_enable_dhcp = (orig_subnet.get('enable_dhcp', False)
if orig_subnet else False)
# DHCPv6 is not yet supported, but slaac is
# 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', default_enable_dhcp) and
subnet.get('ipv6_address_mode') != 'slaac')
subnet.get('ipv6_address_mode') != constants.IPV6_SLAAC)
def _validate_subnet_ip_version(self, subnet):
def _validate_mp_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:
@ -2048,10 +2052,9 @@ class NsxPluginV3Base(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
return
enable_dhcp = subnet.get('enable_dhcp', False)
is_slaac = (subnet.get('ipv6_address_mode') == 'slaac')
is_slaac = (subnet.get('ipv6_address_mode') == constants.IPV6_SLAAC)
if enable_dhcp and not is_slaac:
# No DHCPv6 support yet
# TODO(asarfaty): add ipv6 support for policy plugin
# No DHCPv6 support with the MP DHCP
msg = _("DHCPv6 is not supported")
LOG.error(msg)
raise n_exc.InvalidInput(error_message=msg)
@ -2085,7 +2088,7 @@ class NsxPluginV3Base(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
def _create_subnet_with_mp_dhcp(self, context, subnet):
self._validate_number_of_subnet_static_routes(subnet)
self._validate_host_routes_input(subnet)
self._validate_subnet_ip_version(subnet['subnet'])
self._validate_mp_subnet_ip_version(subnet['subnet'])
net_id = subnet['subnet']['network_id']
network = self._get_network(context, net_id)
@ -2505,6 +2508,7 @@ class NsxPluginV3Base(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
def _build_static_routes(self, gateway_ip, cidr, host_routes):
# The following code is based on _generate_opts_per_subnet() in
# neutron/agent/linux/dhcp.py. It prepares DHCP options for a subnet.
# This code is for IPv4 only (IPv6 dhcp does not support options)
# Add route for directly connected network.
static_routes = [{'network': cidr, 'next_hop': '0.0.0.0'}]
@ -2653,13 +2657,11 @@ 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 and subnet.ipv6_address_mode != 'slaac':
if (subnet.enable_dhcp and
subnet.ipv6_address_mode != constants.IPV6_SLAAC):
return False
return True
def _has_dhcp_enabled_subnet(self, context, network):
return not self._has_no_dhcp_enabled_subnet(context, network)
def _has_single_dhcp_enabled_subnet(self, context, network):
# Check if there is only one DHCP-enabled subnet in the network.
count = 0

610
vmware_nsx/plugins/nsx_p/plugin.py
View File

@ -128,6 +128,8 @@ SEG_SECURITY_PROFILE_ID = (
policy_defs.SegmentSecurityProfileDef.DEFAULT_PROFILE)
SLAAC_NDRA_PROFILE_ID = 'neutron-slaac-profile'
NO_SLAAC_NDRA_PROFILE_ID = 'neutron-no-slaac-profile'
STATELESS_DHCP_NDRA_PROFILE_ID = 'neutron-stateless-dhcp-profile'
STATEFUL_DHCP_NDRA_PROFILE_ID = 'neutron-stateful-dhcp-profile'
IPV6_RA_SERVICE = 'neutron-ipv6-ra'
IPV6_ROUTER_ADV_RULE_NAME = 'all-ipv6'
@ -364,9 +366,11 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
default_az = self.get_default_az()
if default_az.use_policy_dhcp:
self.use_policy_dhcp = True
LOG.info("The policy plugin will use policy based DHCP v4/6")
else:
self._init_native_dhcp()
self.use_policy_dhcp = False
LOG.info("The policy plugin will use MP based DHCP v4")
self._init_native_metadata()
@ -466,35 +470,29 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
SEG_SECURITY_PROFILE_ID)
raise nsx_exc.NsxPluginException(err_msg=msg)
# Ipv6 SLAAC NDRA profile (find it or create)
try:
self.nsxpolicy.ipv6_ndra_profile.get(SLAAC_NDRA_PROFILE_ID)
except nsx_lib_exc.ResourceNotFound:
try:
self.nsxpolicy.ipv6_ndra_profile.create_or_overwrite(
SLAAC_NDRA_PROFILE_ID,
profile_id=SLAAC_NDRA_PROFILE_ID,
ra_mode=policy_constants.IPV6_RA_MODE_SLAAC_RA,
tags=self.nsxpolicy.build_v3_api_version_tag())
except nsx_lib_exc.StaleRevision as e:
# This means that another controller is also creating this
LOG.info("Failed to configure ipv6_ndra_profile for SLAAC: %s",
e)
# Find or create all neutron NDRA profiles
ndra_profiles = {
SLAAC_NDRA_PROFILE_ID: policy_constants.IPV6_RA_MODE_SLAAC_RA,
STATELESS_DHCP_NDRA_PROFILE_ID:
policy_constants.IPV6_RA_MODE_SLAAC_DHCP,
STATEFUL_DHCP_NDRA_PROFILE_ID: policy_constants.IPV6_RA_MODE_DHCP,
NO_SLAAC_NDRA_PROFILE_ID: policy_constants.IPV6_RA_MODE_DISABLED
}
# Verify NO SLAAC NDRA profile (find it or create)
try:
self.nsxpolicy.ipv6_ndra_profile.get(NO_SLAAC_NDRA_PROFILE_ID)
except nsx_lib_exc.ResourceNotFound:
for profile in ndra_profiles:
try:
self.nsxpolicy.ipv6_ndra_profile.create_or_overwrite(
NO_SLAAC_NDRA_PROFILE_ID,
profile_id=NO_SLAAC_NDRA_PROFILE_ID,
ra_mode=policy_constants.IPV6_RA_MODE_DISABLED,
tags=self.nsxpolicy.build_v3_api_version_tag())
except nsx_lib_exc.StaleRevision as e:
# This means that another controller is also creating this
LOG.info("Failed to configure ipv6_ndra_profile for NO SLAAC: "
"%s", e)
self.nsxpolicy.ipv6_ndra_profile.get(profile)
except nsx_lib_exc.ResourceNotFound:
try:
self.nsxpolicy.ipv6_ndra_profile.create_or_overwrite(
profile,
profile_id=profile,
ra_mode=ndra_profiles[profile],
tags=self.nsxpolicy.build_v3_api_version_tag())
except nsx_lib_exc.StaleRevision as e:
# This means that another controller is also creating this
LOG.info("Failed to configure ipv6_ndra_profile %s: %s",
profile, e)
self.client_ssl_profile = None
@ -935,6 +933,23 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
return updated_net
def _get_subnets_nd_profile(self, subnets, additional_profile=None):
profiles = []
if additional_profile:
profiles.append(additional_profile)
for sub in subnets:
profiles.append(self._get_subnet_ndra_profile(sub))
# If there is 1 stateful/stateless DHCP subnet (cannot have both)
# use this profile
if STATEFUL_DHCP_NDRA_PROFILE_ID in profiles:
return STATEFUL_DHCP_NDRA_PROFILE_ID
elif STATELESS_DHCP_NDRA_PROFILE_ID in profiles:
return STATELESS_DHCP_NDRA_PROFILE_ID
elif SLAAC_NDRA_PROFILE_ID in profiles:
# if there is slaac subnet and no DHCP subnet use SLAAC
return SLAAC_NDRA_PROFILE_ID
return NO_SLAAC_NDRA_PROFILE_ID
def _update_slaac_on_router(self, context, router_id,
subnet, router_subnets, delete=False):
# TODO(annak): redesign when policy supports downlink-level
@ -946,33 +961,44 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
# 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
slaac_subnet = (subnet.get('ipv6_address_mode') == 'slaac')
# This method assumes that all the v6 subnets have the same
# ipv6_address_mode.
# Otherwise, earlier validation would already fail.
if slaac_subnet and not delete:
# slaac subnet connected - verify slaac is set on router
profile_id = SLAAC_NDRA_PROFILE_ID
if subnet.get('ip_version') == 4:
# This subnet will not affect the ND profile
return
# Fetch other overlay interface networks
# (VLAN advertising is attached on interface level)
ipv6_overlay_subnets = [s for s in router_subnets
if s['id'] != subnet['id'] and
s.get('ip_version') == 6 and
s.get('enable_dhcp') and
self._is_overlay_network(context,
s['network_id'])]
if delete:
router_subnets = self._load_router_subnet_cidrs_from_db(
context.elevated(), router_id)
# check if there is another slaac overlay subnet that needs
# advertising (vlan advertising is attached on interface level)
slaac_subnets = [s for s in router_subnets
if s['id'] != subnet['id'] and
s.get('ipv6_address_mode') == 'slaac' and
self._is_overlay_network(context,
s['network_id'])]
if not slaac_subnets and slaac_subnet:
# this was the last slaac subnet connected -
# 'subnet' was already removed from the router_subnets list before
# calling this method
if ipv6_overlay_subnets:
# If there is another ipv6 overlay - select the profile by its
# address mode
profile_id = self._get_subnets_nd_profile(ipv6_overlay_subnets)
else:
# this was the last ipv6 subnet connected -
# need to disable slaac on router
profile_id = NO_SLAAC_NDRA_PROFILE_ID
else:
profile_id = self._get_subnet_ndra_profile(subnet)
# Check the other subnets too
if (ipv6_overlay_subnets and
profile_id in [NO_SLAAC_NDRA_PROFILE_ID,
SLAAC_NDRA_PROFILE_ID]):
profile_id = self._get_subnets_nd_profile(
ipv6_overlay_subnets, additional_profile=profile_id)
if profile_id:
self.nsxpolicy.tier1.update(router_id,
ipv6_ndra_profile_id=profile_id)
self.nsxpolicy.tier1.update(router_id, ipv6_ndra_profile_id=profile_id)
def _validate_net_dhcp_edge_cluster(self, context, network, az):
"""Validate that the dhcp server edge cluster match the one of
@ -1000,15 +1026,25 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
raise n_exc.InvalidInput(error_message=msg)
def _create_subnet_dhcp_port(self, context, az, network, subnet):
port = self._get_net_dhcp_port(context, network['id'])
if port:
# If the port already exist (with another subnet) - update it with
# the additional ip
port['fixed_ips'].append({'subnet_id': subnet['id']})
super(NsxPolicyPlugin, self).update_port(
context, port['id'],
{'port': {'fixed_ips': port['fixed_ips']}})
return
port_data = {
"name": "",
"admin_state_up": True,
"device_id": network['id'],
"device_owner": const.DEVICE_OWNER_DHCP,
"network_id": network['id'],
"tenant_id": network["tenant_id"],
"tenant_id": network['tenant_id'],
"mac_address": const.ATTR_NOT_SPECIFIED,
"fixed_ips": [{"subnet_id": subnet['id']}],
"fixed_ips": [{'subnet_id': subnet['id']}],
psec.PORTSECURITY: False
}
# Create the DHCP port (on neutron only) and update its port security
@ -1020,13 +1056,23 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
self._process_portbindings_create_and_update(
context, port_data, neutron_port)
def _delete_subnet_dhcp_port(self, context, net_id):
dhcp_port = self._get_sunbet_dhcp_port(context, net_id)
def _delete_subnet_dhcp_port(self, context, net_id, subnet_id=None):
dhcp_port = self._get_net_dhcp_port(context, net_id)
if dhcp_port:
if subnet_id:
# deleting just this subnets dhcp
if len(dhcp_port['fixed_ips']) > 1:
new_fixed_ips = [ip for ip in dhcp_port['fixed_ips']
if ip['subnet_id'] != subnet_id]
super(NsxPolicyPlugin, self).update_port(
context, dhcp_port['id'],
{'port': {'fixed_ips': new_fixed_ips}})
return
# Delete the port itself
self.delete_port(context, dhcp_port['id'],
force_delete_dhcp=True)
def _get_sunbet_dhcp_port(self, context, net_id):
def _get_net_dhcp_port(self, context, net_id):
filters = {
'network_id': [net_id],
'device_owner': [const.DEVICE_OWNER_DHCP]
@ -1035,7 +1081,7 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
return dhcp_ports[0] if dhcp_ports else None
def _get_sunbet_dhcp_server_ip(self, context, net_id, dhcp_subnet_id):
dhcp_port = self._get_sunbet_dhcp_port(context, net_id)
dhcp_port = self._get_net_dhcp_port(context, net_id)
if dhcp_port:
dhcp_server_ips = [fip['ip_address']
for fip in dhcp_port['fixed_ips']
@ -1044,33 +1090,39 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
return dhcp_server_ips[0]
def _is_dhcp_network(self, context, net_id):
dhcp_port = self._get_sunbet_dhcp_port(context, net_id)
dhcp_port = self._get_net_dhcp_port(context, net_id)
return True if dhcp_port else False
def _get_segment_subnets(self, context, net_id, net_az=None,
interface_subnets=None, **kwargs):
"""Get list of segmentSubnet objects to put on the segment
interface_subnets=None,
deleted_dhcp_subnets=None):
"""Get an updated list of segmentSubnet objects to put on the segment
Including router interface subnets (for overlay networks) &
DHCP subnet (if using policy DHCP)
DHCP subnets (if using policy v4/v6 DHCP)
"""
dhcp_subnet = None
if 'dhcp_subnet' in kwargs:
dhcp_subnet = kwargs['dhcp_subnet']
else:
# Get it from the network
if self.use_policy_dhcp:
# TODO(asarfaty): Add ipv6 support
dhcp_subnets = []
if self.use_policy_dhcp:
# Find networks DHCP enabled subnets
with db_api.CONTEXT_READER.using(context):
network = self._get_network(context, net_id)
for subnet in network.subnets:
if subnet.enable_dhcp and subnet.ip_version == 4:
dhcp_subnet = self.get_subnet(context, subnet.id)
for subnet in network.subnets:
if(subnet.enable_dhcp and
(subnet.ip_version == 4 or
subnet.ipv6_address_mode != const.IPV6_SLAAC)):
if (deleted_dhcp_subnets and
subnet.id in deleted_dhcp_subnets):
# Skip this one as it is being deleted
continue
dhcp_subnets.append(self.get_subnet(context, subnet.id))
if len(dhcp_subnets) == 2:
# A network an have at most 2 DHCP subnets
break
router_subnets = []
if interface_subnets:
router_subnets = interface_subnets
else:
# Get it from the network, only if overlay
# Get networks overlay router interfaces
if self._is_overlay_network(context, net_id):
router_ids = self._get_network_router_ids(
context.elevated(), net_id)
@ -1081,31 +1133,39 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
seg_subnets = []
dhcp_subnet_id = None
if dhcp_subnet:
dhcp_subnet_ids = []
for dhcp_subnet in dhcp_subnets:
dhcp_subnet_id = dhcp_subnet['id']
dhcp_subnet_ids.append(dhcp_subnet_id)
gw_addr = self._get_gateway_addr_from_subnet(dhcp_subnet)
cidr_prefix = int(dhcp_subnet['cidr'].split('/')[1])
dhcp_server_ip = self._get_sunbet_dhcp_server_ip(
context, net_id, dhcp_subnet_id)
dns_nameservers = dhcp_subnet['dns_nameservers']
if not net_az:
net_az = self.get_network_az_by_net_id(context, net_id)
if (not dns_nameservers or
not validators.is_attr_set(dns_nameservers)):
# Use pre-configured dns server
if not net_az:
net_az = self.get_network_az_by_net_id(context, net_id)
dns_nameservers = net_az.nameservers
dhcp_config = policy_defs.SegmentDhcpConfig(
server_address="%s/%s" % (dhcp_server_ip, cidr_prefix),
dns_servers=dns_nameservers,
is_ipv6=False) # TODO(asarfaty): add ipv6 support
is_ipv6 = True if dhcp_subnet.get('ip_version') == 6 else False
server_ip = "%s/%s" % (dhcp_server_ip, cidr_prefix)
kwargs = {'server_address': server_ip,
'dns_servers': dns_nameservers}
if is_ipv6:
network = self._get_network(context, net_id)
kwargs['domain_names'] = [
self._get_network_dns_domain(net_az, network)]
dhcp_config = policy_defs.SegmentDhcpConfigV6(**kwargs)
else:
dhcp_config = policy_defs.SegmentDhcpConfigV4(**kwargs)
seg_subnet = policy_defs.Subnet(gateway_address=gw_addr,
dhcp_config=dhcp_config)
seg_subnets.append(seg_subnet)
for rtr_subnet in router_subnets:
if rtr_subnet['id'] == dhcp_subnet_id:
if rtr_subnet['id'] in dhcp_subnet_ids:
# Do not add the same subnet twice
continue
if rtr_subnet['network_id'] == net_id:
@ -1124,35 +1184,58 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
net_id = network['id']
segment_id = self._get_network_nsx_segment_id(context, net_id)
seg_subnets = self._get_segment_subnets(
context, net_id, net_az=az, dhcp_subnet=subnet)
seg_subnets = self._get_segment_subnets(context, net_id, net_az=az)
# Update dhcp server config on the segment
self.nsxpolicy.segment.update(
segment_id=segment_id,
dhcp_server_config_id=az._policy_dhcp_server_config,
subnets=seg_subnets)
def _disable_network_dhcp(self, context, network):
def _get_net_dhcp_subnets(self, context, net_id):
net_dhcp_subnets = []
net_obj = self._get_network(context, net_id)
for subnet in net_obj.subnets:
if(subnet.enable_dhcp and
(subnet.ip_version == 4 or
subnet.ipv6_address_mode != const.IPV6_SLAAC)):
# This is a DHCP subnet
net_dhcp_subnets.append(subnet.id)
return net_dhcp_subnets
def _disable_network_dhcp(self, context, network, subnet_id=None):
net_id = network['id']
net_dhcp_subnets = self._get_net_dhcp_subnets(context, net_id)
segment_id = self._get_network_nsx_segment_id(context, net_id)
# Remove dhcp server config from the segment
segment_id = self._get_network_nsx_segment_id(
context, net_id)
seg_subnets = self._get_segment_subnets(
context, net_id, dhcp_subnet=None)
self.nsxpolicy.segment.update(
segment_id,
subnets=seg_subnets,
dhcp_server_config_id=None)
if subnet_id and len(net_dhcp_subnets) > 1:
# remove dhcp only from this subnet
seg_subnets = self._get_segment_subnets(
context, net_id, deleted_dhcp_subnets=[subnet_id])
self.nsxpolicy.segment.update(
segment_id,
subnets=seg_subnets)
self._delete_subnet_dhcp_port(context, net_id, subnet_id=subnet_id)
else:
# Remove dhcp server config completly from the segment
seg_subnets = self._get_segment_subnets(
context, net_id, deleted_dhcp_subnets=net_dhcp_subnets)
self.nsxpolicy.segment.update(
segment_id=segment_id,
subnets=seg_subnets,
dhcp_server_config_id=None)
# Delete the neutron DHCP port (and its bindings)
self._delete_subnet_dhcp_port(context, net_id)
# Delete the neutron DHCP port (and its bindings)
self._delete_subnet_dhcp_port(context, net_id)
def _update_subnet_dhcp(self, context, network, subnet, az):
def _update_nsx_net_dhcp(self, context, network, az, subnet=None):
"""Update the DHCP config on a network
Update the segment DHCP config, as well as the dhcp bindings on the
ports.
If just a specific subnet was modified, update only its ports.
"""
net_id = network['id']
segment_id = self._get_network_nsx_segment_id(context, net_id)
seg_subnets = self._get_segment_subnets(
context, net_id, net_az=az, dhcp_subnet=subnet)
seg_subnets = self._get_segment_subnets(context, net_id, net_az=az)
filters = {'network_id': [net_id]}
ports = self.get_ports(context, filters=filters)
@ -1213,18 +1296,55 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
LOG.error(msg)
raise n_exc.InvalidInput(error_message=msg)
def _init_ipv6_gateway(self, subnet):
# Override neutron decision to verify that also for ipv6 the first
# ip in the cidr is not used, as the NSX does not support xxxx::0 as a
# segment subnet gateway.
if (subnet.get('gateway_ip') is const.ATTR_NOT_SPECIFIED and
subnet.get('ip_version') == const.IP_VERSION_6 and
subnet.get('cidr') and subnet['cidr'] != const.ATTR_NOT_SPECIFIED):
net = netaddr.IPNetwork(subnet['cidr'])
subnet['gateway_ip'] = str(net.network + 1)
def _validate_subnet_host_routes(self, subnet, orig_subnet=None):
self._validate_number_of_subnet_static_routes(subnet)
if orig_subnet:
self._validate_host_routes_input(
subnet,
orig_enable_dhcp=orig_subnet['enable_dhcp'],
orig_host_routes=orig_subnet['host_routes'])
else:
self._validate_host_routes_input(subnet)
# IPv6 subnets cannot support host routes
if (subnet['subnet'].get('ip_version') == 6 or
(orig_subnet and orig_subnet.get('ip_version') == 6)):
if (validators.is_attr_set(subnet['subnet'].get('host_routes')) and
subnet['subnet']['host_routes']):
err_msg = _("Host routes can only be supported with IPv4 "
"subnets")
raise n_exc.InvalidInput(error_message=err_msg)
def _has_dhcp_enabled_subnet(self, context, network, ip_version=4):
for subnet in network.subnets:
if subnet.enable_dhcp and subnet.ip_version == ip_version:
if ip_version == 4:
return True
elif subnet.ipv6_address_mode != const.IPV6_SLAAC:
return True
return False
@nsx_plugin_common.api_replay_mode_wrapper
def create_subnet(self, context, subnet):
if not self.use_policy_dhcp:
# Subnet with MP DHCP
return self._create_subnet_with_mp_dhcp(context, subnet)
self._validate_number_of_subnet_static_routes(subnet)
self._validate_host_routes_input(subnet)
self._validate_subnet_ip_version(subnet['subnet'])
self._validate_subnet_host_routes(subnet)
net_id = subnet['subnet']['network_id']
network = self._get_network(context, net_id)
self._validate_single_ipv6_subnet(context, network, subnet['subnet'])
self._init_ipv6_gateway(subnet['subnet'])
net_az = self.get_network_az_by_net_id(context, net_id)
# Allow manipulation of only 1 subnet of the same network at once
@ -1237,10 +1357,11 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
self._validate_external_subnet(context, net_id)
self._validate_net_dhcp_edge_cluster(context, network, net_az)
self._validate_net_type_with_dhcp(context, network)
if self._has_dhcp_enabled_subnet(context, network):
ip_version = subnet['subnet'].get('ip_version', 4)
if self._has_dhcp_enabled_subnet(context, network, ip_version):
msg = (_("Can not create more than one DHCP-enabled "
"subnet in network %s") % net_id)
"subnet for IPv%(ver)s in network %(net)s") %
{'ver': ip_version, 'net': net_id})
LOG.error(msg)
raise n_exc.InvalidInput(error_message=msg)
self._validate_segment_subnets_num(
@ -1273,7 +1394,8 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
# revert the subnet creation
with excutils.save_and_reraise_exception():
# Try to delete the DHCP port, and the neutron subnet
self._delete_subnet_dhcp_port(context, net_id)
self._delete_subnet_dhcp_port(
context, net_id, subnet_id=created_subnet['id'])
super(NsxPolicyPlugin, self).delete_subnet(
context, created_subnet['id'])
@ -1292,16 +1414,16 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
if self._subnet_with_native_dhcp(subnet):
lock = 'nsxp_network_' + subnet['network_id']
with locking.LockManager.get_lock(lock):
# Check if it is the last DHCP-enabled subnet to delete.
# Remove this subnet DHCP config
network = self._get_network(context, subnet['network_id'])
if self._has_single_dhcp_enabled_subnet(context, network):
try:
self._disable_network_dhcp(context, network)
except Exception as e:
LOG.error("Failed to disable DHCP for "
"network %(id)s. Exception: %(e)s",
{'id': network['id'], 'e': e})
# Continue for the neutron subnet deletion
try:
self._disable_network_dhcp(context, network,
subnet_id=subnet_id)
except Exception as e:
LOG.error("Failed to disable DHCP for "
"network %(id)s. Exception: %(e)s",
{'id': network['id'], 'e': e})
# Continue for the neutron subnet deletion
# Delete neutron subnet
super(NsxPolicyPlugin, self).delete_subnet(context, subnet_id)
@ -1312,11 +1434,7 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
subnet_data = subnet['subnet']
updated_subnet = None
orig_subnet = self.get_subnet(context, subnet_id)
self._validate_number_of_subnet_static_routes(subnet)
self._validate_host_routes_input(
subnet,
orig_enable_dhcp=orig_subnet['enable_dhcp'],
orig_host_routes=orig_subnet['host_routes'])
self._validate_subnet_host_routes(subnet, orig_subnet=orig_subnet)
net_id = orig_subnet['network_id']
network = self._get_network(context, net_id)
@ -1333,10 +1451,12 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
with locking.LockManager.get_lock(lock):
if enable_dhcp:
self._validate_net_type_with_dhcp(context, network)
if self._has_dhcp_enabled_subnet(context, network):
ip_version = orig_subnet.get('ip_version', 4)
if self._has_dhcp_enabled_subnet(context, network,
ip_version):
msg = (_("Can not create more than one DHCP-enabled "
"subnet in network %s") % net_id)
"subnet for IPv%(ver)s in network %(net)s") %
{'net': net_id, 'ver': ip_version})
LOG.error(msg)
raise n_exc.InvalidInput(error_message=msg)
@ -1353,7 +1473,8 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
self._enable_subnet_dhcp(context, network,
updated_subnet, net_az)
else:
self._disable_network_dhcp(context, network)
self._disable_network_dhcp(context, network,
subnet_id=subnet_id)
except (nsx_lib_exc.ManagerError, nsx_exc.NsxPluginException):
# revert the subnet update
with excutils.save_and_reraise_exception():
@ -1373,8 +1494,7 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
('dns_nameservers' in subnet_data or
'gateway_ip' in subnet_data or
'host_routes' in subnet_data)):
self._update_subnet_dhcp(context, network,
updated_subnet, net_az)
self._update_nsx_net_dhcp(context, network, net_az, updated_subnet)
return updated_subnet
@ -1615,6 +1735,18 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
return True
def _filter_ipv6_dhcp_fixed_ips(self, context, fixed_ips):
ips = []
for fixed_ip in fixed_ips:
if netaddr.IPNetwork(fixed_ip['ip_address']).version != 6:
continue
with db_api.CONTEXT_READER.using(context):
subnet = self.get_subnet(context, fixed_ip['subnet_id'])
if (subnet['enable_dhcp'] and
subnet.get('ipv6_address_mode') != 'slaac'):
ips.append(fixed_ip)
return ips
def _add_or_overwrite_port_policy_dhcp_binding(
self, context, port, segment_id, dhcp_subnet=None):
if not utils.is_port_dhcp_configurable(port):
@ -1647,7 +1779,23 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
mac_address=port['mac_address'],
options=options)
# TODO(asarfaty): add ipv6 bindings (without options)
for fixed_ip in self._filter_ipv6_dhcp_fixed_ips(
context, port['fixed_ips']):
# There will be only one ipv6 ip here
binding_id = port['id'] + '-ipv6'
name = 'IPv6 binding for port %s' % port['id']
ip = fixed_ip['ip_address']
if dhcp_subnet:
if fixed_ip['subnet_id'] != dhcp_subnet['id']:
continue
subnet = dhcp_subnet
else:
subnet = self.get_subnet(context, fixed_ip['subnet_id'])
self.nsxpolicy.segment_dhcp_static_bindings.create_or_overwrite_v6(
name, segment_id, binding_id=binding_id,
ip_addresses=[ip],
lease_time=cfg.CONF.nsx_p.dhcp_lease_time,
mac_address=port['mac_address'])
def _add_port_policy_dhcp_binding(self, context, port):
net_id = port['network_id']
@ -1690,7 +1838,7 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
pass
def _update_port_policy_dhcp_binding(self, context, old_port, new_port):
# First check if any IPv4 address in fixed_ips is changed.
# First check if any address in fixed_ips changed.
# Then update DHCP server setting or DHCP static binding
# depending on the port type.
# Note that Neutron allows a port with multiple IPs in the
@ -1707,44 +1855,87 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
# Collect IPv4 DHCP addresses from original and updated fixed_ips
# in the form of [(subnet_id, ip_address)].
old_fixed_ips = set([(fixed_ip['subnet_id'], fixed_ip['ip_address'])
for fixed_ip in self._filter_ipv4_dhcp_fixed_ips(
context, old_port['fixed_ips'])])
new_fixed_ips = set([(fixed_ip['subnet_id'], fixed_ip['ip_address'])
for fixed_ip in self._filter_ipv4_dhcp_fixed_ips(
context, new_port['fixed_ips'])])
old_fixed_v4 = set([(fixed_ip['subnet_id'], fixed_ip['ip_address'])
for fixed_ip in self._filter_ipv4_dhcp_fixed_ips(
context, old_port['fixed_ips'])])
new_fixed_v4 = set([(fixed_ip['subnet_id'], fixed_ip['ip_address'])
for fixed_ip in self._filter_ipv4_dhcp_fixed_ips(
context, new_port['fixed_ips'])])
old_fixed_v6 = set([(fixed_ip['subnet_id'], fixed_ip['ip_address'])
for fixed_ip in self._filter_ipv6_dhcp_fixed_ips(
context, old_port['fixed_ips'])])
new_fixed_v6 = set([(fixed_ip['subnet_id'], fixed_ip['ip_address'])
for fixed_ip in self._filter_ipv6_dhcp_fixed_ips(
context, new_port['fixed_ips'])])
# Find out the subnet/IP differences before and after the update.
ips_to_add = list(new_fixed_ips - old_fixed_ips)
ips_to_delete = list(old_fixed_ips - new_fixed_ips)
ip_change = (ips_to_add or ips_to_delete)
v4_to_add = list(new_fixed_v4 - old_fixed_v4)
v4_to_delete = list(old_fixed_v4 - new_fixed_v4)
v6_to_add = list(new_fixed_v6 - old_fixed_v6)
v6_to_delete = list(old_fixed_v6 - new_fixed_v6)
ip_change = (v4_to_add or v4_to_delete or v6_to_add or v6_to_delete)
if (old_port["device_owner"] == const.DEVICE_OWNER_DHCP and
ip_change):
# Update backend DHCP server address if the IP address of a DHCP
# port is changed.
if len(new_fixed_ips) != 1:
if len(new_fixed_v4) > 1 or len(new_fixed_v6) > 1:
msg = _("Can only configure one IP address on a DHCP server")
LOG.error(msg)
raise n_exc.InvalidInput(error_message=msg)
fixed_ip = list(new_fixed_ips)[0]
subnet_id = fixed_ip[0]
net_id = old_port['network_id']
network = self.get_network(context, net_id)
subnet = self.get_subnet(context, subnet_id)
net_az = self.get_network_az_by_net_id(context, net_id)
self._update_subnet_dhcp(context, network, subnet, net_az)
self._update_nsx_net_dhcp(context, network, net_az)
elif utils.is_port_dhcp_configurable(new_port):
dhcp_opts_changed = (old_port[ext_edo.EXTRADHCPOPTS] !=
new_port[ext_edo.EXTRADHCPOPTS])
if (ip_change or dhcp_opts_changed or
old_port['mac_address'] != new_port['mac_address']):
if new_fixed_ips:
if new_fixed_v4 or new_fixed_v6:
# Recreate the bindings of this port
self._add_port_policy_dhcp_binding(context, new_port)
else:
self._delete_port_policy_dhcp_binding(context, old_port)
def _assert_on_ipv6_port_with_dhcpopts(self, context, port_data,
orig_port=None):
"""IPv6 port only port cannot support EXTRADHCPOPTS"""
# Get the updated EXTRADHCPOPTS
extradhcpopts = None
if ext_edo.EXTRADHCPOPTS in port_data:
extradhcpopts = port_data[ext_edo.EXTRADHCPOPTS]
elif orig_port:
extradhcpopts = orig_port.get(ext_edo.EXTRADHCPOPTS)
if not extradhcpopts:
return
# Get the updated list of fixed ips
fixed_ips = []
if (port_data.get('fixed_ips') and
validators.is_attr_set(port_data['fixed_ips'])):
fixed_ips = port_data['fixed_ips']
elif (orig_port and orig_port.get('fixed_ips') and
validators.is_attr_set(orig_port['fixed_ips'])):
fixed_ips = orig_port['fixed_ips']
# Check if any of the ips belongs to an ipv6 subnet with DHCP
# And no ipv4 subnets
for fixed_ip in fixed_ips:
if netaddr.IPNetwork(fixed_ip['ip_address']).version == 4:
# If there are ipv4 addresses - it is allowed
return
with db_api.CONTEXT_READER.using(context):
subnet = self.get_subnet(context, fixed_ip['subnet_id'])
if (subnet['enable_dhcp'] and
subnet['ipv6_address_mode'] != const.IPV6_SLAAC):
err_msg = (_("%s are not supported for IPv6 ports with "
"DHCP v6") % ext_edo.EXTRADHCPOPTS)
LOG.error(err_msg)
raise n_exc.InvalidInput(error_message=err_msg)
def create_port(self, context, port, l2gw_port_check=False):
port_data = port['port']
# validate the new port parameters
@ -1765,6 +1956,15 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
neutron_db = self.base_create_port(context, port)
port["port"].update(neutron_db)
try:
# Validate ipv6 only after fixed_ips are allocated
self._assert_on_ipv6_port_with_dhcpopts(context, port["port"])
except Exception:
with excutils.save_and_reraise_exception():
# rollback
super(NsxPolicyPlugin, self).delete_port(
context, neutron_db['id'])
self.fix_direct_vnic_port_sec(direct_vnic_type, port_data)
(is_psec_on, has_ip, sgids, psgids) = (
self._create_port_preprocess_security(context, port,
@ -1949,6 +2149,8 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
direct_vnic_type = self._validate_port_vnic_type(
context, port_data, original_port['network_id'])
self._assert_on_ipv6_port_with_dhcpopts(
context, port_data, orig_port=original_port)
updated_port = super(NsxPolicyPlugin, self).update_port(
context, port_id, port)
@ -2617,6 +2819,50 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
LOG.error(msg)
raise n_exc.InvalidInput(error_message=msg)
def _get_subnet_ndra_profile(self, subnet):
ndra_profile_id = NO_SLAAC_NDRA_PROFILE_ID
if subnet.get('ip_version') == 6 and subnet.get('enable_dhcp', False):
# Subnet with DHCP v6 of some kind
addr_mode = subnet.get('ipv6_address_mode')
if addr_mode == const.IPV6_SLAAC:
ndra_profile_id = SLAAC_NDRA_PROFILE_ID
elif addr_mode == const.DHCPV6_STATELESS:
ndra_profile_id = STATELESS_DHCP_NDRA_PROFILE_ID
else:
# Stateful DHCP v6 is the default
ndra_profile_id = STATEFUL_DHCP_NDRA_PROFILE_ID
return ndra_profile_id
def _validate_interfaces_address_mode(self, context, router_id,
router_subnets, subnet):
"""Validate that all the overlay ipv6 interfaces of the router have
the same ipv6_address_mode, when a new subnet is added
"""
if subnet['enable_dhcp']:
subnet_address_mode = subnet.get('ipv6_address_mode',
const.DHCPV6_STATEFUL)
else:
# Slaac and non-dhcp can coexist
subnet_address_mode = const.IPV6_SLAAC
ipv6_overlay_subnets = [s for s in router_subnets
if s['id'] != subnet['id'] and
s.get('ip_version') == 6 and
s.get('enable_dhcp') and
self._is_overlay_network(context,
s['network_id'])]
for rtr_subnet in ipv6_overlay_subnets:
address_mode = rtr_subnet.get('ipv6_address_mode',
const.DHCPV6_STATEFUL)
if address_mode != subnet_address_mode:
msg = (_("Interface network %(net_id)s with address mode "
"%(am)s conflicts with other interfaces of router "
"%(rtr_id)s") % {'net_id': subnet['network_id'],
'am': subnet_address_mode,
'rtr_id': router_id})
LOG.error(msg)
raise n_exc.InvalidInput(error_message=msg)
@nsx_plugin_common.api_replay_mode_wrapper
def add_router_interface(self, context, router_id, interface_info):
# NOTE: In dual stack case, neutron would create a separate interface
@ -2658,6 +2904,12 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
self._validate_router_segment_subnets(context, network_id,
overlay_net, subnet)
if subnet and subnet.get('ip_version') == 6 and overlay_net:
orig_rtr_subnets = self._load_router_subnet_cidrs_from_db(
context.elevated(), router_id)
self._validate_interfaces_address_mode(
context, router_id, orig_rtr_subnets, subnet)
# Update the interface of the neutron router
info = super(NsxPolicyPlugin, self).add_router_interface(
context, router_id, interface_info)
@ -2680,7 +2932,6 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
# overlay interface
pol_subnets = self._get_segment_subnets(
context, network_id, interface_subnets=rtr_subnets)
self.nsxpolicy.segment.update(segment_id,
tier1_id=router_id,
subnets=pol_subnets)
@ -2698,9 +2949,7 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
ip_addresses=[rtr_subnet['gateway_ip']],
prefix_len=prefix_len))
slaac_subnet = (subnet.get('ipv6_address_mode') == 'slaac')
ndra_profile_id = (SLAAC_NDRA_PROFILE_ID if slaac_subnet
else NO_SLAAC_NDRA_PROFILE_ID)
ndra_profile_id = self._get_subnet_ndra_profile(subnet)
self.nsxpolicy.tier1.add_segment_interface(
router_id, segment_id,
segment_id, pol_subnets,
@ -3047,34 +3296,38 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
NSX_P_DEFAULT_GROUP)
except nsx_lib_exc.ResourceNotFound:
LOG.info("Going to create default group & "
"communication map under the default domain")
"communication map")
exists = False
else:
LOG.debug("Verified default group already exist")
return
LOG.info("Going to update default group & "
"communication map")
exists = True
# Create the default group membership criteria to match all neutron
# ports by scope (and no tag)
scope_and_tag = "%s|" % (NSX_P_PORT_RESOURCE_TYPE)
conditions = [self.nsxpolicy.group.build_condition(
cond_val=scope_and_tag,
cond_key=policy_constants.CONDITION_KEY_TAG,
cond_member_type=policy_constants.CONDITION_MEMBER_PORT)]
# Create the default OpenStack group
# (This will not fail if the group already exists)
try:
self.nsxpolicy.group.create_or_overwrite_with_conditions(
name=NSX_P_DEFAULT_GROUP,
domain_id=NSX_P_GLOBAL_DOMAIN_ID,
group_id=NSX_P_DEFAULT_GROUP,
description=NSX_P_DEFAULT_GROUP_DESC,
conditions=conditions)
except Exception as e:
msg = (_("Failed to create NSX default group: %(e)s") % {
'e': e})
raise nsx_exc.NsxPluginException(err_msg=msg)
# Create the group only if not exists - no need to update it
if not exists:
# Create the default group membership criteria to match all
# neutron ports by scope (and no tag)
scope_and_tag = "%s|" % (NSX_P_PORT_RESOURCE_TYPE)
conditions = [self.nsxpolicy.group.build_condition(
cond_val=scope_and_tag,
cond_key=policy_constants.CONDITION_KEY_TAG,
cond_member_type=policy_constants.CONDITION_MEMBER_PORT)]
# Create the default OpenStack group
# (This will not fail if the group already exists)
try:
self.nsxpolicy.group.create_or_overwrite_with_conditions(
name=NSX_P_DEFAULT_GROUP,
domain_id=NSX_P_GLOBAL_DOMAIN_ID,
group_id=NSX_P_DEFAULT_GROUP,
description=NSX_P_DEFAULT_GROUP_DESC,
conditions=conditions)
except Exception as e:
msg = (_("Failed to create NSX default group: %(e)s") % {
'e': e})
raise nsx_exc.NsxPluginException(err_msg=msg)
# create default section and rules
# (even if already exists - may need to update rules)
logged = cfg.CONF.nsx_p.log_security_groups_blocked_traffic
scope = [self.nsxpolicy.group.get_path(
NSX_P_GLOBAL_DOMAIN_ID, NSX_P_DEFAULT_GROUP)]
@ -3116,6 +3369,28 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
direction=nsxlib_consts.IN_OUT,
logged=logged)
rule_id += 1
dhcpv6_server_rule = self.nsxpolicy.comm_map.build_entry(
'DHCPv6 Request', NSX_P_GLOBAL_DOMAIN_ID,
NSX_P_DEFAULT_SECTION,
rule_id, sequence_number=rule_id,
service_ids=['DHCPv6_Server'],
action=policy_constants.ACTION_ALLOW,
ip_protocol=nsxlib_consts.IPV6,
scope=scope,
direction=nsxlib_consts.OUT,
logged=logged)
rule_id += 1
dhcpv6_client_rule = self.nsxpolicy.comm_map.build_entry(
'DHCPv6 Reply', NSX_P_GLOBAL_DOMAIN_ID,
NSX_P_DEFAULT_SECTION,
rule_id, sequence_number=rule_id,
service_ids=['DHCPv6_Client'],
action=policy_constants.ACTION_ALLOW,
ip_protocol=nsxlib_consts.IPV6,
scope=scope,
direction=nsxlib_consts.IN,
logged=logged)
rule_id += 1
block_rule = self.nsxpolicy.comm_map.build_entry(
'Block All', NSX_P_GLOBAL_DOMAIN_ID,
NSX_P_DEFAULT_SECTION,
@ -3124,7 +3399,8 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
scope=scope,
direction=nsxlib_consts.IN_OUT,
logged=logged)
rules = [dhcp_client_rule, dhcp_server_rule, nd_rule, block_rule]
rules = [dhcp_client_rule, dhcp_server_rule, dhcpv6_client_rule,
dhcpv6_server_rule, nd_rule, block_rule]
try:
# This will not fail if the map already exists
self.nsxpolicy.comm_map.create_with_entries(

60
vmware_nsx/tests/unit/common_plugin/common_v3.py
View File

@ -284,38 +284,26 @@ class NsxV3TestSubnets(NsxV3SubnetMixin,
def test_list_subnets_with_parameter(self):
super(NsxV3TestSubnets, self).test_list_subnets_with_parameter()
def test_create_subnet_ipv6_gw_is_nw_start_addr(self):
self.skipTest('No DHCP v6 Support yet')
def test_create_subnet_ipv6_gw_is_nw_start_addr_canonicalize(self):
self.skipTest('No DHCP v6 Support yet')
def test_create_subnet_ipv6_gw_is_nw_end_addr(self):
self.skipTest('No DHCP v6 Support yet')
def test_create_subnet_ipv6_first_ip_owned_by_router(self):
self.skipTest('No DHCP v6 Support yet')
def test_create_subnet_ipv6_first_ip_owned_by_non_router(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')
self.skipTest('Test not suited to the plugin DHCP code')
def test_create_subnet_ipv6_slaac_with_port_not_found(self):
self.skipTest('No DHCP v6 Support yet')
self.skipTest('Test not suited to the plugin DHCP code')
def test_bulk_create_subnet_ipv6_auto_addr_with_port_on_network(self):
self.skipTest('No DHCP v6 Support yet')
self.skipTest('No Multiple v6 subnets support yet')
def test_create_subnet_dhcpv6_stateless_with_ip_already_allocated(self):
self.skipTest('Test not suited to the plugin DHCP code')
def test_create_subnet_ipv6_slaac_with_dhcp_port_on_network(self):
self.skipTest('Test not suited to the plugin DHCP code')
def test_create_subnet_dhcpv6_stateless_with_port_on_network(self):
self.skipTest('Test not suited to the plugin DHCP code')
def test_delete_subnet_port_exists_owned_by_network(self):
self.skipTest('Test not suited to the plugin DHCP code')
def test_create_subnets_bulk_native_ipv6(self):
self.skipTest('Multiple IPv6 subnets on one network is not supported')
@ -380,24 +368,6 @@ class NsxV3TestSubnets(NsxV3SubnetMixin,
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')

419
vmware_nsx/tests/unit/nsx_p/test_plugin.py
View File

@ -29,6 +29,7 @@ from neutron.tests.unit.extensions import test_l3 as test_l3_plugin
from neutron.tests.unit.extensions import test_securitygroup
from neutron_lib.api.definitions import external_net as extnet_apidef
from neutron_lib.api.definitions import extra_dhcp_opt as edo_ext
from neutron_lib.api.definitions import extraroute as xroute_apidef
from neutron_lib.api.definitions import l3_ext_gw_mode as l3_egm_apidef
from neutron_lib.api.definitions import port_security as psec
@ -574,10 +575,6 @@ class NsxPTestPorts(common_v3.NsxV3TestPorts,
def setUp(self, **kwargs):
super(NsxPTestPorts, self).setUp(**kwargs)
@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()
@ -1004,12 +1001,46 @@ class NsxPTestPorts(common_v3.NsxV3TestPorts,
**kwargs)
self.assertEqual(res.status_int, exc.HTTPBadRequest.code)
def test_create_ipv6_port(self):
with self.network(name='net') as network:
self._make_v6_subnet(network, constants.DHCPV6_STATEFUL)
res = self._create_port(self.fmt, net_id=network['network']['id'])
port = self.deserialize(self.fmt, res)
self.assertIn('id', port['port'])
def test_create_ipv6_port_with_extra_dhcp(self):
with self.network(name='net') as network:
self._make_v6_subnet(network, constants.DHCPV6_STATEFUL)
opt_list = [{'opt_name': 'bootfile-name',
'opt_value': 'pxelinux.0'},
{'opt_name': 'tftp-server-address',
'opt_value': '123.123.123.123'}]
params = {edo_ext.EXTRADHCPOPTS: opt_list,
'arg_list': (edo_ext.EXTRADHCPOPTS,)}
self._create_port(self.fmt, network['network']['id'],
exc.HTTPBadRequest.code, **params)
def test_update_ipv6_port_with_extra_dhcp(self):
with self.network(name='net') as network:
self._make_v6_subnet(network, constants.DHCPV6_STATEFUL)
res = self._create_port(self.fmt, net_id=network['network']['id'])
port = self.deserialize(self.fmt, res)
self.assertIn('id', port['port'])
opt_list = [{'opt_name': 'bootfile-name',
'opt_value': 'pxelinux.0'},
{'opt_name': 'tftp-server-address',
'opt_value': '123.123.123.123'}]
data = {'port': {edo_ext.EXTRADHCPOPTS: opt_list}}
req = self.new_update_request('ports', data, port['port']['id'])
res = self.deserialize(self.fmt, req.get_response(self.api))
self.assertIn('NeutronError', res)
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.force_slaac = False
def _create_subnet_bulk(self, fmt, number, net_id, name,
ip_version=4, **kwargs):
@ -1033,17 +1064,6 @@ class NsxPTestSubnets(common_v3.NsxV3TestSubnets,
kwargs.update({'override': overrides})
return self._create_bulk(fmt, number, 'subnet', base_data, **kwargs)
def _test_create_subnet(self, network=None, expected=None, **kwargs):
# Until DHCPv6 is supported, switch all test to slaac-only
#TODO(asarfaty): remove when DHCPv6 is supported
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(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:
data = {'subnet': {'network_id': network['network']['id'],
@ -1084,6 +1104,119 @@ class NsxPTestSubnets(common_v3.NsxV3TestSubnets,
super(NsxPTestSubnets,
self).test_create_subnet_ipv6_slaac_with_port_on_network()
def test_create_subnet_ipv6_gw_values(self):
self.skipTest("IPv6 gateway IP is assigned by the plugin")
def test_create_ipv6_subnet_with_host_routes(self):
# IPv6 host routes are not allowed
with self.network() as network:
data = {'subnet': {'network_id': network['network']['id'],
'cidr': '100::/64',
'ip_version': 6,
'tenant_id': network['network']['tenant_id'],
'host_routes': [{'destination': '200::/64',
'nexthop': '100::16'}]}}
subnet_req = self.new_create_request('subnets', data)
res = subnet_req.get_response(self.api)
self.assertEqual(exc.HTTPClientError.code, res.status_int)
def test_update_ipv6_subnet_with_host_routes(self):
# IPv6 host routes are not allowed
with self.network() as network:
data = {'subnet': {'network_id': network['network']['id'],
'cidr': '100::/64',
'ip_version': 6,
'tenant_id': network['network']['tenant_id']}}
subnet_req = self.new_create_request('subnets', data)
subnet = self.deserialize(self.fmt,
subnet_req.get_response(self.api))
sub_id = subnet['subnet']['id']
# update host routes should fail
data = {'subnet': {'host_routes': [{'destination': '200::/64',
'nexthop': '100::16'}]}}
update_req = self.new_update_request('subnets', data, sub_id)
res = update_req.get_response(self.api)
self.assertEqual(exc.HTTPClientError.code, res.status_int)
def _verify_dhcp_service(self, network_id, tenant_id, enabled):
# Verify if DHCP service is enabled on a network.
port_res = self._list_ports('json', 200, network_id,
tenant_id=tenant_id,
device_owner=constants.DEVICE_OWNER_DHCP)
port_list = self.deserialize('json', port_res)
self.assertEqual(len(port_list['ports']) == 1, enabled)
def test_create_dhcpv6_subnet(self):
with mock.patch("vmware_nsxlib.v3.policy.core_resources."
"NsxPolicySegmentApi.update") as seg_update,\
self.subnet(ip_version=constants.IP_VERSION_6, cidr='fe80::/64',
enable_dhcp=True) as subnet:
self.assertEqual(True, subnet['subnet']['enable_dhcp'])
# verify that the dhcp port was created
self._verify_dhcp_service(subnet['subnet']['network_id'],
subnet['subnet']['tenant_id'], True)
# verify backend calls
seg_update.assert_called_once_with(
dhcp_server_config_id=NSX_DHCP_PROFILE_ID,
segment_id=subnet['subnet']['network_id'],
subnets=[mock.ANY])
def test_subnet_enable_dhcpv6(self):
with self.subnet(ip_version=constants.IP_VERSION_6, cidr='fe80::/64',
enable_dhcp=False) as subnet:
data = {'subnet': {'enable_dhcp': True}}
with mock.patch("vmware_nsxlib.v3.policy.core_resources."
"NsxPolicySegmentApi.update") as seg_update:
req = self.new_update_request('subnets', data,
subnet['subnet']['id'])
res = self.deserialize('json', req.get_response(self.api))
self.assertEqual(True, res['subnet']['enable_dhcp'])
# verify that the dhcp port was created
self._verify_dhcp_service(subnet['subnet']['network_id'],
subnet['subnet']['tenant_id'], True)
# verify backend calls
seg_update.assert_called_once_with(
dhcp_server_config_id=NSX_DHCP_PROFILE_ID,
segment_id=subnet['subnet']['network_id'],
subnets=[mock.ANY])
def test_subnet_disable_dhcpv6(self):
with self.subnet(ip_version=constants.IP_VERSION_6, cidr='fe80::/64',
enable_dhcp=True) as subnet:
data = {'subnet': {'enable_dhcp': False}}
with mock.patch("vmware_nsxlib.v3.policy.core_resources."
"NsxPolicySegmentApi.update") as seg_update:
req = self.new_update_request('subnets', data,
subnet['subnet']['id'])
res = self.deserialize('json', req.get_response(self.api))
self.assertEqual(False, res['subnet']['enable_dhcp'])
# verify that the dhcp port was deleted
self._verify_dhcp_service(subnet['subnet']['network_id'],
subnet['subnet']['tenant_id'], False)
# verify backend calls
seg_update.assert_called_once_with(
dhcp_server_config_id=None,
segment_id=subnet['subnet']['network_id'],
subnets=[])
def test_delete_ipv6_dhcp_subnet(self):
with self.subnet(ip_version=constants.IP_VERSION_6, cidr='fe80::/64',
enable_dhcp=True) as subnet:
with mock.patch("vmware_nsxlib.v3.policy.core_resources."
"NsxPolicySegmentApi.update") as seg_update:
req = self.new_delete_request(
'subnets', subnet['subnet']['id'])
res = req.get_response(self.api)
self.assertEqual(exc.HTTPNoContent.code, res.status_int)
# verify that the dhcp port was deleted
self._verify_dhcp_service(subnet['subnet']['network_id'],
subnet['subnet']['tenant_id'], False)
# verify backend calls
seg_update.assert_called_once_with(
dhcp_server_config_id=None,
segment_id=subnet['subnet']['network_id'],
subnets=[])
class NsxPTestSecurityGroup(common_v3.FixExternalNetBaseTest,
NsxPPluginTestCaseMixin,
@ -1471,9 +1604,6 @@ class NsxPTestL3NatTestCase(NsxPTestL3NatTest,
def test_router_add_gateway_multiple_subnets_ipv6(self):
self.skipTest('not supported')
def test_router_add_interface_ipv6_subnet(self):
self.skipTest('DHCPv6 not supported')
def test_slaac_profile_single_subnet(self):
with mock.patch("vmware_nsxlib.v3.policy.core_resources."
"NsxPolicyTier1Api.update") as t1_update:
@ -2268,6 +2398,8 @@ class NsxPTestL3NatTestCase(NsxPTestL3NatTest,
False, pol_const.NAT_FIREWALL_MATCH_EXTERNAL)
def test_router_interface_with_dhcp_subnet(self):
# Policy DHCP does not allow 1 dhcp subnet and another router
# interface subnet on the same overlay network
with self.router() as r,\
self.network() as net,\
self.subnet(cidr='20.0.0.0/24', network=net),\
@ -2277,3 +2409,252 @@ class NsxPTestL3NatTestCase(NsxPTestL3NatTest,
'add', r['router']['id'],
if_subnet['subnet']['id'], None,
expected_code=exc.HTTPBadRequest.code)
def test_router_interface_ndprofile_ipv4(self):
with self.router() as r,\
self.network() as net,\
self.subnet(cidr='20.0.0.0/24', network=net) as if_subnet,\
mock.patch("vmware_nsxlib.v3.policy.core_resources."
"NsxPolicyTier1Api.update") as t1_update:
# Adding ipv4 interface
self._router_interface_action(
'add', r['router']['id'],
if_subnet['subnet']['id'], None)
t1_update.assert_not_called()
# Removing ipv4 interface
self._router_interface_action(
'remove', r['router']['id'],
if_subnet['subnet']['id'], None)
t1_update.assert_not_called()
def _test_router_interface_ndprofile(self, profile_with,
enable_dhcp=True, mode='slaac'):
with self.router() as r,\
self.network() as net,\
self.subnet(cidr='2001::/64', network=net,
ip_version=6, enable_dhcp=enable_dhcp,
ipv6_address_mode=mode,
ipv6_ra_mode=mode) as if_subnet,\
mock.patch("vmware_nsxlib.v3.policy.core_resources."
"NsxPolicyTier1Api.update") as t1_update:
# Adding subnet interface
self._router_interface_action(
'add', r['router']['id'],
if_subnet['subnet']['id'], None)
t1_update.assert_called_with(
r['router']['id'],
ipv6_ndra_profile_id=profile_with)
t1_update.reset_mock()
# Removing subnet interface
self._router_interface_action(
'remove', r['router']['id'],
if_subnet['subnet']['id'], None)
t1_update.assert_called_with(
r['router']['id'],
ipv6_ndra_profile_id=nsx_plugin.NO_SLAAC_NDRA_PROFILE_ID)
def test_router_interface_ndprofile_no_dhcp(self):
self._test_router_interface_ndprofile(
nsx_plugin.NO_SLAAC_NDRA_PROFILE_ID,
enable_dhcp=False, mode=None)
def test_router_interface_ndprofile_slaac(self):
self._test_router_interface_ndprofile(
nsx_plugin.SLAAC_NDRA_PROFILE_ID,
enable_dhcp=True, mode=constants.IPV6_SLAAC)
def test_router_interface_ndprofile_stateful(self):
self._test_router_interface_ndprofile(
nsx_plugin.STATEFUL_DHCP_NDRA_PROFILE_ID,
enable_dhcp=True, mode=constants.DHCPV6_STATEFUL)
def test_router_interface_ndprofile_stateless(self):
self._test_router_interface_ndprofile(
nsx_plugin.STATELESS_DHCP_NDRA_PROFILE_ID,
enable_dhcp=True, mode=constants.DHCPV6_STATELESS)
def _test_router_interfaces_ndprofile(self, sub1_enable_dhcp, sub1_mode,
sub2_enable_dhcp, sub2_mode,
sub1_profile, mixed_profile=None,
successful=True,
sub1_ipversion=6, sub2_ipversion=6):
cidr1 = '2001::/64' if sub1_ipversion == 6 else '201.0.0.0/24'
cidr2 = '2002::/64' if sub2_ipversion == 6 else '202.0.0.0/24'
with self.router() as r,\
self.network() as net1, self.network() as net2,\
self.subnet(cidr=cidr1, network=net1,
ip_version=sub1_ipversion,
enable_dhcp=sub1_enable_dhcp,
ipv6_address_mode=sub1_mode,
ipv6_ra_mode=sub1_mode) as sub1,\
self.subnet(cidr=cidr2, network=net2,
ip_version=sub2_ipversion,
enable_dhcp=sub2_enable_dhcp,
ipv6_address_mode=sub2_mode,
ipv6_ra_mode=sub2_mode) as sub2,\
mock.patch("vmware_nsxlib.v3.policy.core_resources."
"NsxPolicyTier1Api.update") as t1_update:
# Adding first interface
self._router_interface_action(
'add', r['router']['id'],
sub1['subnet']['id'], None)
if sub1_ipversion == 6:
t1_update.assert_called_with(
r['router']['id'],
ipv6_ndra_profile_id=sub1_profile)
t1_update.reset_mock()
else:
t1_update.assert_not_called()
# Adding the 2nd interface
expected_code = (exc.HTTPBadRequest.code if not successful
else exc.HTTPOk.code)
self._router_interface_action(
'add', r['router']['id'],
sub2['subnet']['id'], None,
expected_code=expected_code)
if not successful:
return
if sub2_ipversion == 6:
t1_update.assert_called_with(
r['router']['id'],
ipv6_ndra_profile_id=mixed_profile)
t1_update.reset_mock()
else:
t1_update.assert_not_called()
# Removing the 2nd interface
self._router_interface_action(
'remove', r['router']['id'],
sub2['subnet']['id'], None)
if sub2_ipversion == 6:
t1_update.assert_called_with(
r['router']['id'],
ipv6_ndra_profile_id=sub1_profile)
else:
t1_update.assert_not_called()
def test_router_interfaces_ndprofile_slaac_slaac(self):
self._test_router_interfaces_ndprofile(
True, constants.IPV6_SLAAC,
True, constants.IPV6_SLAAC,
nsx_plugin.SLAAC_NDRA_PROFILE_ID,
nsx_plugin.SLAAC_NDRA_PROFILE_ID)
def test_router_interfaces_ndprofile_slaac_stateful(self):
self._test_router_interfaces_ndprofile(
True, constants.IPV6_SLAAC,
True, constants.DHCPV6_STATEFUL,
nsx_plugin.SLAAC_NDRA_PROFILE_ID,
None, successful=False)
def test_router_interfaces_ndprofile_slaac_stateless(self):
self._test_router_interfaces_ndprofile(
True, constants.IPV6_SLAAC,
True, constants.DHCPV6_STATELESS,
nsx_plugin.SLAAC_NDRA_PROFILE_ID,
None, successful=False)
def test_router_interfaces_ndprofile_disabled_stateful(self):
self._test_router_interfaces_ndprofile(
False, None,
True, constants.DHCPV6_STATEFUL,
nsx_plugin.NO_SLAAC_NDRA_PROFILE_ID,
nsx_plugin.STATEFUL_DHCP_NDRA_PROFILE_ID)
def test_router_interfaces_ndprofile_disabled_stateless(self):
self._test_router_interfaces_ndprofile(
False, None,
True, constants.DHCPV6_STATELESS,
nsx_plugin.NO_SLAAC_NDRA_PROFILE_ID,
nsx_plugin.STATELESS_DHCP_NDRA_PROFILE_ID)
def test_router_interfaces_ndprofile_stateful_stateless(self):
self._test_router_interfaces_ndprofile(
True, constants.DHCPV6_STATEFUL,
True, constants.DHCPV6_STATELESS,
nsx_plugin.STATEFUL_DHCP_NDRA_PROFILE_ID,
None, successful=False)
def test_router_interfaces_ndprofile_v4_stateless(self):
self._test_router_interfaces_ndprofile(
True, None,
True, constants.DHCPV6_STATELESS,
nsx_plugin.NO_SLAAC_NDRA_PROFILE_ID,
nsx_plugin.STATELESS_DHCP_NDRA_PROFILE_ID,
sub1_ipversion=4)
def test_router_interfaces_ndprofile_stateless_v4(self):
self._test_router_interfaces_ndprofile(
True, constants.DHCPV6_STATELESS,
True, None,
nsx_plugin.STATELESS_DHCP_NDRA_PROFILE_ID,
nsx_plugin.STATELESS_DHCP_NDRA_PROFILE_ID,
sub2_ipversion=4)
def _test_router_vlan_interface_ndprofile(self, profile_with,
enable_dhcp=True, mode='slaac'):
providernet_args = {pnet.NETWORK_TYPE: 'vlan',
pnet.SEGMENTATION_ID: 11}
with mock.patch('vmware_nsxlib.v3.policy.core_resources.'
'NsxPolicyTransportZoneApi.get_transport_type',
return_value=nsx_constants.TRANSPORT_TYPE_VLAN), \
self.network(name='vlan_net',
providernet_args=providernet_args,
arg_list=(pnet.NETWORK_TYPE,
pnet.SEGMENTATION_ID)) as net,\
self.router() as r,\
self.subnet(cidr='2001::/64', network=net,
ip_version=6, enable_dhcp=enable_dhcp,
ipv6_address_mode=mode,
ipv6_ra_mode=mode) as if_subnet,\
self._create_l3_ext_network() as ext_net,\
self.subnet(network=ext_net, cidr='10.0.0.0/16',
enable_dhcp=False) as ext_sub,\
mock.patch("vmware_nsxlib.v3.policy.core_resources."
"NsxPolicyTier1Api.add_segment_interface") as t1_add,\
mock.patch("vmware_nsxlib.v3.policy.core_resources."
"NsxPolicyTier1Api.remove_segment_interface") as t1_del:
# Add router GW
self._add_external_gateway_to_router(
r['router']['id'],
ext_sub['subnet']['network_id'])
# Adding subnet interface
self._router_interface_action(
'add', r['router']['id'],
if_subnet['subnet']['id'], None)
t1_add.assert_called_once_with(
r['router']['id'], mock.ANY, mock.ANY, [mock.ANY],
profile_with)
# Removing subnet interface
self._router_interface_action(
'remove', r['router']['id'],
if_subnet['subnet']['id'], None)
t1_del.assert_called_once_with(r['router']['id'], mock.ANY)
def test_router_vlan_interface_ndprofile_no_dhcp(self):
self._test_router_vlan_interface_ndprofile(
nsx_plugin.NO_SLAAC_NDRA_PROFILE_ID,
enable_dhcp=False, mode=None)
def test_router_vlan_interface_ndprofile_slaac(self):
self._test_router_vlan_interface_ndprofile(
nsx_plugin.SLAAC_NDRA_PROFILE_ID,
enable_dhcp=True, mode=constants.IPV6_SLAAC)
def test_router_vlan_interface_ndprofile_stateful(self):
self._test_router_vlan_interface_ndprofile(
nsx_plugin.STATEFUL_DHCP_NDRA_PROFILE_ID,
enable_dhcp=True, mode=constants.DHCPV6_STATEFUL)
def test_router_vlan_interface_ndprofile_stateless(self):
self._test_router_vlan_interface_ndprofile(
nsx_plugin.STATELESS_DHCP_NDRA_PROFILE_ID,
enable_dhcp=True, mode=constants.DHCPV6_STATELESS)

63
vmware_nsx/tests/unit/nsx_p/test_policy_dhcp_metadata.py
View File

@ -104,8 +104,8 @@ class NsxPolicyDhcpTestCase(test_plugin.NsxPPluginTestCaseMixin):
'host_routes': host_routes,
'shared': shared}}
def _bind_name(self, port):
return 'IPv4 binding for port %s' % port['port']['id']
def _bind_name(self, port, ip_version=4):
return 'IPv%s binding for port %s' % (ip_version, port['port']['id'])
def _verify_dhcp_service(self, network_id, tenant_id, enabled):
# Verify if DHCP service is enabled on a network.
@ -174,6 +174,39 @@ class NsxPolicyDhcpTestCase(test_plugin.NsxPPluginTestCaseMixin):
binding_id=port['port']['id'] + '-ipv4',
**binding_data)
def _verify_dhcp_binding_v6(self, subnet, port_data, update_data,
assert_data):
# Verify if DHCP-v6 binding is updated.
with mock.patch('vmware_nsxlib.v3.policy.core_resources.'
'SegmentDhcpStaticBindingConfigApi.'
'create_or_overwrite_v6') as update_dhcp_binding:
device_owner = constants.DEVICE_OWNER_COMPUTE_PREFIX + 'None'
device_id = uuidutils.generate_uuid()
with self.port(subnet=subnet, device_owner=device_owner,
device_id=device_id, **port_data) as port:
binding_name = self._bind_name(port, 6)
ip_address = port['port']['fixed_ips'][0]['ip_address']
binding_data = {'mac_address': port['port']['mac_address'],
'ip_addresses': [ip_address],
'lease_time': 86400}
# Verify the initial bindings call.
update_dhcp_binding.assert_called_once_with(
binding_name, subnet['subnet']['network_id'],
binding_id=port['port']['id'] + '-ipv6',
**binding_data)
update_dhcp_binding.reset_mock()
# Update the port with provided data.
self.plugin.update_port(
context.get_admin_context(), port['port']['id'],
update_data)
# Extend basic binding data with to-be-asserted data.
binding_data.update(assert_data)
# Verify the update call.
update_dhcp_binding.assert_called_once_with(
binding_name, subnet['subnet']['network_id'],
binding_id=port['port']['id'] + '-ipv6',
**binding_data)
def test_dhcp_service_with_create_network(self):
# Test if DHCP service is disabled on a network when it is created.
with self.network() as network:
@ -569,6 +602,20 @@ class NsxPolicyDhcpTestCase(test_plugin.NsxPPluginTestCaseMixin):
self._verify_dhcp_binding(subnet, port_data, update_data,
assert_data)
def test_dhcp_binding_v6_with_update_port_ip(self):
# Test if DHCP binding is updated when the IP of the associated
# compute port is changed.
with self.subnet(ip_version=6, cidr='101::/64',
enable_dhcp=True) as subnet:
port_data = {'fixed_ips': [{'subnet_id': subnet['subnet']['id'],
'ip_address': '101::3'}]}
new_ip = '101::4'
update_data = {'port': {'fixed_ips': [
{'subnet_id': subnet['subnet']['id'], 'ip_address': new_ip}]}}
assert_data = {'ip_addresses': [new_ip]}
self._verify_dhcp_binding_v6(subnet, port_data, update_data,
assert_data)
def test_dhcp_binding_with_update_port_mac(self):
# Test if DHCP binding is updated when the Mac of the associated
# compute port is changed.
@ -874,6 +921,18 @@ class NsxPolicyDhcpTestCase(test_plugin.NsxPPluginTestCaseMixin):
context.get_admin_context(), neutron_subnet['id'],
{'subnet': {'enable_dhcp': True}})
def test_create_subnet_with_dhcp_v6_port(self):
with self.subnet(enable_dhcp=True, ip_version=6,
cidr="2002::/64") as subnet:
# find the dhcp port and verify it has port security disabled
ports = self.plugin.get_ports(
context.get_admin_context())
self.assertEqual(1, len(ports))
self.assertEqual('network:dhcp', ports[0]['device_owner'])
self.assertEqual(subnet['subnet']['network_id'],
ports[0]['network_id'])
self.assertEqual(False, ports[0]['port_security_enabled'])
class NsxPolicyMetadataTestCase(test_plugin.NsxPPluginTestCaseMixin):
"""Test native metadata config when using MP MDProxy"""

30
vmware_nsx/tests/unit/nsx_v3/test_plugin.py
View File

@ -1007,6 +1007,36 @@ class TestSubnetsV2(common_v3.NsxV3TestSubnets, NsxV3PluginTestCaseMixin):
self.plugin.update_subnet,
context.get_admin_context(), subnet['id'], data)
def test_create_subnet_ipv6_gw_is_nw_start_addr(self):
self.skipTest('No DHCP v6 Support yet')
def test_create_subnet_ipv6_gw_is_nw_start_addr_canonicalize(self):
self.skipTest('No DHCP v6 Support yet')
def test_create_subnet_ipv6_gw_is_nw_end_addr(self):
self.skipTest('No DHCP v6 Support yet')
def test_create_subnet_ipv6_first_ip_owned_by_router(self):
self.skipTest('No DHCP v6 Support yet')
def test_create_subnet_ipv6_first_ip_owned_by_non_router(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_update_subnet_ipv6_ra_mode_fails(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_create_subnet_ipv6_slaac_with_db_reference_error(self):
self.skipTest('No DHCP v6 Support yet')
class TestPortsV2(common_v3.NsxV3SubnetMixin,
common_v3.NsxV3TestPorts, NsxV3PluginTestCaseMixin,