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