500 lines
19 KiB
Python
500 lines
19 KiB
Python
# Copyright 2018 VMware, Inc.
|
|
# All Rights Reserved
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
# not use this file except in compliance with the License. You may obtain
|
|
# a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
# License for the specific language governing permissions and limitations
|
|
# under the License.
|
|
|
|
import contextlib
|
|
|
|
import decorator
|
|
|
|
|
|
from neutron.tests.unit.db import test_db_base_plugin_v2 as test_plugin
|
|
|
|
import netaddr
|
|
from neutron_lib import constants
|
|
from webob import exc
|
|
|
|
|
|
class FixExternalNetBaseTest(object):
|
|
"""Base class providing utilities for handling tests which require updating
|
|
a network to be external, which is not supported for the NSX-v3 and NSX-P
|
|
plugins.
|
|
"""
|
|
def setUp(self, *args, **kwargs):
|
|
self.original_subnet = self.subnet
|
|
self.original_create_subnet = self._create_subnet
|
|
self.original_network = self.network
|
|
self.subnet_calls = []
|
|
super(FixExternalNetBaseTest, self).setUp(*args, **kwargs)
|
|
|
|
def _set_net_external(self, net_id):
|
|
# This action is not supported by the V3 plugin
|
|
pass
|
|
|
|
def _create_external_network(self):
|
|
data = {'network': {'name': 'net1',
|
|
'router:external': 'True',
|
|
'tenant_id': 'tenant_one',
|
|
'provider:physical_network': 'stam'}}
|
|
network_req = self.new_create_request('networks', data)
|
|
network = self.deserialize(self.fmt,
|
|
network_req.get_response(self.api))
|
|
return network
|
|
|
|
def external_subnet(self, network=None, **kwargs):
|
|
# External subnet ,ust have dhcp disabled
|
|
kwargs['enable_dhcp'] = False
|
|
if network:
|
|
return self.original_subnet(network=network, **kwargs)
|
|
ext_net = self._create_external_network()
|
|
return self.original_subnet(network=ext_net, **kwargs)
|
|
|
|
def create_external_subnet(self, *args, **kwargs):
|
|
kwargs['enable_dhcp'] = False
|
|
return super(FixExternalNetBaseTest, self)._create_subnet(
|
|
*args, **kwargs)
|
|
|
|
def no_dhcp_subnet(self, *args, **kwargs):
|
|
if 'enable_dhcp' in kwargs:
|
|
return self.original_subnet(*args, **kwargs)
|
|
return self.original_subnet(*args, enable_dhcp=False, **kwargs)
|
|
|
|
def external_subnet_by_list(self, *args, **kwargs):
|
|
if len(self.subnet_calls) > 0:
|
|
result = self.subnet_calls[0](*args, **kwargs)
|
|
del self.subnet_calls[0]
|
|
else:
|
|
# back to normal
|
|
self.subnet = self.original_subnet
|
|
result = self.subnet(*args, **kwargs)
|
|
return result
|
|
|
|
@contextlib.contextmanager
|
|
def floatingip_with_assoc(self, port_id=None, fmt=None, fixed_ip=None,
|
|
public_cidr='11.0.0.0/24', set_context=False,
|
|
tenant_id=None, **kwargs):
|
|
# Override super implementation to avoid changing the network to
|
|
# external after creation
|
|
with self._create_l3_ext_network() as ext_net,\
|
|
self.subnet(network=ext_net, cidr=public_cidr,
|
|
set_context=set_context,
|
|
tenant_id=tenant_id,
|
|
enable_dhcp=False) as public_sub:
|
|
private_port = None
|
|
if port_id:
|
|
private_port = self._show('ports', port_id)
|
|
with test_plugin.optional_ctx(
|
|
private_port, self.port,
|
|
set_context=set_context,
|
|
tenant_id=tenant_id) as private_port:
|
|
with self.router(set_context=set_context,
|
|
tenant_id=tenant_id) as r:
|
|
sid = private_port['port']['fixed_ips'][0]['subnet_id']
|
|
private_sub = {'subnet': {'id': sid}}
|
|
floatingip = None
|
|
|
|
self._add_external_gateway_to_router(
|
|
r['router']['id'],
|
|
public_sub['subnet']['network_id'])
|
|
self._router_interface_action(
|
|
'add', r['router']['id'],
|
|
private_sub['subnet']['id'], None)
|
|
|
|
floatingip = self._make_floatingip(
|
|
fmt or self.fmt,
|
|
public_sub['subnet']['network_id'],
|
|
port_id=private_port['port']['id'],
|
|
fixed_ip=fixed_ip,
|
|
tenant_id=tenant_id,
|
|
set_context=set_context,
|
|
**kwargs)
|
|
yield floatingip
|
|
|
|
if floatingip:
|
|
self._delete('floatingips',
|
|
floatingip['floatingip']['id'])
|
|
|
|
@contextlib.contextmanager
|
|
def floatingip_no_assoc(self, private_sub, fmt=None,
|
|
set_context=False, flavor_id=None, **kwargs):
|
|
# override super code to create an external subnet in advanced
|
|
with self.external_subnet(cidr='12.0.0.0/24') as public_sub:
|
|
with self.floatingip_no_assoc_with_public_sub(
|
|
private_sub, fmt, set_context, public_sub,
|
|
flavor_id, **kwargs) as (f, r):
|
|
# Yield only the floating ip object
|
|
yield f
|
|
|
|
|
|
# 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
|
|
|
|
|
|
@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):
|
|
self.subnet_calls.append(self.subnet)
|
|
self.subnet_calls.append(self.external_subnet)
|
|
|
|
|
|
def call_with_subnet_calls(self, f, *args, **kwargs):
|
|
self.subnet = self.external_subnet_by_list
|
|
result = f(*args, **kwargs)
|
|
self.subnet = self.original_subnet
|
|
return result
|
|
|
|
|
|
@decorator.decorator
|
|
def with_external_subnet_once(f, *args, **kwargs):
|
|
obj = args[0]
|
|
init_subnet_calls(obj, 1)
|
|
return call_with_subnet_calls(obj, f, *args, **kwargs)
|
|
|
|
|
|
@decorator.decorator
|
|
def with_external_subnet_second_time(f, *args, **kwargs):
|
|
obj = args[0]
|
|
init_subnet_calls(obj, 2)
|
|
return call_with_subnet_calls(obj, f, *args, **kwargs)
|
|
|
|
|
|
@decorator.decorator
|
|
def with_external_subnet_third_time(f, *args, **kwargs):
|
|
obj = args[0]
|
|
init_subnet_calls(obj, 3)
|
|
return call_with_subnet_calls(obj, f, *args, **kwargs)
|
|
|
|
|
|
@decorator.decorator
|
|
def with_external_network(f, *args, **kwargs):
|
|
obj = args[0]
|
|
obj.network = obj.external_network
|
|
obj.subnet = obj.external_subnet
|
|
obj._create_subnet = obj.create_external_subnet
|
|
result = f(*args, **kwargs)
|
|
obj._create_subnet = obj.original_create_subnet
|
|
obj.subnet = obj.original_subnet
|
|
obj.network = obj.original_network
|
|
return result
|
|
|
|
|
|
# Override subnet creation in some tests to create a subnet with dhcp
|
|
# disabled
|
|
@decorator.decorator
|
|
def with_no_dhcp_subnet(f, *args, **kwargs):
|
|
obj = args[0]
|
|
obj.subnet = obj.no_dhcp_subnet
|
|
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_filtering_by_cidr_used_on_create(self):
|
|
super(NsxV3TestSubnets,
|
|
self).test_list_subnets_filtering_by_cidr_used_on_create()
|
|
|
|
@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_pd_gw_values(self):
|
|
self.skipTest('Test not suited to the plugin DHCP code')
|
|
|
|
def test_create_subnet_ipv6_slaac_with_port_not_found(self):
|
|
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 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')
|
|
|
|
@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_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_create_port_additional_ip(self):
|
|
"""Test that creation of port with additional IP fails."""
|
|
with self.subnet() as subnet:
|
|
data = {'port': {'network_id': subnet['subnet']['network_id'],
|
|
'tenant_id': subnet['subnet']['tenant_id'],
|
|
'fixed_ips': [{'subnet_id':
|
|
subnet['subnet']['id']},
|
|
{'subnet_id':
|
|
subnet['subnet']['id']}]}}
|
|
port_req = self.new_create_request('ports', data)
|
|
res = port_req.get_response(self.api)
|
|
self.assertEqual(exc.HTTPBadRequest.code,
|
|
res.status_int)
|
|
|
|
def test_create_port_additional_ip_no_dhcp(self):
|
|
"""Creation of port with additional IP succeeds with DHCP off."""
|
|
with self.subnet(enable_dhcp=False) as subnet:
|
|
data = {'port': {'network_id': subnet['subnet']['network_id'],
|
|
'tenant_id': subnet['subnet']['tenant_id'],
|
|
'fixed_ips': [{'subnet_id':
|
|
subnet['subnet']['id']},
|
|
{'subnet_id':
|
|
subnet['subnet']['id']}]}}
|
|
port_req = self.new_create_request('ports', data)
|
|
res = port_req.get_response(self.api)
|
|
port = self.deserialize(self.fmt, res)
|
|
ips = port['port']['fixed_ips']
|
|
self.assertEqual(2, len(ips))
|
|
self.assertNotEqual(ips[0]['ip_address'],
|
|
ips[1]['ip_address'])
|
|
network_ip_net = netaddr.IPNetwork(subnet['subnet']['cidr'])
|
|
self.assertIn(ips[0]['ip_address'], network_ip_net)
|
|
self.assertIn(ips[1]['ip_address'], network_ip_net)
|
|
|
|
def test_update_port_add_additional_ip(self):
|
|
"""Test update of port with additional IP fails."""
|
|
with self.subnet() as subnet:
|
|
with self.port(subnet=subnet) as port:
|
|
data = {'port': {'admin_state_up': False,
|
|
'fixed_ips': [{'subnet_id':
|
|
subnet['subnet']['id']},
|
|
{'subnet_id':
|
|
subnet['subnet']['id']}]}}
|
|
req = self.new_update_request('ports', data,
|
|
port['port']['id'])
|
|
res = req.get_response(self.api)
|
|
self.assertEqual(exc.HTTPBadRequest.code,
|
|
res.status_int)
|
|
|
|
def test_update_port_add_additional_ip_no_dhcp(self):
|
|
"""Test update of port with additional IP succeeds if DHCP is off."""
|
|
with self.subnet(enable_dhcp=False) as subnet:
|
|
with self.port(subnet=subnet) as port:
|
|
data = {'port': {'admin_state_up': False,
|
|
'fixed_ips': [{'subnet_id':
|
|
subnet['subnet']['id']},
|
|
{'subnet_id':
|
|
subnet['subnet']['id']}]}}
|
|
req = self.new_update_request('ports', data,
|
|
port['port']['id'])
|
|
res = req.get_response(self.api)
|
|
port = self.deserialize(self.fmt, res)
|
|
ips = port['port']['fixed_ips']
|
|
self.assertEqual(2, len(ips))
|
|
self.assertNotEqual(ips[0]['ip_address'],
|
|
ips[1]['ip_address'])
|
|
network_ip_net = netaddr.IPNetwork(subnet['subnet']['cidr'])
|
|
self.assertIn(ips[0]['ip_address'], network_ip_net)
|
|
self.assertIn(ips[1]['ip_address'], network_ip_net)
|
|
|
|
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')
|