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, RouterCreationError)
  25 from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase
  26 from snaps.openstack.utils import neutron_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.assertIsNone(settings.admin_state_up)
  55         self.assertIsNone(settings.enable_snat)
  56         self.assertIsNone(settings.external_fixed_ips)
  57         self.assertIsNotNone(settings.internal_subnets)
  58         self.assertTrue(isinstance(settings.internal_subnets, list))
  59         self.assertEqual(0, len(settings.internal_subnets))
  60         self.assertIsNotNone(settings.port_settings)
  61         self.assertTrue(isinstance(settings.port_settings, list))
  62         self.assertEqual(0, len(settings.port_settings))
  63
  64     def test_config_with_name_only(self):
  65         settings = RouterSettings(**{'name': 'foo'})
  66         self.assertEqual('foo', settings.name)
  67         self.assertIsNone(settings.project_name)
  68         self.assertIsNone(settings.external_gateway)
  69         self.assertIsNone(settings.admin_state_up)
  70         self.assertIsNone(settings.enable_snat)
  71         self.assertIsNone(settings.external_fixed_ips)
  72         self.assertIsNotNone(settings.internal_subnets)
  73         self.assertTrue(isinstance(settings.internal_subnets, list))
  74         self.assertEqual(0, len(settings.internal_subnets))
  75         self.assertIsNotNone(settings.port_settings)
  76         self.assertTrue(isinstance(settings.port_settings, list))
  77         self.assertEqual(0, len(settings.port_settings))
  78
  79     def test_all(self):
  80         port_settings = PortSettings(name='foo', network_name='bar')
  81         settings = RouterSettings(
  82             name='foo', project_name='bar', external_gateway='foo_gateway',
  83             admin_state_up=True, enable_snat=False, external_fixed_ips=['ip1'],
  84             internal_subnets=['10.0.0.1/24'], interfaces=[port_settings])
  85         self.assertEqual('foo', settings.name)
  86         self.assertEqual('bar', settings.project_name)
  87         self.assertEqual('foo_gateway', settings.external_gateway)
  88         self.assertTrue(settings.admin_state_up)
  89         self.assertFalse(settings.enable_snat)
  90         self.assertEqual(['ip1'], settings.external_fixed_ips)
  91         self.assertIsNotNone(settings.internal_subnets)
  92         self.assertTrue(isinstance(settings.internal_subnets, list))
  93         self.assertEqual(1, len(settings.internal_subnets))
  94         self.assertEqual(['10.0.0.1/24'], settings.internal_subnets)
  95         self.assertEqual([port_settings], settings.port_settings)
  96
  97     def test_config_all(self):
  98         settings = RouterSettings(
  99             **{'name': 'foo', 'project_name': 'bar',
 100                'external_gateway': 'foo_gateway', 'admin_state_up': True,
 101                'enable_snat': False, 'external_fixed_ips': ['ip1'],
 102                'internal_subnets': ['10.0.0.1/24'],
 103                'interfaces':
 104                    [{'port': {'name': 'foo-port',
 105                               'network_name': 'bar-net'}}]})
 106         self.assertEqual('foo', settings.name)
 107         self.assertEqual('bar', settings.project_name)
 108         self.assertEqual('foo_gateway', settings.external_gateway)
 109         self.assertTrue(settings.admin_state_up)
 110         self.assertFalse(settings.enable_snat)
 111         self.assertEqual(['ip1'], settings.external_fixed_ips)
 112         self.assertIsNotNone(settings.internal_subnets)
 113         self.assertTrue(isinstance(settings.internal_subnets, list))
 114         self.assertEqual(1, len(settings.internal_subnets))
 115         self.assertEqual(['10.0.0.1/24'], settings.internal_subnets)
 116         self.assertEqual([PortSettings(**{'name': 'foo-port',
 117                                           'network_name': 'bar-net'})],
 118                          settings.port_settings)
 119
 120
 121 class CreateRouterSuccessTests(OSIntegrationTestCase):
 122     """
 123     Class for testing routers with various positive scenarios expected to
 124     succeed
 125     """
 126
 127     def setUp(self):
 128         """
 129         Initializes objects used for router testing
 130         """
 131         super(self.__class__, self).__start__()
 132
 133         self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
 134         self.router_creator = None
 135         self.network_creator1 = None
 136         self.network_creator2 = None
 137         self.neutron = neutron_utils.neutron_client(self.os_creds)
 138
 139     def tearDown(self):
 140         """
 141         Cleans the remote OpenStack objects used for router testing
 142         """
 143         if self.router_creator:
 144             self.router_creator.clean()
 145
 146         if self.network_creator1:
 147             self.network_creator1.clean()
 148
 149         if self.network_creator2:
 150             self.network_creator2.clean()
 151
 152         super(self.__class__, self).__clean__()
 153
 154     def test_create_router_vanilla(self):
 155         """
 156         Test creation of a most basic router with minimal options.
 157         """
 158         router_settings = RouterSettings(name=self.guid + '-pub-router',
 159                                          external_gateway=self.ext_net_name)
 160
 161         self.router_creator = create_router.OpenStackRouter(self.os_creds,
 162                                                             router_settings)
 163         self.router_creator.create()
 164
 165         router = neutron_utils.get_router_by_name(self.neutron,
 166                                                   router_settings.name)
 167         self.assertIsNotNone(router)
 168
 169         self.assertTrue(verify_router_attributes(
 170             router, self.router_creator, ext_gateway=self.ext_net_name))
 171
 172     def test_create_delete_router(self):
 173         """
 174         Test that clean() will not raise an exception if the router is deleted
 175         by another process.
 176         """
 177         self.router_settings = RouterSettings(
 178             name=self.guid + '-pub-router', external_gateway=self.ext_net_name)
 179
 180         self.router_creator = create_router.OpenStackRouter(
 181             self.os_creds, self.router_settings)
 182         created_router = self.router_creator.create()
 183         self.assertIsNotNone(created_router)
 184         retrieved_router = neutron_utils.get_router_by_name(
 185             self.neutron, self.router_settings.name)
 186         self.assertIsNotNone(retrieved_router)
 187
 188         neutron_utils.delete_router(self.neutron, created_router)
 189
 190         retrieved_router = neutron_utils.get_router_by_name(
 191             self.neutron, self.router_settings.name)
 192         self.assertIsNone(retrieved_router)
 193
 194         # Should not raise an exception
 195         self.router_creator.clean()
 196
 197     def test_create_router_admin_state_false(self):
 198         """
 199         Test creation of a basic router with admin state down.
 200         """
 201         router_settings = RouterSettings(name=self.guid + '-pub-router',
 202                                          admin_state_up=False)
 203
 204         self.router_creator = create_router.OpenStackRouter(self.os_creds,
 205                                                             router_settings)
 206         self.router_creator.create()
 207
 208         router = neutron_utils.get_router_by_name(self.neutron,
 209                                                   router_settings.name)
 210         self.assertIsNotNone(router)
 211
 212         self.assertTrue(verify_router_attributes(router, self.router_creator,
 213                                                  admin_state=False))
 214
 215     def test_create_router_admin_state_True(self):
 216         """
 217         Test creation of a basic router with admin state Up.
 218         """
 219         router_settings = RouterSettings(name=self.guid + '-pub-router',
 220                                          admin_state_up=True)
 221
 222         self.router_creator = create_router.OpenStackRouter(self.os_creds,
 223                                                             router_settings)
 224         self.router_creator.create()
 225
 226         router = neutron_utils.get_router_by_name(self.neutron,
 227                                                   router_settings.name)
 228         self.assertIsNotNone(router)
 229
 230         self.assertTrue(verify_router_attributes(router, self.router_creator,
 231                                                  admin_state=True))
 232
 233     def test_create_router_private_network(self):
 234         """
 235         Test creation of a router connected with two private networks and no
 236         external gateway
 237         """
 238         network_settings1 = NetworkSettings(
 239             name=self.guid + '-pub-net1',
 240             subnet_settings=[
 241                 create_network.SubnetSettings(
 242                     cidr=cidr1, name=self.guid + '-pub-subnet1',
 243                     gateway_ip=static_gateway_ip1)])
 244         network_settings2 = NetworkSettings(
 245             name=self.guid + '-pub-net2',
 246             subnet_settings=[
 247                 create_network.SubnetSettings(
 248                     cidr=cidr2, name=self.guid + '-pub-subnet2',
 249                     gateway_ip=static_gateway_ip2)])
 250
 251         self.network_creator1 = OpenStackNetwork(self.os_creds,
 252                                                  network_settings1)
 253         self.network_creator2 = OpenStackNetwork(self.os_creds,
 254                                                  network_settings2)
 255
 256         self.network_creator1.create()
 257         self.network_creator2.create()
 258
 259         port_settings = [
 260             create_network.PortSettings(name=self.guid + '-port1', ip_addrs=[
 261                 {'subnet_name': network_settings1.subnet_settings[0].name,
 262                  'ip': static_gateway_ip1}],
 263                                         network_name=network_settings1.name),
 264             create_network.PortSettings(name=self.guid + '-port2', ip_addrs=[
 265                 {'subnet_name': network_settings2.subnet_settings[0].name,
 266                  'ip': static_gateway_ip2}],
 267                                           network_name=network_settings2.name)]
 268
 269         router_settings = RouterSettings(name=self.guid + '-pub-router',
 270                                          port_settings=port_settings)
 271         self.router_creator = create_router.OpenStackRouter(self.os_creds,
 272                                                             router_settings)
 273         self.router_creator.create()
 274
 275         router = neutron_utils.get_router_by_name(self.neutron,
 276                                                   router_settings.name)
 277
 278         self.assertTrue(verify_router_attributes(router, self.router_creator))
 279
 280     def test_create_router_external_network(self):
 281         """
 282         Test creation of a router connected to an external network and a
 283         private network.
 284         """
 285         network_settings = NetworkSettings(
 286             name=self.guid + '-pub-net1',
 287             subnet_settings=[
 288                 create_network.SubnetSettings(
 289                     cidr=cidr1, name=self.guid + '-pub-subnet1',
 290                     gateway_ip=static_gateway_ip1)])
 291         self.network_creator1 = OpenStackNetwork(self.os_creds,
 292                                                  network_settings)
 293         self.network_creator1.create()
 294
 295         port_settings = [
 296             create_network.PortSettings(name=self.guid + '-port1', ip_addrs=[
 297                 {'subnet_name': network_settings.subnet_settings[0].name,
 298                  'ip': static_gateway_ip1}],
 299                                         network_name=network_settings.name)]
 300
 301         router_settings = RouterSettings(name=self.guid + '-pub-router',
 302                                          external_gateway=self.ext_net_name,
 303                                          port_settings=port_settings)
 304         self.router_creator = create_router.OpenStackRouter(self.os_creds,
 305                                                             router_settings)
 306         self.router_creator.create()
 307
 308         router = neutron_utils.get_router_by_name(self.neutron,
 309                                                   router_settings.name)
 310
 311         self.assertTrue(verify_router_attributes(
 312             router, self.router_creator, ext_gateway=self.ext_net_name))
 313
 314
 315 class CreateRouterNegativeTests(OSIntegrationTestCase):
 316     """
 317     Class for testing routers with various negative scenarios expected to fail.
 318     """
 319
 320     def setUp(self):
 321         """
 322         Initializes objects used for router testing
 323         """
 324         super(self.__class__, self).__start__()
 325
 326         self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
 327         self.router_creator = None
 328
 329     def tearDown(self):
 330         """
 331         Cleans the remote OpenStack objects used for router testing
 332         """
 333         if self.router_creator:
 334             self.router_creator.clean()
 335
 336         super(self.__class__, self).__clean__()
 337
 338     def test_create_router_noname(self):
 339         """
 340         Test creating a router without a name.
 341         """
 342         with self.assertRaises(RouterSettingsError):
 343             router_settings = RouterSettings(
 344                 name=None, external_gateway=self.ext_net_name)
 345             self.router_creator = create_router.OpenStackRouter(
 346                 self.os_creds, router_settings)
 347             self.router_creator.create()
 348
 349     def test_create_router_invalid_gateway_name(self):
 350         """
 351         Test creating a router without a valid network gateway name.
 352         """
 353         with self.assertRaises(RouterSettingsError):
 354             router_settings = RouterSettings(name=self.guid + '-pub-router',
 355                                              external_gateway="Invalid_name")
 356             self.router_creator = create_router.OpenStackRouter(
 357                 self.os_creds, router_settings)
 358             self.router_creator.create()
 359
 360
 361 def verify_router_attributes(router_operational, router_creator,
 362                              admin_state=True, ext_gateway=None):
 363     """
 364     Helper function to validate the attributes of router created with the one
 365     operational
 366     :param router_operational: Operational Router object returned from neutron
 367                                utils of type snaps.domain.Router
 368     :param router_creator: router_creator object returned from creating a
 369                            router in the router test functions
 370     :param admin_state: True if router is expected to be Up, else False
 371     :param ext_gateway: None if router is not connected to external gateway
 372     :return:
 373     """
 374
 375     router = router_creator.get_router()
 376
 377     if not router_operational:
 378         return False
 379     elif not router_creator:
 380         return False
 381     elif not (router_operational.name == router_creator.router_settings.name):
 382         return False
 383     elif not (router_operational.id == router.id):
 384         return False
 385     elif not (router_operational.status == router.status):
 386         return False
 387     elif not (router_operational.tenant_id == router.tenant_id):
 388         return False
 389     elif not (admin_state == router_operational.admin_state_up):
 390         return False
 391     elif (ext_gateway is None) and \
 392             (router_operational.external_gateway_info is not None):
 393         return False
 394     elif ext_gateway is not None:
 395         if router_operational.external_gateway_info is None:
 396             return False
 397     return True