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