397 lines
16 KiB
Python
397 lines
16 KiB
Python
# 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.
|
|
|
|
from unittest import mock
|
|
|
|
import netaddr
|
|
from neutron_lib import constants
|
|
from neutron_lib import context
|
|
from oslo_config import cfg
|
|
from oslo_utils import netutils
|
|
from oslo_utils import uuidutils
|
|
|
|
from neutron.ipam import driver
|
|
from neutron.ipam import exceptions as ipam_exc
|
|
from neutron.ipam import requests as ipam_req
|
|
from neutron import manager
|
|
from neutron.tests import base
|
|
from neutron.tests.unit.ipam import fake_driver
|
|
|
|
FAKE_IPAM_CLASS = 'neutron.tests.unit.ipam.fake_driver.FakeDriver'
|
|
|
|
|
|
class IpamSubnetRequestTestCase(base.BaseTestCase):
|
|
|
|
def setUp(self):
|
|
super(IpamSubnetRequestTestCase, self).setUp()
|
|
self.tenant_id = uuidutils.generate_uuid()
|
|
self.subnet_id = uuidutils.generate_uuid()
|
|
|
|
|
|
class TestIpamSubnetRequests(IpamSubnetRequestTestCase):
|
|
|
|
def test_subnet_request(self):
|
|
pool = ipam_req.SubnetRequest(self.tenant_id,
|
|
self.subnet_id)
|
|
self.assertEqual(self.tenant_id, pool.tenant_id)
|
|
self.assertEqual(self.subnet_id, pool.subnet_id)
|
|
self.assertIsNone(pool.gateway_ip)
|
|
self.assertIsNone(pool.allocation_pools)
|
|
|
|
def test_subnet_request_gateway(self):
|
|
request = ipam_req.SubnetRequest(self.tenant_id,
|
|
self.subnet_id,
|
|
gateway_ip='1.2.3.1')
|
|
self.assertEqual('1.2.3.1', str(request.gateway_ip))
|
|
|
|
def test_subnet_request_bad_gateway(self):
|
|
self.assertRaises(netaddr.core.AddrFormatError,
|
|
ipam_req.SubnetRequest,
|
|
self.tenant_id,
|
|
self.subnet_id,
|
|
gateway_ip='1.2.3.')
|
|
|
|
def test_subnet_request_with_range(self):
|
|
allocation_pools = [netaddr.IPRange('1.2.3.4', '1.2.3.5'),
|
|
netaddr.IPRange('1.2.3.7', '1.2.3.9')]
|
|
request = ipam_req.SubnetRequest(self.tenant_id,
|
|
self.subnet_id,
|
|
allocation_pools=allocation_pools)
|
|
self.assertEqual(allocation_pools, request.allocation_pools)
|
|
|
|
def test_subnet_request_range_not_list(self):
|
|
self.assertRaises(TypeError,
|
|
ipam_req.SubnetRequest,
|
|
self.tenant_id,
|
|
self.subnet_id,
|
|
allocation_pools=1)
|
|
|
|
def test_subnet_request_bad_range(self):
|
|
self.assertRaises(TypeError,
|
|
ipam_req.SubnetRequest,
|
|
self.tenant_id,
|
|
self.subnet_id,
|
|
allocation_pools=['1.2.3.4'])
|
|
|
|
def test_subnet_request_different_versions(self):
|
|
pools = [netaddr.IPRange('0.0.0.1', '0.0.0.2'),
|
|
netaddr.IPRange('::1', '::2')]
|
|
self.assertRaises(ValueError,
|
|
ipam_req.SubnetRequest,
|
|
self.tenant_id,
|
|
self.subnet_id,
|
|
allocation_pools=pools)
|
|
|
|
def test_subnet_request_overlap(self):
|
|
pools = [netaddr.IPRange('0.0.0.10', '0.0.0.20'),
|
|
netaddr.IPRange('0.0.0.8', '0.0.0.10')]
|
|
self.assertRaises(ValueError,
|
|
ipam_req.SubnetRequest,
|
|
self.tenant_id,
|
|
self.subnet_id,
|
|
allocation_pools=pools)
|
|
|
|
|
|
class TestIpamAnySubnetRequest(IpamSubnetRequestTestCase):
|
|
|
|
def test_subnet_request(self):
|
|
request = ipam_req.AnySubnetRequest(self.tenant_id,
|
|
self.subnet_id,
|
|
constants.IPv4,
|
|
24,
|
|
gateway_ip='0.0.0.1')
|
|
self.assertEqual(24, request.prefixlen)
|
|
|
|
def test_subnet_request_bad_prefix_type(self):
|
|
self.assertRaises(netaddr.core.AddrFormatError,
|
|
ipam_req.AnySubnetRequest,
|
|
self.tenant_id,
|
|
self.subnet_id,
|
|
constants.IPv4,
|
|
'A')
|
|
|
|
def test_subnet_request_bad_prefix(self):
|
|
self.assertRaises(netaddr.core.AddrFormatError,
|
|
ipam_req.AnySubnetRequest,
|
|
self.tenant_id,
|
|
self.subnet_id,
|
|
constants.IPv4,
|
|
33)
|
|
self.assertRaises(netaddr.core.AddrFormatError,
|
|
ipam_req.AnySubnetRequest,
|
|
self.tenant_id,
|
|
self.subnet_id,
|
|
constants.IPv6,
|
|
129)
|
|
|
|
def test_subnet_request_gateway(self):
|
|
request = ipam_req.AnySubnetRequest(self.tenant_id,
|
|
self.subnet_id,
|
|
constants.IPv6,
|
|
64,
|
|
gateway_ip='2000::1')
|
|
self.assertEqual(netaddr.IPAddress('2000::1'), request.gateway_ip)
|
|
|
|
def test_subnet_request_allocation_pool_wrong_version(self):
|
|
pools = [netaddr.IPRange('0.0.0.4', '0.0.0.5')]
|
|
self.assertRaises(ipam_exc.IpamValueInvalid,
|
|
ipam_req.AnySubnetRequest,
|
|
self.tenant_id,
|
|
self.subnet_id,
|
|
constants.IPv6,
|
|
64,
|
|
allocation_pools=pools)
|
|
|
|
def test_subnet_request_allocation_pool_not_in_net(self):
|
|
pools = [netaddr.IPRange('0.0.0.64', '0.0.0.128')]
|
|
self.assertRaises(ipam_exc.IpamValueInvalid,
|
|
ipam_req.AnySubnetRequest,
|
|
self.tenant_id,
|
|
self.subnet_id,
|
|
constants.IPv4,
|
|
25,
|
|
allocation_pools=pools)
|
|
|
|
|
|
class TestIpamSpecificSubnetRequest(IpamSubnetRequestTestCase):
|
|
|
|
def test_subnet_request(self):
|
|
request = ipam_req.SpecificSubnetRequest(self.tenant_id,
|
|
self.subnet_id,
|
|
'1.2.3.0/24',
|
|
gateway_ip='1.2.3.1')
|
|
self.assertEqual(24, request.prefixlen)
|
|
self.assertEqual(netaddr.IPAddress('1.2.3.1'), request.gateway_ip)
|
|
self.assertEqual(netaddr.IPNetwork('1.2.3.0/24'), request.subnet_cidr)
|
|
|
|
def test_subnet_request_gateway(self):
|
|
request = ipam_req.SpecificSubnetRequest(self.tenant_id,
|
|
self.subnet_id,
|
|
'2001::1',
|
|
gateway_ip='2000::1')
|
|
self.assertEqual(netaddr.IPAddress('2000::1'), request.gateway_ip)
|
|
|
|
|
|
class TestAddressRequest(base.BaseTestCase):
|
|
|
|
# This class doesn't test much. At least running through all of the
|
|
# constructors may shake out some trivial bugs.
|
|
|
|
EUI64 = ipam_req.AutomaticAddressRequest.EUI64
|
|
|
|
def test_specific_address_ipv6(self):
|
|
request = ipam_req.SpecificAddressRequest('2000::45')
|
|
self.assertEqual(netaddr.IPAddress('2000::45'), request.address)
|
|
|
|
def test_specific_address_ipv4(self):
|
|
request = ipam_req.SpecificAddressRequest('1.2.3.32')
|
|
self.assertEqual(netaddr.IPAddress('1.2.3.32'), request.address)
|
|
|
|
def test_any_address(self):
|
|
ipam_req.AnyAddressRequest()
|
|
|
|
def test_automatic_address_request_eui64(self):
|
|
subnet_cidr = '2607:f0d0:1002:51::/64'
|
|
port_mac = 'aa:bb:cc:dd:ee:ff'
|
|
eui_addr = str(netutils.get_ipv6_addr_by_EUI64(subnet_cidr,
|
|
port_mac))
|
|
request = ipam_req.AutomaticAddressRequest(
|
|
address_type=self.EUI64,
|
|
prefix=subnet_cidr,
|
|
mac=port_mac)
|
|
self.assertEqual(request.address, netaddr.IPAddress(eui_addr))
|
|
|
|
def test_automatic_address_request_invalid_address_type_raises(self):
|
|
self.assertRaises(ipam_exc.InvalidAddressType,
|
|
ipam_req.AutomaticAddressRequest,
|
|
address_type='kaboom')
|
|
|
|
def test_automatic_address_request_eui64_no_mac_raises(self):
|
|
self.assertRaises(ipam_exc.AddressCalculationFailure,
|
|
ipam_req.AutomaticAddressRequest,
|
|
address_type=self.EUI64,
|
|
prefix='meh')
|
|
|
|
def test_automatic_address_request_eui64_alien_param_raises(self):
|
|
self.assertRaises(ipam_exc.AddressCalculationFailure,
|
|
ipam_req.AutomaticAddressRequest,
|
|
address_type=self.EUI64,
|
|
mac='meh',
|
|
alien='et',
|
|
prefix='meh')
|
|
|
|
|
|
class TestIpamDriverLoader(base.BaseTestCase):
|
|
|
|
def setUp(self):
|
|
super(TestIpamDriverLoader, self).setUp()
|
|
self.ctx = context.get_admin_context()
|
|
|
|
def _verify_fake_ipam_driver_is_loaded(self, driver_name):
|
|
mgr = manager.NeutronManager
|
|
ipam_driver = mgr.load_class_for_provider('neutron.ipam_drivers',
|
|
driver_name)
|
|
|
|
self.assertEqual(
|
|
fake_driver.FakeDriver, ipam_driver,
|
|
"loaded ipam driver should be FakeDriver")
|
|
|
|
def _verify_import_error_is_generated(self, driver_name):
|
|
mgr = manager.NeutronManager
|
|
self.assertRaises(ImportError, mgr.load_class_for_provider,
|
|
'neutron.ipam_drivers',
|
|
driver_name)
|
|
|
|
def test_ipam_driver_is_loaded_by_class(self):
|
|
self._verify_fake_ipam_driver_is_loaded(FAKE_IPAM_CLASS)
|
|
|
|
def test_ipam_driver_is_loaded_by_name(self):
|
|
self._verify_fake_ipam_driver_is_loaded('fake')
|
|
|
|
def test_ipam_driver_raises_import_error(self):
|
|
self._verify_import_error_is_generated(
|
|
'neutron.tests.unit.ipam_req.SomeNonExistentClass')
|
|
|
|
def test_ipam_driver_raises_import_error_for_none(self):
|
|
self._verify_import_error_is_generated(None)
|
|
|
|
def _load_ipam_driver(self, driver_name, subnet_pool_id):
|
|
cfg.CONF.set_override("ipam_driver", driver_name)
|
|
return driver.Pool.get_instance(subnet_pool_id, self.ctx)
|
|
|
|
def test_ipam_driver_is_loaded_from_ipam_driver_config_value(self):
|
|
ipam_driver = self._load_ipam_driver('fake', None)
|
|
self.assertIsInstance(
|
|
ipam_driver, fake_driver.FakeDriver,
|
|
"loaded ipam driver should be of type FakeDriver")
|
|
|
|
@mock.patch(FAKE_IPAM_CLASS)
|
|
def test_ipam_driver_is_loaded_with_subnet_pool_id(self, ipam_mock):
|
|
subnet_pool_id = 'SomePoolID'
|
|
self._load_ipam_driver('fake', subnet_pool_id)
|
|
ipam_mock.assert_called_once_with(subnet_pool_id, self.ctx)
|
|
|
|
|
|
class TestAddressRequestFactory(base.BaseTestCase):
|
|
|
|
def test_specific_address_request_is_loaded(self):
|
|
for address in ('10.12.0.15', 'fffe::1'):
|
|
ip = {'ip_address': address}
|
|
port = {'device_owner':
|
|
constants.DEVICE_OWNER_COMPUTE_PREFIX + 'None'}
|
|
self.assertIsInstance(
|
|
ipam_req.AddressRequestFactory.get_request(None, port, ip),
|
|
ipam_req.SpecificAddressRequest)
|
|
|
|
def test_any_address_request_is_loaded(self):
|
|
for addr in [None, '']:
|
|
ip = {'ip_address': addr}
|
|
port = {'device_owner':
|
|
constants.DEVICE_OWNER_COMPUTE_PREFIX + 'None'}
|
|
self.assertIsInstance(
|
|
ipam_req.AddressRequestFactory.get_request(None, port, ip),
|
|
ipam_req.AnyAddressRequest)
|
|
|
|
def test_automatic_address_request_is_loaded(self):
|
|
ip = {'mac': '6c:62:6d:de:cf:49',
|
|
'subnet_cidr': '2001:470:abcd::/64',
|
|
'eui64_address': True}
|
|
port = {'device_owner': constants.DEVICE_OWNER_COMPUTE_PREFIX + 'None'}
|
|
self.assertIsInstance(
|
|
ipam_req.AddressRequestFactory.get_request(None, port, ip),
|
|
ipam_req.AutomaticAddressRequest)
|
|
|
|
def test_prefernext_address_request_on_dhcp_port(self):
|
|
ip = {}
|
|
port = {'device_owner': constants.DEVICE_OWNER_DHCP}
|
|
self.assertIsInstance(
|
|
ipam_req.AddressRequestFactory.get_request(None, port, ip),
|
|
ipam_req.PreferNextAddressRequest)
|
|
|
|
|
|
class TestSubnetRequestFactory(IpamSubnetRequestTestCase):
|
|
|
|
def _build_subnet_dict(self, id=None, cidr='192.168.1.0/24',
|
|
prefixlen=8, ip_version=constants.IP_VERSION_4):
|
|
subnet = {'cidr': cidr,
|
|
'prefixlen': prefixlen,
|
|
'ip_version': ip_version,
|
|
'tenant_id': self.tenant_id,
|
|
'gateway_ip': None,
|
|
'allocation_pools': None,
|
|
'id': id or self.subnet_id}
|
|
subnetpool = {'ip_version': ip_version,
|
|
'default_prefixlen': prefixlen}
|
|
return subnet, subnetpool
|
|
|
|
def test_specific_subnet_request_is_loaded(self):
|
|
addresses = [
|
|
'10.12.0.15/24',
|
|
'10.12.0.0/24',
|
|
'fffe::1/64',
|
|
'fffe::/64']
|
|
for address in addresses:
|
|
subnet, subnetpool = self._build_subnet_dict(cidr=address)
|
|
self.assertIsInstance(
|
|
ipam_req.SubnetRequestFactory.get_request(None,
|
|
subnet,
|
|
subnetpool),
|
|
ipam_req.SpecificSubnetRequest)
|
|
|
|
def test_any_address_request_is_loaded_for_ipv4(self):
|
|
subnet, subnetpool = self._build_subnet_dict(
|
|
cidr=None, ip_version=constants.IP_VERSION_4)
|
|
self.assertIsInstance(
|
|
ipam_req.SubnetRequestFactory.get_request(None,
|
|
subnet,
|
|
subnetpool),
|
|
ipam_req.AnySubnetRequest)
|
|
|
|
def test_any_address_request_is_loaded_for_ipv6(self):
|
|
subnet, subnetpool = self._build_subnet_dict(
|
|
cidr=None, ip_version=constants.IP_VERSION_6)
|
|
self.assertIsInstance(
|
|
ipam_req.SubnetRequestFactory.get_request(None,
|
|
subnet,
|
|
subnetpool),
|
|
ipam_req.AnySubnetRequest)
|
|
|
|
def test_args_are_passed_to_specific_request(self):
|
|
subnet, subnetpool = self._build_subnet_dict()
|
|
request = ipam_req.SubnetRequestFactory.get_request(None,
|
|
subnet,
|
|
subnetpool)
|
|
self.assertIsInstance(request,
|
|
ipam_req.SpecificSubnetRequest)
|
|
self.assertEqual(self.tenant_id, request.tenant_id)
|
|
self.assertEqual(self.subnet_id, request.subnet_id)
|
|
self.assertIsNone(request.gateway_ip)
|
|
self.assertIsNone(request.allocation_pools)
|
|
|
|
|
|
class TestGetRequestFactory(base.BaseTestCase):
|
|
|
|
def setUp(self):
|
|
super(TestGetRequestFactory, self).setUp()
|
|
cfg.CONF.set_override('ipam_driver', 'fake')
|
|
self.driver = driver.Pool.get_instance(None, None)
|
|
|
|
def test_get_subnet_request_factory(self):
|
|
self.assertEqual(
|
|
self.driver.get_subnet_request_factory(),
|
|
ipam_req.SubnetRequestFactory)
|
|
|
|
def test_get_address_request_factory(self):
|
|
self.assertEqual(
|
|
self.driver.get_address_request_factory(),
|
|
ipam_req.AddressRequestFactory)
|