snaps/openstack/tests/create_router_tests.py

   1 # Copyright (c) 2017 Cable Television Laboratories, Inc. ("CableLabs")
   2 #                    and others.  All rights reserved.
   3 #
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain 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,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15 import unittest
  16 import uuid
  17
  18 from snaps.config.network import PortConfig, NetworkConfig
  19 from snaps.config.router import RouterConfigError, RouterConfig
  20 from snaps.openstack import create_network
  21 from snaps.openstack import create_router
  22 from snaps.openstack.create_network import OpenStackNetwork
  23 from snaps.openstack.create_router import RouterSettings
  24 from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase
  25 from snaps.openstack.utils import neutron_utils, settings_utils
  26
  27 __author__ = 'mmakati'
  28
  29 cidr1 = '10.200.201.0/24'
  30 cidr2 = '10.200.202.0/24'
  31 static_gateway_ip1 = '10.200.201.1'
  32 static_gateway_ip2 = '10.200.202.1'
  33
  34
  35 class RouterSettingsUnitTests(unittest.TestCase):
  36     """
  37     Class for testing the RouterSettings class
  38     """
  39
  40     def test_no_params(self):
  41         with self.assertRaises(RouterConfigError):
  42             RouterSettings()
  43
  44     def test_empty_config(self):
  45         with self.assertRaises(RouterConfigError):
  46             RouterSettings(**dict())
  47
  48     def test_name_only(self):
  49         settings = RouterSettings(name='foo')
  50         self.assertEqual('foo', settings.name)
  51         self.assertIsNone(settings.project_name)
  52         self.assertIsNone(settings.external_gateway)
  53         self.assertTrue(settings.admin_state_up)
  54         self.assertIsNone(settings.enable_snat)
  55         self.assertIsNotNone(settings.internal_subnets)
  56         self.assertTrue(isinstance(settings.internal_subnets, list))
  57         self.assertEqual(0, len(settings.internal_subnets))
  58         self.assertIsNotNone(settings.port_settings)
  59         self.assertTrue(isinstance(settings.port_settings, list))
  60         self.assertEqual(0, len(settings.port_settings))
  61
  62     def test_config_with_name_only(self):
  63         settings = RouterSettings(**{'name': 'foo'})
  64         self.assertEqual('foo', settings.name)
  65         self.assertIsNone(settings.project_name)
  66         self.assertIsNone(settings.external_gateway)
  67         self.assertTrue(settings.admin_state_up)
  68         self.assertIsNone(settings.enable_snat)
  69         self.assertIsNotNone(settings.internal_subnets)
  70         self.assertTrue(isinstance(settings.internal_subnets, list))
  71         self.assertEqual(0, len(settings.internal_subnets))
  72         self.assertIsNotNone(settings.port_settings)
  73         self.assertTrue(isinstance(settings.port_settings, list))
  74         self.assertEqual(0, len(settings.port_settings))
  75
  76     def test_all(self):
  77         port_settings = PortConfig(name='foo', network_name='bar')
  78         settings = RouterSettings(
  79             name='foo', project_name='bar', external_gateway='foo_gateway',
  80             admin_state_up=True, enable_snat=False,
  81             internal_subnets=['10.0.0.1/24'], interfaces=[port_settings])
  82         self.assertEqual('foo', settings.name)
  83         self.assertEqual('bar', settings.project_name)
  84         self.assertEqual('foo_gateway', settings.external_gateway)
  85         self.assertTrue(settings.admin_state_up)
  86         self.assertFalse(settings.enable_snat)
  87         self.assertIsNotNone(settings.internal_subnets)
  88         self.assertTrue(isinstance(settings.internal_subnets, list))
  89         self.assertEqual(1, len(settings.internal_subnets))
  90         self.assertEqual(['10.0.0.1/24'], settings.internal_subnets)
  91         self.assertEqual([port_settings], settings.port_settings)
  92
  93     def test_config_all(self):
  94         settings = RouterSettings(
  95             **{'name': 'foo', 'project_name': 'bar',
  96                'external_gateway': 'foo_gateway', 'admin_state_up': True,
  97                'enable_snat': False, 'internal_subnets': ['10.0.0.1/24'],
  98                'interfaces':
  99                    [{'port': {'name': 'foo-port',
 100                               'network_name': 'bar-net'}}]})
 101         self.assertEqual('foo', settings.name)
 102         self.assertEqual('bar', settings.project_name)
 103         self.assertEqual('foo_gateway', settings.external_gateway)
 104         self.assertTrue(settings.admin_state_up)
 105         self.assertFalse(settings.enable_snat)
 106         self.assertIsNotNone(settings.internal_subnets)
 107         self.assertTrue(isinstance(settings.internal_subnets, list))
 108         self.assertEqual(1, len(settings.internal_subnets))
 109         self.assertEqual(['10.0.0.1/24'], settings.internal_subnets)
 110         self.assertEqual(
 111             [PortConfig(**{'name': 'foo-port', 'network_name': 'bar-net'})],
 112             settings.port_settings)
 113
 114
 115 class CreateRouterSuccessTests(OSIntegrationTestCase):
 116     """
 117     Class for testing routers with various positive scenarios expected to
 118     succeed
 119     """
 120
 121     def setUp(self):
 122         """
 123         Initializes objects used for router testing
 124         """
 125         super(self.__class__, self).__start__()
 126
 127         self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
 128         self.router_creator = None
 129         self.network_creator1 = None
 130         self.network_creator2 = None
 131         self.neutron = neutron_utils.neutron_client(self.os_creds)
 132
 133     def tearDown(self):
 134         """
 135         Cleans the remote OpenStack objects used for router testing
 136         """
 137         if self.router_creator:
 138             self.router_creator.clean()
 139
 140         if self.network_creator1:
 141             self.network_creator1.clean()
 142
 143         if self.network_creator2:
 144             self.network_creator2.clean()
 145
 146         super(self.__class__, self).__clean__()
 147
 148     def test_create_router_vanilla(self):
 149         """
 150         Test creation of a most basic router with minimal options.
 151         """
 152         router_settings = RouterConfig(
 153             name=self.guid + '-pub-router', external_gateway=self.ext_net_name)
 154
 155         self.router_creator = create_router.OpenStackRouter(self.os_creds,
 156                                                             router_settings)
 157         self.router_creator.create()
 158
 159         router = neutron_utils.get_router(self.neutron,
 160                                           router_settings=router_settings)
 161         self.assertIsNotNone(router)
 162
 163         self.assertEqual(self.router_creator.get_router(), router)
 164
 165         self.check_router_recreation(router, router_settings)
 166
 167     def test_create_router_admin_user_to_new_project(self):
 168         """
 169         Test creation of a most basic router with the admin user pointing
 170         to the new project.
 171         """
 172         router_settings = RouterConfig(
 173             name=self.guid + '-pub-router', external_gateway=self.ext_net_name,
 174             project_name=self.os_creds.project_name)
 175
 176         self.router_creator = create_router.OpenStackRouter(
 177             self.admin_os_creds, router_settings)
 178         self.router_creator.create()
 179
 180         router = neutron_utils.get_router(self.neutron,
 181                                           router_settings=router_settings)
 182         self.assertIsNotNone(router)
 183
 184         self.assertEqual(self.router_creator.get_router(), router)
 185
 186         self.check_router_recreation(router, router_settings)
 187
 188     def test_create_router_new_user_to_admin_project(self):
 189         """
 190         Test creation of a most basic router with the new user pointing
 191         to the admin project.
 192         """
 193         router_settings = RouterConfig(
 194             name=self.guid + '-pub-router', external_gateway=self.ext_net_name,
 195             project_name=self.admin_os_creds.project_name)
 196
 197         self.router_creator = create_router.OpenStackRouter(
 198             self.os_creds, router_settings)
 199         self.router_creator.create()
 200
 201         router = neutron_utils.get_router(self.neutron,
 202                                           router_settings=router_settings)
 203         self.assertIsNotNone(router)
 204
 205         self.assertEqual(self.router_creator.get_router(), router)
 206
 207         self.check_router_recreation(router, router_settings)
 208
 209     def test_create_delete_router(self):
 210         """
 211         Test that clean() will not raise an exception if the router is deleted
 212         by another process.
 213         """
 214         self.router_settings = RouterConfig(
 215             name=self.guid + '-pub-router', external_gateway=self.ext_net_name)
 216
 217         self.router_creator = create_router.OpenStackRouter(
 218             self.os_creds, self.router_settings)
 219         created_router = self.router_creator.create()
 220         self.assertIsNotNone(created_router)
 221         retrieved_router = neutron_utils.get_router(
 222             self.neutron, router_settings=self.router_settings)
 223         self.assertIsNotNone(retrieved_router)
 224
 225         neutron_utils.delete_router(self.neutron, created_router)
 226
 227         retrieved_router = neutron_utils.get_router(
 228             self.neutron, router_settings=self.router_settings)
 229         self.assertIsNone(retrieved_router)
 230
 231         # Should not raise an exception
 232         self.router_creator.clean()
 233
 234     def test_create_router_admin_state_false(self):
 235         """
 236         Test creation of a basic router with admin state down.
 237         """
 238         router_settings = RouterConfig(
 239             name=self.guid + '-pub-router', admin_state_up=False)
 240
 241         self.router_creator = create_router.OpenStackRouter(self.os_creds,
 242                                                             router_settings)
 243         self.router_creator.create()
 244
 245         router = neutron_utils.get_router(self.neutron,
 246                                           router_settings=router_settings)
 247         self.assertIsNotNone(router)
 248
 249         self.assertEqual(self.router_creator.get_router(), router)
 250
 251         self.check_router_recreation(router, router_settings)
 252
 253     def test_create_router_admin_state_True(self):
 254         """
 255         Test creation of a basic router with admin state Up.
 256         """
 257         router_settings = RouterConfig(
 258             name=self.guid + '-pub-router', admin_state_up=True)
 259
 260         self.router_creator = create_router.OpenStackRouter(
 261             self.os_creds, router_settings)
 262         self.router_creator.create()
 263
 264         router = neutron_utils.get_router(
 265             self.neutron, router_settings=router_settings)
 266         self.assertIsNotNone(router)
 267
 268         self.assertEqual(self.router_creator.get_router(), router)
 269
 270         self.check_router_recreation(router, router_settings)
 271
 272     def test_create_router_private_network(self):
 273         """
 274         Test creation of a router connected with two private networks and no
 275         external gateway
 276         """
 277         network_settings1 = NetworkConfig(
 278             name=self.guid + '-pub-net1',
 279             subnet_settings=[
 280                 create_network.SubnetConfig(
 281                     cidr=cidr1, name=self.guid + '-pub-subnet1',
 282                     gateway_ip=static_gateway_ip1)])
 283         network_settings2 = NetworkConfig(
 284             name=self.guid + '-pub-net2',
 285             subnet_settings=[
 286                 create_network.SubnetConfig(
 287                     cidr=cidr2, name=self.guid + '-pub-subnet2',
 288                     gateway_ip=static_gateway_ip2)])
 289
 290         self.network_creator1 = OpenStackNetwork(self.os_creds,
 291                                                  network_settings1)
 292         self.network_creator2 = OpenStackNetwork(self.os_creds,
 293                                                  network_settings2)
 294
 295         self.network_creator1.create()
 296         self.network_creator2.create()
 297
 298         port_settings = [
 299             create_network.PortConfig(
 300                 name=self.guid + '-port1',
 301                 ip_addrs=[{
 302                     'subnet_name':
 303                         network_settings1.subnet_settings[0].name,
 304                     'ip': static_gateway_ip1
 305                 }],
 306                 network_name=network_settings1.name,
 307                 project_name=self.os_creds.project_name),
 308             create_network.PortConfig(
 309                 name=self.guid + '-port2',
 310                 ip_addrs=[{
 311                     'subnet_name': network_settings2.subnet_settings[0].name,
 312                     'ip': static_gateway_ip2
 313                 }],
 314                 network_name=network_settings2.name,
 315                 project_name=self.os_creds.project_name)]
 316
 317         router_settings = RouterConfig(
 318             name=self.guid + '-pub-router', port_settings=port_settings)
 319         self.router_creator = create_router.OpenStackRouter(self.os_creds,
 320                                                             router_settings)
 321         self.router_creator.create()
 322
 323         router = neutron_utils.get_router(
 324             self.neutron, router_settings=router_settings)
 325
 326         self.assertEqual(router, self.router_creator.get_router())
 327
 328         # Instantiate second identical creator to ensure a second router
 329         # has not been created
 330         router_creator2 = create_router.OpenStackRouter(
 331             self.os_creds, router_settings)
 332         router2 = router_creator2.create()
 333         self.assertIsNotNone(self.router_creator.get_router(), router2)
 334
 335         self.check_router_recreation(router2, router_settings)
 336
 337     def test_create_router_external_network(self):
 338         """
 339         Test creation of a router connected to an external network and a
 340         private network.
 341         """
 342         network_settings = NetworkConfig(
 343             name=self.guid + '-pub-net1',
 344             subnet_settings=[
 345                 create_network.SubnetConfig(
 346                     cidr=cidr1, name=self.guid + '-pub-subnet1',
 347                     gateway_ip=static_gateway_ip1)])
 348         self.network_creator1 = OpenStackNetwork(self.os_creds,
 349                                                  network_settings)
 350         self.network_creator1.create()
 351
 352         port_settings = [
 353             create_network.PortConfig(
 354                 name=self.guid + '-port1',
 355                 ip_addrs=[{
 356                     'subnet_name': network_settings.subnet_settings[0].name,
 357                     'ip': static_gateway_ip1}],
 358                 network_name=network_settings.name,
 359                 project_name=self.os_creds.project_name)]
 360
 361         router_settings = RouterConfig(
 362             name=self.guid + '-pub-router', external_gateway=self.ext_net_name,
 363             port_settings=port_settings)
 364         self.router_creator = create_router.OpenStackRouter(
 365             self.os_creds, router_settings)
 366         self.router_creator.create()
 367
 368         router = neutron_utils.get_router(
 369             self.neutron, router_settings=router_settings)
 370
 371         self.assertEquals(router, self.router_creator.get_router())
 372
 373         self.check_router_recreation(router, router_settings)
 374
 375     def check_router_recreation(self, router, orig_settings):
 376         """
 377         Validates the derived RouterConfig with the original
 378         :param router: the Router domain object to test
 379         :param orig_settings: the original RouterConfig object that was
 380                               responsible for creating the router
 381         :return: the derived RouterConfig object
 382         """
 383         derived_settings = settings_utils.create_router_config(
 384             self.neutron, router)
 385         self.assertIsNotNone(derived_settings)
 386         self.assertEqual(
 387             orig_settings.enable_snat, derived_settings.enable_snat)
 388         self.assertEqual(orig_settings.external_gateway,
 389                          derived_settings.external_gateway)
 390         self.assertEqual(orig_settings.name, derived_settings.name)
 391         self.assertEqual(orig_settings.internal_subnets,
 392                          derived_settings.internal_subnets)
 393
 394         if orig_settings.external_gateway:
 395             self.assertEqual(len(orig_settings.port_settings),
 396                              len(derived_settings.port_settings))
 397         else:
 398             self.assertEqual(len(orig_settings.port_settings),
 399                              len(derived_settings.port_settings))
 400
 401         if len(orig_settings.port_settings) > 0:
 402             self.assertEqual(orig_settings.port_settings[0].name,
 403                              derived_settings.port_settings[0].name)
 404
 405         if len(orig_settings.port_settings) > 1:
 406             self.assertEqual(orig_settings.port_settings[1].name,
 407                              derived_settings.port_settings[1].name)
 408
 409         return derived_settings
 410
 411
 412 class CreateRouterNegativeTests(OSIntegrationTestCase):
 413     """
 414     Class for testing routers with various negative scenarios expected to fail.
 415     """
 416
 417     def setUp(self):
 418         """
 419         Initializes objects used for router testing
 420         """
 421         super(self.__class__, self).__start__()
 422
 423         self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
 424         self.router_creator = None
 425
 426     def tearDown(self):
 427         """
 428         Cleans the remote OpenStack objects used for router testing
 429         """
 430         if self.router_creator:
 431             self.router_creator.clean()
 432
 433         super(self.__class__, self).__clean__()
 434
 435     def test_create_router_noname(self):
 436         """
 437         Test creating a router without a name.
 438         """
 439         with self.assertRaises(RouterConfigError):
 440             router_settings = RouterConfig(
 441                 name=None, external_gateway=self.ext_net_name)
 442             self.router_creator = create_router.OpenStackRouter(
 443                 self.os_creds, router_settings)
 444             self.router_creator.create()
 445
 446     def test_create_router_invalid_gateway_name(self):
 447         """
 448         Test creating a router without a valid network gateway name.
 449         """
 450         with self.assertRaises(RouterConfigError):
 451             router_settings = RouterConfig(
 452                 name=self.guid + '-pub-router',
 453                 external_gateway="Invalid_name")
 454             self.router_creator = create_router.OpenStackRouter(
 455                 self.os_creds, router_settings)
 456             self.router_creator.create()