neutron/tests/unit/extensions/test_l3_ext_gw_mode.py

   1 # Copyright 2013 VMware, Inc.
   2 # All rights reserved.
   3 #
   4 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
   5 #    not use this file except in compliance with the License. You may obtain
   6 #    a copy of the License at
   7 #
   8 #         http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 #    Unless required by applicable law or agreed to in writing, software
  11 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  12 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  13 #    License for the specific language governing permissions and limitations
  14 #    under the License.
  15 #
  16
  17 import mock
  18 from oslo_config import cfg
  19 from oslo_utils import uuidutils
  20 import testscenarios
  21 from webob import exc
  22
  23 from neutron.common import constants
  24 from neutron.db import api as db_api
  25 from neutron.db import external_net_db
  26 from neutron.db import l3_db
  27 from neutron.db import l3_gwmode_db
  28 from neutron.db import models_v2
  29 from neutron.extensions import l3
  30 from neutron.extensions import l3_ext_gw_mode
  31 from neutron.tests import base
  32 from neutron.tests.unit.db import test_db_base_plugin_v2
  33 from neutron.tests.unit.extensions import test_l3
  34 from neutron.tests.unit import testlib_api
  35
  36 _uuid = uuidutils.generate_uuid
  37 FAKE_GW_PORT_ID = _uuid()
  38 FAKE_GW_PORT_MAC = 'aa:bb:cc:dd:ee:ff'
  39 FAKE_FIP_EXT_PORT_ID = _uuid()
  40 FAKE_FIP_EXT_PORT_MAC = '11:22:33:44:55:66'
  41 FAKE_FIP_INT_PORT_ID = _uuid()
  42 FAKE_FIP_INT_PORT_MAC = 'aa:aa:aa:aa:aa:aa'
  43 FAKE_ROUTER_PORT_ID = _uuid()
  44 FAKE_ROUTER_PORT_MAC = 'bb:bb:bb:bb:bb:bb'
  45
  46
  47 class TestExtensionManager(object):
  48
  49     def get_resources(self):
  50         # Simulate extension of L3 attribute map
  51         for key in l3.RESOURCE_ATTRIBUTE_MAP.keys():
  52             l3.RESOURCE_ATTRIBUTE_MAP[key].update(
  53                 l3_ext_gw_mode.EXTENDED_ATTRIBUTES_2_0.get(key, {}))
  54         return l3.L3.get_resources()
  55
  56     def get_actions(self):
  57         return []
  58
  59     def get_request_extensions(self):
  60         return []
  61
  62
  63 # A simple class for making a concrete class out of the mixin
  64 # for the case of a plugin that integrates l3 routing.
  65 class TestDbIntPlugin(test_l3.TestL3NatIntPlugin,
  66                       l3_gwmode_db.L3_NAT_db_mixin):
  67
  68     supported_extension_aliases = ["external-net", "router", "ext-gw-mode"]
  69
  70
  71 # A simple class for making a concrete class out of the mixin
  72 # for the case of a l3 router service plugin
  73 class TestDbSepPlugin(test_l3.TestL3NatServicePlugin,
  74                       l3_gwmode_db.L3_NAT_db_mixin):
  75
  76     supported_extension_aliases = ["router", "ext-gw-mode"]
  77
  78
  79 class TestGetEnableSnat(testscenarios.WithScenarios, base.BaseTestCase):
  80     scenarios = [
  81         ('enabled', {'enable_snat_by_default': True}),
  82         ('disabled', {'enable_snat_by_default': False})]
  83
  84     def setUp(self):
  85         super(TestGetEnableSnat, self).setUp()
  86         self.config(enable_snat_by_default=self.enable_snat_by_default)
  87
  88     def _test_get_enable_snat(self, expected, info):
  89         observed = l3_gwmode_db.L3_NAT_dbonly_mixin._get_enable_snat(info)
  90         self.assertEqual(expected, observed)
  91
  92     def test_get_enable_snat_without_gw_info(self):
  93         self._test_get_enable_snat(self.enable_snat_by_default, {})
  94
  95     def test_get_enable_snat_without_enable_snat(self):
  96         info = {'network_id': _uuid()}
  97         self._test_get_enable_snat(self.enable_snat_by_default, info)
  98
  99     def test_get_enable_snat_with_snat_enabled(self):
 100         self._test_get_enable_snat(True, {'enable_snat': True})
 101
 102     def test_get_enable_snat_with_snat_disabled(self):
 103         self._test_get_enable_snat(False, {'enable_snat': False})
 104
 105
 106 class TestL3GwModeMixin(testlib_api.SqlTestCase):
 107
 108     def setUp(self):
 109         super(TestL3GwModeMixin, self).setUp()
 110         plugin = __name__ + '.' + TestDbIntPlugin.__name__
 111         self.setup_coreplugin(plugin)
 112         self.target_object = TestDbIntPlugin()
 113         # Patch the context
 114         ctx_patcher = mock.patch('neutron.context', autospec=True)
 115         mock_context = ctx_patcher.start()
 116         self.context = mock_context.get_admin_context()
 117         # This ensure also calls to elevated work in unit tests
 118         self.context.elevated.return_value = self.context
 119         self.context.session = db_api.get_session()
 120         # Create sample data for tests
 121         self.ext_net_id = _uuid()
 122         self.int_net_id = _uuid()
 123         self.int_sub_id = _uuid()
 124         self.tenant_id = 'the_tenant'
 125         self.network = models_v2.Network(
 126             id=self.ext_net_id,
 127             tenant_id=self.tenant_id,
 128             admin_state_up=True,
 129             status=constants.NET_STATUS_ACTIVE)
 130         self.net_ext = external_net_db.ExternalNetwork(
 131             network_id=self.ext_net_id)
 132         self.context.session.add(self.network)
 133         # The following is to avoid complaints from SQLite on
 134         # foreign key violations
 135         self.context.session.flush()
 136         self.context.session.add(self.net_ext)
 137         self.router = l3_db.Router(
 138             id=_uuid(),
 139             name=None,
 140             tenant_id=self.tenant_id,
 141             admin_state_up=True,
 142             status=constants.NET_STATUS_ACTIVE,
 143             enable_snat=True,
 144             gw_port_id=None)
 145         self.context.session.add(self.router)
 146         self.context.session.flush()
 147         self.router_gw_port = models_v2.Port(
 148             id=FAKE_GW_PORT_ID,
 149             tenant_id=self.tenant_id,
 150             device_id=self.router.id,
 151             device_owner=l3_db.DEVICE_OWNER_ROUTER_GW,
 152             admin_state_up=True,
 153             status=constants.PORT_STATUS_ACTIVE,
 154             mac_address=FAKE_GW_PORT_MAC,
 155             network_id=self.ext_net_id)
 156         self.router.gw_port_id = self.router_gw_port.id
 157         self.context.session.add(self.router)
 158         self.context.session.add(self.router_gw_port)
 159         self.context.session.flush()
 160         self.fip_ext_port = models_v2.Port(
 161             id=FAKE_FIP_EXT_PORT_ID,
 162             tenant_id=self.tenant_id,
 163             admin_state_up=True,
 164             device_id=self.router.id,
 165             device_owner=l3_db.DEVICE_OWNER_FLOATINGIP,
 166             status=constants.PORT_STATUS_ACTIVE,
 167             mac_address=FAKE_FIP_EXT_PORT_MAC,
 168             network_id=self.ext_net_id)
 169         self.context.session.add(self.fip_ext_port)
 170         self.context.session.flush()
 171         self.int_net = models_v2.Network(
 172             id=self.int_net_id,
 173             tenant_id=self.tenant_id,
 174             admin_state_up=True,
 175             status=constants.NET_STATUS_ACTIVE)
 176         self.int_sub = models_v2.Subnet(
 177             id=self.int_sub_id,
 178             tenant_id=self.tenant_id,
 179             ip_version=4,
 180             cidr='3.3.3.0/24',
 181             gateway_ip='3.3.3.1',
 182             network_id=self.int_net_id)
 183         self.router_port = models_v2.Port(
 184             id=FAKE_ROUTER_PORT_ID,
 185             tenant_id=self.tenant_id,
 186             admin_state_up=True,
 187             device_id=self.router.id,
 188             device_owner=l3_db.DEVICE_OWNER_ROUTER_INTF,
 189             status=constants.PORT_STATUS_ACTIVE,
 190             mac_address=FAKE_ROUTER_PORT_MAC,
 191             network_id=self.int_net_id)
 192         self.router_port_ip_info = models_v2.IPAllocation(
 193             port_id=self.router_port.id,
 194             network_id=self.int_net.id,
 195             subnet_id=self.int_sub_id,
 196             ip_address='3.3.3.1')
 197         self.context.session.add(self.int_net)
 198         self.context.session.add(self.int_sub)
 199         self.context.session.add(self.router_port)
 200         self.context.session.add(self.router_port_ip_info)
 201         self.context.session.flush()
 202         self.fip_int_port = models_v2.Port(
 203             id=FAKE_FIP_INT_PORT_ID,
 204             tenant_id=self.tenant_id,
 205             admin_state_up=True,
 206             device_id='something',
 207             device_owner=constants.DEVICE_OWNER_COMPUTE_PREFIX + 'nova',
 208             status=constants.PORT_STATUS_ACTIVE,
 209             mac_address=FAKE_FIP_INT_PORT_MAC,
 210             network_id=self.int_net_id)
 211         self.fip_int_ip_info = models_v2.IPAllocation(
 212             port_id=self.fip_int_port.id,
 213             network_id=self.int_net.id,
 214             subnet_id=self.int_sub_id,
 215             ip_address='3.3.3.3')
 216         self.fip = l3_db.FloatingIP(
 217             id=_uuid(),
 218             floating_ip_address='1.1.1.2',
 219             floating_network_id=self.ext_net_id,
 220             floating_port_id=FAKE_FIP_EXT_PORT_ID,
 221             fixed_port_id=None,
 222             fixed_ip_address=None,
 223             router_id=None)
 224         self.context.session.add(self.fip_int_port)
 225         self.context.session.add(self.fip_int_ip_info)
 226         self.context.session.add(self.fip)
 227         self.context.session.flush()
 228         self.fip_request = {'port_id': FAKE_FIP_INT_PORT_ID,
 229                             'tenant_id': self.tenant_id}
 230
 231     def _get_gwports_dict(self, gw_ports):
 232         return dict((gw_port['id'], gw_port)
 233                     for gw_port in gw_ports)
 234
 235     def _reset_ext_gw(self):
 236         # Reset external gateway
 237         self.router.gw_port_id = None
 238         self.context.session.add(self.router)
 239         self.context.session.flush()
 240
 241     def _test_update_router_gw(self, current_enable_snat, gw_info=None,
 242                                expected_enable_snat=True):
 243         if not current_enable_snat:
 244             previous_gw_info = {'network_id': self.ext_net_id,
 245                                 'enable_snat': current_enable_snat}
 246             self.target_object._update_router_gw_info(
 247                 self.context, self.router.id, previous_gw_info)
 248
 249         self.target_object._update_router_gw_info(
 250             self.context, self.router.id, gw_info)
 251         router = self.target_object._get_router(
 252             self.context, self.router.id)
 253         try:
 254             self.assertEqual(FAKE_GW_PORT_ID,
 255                              router.gw_port.id)
 256             self.assertEqual(FAKE_GW_PORT_MAC,
 257                              router.gw_port.mac_address)
 258         except AttributeError:
 259             self.assertIsNone(router.gw_port)
 260         self.assertEqual(expected_enable_snat, router.enable_snat)
 261
 262     def test_update_router_gw_with_gw_info_none(self):
 263         self._test_update_router_gw(current_enable_snat=True)
 264
 265     def test_update_router_gw_without_info_and_snat_disabled_previously(self):
 266         self._test_update_router_gw(current_enable_snat=False)
 267
 268     def test_update_router_gw_with_network_only(self):
 269         info = {'network_id': self.ext_net_id}
 270         self._test_update_router_gw(current_enable_snat=True, gw_info=info)
 271
 272     def test_update_router_gw_with_network_and_snat_disabled_previously(self):
 273         info = {'network_id': self.ext_net_id}
 274         self._test_update_router_gw(current_enable_snat=False, gw_info=info)
 275
 276     def test_update_router_gw_with_snat_disabled(self):
 277         info = {'network_id': self.ext_net_id,
 278                 'enable_snat': False}
 279         self._test_update_router_gw(
 280             current_enable_snat=True, gw_info=info, expected_enable_snat=False)
 281
 282     def test_update_router_gw_with_snat_enabled(self):
 283         info = {'network_id': self.ext_net_id,
 284                 'enable_snat': True}
 285         self._test_update_router_gw(current_enable_snat=False, gw_info=info)
 286
 287     def test_make_router_dict_no_ext_gw(self):
 288         self._reset_ext_gw()
 289         router_dict = self.target_object._make_router_dict(self.router)
 290         self.assertIsNone(router_dict[l3.EXTERNAL_GW_INFO])
 291
 292     def test_make_router_dict_with_ext_gw(self):
 293         router_dict = self.target_object._make_router_dict(self.router)
 294         self.assertEqual({'network_id': self.ext_net_id,
 295                           'enable_snat': True,
 296                           'external_fixed_ips': []},
 297                          router_dict[l3.EXTERNAL_GW_INFO])
 298
 299     def test_make_router_dict_with_ext_gw_snat_disabled(self):
 300         self.router.enable_snat = False
 301         router_dict = self.target_object._make_router_dict(self.router)
 302         self.assertEqual({'network_id': self.ext_net_id,
 303                           'enable_snat': False,
 304                           'external_fixed_ips': []},
 305                          router_dict[l3.EXTERNAL_GW_INFO])
 306
 307     def test_build_routers_list_no_ext_gw(self):
 308         self._reset_ext_gw()
 309         router_dict = self.target_object._make_router_dict(self.router)
 310         routers = self.target_object._build_routers_list(self.context,
 311                                                          [router_dict],
 312                                                          [])
 313         self.assertEqual(1, len(routers))
 314         router = routers[0]
 315         self.assertIsNone(router.get('gw_port'))
 316         self.assertIsNone(router.get('enable_snat'))
 317
 318     def test_build_routers_list_with_ext_gw(self):
 319         router_dict = self.target_object._make_router_dict(self.router)
 320         routers = self.target_object._build_routers_list(
 321             self.context, [router_dict],
 322             self._get_gwports_dict([self.router.gw_port]))
 323         self.assertEqual(1, len(routers))
 324         router = routers[0]
 325         self.assertIsNotNone(router.get('gw_port'))
 326         self.assertEqual(FAKE_GW_PORT_ID, router['gw_port']['id'])
 327         self.assertTrue(router.get('enable_snat'))
 328
 329     def test_build_routers_list_with_ext_gw_snat_disabled(self):
 330         self.router.enable_snat = False
 331         router_dict = self.target_object._make_router_dict(self.router)
 332         routers = self.target_object._build_routers_list(
 333             self.context, [router_dict],
 334             self._get_gwports_dict([self.router.gw_port]))
 335         self.assertEqual(1, len(routers))
 336         router = routers[0]
 337         self.assertIsNotNone(router.get('gw_port'))
 338         self.assertEqual(FAKE_GW_PORT_ID, router['gw_port']['id'])
 339         self.assertFalse(router.get('enable_snat'))
 340
 341     def test_build_routers_list_with_gw_port_mismatch(self):
 342         router_dict = self.target_object._make_router_dict(self.router)
 343         routers = self.target_object._build_routers_list(
 344             self.context, [router_dict], {})
 345         self.assertEqual(1, len(routers))
 346         router = routers[0]
 347         self.assertIsNone(router.get('gw_port'))
 348         self.assertIsNone(router.get('enable_snat'))
 349
 350
 351 class ExtGwModeIntTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase,
 352                            test_l3.L3NatTestCaseMixin):
 353
 354     def setUp(self, plugin=None, svc_plugins=None, ext_mgr=None):
 355         # Store l3 resource attribute map as it will be updated
 356         self._l3_attribute_map_bk = {}
 357         for item in l3.RESOURCE_ATTRIBUTE_MAP:
 358             self._l3_attribute_map_bk[item] = (
 359                 l3.RESOURCE_ATTRIBUTE_MAP[item].copy())
 360         plugin = plugin or (
 361             'neutron.tests.unit.extensions.test_l3_ext_gw_mode.'
 362             'TestDbIntPlugin')
 363         # for these tests we need to enable overlapping ips
 364         cfg.CONF.set_default('allow_overlapping_ips', True)
 365         ext_mgr = ext_mgr or TestExtensionManager()
 366         super(ExtGwModeIntTestCase, self).setUp(plugin=plugin,
 367                                                 ext_mgr=ext_mgr,
 368                                                 service_plugins=svc_plugins)
 369         self.addCleanup(self.restore_l3_attribute_map)
 370
 371     def restore_l3_attribute_map(self):
 372         l3.RESOURCE_ATTRIBUTE_MAP = self._l3_attribute_map_bk
 373
 374     def tearDown(self):
 375         super(ExtGwModeIntTestCase, self).tearDown()
 376
 377     def _set_router_external_gateway(self, router_id, network_id,
 378                                      snat_enabled=None,
 379                                      expected_code=exc.HTTPOk.code,
 380                                      neutron_context=None):
 381         ext_gw_info = {'network_id': network_id}
 382         # Need to set enable_snat also if snat_enabled == False
 383         if snat_enabled is not None:
 384             ext_gw_info['enable_snat'] = snat_enabled
 385         return self._update('routers', router_id,
 386                             {'router': {'external_gateway_info':
 387                                         ext_gw_info}},
 388                             expected_code=expected_code,
 389                             neutron_context=neutron_context)
 390
 391     def test_router_create_show_no_ext_gwinfo(self):
 392         name = 'router1'
 393         tenant_id = _uuid()
 394         expected_value = [('name', name), ('tenant_id', tenant_id),
 395                           ('admin_state_up', True), ('status', 'ACTIVE'),
 396                           ('external_gateway_info', None)]
 397         with self.router(name=name, admin_state_up=True,
 398                          tenant_id=tenant_id) as router:
 399             res = self._show('routers', router['router']['id'])
 400             for k, v in expected_value:
 401                 self.assertEqual(res['router'][k], v)
 402
 403     def _test_router_create_show_ext_gwinfo(self, snat_input_value,
 404                                             snat_expected_value):
 405         name = 'router1'
 406         tenant_id = _uuid()
 407         with self.subnet() as s:
 408             ext_net_id = s['subnet']['network_id']
 409             self._set_net_external(ext_net_id)
 410             input_value = {'network_id': ext_net_id}
 411             if snat_input_value in (True, False):
 412                 input_value['enable_snat'] = snat_input_value
 413             expected_value = [('name', name), ('tenant_id', tenant_id),
 414                               ('admin_state_up', True), ('status', 'ACTIVE'),
 415                               ('external_gateway_info',
 416                                {'network_id': ext_net_id,
 417                                 'enable_snat': snat_expected_value,
 418                                 'external_fixed_ips': [{
 419                                     'ip_address': mock.ANY,
 420                                     'subnet_id': s['subnet']['id']}]})]
 421             with self.router(
 422                 name=name, admin_state_up=True, tenant_id=tenant_id,
 423                 external_gateway_info=input_value) as router:
 424                 res = self._show('routers', router['router']['id'])
 425                 for k, v in expected_value:
 426                     self.assertEqual(res['router'][k], v)
 427
 428     def test_router_create_show_ext_gwinfo_default(self):
 429         self._test_router_create_show_ext_gwinfo(None, True)
 430
 431     def test_router_create_show_ext_gwinfo_with_snat_enabled(self):
 432         self._test_router_create_show_ext_gwinfo(True, True)
 433
 434     def test_router_create_show_ext_gwinfo_with_snat_disabled(self):
 435         self._test_router_create_show_ext_gwinfo(False, False)
 436
 437     def _test_router_update_ext_gwinfo(self, snat_input_value,
 438                                        snat_expected_value=False,
 439                                        expected_http_code=exc.HTTPOk.code):
 440         with self.router() as r:
 441             with self.subnet() as s:
 442                 try:
 443                     ext_net_id = s['subnet']['network_id']
 444                     self._set_net_external(ext_net_id)
 445                     self._set_router_external_gateway(
 446                         r['router']['id'], ext_net_id,
 447                         snat_enabled=snat_input_value,
 448                         expected_code=expected_http_code)
 449                     if expected_http_code != exc.HTTPOk.code:
 450                         return
 451                     body = self._show('routers', r['router']['id'])
 452                     res_gw_info = body['router']['external_gateway_info']
 453                     self.assertEqual(res_gw_info['network_id'], ext_net_id)
 454                     self.assertEqual(res_gw_info['enable_snat'],
 455                                      snat_expected_value)
 456                 finally:
 457                     self._remove_external_gateway_from_router(
 458                         r['router']['id'], ext_net_id)
 459
 460     def test_router_update_ext_gwinfo_default(self):
 461         self._test_router_update_ext_gwinfo(None, True)
 462
 463     def test_router_update_ext_gwinfo_with_snat_enabled(self):
 464         self._test_router_update_ext_gwinfo(True, True)
 465
 466     def test_router_update_ext_gwinfo_with_snat_disabled(self):
 467         self._test_router_update_ext_gwinfo(False, False)
 468
 469     def test_router_update_ext_gwinfo_with_invalid_snat_setting(self):
 470         self._test_router_update_ext_gwinfo(
 471             'xxx', None, expected_http_code=exc.HTTPBadRequest.code)
 472
 473
 474 class ExtGwModeSepTestCase(ExtGwModeIntTestCase):
 475
 476     def setUp(self, plugin=None):
 477         # Store l3 resource attribute map as it will be updated
 478         self._l3_attribute_map_bk = {}
 479         for item in l3.RESOURCE_ATTRIBUTE_MAP:
 480             self._l3_attribute_map_bk[item] = (
 481                 l3.RESOURCE_ATTRIBUTE_MAP[item].copy())
 482         plugin = plugin or (
 483             'neutron.tests.unit.extensions.test_l3.TestNoL3NatPlugin')
 484         # the L3 service plugin
 485         l3_plugin = ('neutron.tests.unit.extensions.test_l3_ext_gw_mode.'
 486                      'TestDbSepPlugin')
 487         svc_plugins = {'l3_plugin_name': l3_plugin}
 488         # for these tests we need to enable overlapping ips
 489         cfg.CONF.set_default('allow_overlapping_ips', True)
 490         super(ExtGwModeSepTestCase, self).setUp(plugin=plugin,
 491                                                 svc_plugins=svc_plugins)
 492         self.addCleanup(self.restore_l3_attribute_map)