Ensure project for routers are handled properly.
[snaps.git] / 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, PortConfigError
19 from snaps.config.router import RouterConfigError, RouterConfig
20 from snaps.config.security_group import SecurityGroupConfig
21 from snaps.openstack import create_network
22 from snaps.openstack import create_router
23 from snaps.openstack.create_network import OpenStackNetwork
24 from snaps.openstack.create_router import RouterSettings, OpenStackRouter
25 from snaps.openstack.create_security_group import OpenStackSecurityGroup
26 from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase
27 from snaps.openstack.utils import neutron_utils, settings_utils
28
29 __author__ = 'mmakati'
30
31 cidr1 = '10.200.201.0/24'
32 cidr2 = '10.200.202.0/24'
33 static_gateway_ip1 = '10.200.201.1'
34 static_gateway_ip2 = '10.200.202.1'
35
36
37 class RouterSettingsUnitTests(unittest.TestCase):
38     """
39     Class for testing the RouterSettings class
40     """
41
42     def test_no_params(self):
43         with self.assertRaises(RouterConfigError):
44             RouterSettings()
45
46     def test_empty_config(self):
47         with self.assertRaises(RouterConfigError):
48             RouterSettings(**dict())
49
50     def test_name_only(self):
51         settings = RouterSettings(name='foo')
52         self.assertEqual('foo', settings.name)
53         self.assertIsNone(settings.project_name)
54         self.assertIsNone(settings.external_gateway)
55         self.assertTrue(settings.admin_state_up)
56         self.assertIsNone(settings.enable_snat)
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.assertTrue(settings.admin_state_up)
70         self.assertIsNone(settings.enable_snat)
71         self.assertIsNotNone(settings.internal_subnets)
72         self.assertTrue(isinstance(settings.internal_subnets, list))
73         self.assertEqual(0, len(settings.internal_subnets))
74         self.assertIsNotNone(settings.port_settings)
75         self.assertTrue(isinstance(settings.port_settings, list))
76         self.assertEqual(0, len(settings.port_settings))
77
78     def test_all(self):
79         port_settings = PortConfig(name='foo', network_name='bar')
80         settings = RouterSettings(
81             name='foo', project_name='bar', external_gateway='foo_gateway',
82             admin_state_up=True, enable_snat=False,
83             internal_subnets=['10.0.0.1/24'], interfaces=[port_settings])
84         self.assertEqual('foo', settings.name)
85         self.assertEqual('bar', settings.project_name)
86         self.assertEqual('foo_gateway', settings.external_gateway)
87         self.assertTrue(settings.admin_state_up)
88         self.assertFalse(settings.enable_snat)
89         self.assertIsNotNone(settings.internal_subnets)
90         self.assertTrue(isinstance(settings.internal_subnets, list))
91         self.assertEqual(1, len(settings.internal_subnets))
92         self.assertEqual(['10.0.0.1/24'], settings.internal_subnets)
93         self.assertEqual([port_settings], settings.port_settings)
94
95     def test_config_all(self):
96         settings = RouterSettings(
97             **{'name': 'foo', 'project_name': 'bar',
98                'external_gateway': 'foo_gateway', 'admin_state_up': True,
99                'enable_snat': False, 'internal_subnets': ['10.0.0.1/24'],
100                'interfaces':
101                    [{'port': {'name': 'foo-port',
102                               'network_name': 'bar-net'}}]})
103         self.assertEqual('foo', settings.name)
104         self.assertEqual('bar', settings.project_name)
105         self.assertEqual('foo_gateway', settings.external_gateway)
106         self.assertTrue(settings.admin_state_up)
107         self.assertFalse(settings.enable_snat)
108         self.assertIsNotNone(settings.internal_subnets)
109         self.assertTrue(isinstance(settings.internal_subnets, list))
110         self.assertEqual(1, len(settings.internal_subnets))
111         self.assertEqual(['10.0.0.1/24'], settings.internal_subnets)
112         self.assertEqual(
113             [PortConfig(**{'name': 'foo-port', 'network_name': 'bar-net'})],
114             settings.port_settings)
115
116
117 class CreateRouterSuccessTests(OSIntegrationTestCase):
118     """
119     Class for testing routers with various positive scenarios expected to
120     succeed
121     """
122
123     def setUp(self):
124         """
125         Initializes objects used for router testing
126         """
127         super(self.__class__, self).__start__()
128
129         self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
130         self.router_creator = None
131         self.network_creator1 = None
132         self.network_creator2 = None
133         self.neutron = neutron_utils.neutron_client(self.os_creds)
134
135     def tearDown(self):
136         """
137         Cleans the remote OpenStack objects used for router testing
138         """
139         if self.router_creator:
140             self.router_creator.clean()
141
142         if self.network_creator1:
143             self.network_creator1.clean()
144
145         if self.network_creator2:
146             self.network_creator2.clean()
147
148         super(self.__class__, self).__clean__()
149
150     def test_create_router_vanilla(self):
151         """
152         Test creation of a most basic router with minimal options.
153         """
154         router_settings = RouterConfig(
155             name=self.guid + '-pub-router', external_gateway=self.ext_net_name)
156
157         self.router_creator = create_router.OpenStackRouter(
158             self.os_creds, router_settings)
159         self.router_creator.create()
160
161         router = neutron_utils.get_router(
162             self.neutron, self.keystone, router_settings=router_settings,
163             project_name=self.os_creds.project_name)
164         self.assertIsNotNone(router)
165
166         self.assertEqual(self.router_creator.get_router(), router)
167
168         self.check_router_recreation(router, router_settings)
169
170     def test_create_router_admin_user_to_new_project(self):
171         """
172         Test creation of a most basic router with the admin user pointing
173         to the new project.
174         """
175         router_settings = RouterConfig(
176             name=self.guid + '-pub-router', external_gateway=self.ext_net_name,
177             project_name=self.os_creds.project_name)
178
179         self.router_creator = create_router.OpenStackRouter(
180             self.admin_os_creds, router_settings)
181         self.router_creator.create()
182
183         router = neutron_utils.get_router(
184             self.neutron, self.keystone, router_settings=router_settings,
185             project_name=self.os_creds.project_name)
186         self.assertIsNotNone(router)
187
188         self.assertEqual(self.router_creator.get_router().id, router.id)
189
190         self.check_router_recreation(router, router_settings)
191
192     def test_create_router_new_user_as_admin_project(self):
193         """
194         Test creation of a most basic router with the new user pointing
195         to the admin project.
196         """
197         router_settings = RouterConfig(
198             name=self.guid + '-pub-router', external_gateway=self.ext_net_name,
199             project_name=self.os_creds.project_name)
200
201         self.router_creator = create_router.OpenStackRouter(
202             self.admin_os_creds, router_settings)
203         self.router_creator.create()
204
205         router = neutron_utils.get_router(
206             self.neutron, self.keystone, router_settings=router_settings,
207             project_name=self.os_creds.project_name)
208         self.assertIsNotNone(router)
209
210         self.assertEqual(self.router_creator.get_router().id, router.id)
211
212         self.check_router_recreation(router, router_settings)
213
214     def test_create_delete_router(self):
215         """
216         Test that clean() will not raise an exception if the router is deleted
217         by another process.
218         """
219         self.router_settings = RouterConfig(
220             name=self.guid + '-pub-router', external_gateway=self.ext_net_name)
221
222         self.router_creator = create_router.OpenStackRouter(
223             self.os_creds, self.router_settings)
224         created_router = self.router_creator.create()
225         self.assertIsNotNone(created_router)
226         retrieved_router = neutron_utils.get_router(
227             self.neutron, self.keystone, router_settings=self.router_settings,
228             project_name=self.os_creds.project_name)
229         self.assertIsNotNone(retrieved_router)
230
231         neutron_utils.delete_router(self.neutron, created_router)
232
233         retrieved_router = neutron_utils.get_router(
234             self.neutron, self.keystone, router_settings=self.router_settings,
235             project_name=self.os_creds.project_name)
236         self.assertIsNone(retrieved_router)
237
238         # Should not raise an exception
239         self.router_creator.clean()
240
241     def test_create_router_admin_state_false(self):
242         """
243         Test creation of a basic router with admin state down.
244         """
245         router_settings = RouterConfig(
246             name=self.guid + '-pub-router', admin_state_up=False)
247
248         self.router_creator = create_router.OpenStackRouter(self.os_creds,
249                                                             router_settings)
250         self.router_creator.create()
251
252         router = neutron_utils.get_router(
253             self.neutron, self.keystone, router_settings=router_settings,
254             project_name=self.os_creds.project_name)
255         self.assertIsNotNone(router)
256
257         self.assertEqual(self.router_creator.get_router(), router)
258
259         self.check_router_recreation(router, router_settings)
260
261     def test_create_router_admin_state_True(self):
262         """
263         Test creation of a basic router with admin state Up.
264         """
265         router_settings = RouterConfig(
266             name=self.guid + '-pub-router', admin_state_up=True)
267
268         self.router_creator = create_router.OpenStackRouter(
269             self.os_creds, router_settings)
270         self.router_creator.create()
271
272         router = neutron_utils.get_router(
273             self.neutron, self.keystone, router_settings=router_settings,
274             project_name=self.os_creds.project_name)
275         self.assertIsNotNone(router)
276
277         self.assertEqual(self.router_creator.get_router(), router)
278
279         self.check_router_recreation(router, router_settings)
280
281     def test_create_router_private_network(self):
282         """
283         Test creation of a router connected with two private networks and no
284         external gateway
285         """
286         network_settings1 = NetworkConfig(
287             name=self.guid + '-pub-net1',
288             subnet_settings=[
289                 create_network.SubnetConfig(
290                     cidr=cidr1, name=self.guid + '-pub-subnet1',
291                     gateway_ip=static_gateway_ip1)])
292         network_settings2 = NetworkConfig(
293             name=self.guid + '-pub-net2',
294             subnet_settings=[
295                 create_network.SubnetConfig(
296                     cidr=cidr2, name=self.guid + '-pub-subnet2',
297                     gateway_ip=static_gateway_ip2)])
298
299         self.network_creator1 = OpenStackNetwork(self.os_creds,
300                                                  network_settings1)
301         self.network_creator2 = OpenStackNetwork(self.os_creds,
302                                                  network_settings2)
303
304         self.network_creator1.create()
305         self.network_creator2.create()
306
307         port_settings = [
308             create_network.PortConfig(
309                 name=self.guid + '-port1',
310                 ip_addrs=[{
311                     'subnet_name':
312                         network_settings1.subnet_settings[0].name,
313                     'ip': static_gateway_ip1
314                 }],
315                 network_name=network_settings1.name),
316             create_network.PortConfig(
317                 name=self.guid + '-port2',
318                 ip_addrs=[{
319                     'subnet_name': network_settings2.subnet_settings[0].name,
320                     'ip': static_gateway_ip2
321                 }],
322                 network_name=network_settings2.name)]
323
324         router_settings = RouterConfig(
325             name=self.guid + '-pub-router', port_settings=port_settings)
326         self.router_creator = create_router.OpenStackRouter(
327             self.os_creds, router_settings)
328         self.router_creator.create()
329
330         router = neutron_utils.get_router(
331             self.neutron, self.keystone, router_settings=router_settings,
332             project_name=self.os_creds.project_name)
333
334         self.assertEqual(router, self.router_creator.get_router())
335
336         # Instantiate second identical creator to ensure a second router
337         # has not been created
338         router_creator2 = create_router.OpenStackRouter(
339             self.os_creds, router_settings)
340         router2 = router_creator2.create()
341         self.assertIsNotNone(self.router_creator.get_router(), router2)
342
343         self.check_router_recreation(router2, router_settings)
344
345     def test_create_router_external_network(self):
346         """
347         Test creation of a router connected to an external network and a
348         private network.
349         """
350         network_settings = NetworkConfig(
351             name=self.guid + '-pub-net1',
352             subnet_settings=[
353                 create_network.SubnetConfig(
354                     cidr=cidr1, name=self.guid + '-pub-subnet1',
355                     gateway_ip=static_gateway_ip1)])
356         self.network_creator1 = OpenStackNetwork(self.os_creds,
357                                                  network_settings)
358         self.network_creator1.create()
359
360         port_settings = [
361             create_network.PortConfig(
362                 name=self.guid + '-port1',
363                 ip_addrs=[{
364                     'subnet_name': network_settings.subnet_settings[0].name,
365                     'ip': static_gateway_ip1}],
366                 network_name=network_settings.name)]
367
368         router_settings = RouterConfig(
369             name=self.guid + '-pub-router', external_gateway=self.ext_net_name,
370             port_settings=port_settings)
371         self.router_creator = create_router.OpenStackRouter(
372             self.os_creds, router_settings)
373         self.router_creator.create()
374
375         router = neutron_utils.get_router(
376             self.neutron, self.keystone, router_settings=router_settings,
377             project_name=self.os_creds.project_name)
378
379         self.assertEquals(router, self.router_creator.get_router())
380
381         self.check_router_recreation(router, router_settings)
382
383     def check_router_recreation(self, router, orig_settings):
384         """
385         Validates the derived RouterConfig with the original
386         :param router: the Router domain object to test
387         :param orig_settings: the original RouterConfig object that was
388                               responsible for creating the router
389         :return: the derived RouterConfig object
390         """
391         derived_settings = settings_utils.create_router_config(
392             self.neutron, router)
393         self.assertIsNotNone(derived_settings)
394         self.assertEqual(
395             orig_settings.enable_snat, derived_settings.enable_snat)
396         self.assertEqual(orig_settings.external_gateway,
397                          derived_settings.external_gateway)
398         self.assertEqual(orig_settings.name, derived_settings.name)
399         self.assertEqual(orig_settings.internal_subnets,
400                          derived_settings.internal_subnets)
401
402         if orig_settings.external_gateway:
403             self.assertEqual(len(orig_settings.port_settings),
404                              len(derived_settings.port_settings))
405         else:
406             self.assertEqual(len(orig_settings.port_settings),
407                              len(derived_settings.port_settings))
408
409         if len(orig_settings.port_settings) > 0:
410             self.assertEqual(orig_settings.port_settings[0].name,
411                              derived_settings.port_settings[0].name)
412
413         if len(orig_settings.port_settings) > 1:
414             self.assertEqual(orig_settings.port_settings[1].name,
415                              derived_settings.port_settings[1].name)
416
417         return derived_settings
418
419
420 class CreateRouterNegativeTests(OSIntegrationTestCase):
421     """
422     Class for testing routers with various negative scenarios expected to fail.
423     """
424
425     def setUp(self):
426         """
427         Initializes objects used for router testing
428         """
429         super(self.__class__, self).__start__()
430
431         self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
432         self.network_creator = None
433         self.router_creator = None
434
435     def tearDown(self):
436         """
437         Cleans the remote OpenStack objects used for router testing
438         """
439         if self.router_creator:
440             self.router_creator.clean()
441
442         if self.network_creator:
443             self.network_creator.clean()
444
445         super(self.__class__, self).__clean__()
446
447     def test_create_router_noname(self):
448         """
449         Test creating a router without a name.
450         """
451         with self.assertRaises(RouterConfigError):
452             router_settings = RouterConfig(
453                 name=None, external_gateway=self.ext_net_name)
454             self.router_creator = create_router.OpenStackRouter(
455                 self.os_creds, router_settings)
456             self.router_creator.create()
457
458     def test_create_router_invalid_gateway_name(self):
459         """
460         Test creating a router without a valid network gateway name.
461         """
462         with self.assertRaises(RouterConfigError):
463             router_settings = RouterConfig(
464                 name=self.guid + '-pub-router',
465                 external_gateway="Invalid_name")
466             self.router_creator = create_router.OpenStackRouter(
467                 self.os_creds, router_settings)
468             self.router_creator.create()
469
470     def test_create_router_admin_ports(self):
471         """
472         Test creation of a router with ports to subnets owned by the admin
473         project
474         """
475         network_settings = NetworkConfig(
476             name=self.guid + '-pub-net1',
477             subnet_settings=[
478                 create_network.SubnetConfig(
479                     cidr=cidr1, name=self.guid + '-pub-subnet1',
480                     gateway_ip=static_gateway_ip1)])
481         self.network_creator = OpenStackNetwork(
482             self.admin_os_creds, network_settings)
483         self.network_creator.create()
484
485         port_settings = [
486             create_network.PortConfig(
487                 name=self.guid + '-port1',
488                 ip_addrs=[{
489                     'subnet_name': network_settings.subnet_settings[0].name,
490                     'ip': static_gateway_ip1}],
491                 network_name=network_settings.name)]
492
493         router_settings = RouterConfig(
494             name=self.guid + '-pub-router', external_gateway=self.ext_net_name,
495             port_settings=port_settings)
496         self.router_creator = create_router.OpenStackRouter(
497             self.os_creds, router_settings)
498
499         with self.assertRaises(PortConfigError):
500             self.router_creator.create()
501
502
503 class CreateMultipleRouterTests(OSIntegrationTestCase):
504     """
505     Test for the OpenStackRouter class and how it interacts with routers
506     groups within other projects with the same name
507     """
508
509     def setUp(self):
510         """
511         Initializes objects used for router testing
512         """
513         super(self.__class__, self).__start__()
514
515         self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
516         self.admin_router_creator = None
517         self.proj_router_creator = None
518         self.neutron = neutron_utils.neutron_client(self.os_creds)
519
520         network_settings = NetworkConfig(
521             name=self.guid + '-pub-net', shared=True,
522             subnet_settings=[
523                 create_network.SubnetConfig(
524                     cidr=cidr1, name=self.guid + '-pub-subnet',
525                     gateway_ip=static_gateway_ip1)])
526
527         self.network_creator = OpenStackNetwork(
528             self.admin_os_creds, network_settings)
529         self.network_creator.create()
530
531     def tearDown(self):
532         """
533         Cleans the remote OpenStack objects used for router testing
534         """
535         if self.admin_router_creator:
536             self.admin_router_creator.clean()
537
538         if self.proj_router_creator:
539             self.proj_router_creator.clean()
540
541         if self.network_creator:
542             self.network_creator.clean()
543
544         super(self.__class__, self).__clean__()
545
546     def test_router_same_name_diff_proj(self):
547         """
548         Tests the creation of an OpenStackNetwork with the same name
549         within a different project/tenant when not configured but implied by
550         the OSCreds.
551         """
552         # Create Router
553
554         router_config = RouterConfig(name=self.guid + '-router')
555         self.admin_router_creator = OpenStackRouter(
556             self.admin_os_creds, router_config)
557         self.admin_router_creator.create()
558
559         self.proj_router_creator = OpenStackRouter(
560             self.os_creds, router_config)
561         self.proj_router_creator.create()
562
563         self.assertNotEqual(
564             self.admin_router_creator.get_router().id,
565             self.proj_router_creator.get_router().id)
566
567         admin_creator2 = OpenStackRouter(
568             self.admin_os_creds, router_config)
569         admin_creator2.create()
570         self.assertEqual(
571             self.admin_router_creator.get_router(),
572             admin_creator2.get_router())
573
574         proj_creator2 = OpenStackRouter(self.os_creds, router_config)
575         proj_creator2.create()
576         self.assertEqual(self.proj_router_creator.get_router(),
577                          proj_creator2.get_router())
578
579     def test_router_create_by_admin_to_different_project(self):
580         """
581         Tests the creation of an OpenStackRouter by the admin user and
582         initialize again with tenant credentials.
583         """
584         # Create Network
585
586         admin_router_config = RouterConfig(
587             name=self.guid + '-router',
588             project_name=self.os_creds.project_name)
589
590         self.admin_router_creator = OpenStackRouter(
591             self.admin_os_creds, admin_router_config)
592         self.admin_router_creator.create()
593
594         proj_router_config = RouterConfig(
595             name=self.guid + '-router',
596             project_name=self.os_creds.project_name)
597
598         self.proj_router_creator = OpenStackRouter(
599             self.os_creds, proj_router_config)
600         self.proj_router_creator.create()
601
602         self.assertEqual(
603             self.admin_router_creator.get_router().id,
604             self.proj_router_creator.get_router().id)
605
606
607 class CreateRouterSecurityGroupTests(OSIntegrationTestCase):
608     """
609     Class for testing routers with ports containing security groups
610     """
611
612     def setUp(self):
613         """
614         Initializes objects used for router testing
615         """
616         super(self.__class__, self).__start__()
617
618         self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
619         self.router_creator = None
620         self.network_creator = None
621
622         self.sec_grp_creator = OpenStackSecurityGroup(
623             self.os_creds, SecurityGroupConfig(name=self.guid + '-sec_grp'))
624         self.sec_grp_creator.create()
625
626         self.neutron = neutron_utils.neutron_client(self.os_creds)
627
628     def tearDown(self):
629         """
630         Cleans the remote OpenStack objects used for router testing
631         """
632         if self.router_creator:
633             self.router_creator.clean()
634
635         if self.network_creator:
636             self.network_creator.clean()
637
638         if self.sec_grp_creator:
639             self.sec_grp_creator.clean()
640
641         super(self.__class__, self).__clean__()
642
643     def test_create_router_secure_port(self):
644         """
645         Test creation of a router with a port that has a security group.
646         """
647         network_settings = NetworkConfig(
648             name=self.guid + '-pub-net1',
649             subnet_settings=[
650                 create_network.SubnetConfig(
651                     cidr=cidr1, name=self.guid + '-pub-subnet1')])
652         self.network_creator = OpenStackNetwork(
653             self.os_creds, network_settings)
654         self.network_creator.create()
655
656         port_settings = [
657             create_network.PortConfig(
658                 name=self.guid + '-port1',
659                 ip_addrs=[{
660                     'subnet_name': network_settings.subnet_settings[0].name,
661                     'ip': static_gateway_ip1}],
662                 network_name=network_settings.name,
663                 security_groups=[self.sec_grp_creator.sec_grp_settings.name])]
664
665         router_settings = RouterConfig(
666             name=self.guid + '-pub-router', external_gateway=self.ext_net_name,
667             port_settings=port_settings)
668         self.router_creator = create_router.OpenStackRouter(
669             self.os_creds, router_settings)
670         self.router_creator.create()