neutron/tests/unit/ipam/test_requests.py

   1 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
   2 #    not use this file except in compliance with the License. You may obtain
   3 #    a copy of the License at
   4 #
   5 #         http://www.apache.org/licenses/LICENSE-2.0
   6 #
   7 #    Unless required by applicable law or agreed to in writing, software
   8 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   9 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  10 #    License for the specific language governing permissions and limitations
  11 #    under the License.
  12
  13 import mock
  14 import netaddr
  15 from oslo_config import cfg
  16 from oslo_utils import uuidutils
  17
  18 from neutron.common import constants
  19 from neutron.common import ipv6_utils
  20 from neutron import context
  21 from neutron.ipam import driver
  22 from neutron.ipam import exceptions as ipam_exc
  23 from neutron.ipam import requests as ipam_req
  24 from neutron import manager
  25 from neutron.tests import base
  26 from neutron.tests.unit.ipam import fake_driver
  27
  28 FAKE_IPAM_CLASS = 'neutron.tests.unit.ipam.fake_driver.FakeDriver'
  29
  30
  31 class IpamSubnetRequestTestCase(base.BaseTestCase):
  32
  33     def setUp(self):
  34         super(IpamSubnetRequestTestCase, self).setUp()
  35         self.tenant_id = uuidutils.generate_uuid()
  36         self.subnet_id = uuidutils.generate_uuid()
  37
  38
  39 class TestIpamSubnetRequests(IpamSubnetRequestTestCase):
  40
  41     def test_subnet_request(self):
  42         pool = ipam_req.SubnetRequest(self.tenant_id,
  43                                   self.subnet_id)
  44         self.assertEqual(self.tenant_id, pool.tenant_id)
  45         self.assertEqual(self.subnet_id, pool.subnet_id)
  46         self.assertIsNone(pool.gateway_ip)
  47         self.assertIsNone(pool.allocation_pools)
  48
  49     def test_subnet_request_gateway(self):
  50         request = ipam_req.SubnetRequest(self.tenant_id,
  51                                      self.subnet_id,
  52                                      gateway_ip='1.2.3.1')
  53         self.assertEqual('1.2.3.1', str(request.gateway_ip))
  54
  55     def test_subnet_request_bad_gateway(self):
  56         self.assertRaises(netaddr.core.AddrFormatError,
  57                           ipam_req.SubnetRequest,
  58                           self.tenant_id,
  59                           self.subnet_id,
  60                           gateway_ip='1.2.3.')
  61
  62     def test_subnet_request_with_range(self):
  63         allocation_pools = [netaddr.IPRange('1.2.3.4', '1.2.3.5'),
  64                             netaddr.IPRange('1.2.3.7', '1.2.3.9')]
  65         request = ipam_req.SubnetRequest(self.tenant_id,
  66                                      self.subnet_id,
  67                                      allocation_pools=allocation_pools)
  68         self.assertEqual(allocation_pools, request.allocation_pools)
  69
  70     def test_subnet_request_range_not_list(self):
  71         self.assertRaises(TypeError,
  72                           ipam_req.SubnetRequest,
  73                           self.tenant_id,
  74                           self.subnet_id,
  75                           allocation_pools=1)
  76
  77     def test_subnet_request_bad_range(self):
  78         self.assertRaises(TypeError,
  79                           ipam_req.SubnetRequest,
  80                           self.tenant_id,
  81                           self.subnet_id,
  82                           allocation_pools=['1.2.3.4'])
  83
  84     def test_subnet_request_different_versions(self):
  85         pools = [netaddr.IPRange('0.0.0.1', '0.0.0.2'),
  86                  netaddr.IPRange('::1', '::2')]
  87         self.assertRaises(ValueError,
  88                           ipam_req.SubnetRequest,
  89                           self.tenant_id,
  90                           self.subnet_id,
  91                           allocation_pools=pools)
  92
  93     def test_subnet_request_overlap(self):
  94         pools = [netaddr.IPRange('0.0.0.10', '0.0.0.20'),
  95                  netaddr.IPRange('0.0.0.8', '0.0.0.10')]
  96         self.assertRaises(ValueError,
  97                           ipam_req.SubnetRequest,
  98                           self.tenant_id,
  99                           self.subnet_id,
 100                           allocation_pools=pools)
 101
 102
 103 class TestIpamAnySubnetRequest(IpamSubnetRequestTestCase):
 104
 105     def test_subnet_request(self):
 106         request = ipam_req.AnySubnetRequest(self.tenant_id,
 107                                         self.subnet_id,
 108                                         constants.IPv4,
 109                                         24,
 110                                         gateway_ip='0.0.0.1')
 111         self.assertEqual(24, request.prefixlen)
 112
 113     def test_subnet_request_bad_prefix_type(self):
 114         self.assertRaises(netaddr.core.AddrFormatError,
 115                           ipam_req.AnySubnetRequest,
 116                           self.tenant_id,
 117                           self.subnet_id,
 118                           constants.IPv4,
 119                           'A')
 120
 121     def test_subnet_request_bad_prefix(self):
 122         self.assertRaises(netaddr.core.AddrFormatError,
 123                           ipam_req.AnySubnetRequest,
 124                           self.tenant_id,
 125                           self.subnet_id,
 126                           constants.IPv4,
 127                           33)
 128         self.assertRaises(netaddr.core.AddrFormatError,
 129                           ipam_req.AnySubnetRequest,
 130                           self.tenant_id,
 131                           self.subnet_id,
 132                           constants.IPv6,
 133                           129)
 134
 135     def test_subnet_request_bad_gateway(self):
 136         self.assertRaises(ValueError,
 137                           ipam_req.AnySubnetRequest,
 138                           self.tenant_id,
 139                           self.subnet_id,
 140                           constants.IPv6,
 141                           64,
 142                           gateway_ip='2000::1')
 143
 144     def test_subnet_request_allocation_pool_wrong_version(self):
 145         pools = [netaddr.IPRange('0.0.0.4', '0.0.0.5')]
 146         self.assertRaises(ValueError,
 147                           ipam_req.AnySubnetRequest,
 148                           self.tenant_id,
 149                           self.subnet_id,
 150                           constants.IPv6,
 151                           64,
 152                           allocation_pools=pools)
 153
 154     def test_subnet_request_allocation_pool_not_in_net(self):
 155         pools = [netaddr.IPRange('0.0.0.64', '0.0.0.128')]
 156         self.assertRaises(ValueError,
 157                           ipam_req.AnySubnetRequest,
 158                           self.tenant_id,
 159                           self.subnet_id,
 160                           constants.IPv4,
 161                           25,
 162                           allocation_pools=pools)
 163
 164
 165 class TestIpamSpecificSubnetRequest(IpamSubnetRequestTestCase):
 166
 167     def test_subnet_request(self):
 168         request = ipam_req.SpecificSubnetRequest(self.tenant_id,
 169                                              self.subnet_id,
 170                                              '1.2.3.0/24',
 171                                              gateway_ip='1.2.3.1')
 172         self.assertEqual(24, request.prefixlen)
 173         self.assertEqual(netaddr.IPAddress('1.2.3.1'), request.gateway_ip)
 174         self.assertEqual(netaddr.IPNetwork('1.2.3.0/24'), request.subnet_cidr)
 175
 176     def test_subnet_request_bad_gateway(self):
 177         self.assertRaises(ValueError,
 178                           ipam_req.SpecificSubnetRequest,
 179                           self.tenant_id,
 180                           self.subnet_id,
 181                           '2001::1',
 182                           gateway_ip='2000::1')
 183
 184
 185 class TestAddressRequest(base.BaseTestCase):
 186
 187     # This class doesn't test much.  At least running through all of the
 188     # constructors may shake out some trivial bugs.
 189
 190     EUI64 = ipam_req.AutomaticAddressRequest.EUI64
 191
 192     def setUp(self):
 193         super(TestAddressRequest, self).setUp()
 194
 195     def test_specific_address_ipv6(self):
 196         request = ipam_req.SpecificAddressRequest('2000::45')
 197         self.assertEqual(netaddr.IPAddress('2000::45'), request.address)
 198
 199     def test_specific_address_ipv4(self):
 200         request = ipam_req.SpecificAddressRequest('1.2.3.32')
 201         self.assertEqual(netaddr.IPAddress('1.2.3.32'), request.address)
 202
 203     def test_any_address(self):
 204         ipam_req.AnyAddressRequest()
 205
 206     def test_automatic_address_request_eui64(self):
 207         subnet_cidr = '2607:f0d0:1002:51::/64'
 208         port_mac = 'aa:bb:cc:dd:ee:ff'
 209         eui_addr = str(ipv6_utils.get_ipv6_addr_by_EUI64(subnet_cidr,
 210                                                          port_mac))
 211         request = ipam_req.AutomaticAddressRequest(
 212             address_type=self.EUI64,
 213             prefix=subnet_cidr,
 214             mac=port_mac)
 215         self.assertEqual(request.address, netaddr.IPAddress(eui_addr))
 216
 217     def test_automatic_address_request_invalid_address_type_raises(self):
 218         self.assertRaises(ipam_exc.InvalidAddressType,
 219                           ipam_req.AutomaticAddressRequest,
 220                           address_type='kaboom')
 221
 222     def test_automatic_address_request_eui64_no_mac_raises(self):
 223         self.assertRaises(ipam_exc.AddressCalculationFailure,
 224                           ipam_req.AutomaticAddressRequest,
 225                           address_type=self.EUI64,
 226                           prefix='meh')
 227
 228     def test_automatic_address_request_eui64_alien_param_raises(self):
 229         self.assertRaises(ipam_exc.AddressCalculationFailure,
 230                           ipam_req.AutomaticAddressRequest,
 231                           address_type=self.EUI64,
 232                           mac='meh',
 233                           alien='et',
 234                           prefix='meh')
 235
 236
 237 class TestIpamDriverLoader(base.BaseTestCase):
 238
 239     def setUp(self):
 240         super(TestIpamDriverLoader, self).setUp()
 241         self.ctx = context.get_admin_context()
 242
 243     def _verify_fake_ipam_driver_is_loaded(self, driver_name):
 244         mgr = manager.NeutronManager
 245         ipam_driver = mgr.load_class_for_provider('neutron.ipam_drivers',
 246                                                   driver_name)
 247
 248         self.assertEqual(
 249             fake_driver.FakeDriver, ipam_driver,
 250             "loaded ipam driver should be FakeDriver")
 251
 252     def _verify_import_error_is_generated(self, driver_name):
 253         mgr = manager.NeutronManager
 254         self.assertRaises(ImportError, mgr.load_class_for_provider,
 255                           'neutron.ipam_drivers',
 256                           driver_name)
 257
 258     def test_ipam_driver_is_loaded_by_class(self):
 259         self._verify_fake_ipam_driver_is_loaded(FAKE_IPAM_CLASS)
 260
 261     def test_ipam_driver_is_loaded_by_name(self):
 262         self._verify_fake_ipam_driver_is_loaded('fake')
 263
 264     def test_ipam_driver_raises_import_error(self):
 265         self._verify_import_error_is_generated(
 266             'neutron.tests.unit.ipam_req.SomeNonExistentClass')
 267
 268     def test_ipam_driver_raises_import_error_for_none(self):
 269         self._verify_import_error_is_generated(None)
 270
 271     def _load_ipam_driver(self, driver_name, subnet_pool_id):
 272         cfg.CONF.set_override("ipam_driver", driver_name)
 273         return driver.Pool.get_instance(subnet_pool_id, self.ctx)
 274
 275     def test_ipam_driver_is_loaded_from_ipam_driver_config_value(self):
 276         ipam_driver = self._load_ipam_driver('fake', None)
 277         self.assertIsInstance(
 278             ipam_driver, fake_driver.FakeDriver,
 279             "loaded ipam driver should be of type FakeDriver")
 280
 281     @mock.patch(FAKE_IPAM_CLASS)
 282     def test_ipam_driver_is_loaded_with_subnet_pool_id(self, ipam_mock):
 283         subnet_pool_id = 'SomePoolID'
 284         self._load_ipam_driver('fake', subnet_pool_id)
 285         ipam_mock.assert_called_once_with(subnet_pool_id, self.ctx)
 286
 287
 288 class TestAddressRequestFactory(base.BaseTestCase):
 289
 290     def test_specific_address_request_is_loaded(self):
 291         for address in ('10.12.0.15', 'fffe::1'):
 292             ip = {'ip_address': address}
 293             self.assertIsInstance(
 294                 ipam_req.AddressRequestFactory.get_request(None, None, ip),
 295                 ipam_req.SpecificAddressRequest)
 296
 297     def test_any_address_request_is_loaded(self):
 298         for addr in [None, '']:
 299             ip = {'ip_address': addr}
 300             self.assertIsInstance(
 301                 ipam_req.AddressRequestFactory.get_request(None, None, ip),
 302                 ipam_req.AnyAddressRequest)
 303
 304     def test_automatic_address_request_is_loaded(self):
 305         ip = {'mac': '6c:62:6d:de:cf:49',
 306               'subnet_cidr': '2001:470:abcd::/64',
 307               'eui64_address': True}
 308         self.assertIsInstance(
 309             ipam_req.AddressRequestFactory.get_request(None, None, ip),
 310             ipam_req.AutomaticAddressRequest)
 311
 312
 313 class TestSubnetRequestFactory(IpamSubnetRequestTestCase):
 314
 315     def _build_subnet_dict(self, id=None, cidr='192.168.1.0/24',
 316                            prefixlen=8, ip_version=4):
 317         subnet = {'cidr': cidr,
 318                   'prefixlen': prefixlen,
 319                   'ip_version': ip_version,
 320                   'tenant_id': self.tenant_id,
 321                   'gateway_ip': None,
 322                   'allocation_pools': None,
 323                   'id': id or self.subnet_id}
 324         subnetpool = {'ip_version': ip_version,
 325                       'default_prefixlen': prefixlen}
 326         return subnet, subnetpool
 327
 328     def test_specific_subnet_request_is_loaded(self):
 329         addresses = [
 330             '10.12.0.15/24',
 331             '10.12.0.0/24',
 332             'fffe::1/64',
 333             'fffe::/64']
 334         for address in addresses:
 335             subnet, subnetpool = self._build_subnet_dict(cidr=address)
 336             self.assertIsInstance(
 337                 ipam_req.SubnetRequestFactory.get_request(None,
 338                                                           subnet,
 339                                                           subnetpool),
 340                 ipam_req.SpecificSubnetRequest)
 341
 342     def test_any_address_request_is_loaded_for_ipv4(self):
 343         subnet, subnetpool = self._build_subnet_dict(cidr=None, ip_version=4)
 344         self.assertIsInstance(
 345             ipam_req.SubnetRequestFactory.get_request(None,
 346                                                       subnet,
 347                                                       subnetpool),
 348             ipam_req.AnySubnetRequest)
 349
 350     def test_any_address_request_is_loaded_for_ipv6(self):
 351         subnet, subnetpool = self._build_subnet_dict(cidr=None, ip_version=6)
 352         self.assertIsInstance(
 353             ipam_req.SubnetRequestFactory.get_request(None,
 354                                                       subnet,
 355                                                       subnetpool),
 356             ipam_req.AnySubnetRequest)
 357
 358     def test_args_are_passed_to_specific_request(self):
 359         subnet, subnetpool = self._build_subnet_dict()
 360         request = ipam_req.SubnetRequestFactory.get_request(None,
 361                                                             subnet,
 362                                                             subnetpool)
 363         self.assertIsInstance(request,
 364                               ipam_req.SpecificSubnetRequest)
 365         self.assertEqual(self.tenant_id, request.tenant_id)
 366         self.assertEqual(self.subnet_id, request.subnet_id)
 367         self.assertIsNone(request.gateway_ip)
 368         self.assertIsNone(request.allocation_pools)
 369
 370
 371 class TestGetRequestFactory(base.BaseTestCase):
 372
 373     def setUp(self):
 374         super(TestGetRequestFactory, self).setUp()
 375         cfg.CONF.set_override('ipam_driver', 'fake')
 376         self.driver = driver.Pool.get_instance(None, None)
 377
 378     def test_get_subnet_request_factory(self):
 379         self.assertEqual(
 380             self.driver.get_subnet_request_factory(),
 381             ipam_req.SubnetRequestFactory)
 382
 383     def test_get_address_request_factory(self):
 384         self.assertEqual(
 385             self.driver.get_address_request_factory(),
 386             ipam_req.AddressRequestFactory)