neutron/tests/unit/extensions/test_address_scope.py

   1 # Copyright (c) 2015 Red Hat, Inc.
   2 #
   3 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
   4 #    not use this file except in compliance with the License. You may obtain
   5 #    a copy of the License at
   6 #
   7 #         http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 #    Unless required by applicable law or agreed to in writing, software
  10 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  11 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  12 #    License for the specific language governing permissions and limitations
  13 #    under the License.
  14
  15 import contextlib
  16 import netaddr
  17
  18 import webob.exc
  19
  20 from neutron.api.v2 import attributes as attr
  21 from neutron.common import constants
  22 from neutron import context
  23 from neutron.db import address_scope_db
  24 from neutron.db import db_base_plugin_v2
  25 from neutron.extensions import address_scope as ext_address_scope
  26 from neutron.tests.unit.db import test_db_base_plugin_v2
  27
  28 DB_PLUGIN_KLASS = ('neutron.tests.unit.extensions.test_address_scope.'
  29                    'AddressScopeTestPlugin')
  30
  31
  32 class AddressScopeTestExtensionManager(object):
  33
  34     def get_resources(self):
  35         # Add the resources to the global attribute map
  36         # This is done here as the setup process won't
  37         # initialize the main API router which extends
  38         # the global attribute map
  39         attr.RESOURCE_ATTRIBUTE_MAP.update(
  40             ext_address_scope.RESOURCE_ATTRIBUTE_MAP)
  41         return ext_address_scope.Address_scope.get_resources()
  42
  43     def get_actions(self):
  44         return []
  45
  46     def get_request_extensions(self):
  47         return []
  48
  49
  50 class AddressScopeTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
  51
  52     def _create_address_scope(self, fmt, ip_version=constants.IP_VERSION_4,
  53                               expected_res_status=None, admin=False, **kwargs):
  54         address_scope = {'address_scope': {}}
  55         address_scope['address_scope']['ip_version'] = ip_version
  56         for k, v in kwargs.items():
  57             address_scope['address_scope'][k] = str(v)
  58
  59         address_scope_req = self.new_create_request('address-scopes',
  60                                                     address_scope, fmt)
  61
  62         if not admin:
  63             neutron_context = context.Context('', kwargs.get('tenant_id',
  64                                                              self._tenant_id))
  65             address_scope_req.environ['neutron.context'] = neutron_context
  66
  67         address_scope_res = address_scope_req.get_response(self.ext_api)
  68         if expected_res_status:
  69             self.assertEqual(address_scope_res.status_int, expected_res_status)
  70         return address_scope_res
  71
  72     def _make_address_scope(self, fmt, ip_version, admin=False, **kwargs):
  73         res = self._create_address_scope(fmt, ip_version,
  74                                          admin=admin, **kwargs)
  75         if res.status_int >= webob.exc.HTTPClientError.code:
  76             raise webob.exc.HTTPClientError(code=res.status_int)
  77         return self.deserialize(fmt, res)
  78
  79     @contextlib.contextmanager
  80     def address_scope(self, ip_version=constants.IP_VERSION_4,
  81                       admin=False, **kwargs):
  82         addr_scope = self._make_address_scope(self.fmt, ip_version,
  83                                               admin, **kwargs)
  84         yield addr_scope
  85
  86     def _test_create_address_scope(self, ip_version=constants.IP_VERSION_4,
  87                                    admin=False, expected=None, **kwargs):
  88         keys = kwargs.copy()
  89         keys.setdefault('tenant_id', self._tenant_id)
  90         with self.address_scope(ip_version,
  91                                 admin=admin, **keys) as addr_scope:
  92             keys['ip_version'] = ip_version
  93             self._validate_resource(addr_scope, keys, 'address_scope')
  94             if expected:
  95                 self._compare_resource(addr_scope, expected, 'address_scope')
  96         return addr_scope
  97
  98     def _test_update_address_scope(self, addr_scope_id, data, admin=False,
  99                                    expected=None, tenant_id=None):
 100         update_req = self.new_update_request(
 101             'address-scopes', data, addr_scope_id)
 102         if not admin:
 103             neutron_context = context.Context('', tenant_id or self._tenant_id)
 104             update_req.environ['neutron.context'] = neutron_context
 105
 106         update_res = update_req.get_response(self.ext_api)
 107         if expected:
 108             addr_scope = self.deserialize(self.fmt, update_res)
 109             self._compare_resource(addr_scope, expected, 'address_scope')
 110             return addr_scope
 111
 112         return update_res
 113
 114
 115 class AddressScopeTestPlugin(db_base_plugin_v2.NeutronDbPluginV2,
 116                              address_scope_db.AddressScopeDbMixin):
 117     __native_pagination_support = True
 118     __native_sorting_support = True
 119
 120     supported_extension_aliases = ["address-scope"]
 121
 122
 123 class TestAddressScope(AddressScopeTestCase):
 124
 125     def setUp(self):
 126         plugin = DB_PLUGIN_KLASS
 127         ext_mgr = AddressScopeTestExtensionManager()
 128         super(TestAddressScope, self).setUp(plugin=plugin, ext_mgr=ext_mgr)
 129
 130     def test_create_address_scope_ipv4(self):
 131         expected_addr_scope = {'name': 'foo-address-scope',
 132                                'tenant_id': self._tenant_id,
 133                                'shared': False,
 134                                'ip_version': constants.IP_VERSION_4}
 135         self._test_create_address_scope(name='foo-address-scope',
 136                                         expected=expected_addr_scope)
 137
 138     def test_create_address_scope_ipv6(self):
 139         expected_addr_scope = {'name': 'foo-address-scope',
 140                                'tenant_id': self._tenant_id,
 141                                'shared': False,
 142                                'ip_version': constants.IP_VERSION_6}
 143         self._test_create_address_scope(constants.IP_VERSION_6,
 144                                         name='foo-address-scope',
 145                                         expected=expected_addr_scope)
 146
 147     def test_create_address_scope_empty_name(self):
 148         expected_addr_scope = {'name': '',
 149                                'tenant_id': self._tenant_id,
 150                                'shared': False}
 151         self._test_create_address_scope(name='', expected=expected_addr_scope)
 152
 153         # no name specified
 154         self._test_create_address_scope(expected=expected_addr_scope)
 155
 156     def test_create_address_scope_shared_admin(self):
 157         expected_addr_scope = {'name': 'foo-address-scope', 'shared': True}
 158         self._test_create_address_scope(name='foo-address-scope', admin=True,
 159                                         shared=True,
 160                                         expected=expected_addr_scope)
 161
 162     def test_created_address_scope_shared_non_admin(self):
 163         res = self._create_address_scope(self.fmt, name='foo-address-scope',
 164                                          tenant_id=self._tenant_id,
 165                                          admin=False, shared=True)
 166         self.assertEqual(webob.exc.HTTPForbidden.code, res.status_int)
 167
 168     def test_created_address_scope_specify_id(self):
 169         res = self._create_address_scope(self.fmt, name='foo-address-scope',
 170                                          id='foo-id')
 171         self.assertEqual(webob.exc.HTTPClientError.code, res.status_int)
 172
 173     def test_delete_address_scope(self):
 174         with self.address_scope(name='foo-address-scope') as addr_scope:
 175             self._delete('address-scopes', addr_scope['address_scope']['id'])
 176             self._show('address-scopes', addr_scope['address_scope']['id'],
 177                        expected_code=webob.exc.HTTPNotFound.code)
 178
 179     def test_update_address_scope(self):
 180         addr_scope = self._test_create_address_scope(name='foo-address-scope')
 181         data = {'address_scope': {'name': 'bar-address-scope'}}
 182         self._test_update_address_scope(addr_scope['address_scope']['id'],
 183                                         data, expected=data['address_scope'])
 184
 185     def test_update_address_scope_shared_true_admin(self):
 186         addr_scope = self._test_create_address_scope(name='foo-address-scope')
 187         data = {'address_scope': {'shared': True}}
 188         self._test_update_address_scope(addr_scope['address_scope']['id'],
 189                                         data, admin=True,
 190                                         expected=data['address_scope'])
 191
 192     def test_update_address_scope_shared_true_non_admin(self):
 193         addr_scope = self._test_create_address_scope(name='foo-address-scope')
 194         data = {'address_scope': {'shared': True}}
 195         res = self._test_update_address_scope(
 196             addr_scope['address_scope']['id'], data, admin=False)
 197         self.assertEqual(webob.exc.HTTPForbidden.code, res.status_int)
 198
 199     def test_update_address_scope_shared_false_admin(self):
 200         addr_scope = self._test_create_address_scope(name='foo-address-scope',
 201                                                      admin=True, shared=True)
 202         data = {'address_scope': {'shared': False}}
 203         res = self._test_update_address_scope(
 204             addr_scope['address_scope']['id'], data, admin=True)
 205         self.assertEqual(webob.exc.HTTPClientError.code, res.status_int)
 206
 207     def test_get_address_scope(self):
 208         addr_scope = self._test_create_address_scope(name='foo-address-scope')
 209         req = self.new_show_request('address-scopes',
 210                                     addr_scope['address_scope']['id'])
 211         res = self.deserialize(self.fmt, req.get_response(self.ext_api))
 212         self.assertEqual(addr_scope['address_scope']['id'],
 213                          res['address_scope']['id'])
 214
 215     def test_get_address_scope_different_tenants_not_shared(self):
 216         addr_scope = self._test_create_address_scope(name='foo-address-scope')
 217         req = self.new_show_request('address-scopes',
 218                                     addr_scope['address_scope']['id'])
 219         neutron_context = context.Context('', 'not-the-owner')
 220         req.environ['neutron.context'] = neutron_context
 221         res = req.get_response(self.ext_api)
 222         self.assertEqual(webob.exc.HTTPNotFound.code, res.status_int)
 223
 224     def test_get_address_scope_different_tenants_shared(self):
 225         addr_scope = self._test_create_address_scope(name='foo-address-scope',
 226                                                      shared=True, admin=True)
 227         req = self.new_show_request('address-scopes',
 228                                     addr_scope['address_scope']['id'])
 229         neutron_context = context.Context('', 'test-tenant-2')
 230         req.environ['neutron.context'] = neutron_context
 231         res = self.deserialize(self.fmt, req.get_response(self.ext_api))
 232         self.assertEqual(addr_scope['address_scope']['id'],
 233                          res['address_scope']['id'])
 234
 235     def test_list_address_scopes(self):
 236         self._test_create_address_scope(name='foo-address-scope')
 237         self._test_create_address_scope(constants.IP_VERSION_6,
 238                                         name='bar-address-scope')
 239         res = self._list('address-scopes')
 240         self.assertEqual(2, len(res['address_scopes']))
 241
 242     def test_list_address_scopes_different_tenants_shared(self):
 243         self._test_create_address_scope(name='foo-address-scope', shared=True,
 244                                         admin=True)
 245         admin_res = self._list('address-scopes')
 246         mortal_res = self._list(
 247             'address-scopes',
 248             neutron_context=context.Context('', 'not-the-owner'))
 249         self.assertEqual(1, len(admin_res['address_scopes']))
 250         self.assertEqual(1, len(mortal_res['address_scopes']))
 251
 252     def test_list_address_scopes_different_tenants_not_shared(self):
 253         self._test_create_address_scope(constants.IP_VERSION_6,
 254                                         name='foo-address-scope')
 255         admin_res = self._list('address-scopes')
 256         mortal_res = self._list(
 257             'address-scopes',
 258             neutron_context=context.Context('', 'not-the-owner'))
 259         self.assertEqual(1, len(admin_res['address_scopes']))
 260         self.assertEqual(0, len(mortal_res['address_scopes']))
 261
 262
 263 class TestSubnetPoolsWithAddressScopes(AddressScopeTestCase):
 264     def setUp(self):
 265         plugin = DB_PLUGIN_KLASS
 266         ext_mgr = AddressScopeTestExtensionManager()
 267         super(TestSubnetPoolsWithAddressScopes, self).setUp(plugin=plugin,
 268                                                             ext_mgr=ext_mgr)
 269
 270     def _test_create_subnetpool(self, prefixes, expected=None,
 271                                 admin=False, **kwargs):
 272         keys = kwargs.copy()
 273         keys.setdefault('tenant_id', self._tenant_id)
 274         with self.subnetpool(prefixes, admin, **keys) as subnetpool:
 275             self._validate_resource(subnetpool, keys, 'subnetpool')
 276             if expected:
 277                 self._compare_resource(subnetpool, expected, 'subnetpool')
 278         return subnetpool
 279
 280     def test_create_subnetpool_associate_address_scope(self):
 281         with self.address_scope(name='foo-address-scope') as addr_scope:
 282             address_scope_id = addr_scope['address_scope']['id']
 283             subnet = netaddr.IPNetwork('10.10.10.0/24')
 284             expected = {'address_scope_id': address_scope_id}
 285             self._test_create_subnetpool([subnet.cidr], expected=expected,
 286                                          name='foo-subnetpool',
 287                                          min_prefixlen='21',
 288                                          address_scope_id=address_scope_id)
 289
 290     def test_create_subnetpool_associate_invalid_address_scope(self):
 291         self.assertRaises(
 292             webob.exc.HTTPClientError, self._test_create_subnetpool, [],
 293             min_prefixlen='21', address_scope_id='foo-addr-scope-id')
 294
 295     def test_create_subnetpool_assoc_address_scope_with_prefix_intersect(self):
 296         with self.address_scope(name='foo-address-scope') as addr_scope:
 297             address_scope_id = addr_scope['address_scope']['id']
 298             subnet = netaddr.IPNetwork('10.10.10.0/24')
 299             expected = {'address_scope_id': address_scope_id}
 300             self._test_create_subnetpool([subnet.cidr], expected=expected,
 301                                          name='foo-subnetpool',
 302                                          min_prefixlen='21',
 303                                          address_scope_id=address_scope_id)
 304             overlap_subnet = netaddr.IPNetwork('10.10.10.10/24')
 305             self.assertRaises(
 306                 webob.exc.HTTPClientError, self._test_create_subnetpool,
 307                 [overlap_subnet.cidr], min_prefixlen='21',
 308                 address_scope_id=address_scope_id)
 309
 310     def test_update_subnetpool_associate_address_scope(self):
 311         subnet = netaddr.IPNetwork('10.10.10.0/24')
 312         initial_subnetpool = self._test_create_subnetpool([subnet.cidr],
 313                                                           name='foo-sp',
 314                                                           min_prefixlen='21')
 315         with self.address_scope(name='foo-address-scope') as addr_scope:
 316             address_scope_id = addr_scope['address_scope']['id']
 317             data = {'subnetpool': {'address_scope_id': address_scope_id}}
 318             req = self.new_update_request(
 319                 'subnetpools', data, initial_subnetpool['subnetpool']['id'])
 320             api = self._api_for_resource('subnetpools')
 321             res = self.deserialize(self.fmt, req.get_response(api))
 322             self._compare_resource(res, data['subnetpool'], 'subnetpool')
 323
 324     def test_update_subnetpool_associate_invalid_address_scope(self):
 325         subnet = netaddr.IPNetwork('10.10.10.0/24')
 326         initial_subnetpool = self._test_create_subnetpool([subnet.cidr],
 327                                                           name='foo-sp',
 328                                                           min_prefixlen='21')
 329         data = {'subnetpool': {'address_scope_id': 'foo-addr-scope-id'}}
 330         req = self.new_update_request(
 331             'subnetpools', data, initial_subnetpool['subnetpool']['id'])
 332         api = self._api_for_resource('subnetpools')
 333         res = req.get_response(api)
 334         self.assertEqual(webob.exc.HTTPClientError.code, res.status_int)
 335
 336     def test_update_subnetpool_disassociate_address_scope(self):
 337         with self.address_scope(name='foo-address-scope') as addr_scope:
 338             address_scope_id = addr_scope['address_scope']['id']
 339             subnet = netaddr.IPNetwork('10.10.10.0/24')
 340             expected = {'address_scope_id': address_scope_id}
 341             initial_subnetpool = self._test_create_subnetpool(
 342                 [subnet.cidr], expected=expected, name='foo-sp',
 343                 min_prefixlen='21', address_scope_id=address_scope_id)
 344
 345             data = {'subnetpool': {'address_scope_id': None}}
 346             req = self.new_update_request(
 347                 'subnetpools', data, initial_subnetpool['subnetpool']['id'])
 348             api = self._api_for_resource('subnetpools')
 349             res = self.deserialize(self.fmt, req.get_response(api))
 350             self._compare_resource(res, data['subnetpool'], 'subnetpool')
 351
 352     def test_update_subnetpool_associate_another_address_scope(self):
 353         with self.address_scope(name='foo-address-scope') as addr_scope:
 354             address_scope_id = addr_scope['address_scope']['id']
 355             subnet = netaddr.IPNetwork('10.10.10.0/24')
 356             expected = {'address_scope_id': address_scope_id}
 357             initial_subnetpool = self._test_create_subnetpool(
 358                 [subnet.cidr], expected=expected, name='foo-sp',
 359                 min_prefixlen='21', address_scope_id=address_scope_id)
 360
 361             with self.address_scope(name='foo-address-scope') as other_a_s:
 362                 other_a_s_id = other_a_s['address_scope']['id']
 363                 update_data = {'subnetpool': {'address_scope_id':
 364                                               other_a_s_id}}
 365                 req = self.new_update_request(
 366                     'subnetpools', update_data,
 367                     initial_subnetpool['subnetpool']['id'])
 368                 api = self._api_for_resource('subnetpools')
 369                 res = self.deserialize(self.fmt, req.get_response(api))
 370                 self._compare_resource(res, update_data['subnetpool'],
 371                                        'subnetpool')
 372
 373     def test_delete_address_scope_in_use(self):
 374         with self.address_scope(name='foo-address-scope') as addr_scope:
 375             address_scope_id = addr_scope['address_scope']['id']
 376             subnet = netaddr.IPNetwork('10.10.10.0/24')
 377             expected = {'address_scope_id': address_scope_id}
 378             self._test_create_subnetpool([subnet.cidr], expected=expected,
 379                                          name='foo-subnetpool',
 380                                          min_prefixlen='21',
 381                                          address_scope_id=address_scope_id)
 382             self._delete('address-scopes', address_scope_id,
 383                          expected_code=webob.exc.HTTPConflict.code)
 384
 385     def test_add_subnetpool_address_scope_wrong_address_family(self):
 386         with self.address_scope(constants.IP_VERSION_6,
 387                                 name='foo-address-scope') as addr_scope:
 388             address_scope_id = addr_scope['address_scope']['id']
 389             subnet = netaddr.IPNetwork('10.10.10.0/24')
 390             self.assertRaises(webob.exc.HTTPClientError,
 391                               self._test_create_subnetpool,
 392                               [subnet.cidr], name='foo-subnetpool',
 393                               min_prefixlen='21',
 394                               address_scope_id=address_scope_id)
 395
 396     def test_update_subnetpool_associate_address_scope_wrong_family(self):
 397         with self.address_scope(constants.IP_VERSION_6,
 398                                 name='foo-address-scope') as addr_scope:
 399             address_scope_id = addr_scope['address_scope']['id']
 400             subnet = netaddr.IPNetwork('2001:db8::/64')
 401             expected = {'address_scope_id': address_scope_id}
 402             initial_subnetpool = self._test_create_subnetpool(
 403                 [subnet.cidr], expected=expected, name='foo-sp',
 404                 min_prefixlen='64', address_scope_id=address_scope_id)
 405
 406             with self.address_scope(name='foo-address-scope') as other_a_s:
 407                 other_a_s_id = other_a_s['address_scope']['id']
 408                 update_data = {'subnetpool': {'address_scope_id':
 409                                               other_a_s_id}}
 410                 req = self.new_update_request(
 411                     'subnetpools', update_data,
 412                     initial_subnetpool['subnetpool']['id'])
 413                 api = self._api_for_resource('subnetpools')
 414                 res = req.get_response(api)
 415                 self.assertEqual(webob.exc.HTTPBadRequest.code,
 416                                  res.status_int)