Merge "Changes to ProjectSettings constructors to use kwargs."
[snaps.git] / snaps / openstack / tests / create_network_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_router
19 from snaps.openstack.create_network import (OpenStackNetwork, NetworkSettings,
20                                             SubnetSettings, PortSettings)
21 from snaps.openstack.tests import openstack_tests
22 from snaps.openstack.tests.os_source_file_test import (OSIntegrationTestCase,
23                                                        OSComponentTestCase)
24 from snaps.openstack.utils import neutron_utils
25 from snaps.openstack.utils.tests import neutron_utils_tests
26
27 __author__ = 'spisarski'
28
29
30 class NetworkSettingsUnitTests(unittest.TestCase):
31     """
32     Tests the construction of the NetworkSettings class
33     """
34
35     def test_no_params(self):
36         with self.assertRaises(Exception):
37             NetworkSettings()
38
39     def test_empty_config(self):
40         with self.assertRaises(Exception):
41             NetworkSettings(**dict())
42
43     def test_name_only(self):
44         settings = NetworkSettings(name='foo')
45         self.assertEqual('foo', settings.name)
46         self.assertTrue(settings.admin_state_up)
47         self.assertIsNone(settings.shared)
48         self.assertIsNone(settings.project_name)
49         self.assertFalse(settings.external)
50         self.assertIsNone(settings.network_type)
51         self.assertEqual(0, len(settings.subnet_settings))
52
53     def test_config_with_name_only(self):
54         settings = NetworkSettings(**{'name': 'foo'})
55         self.assertEqual('foo', settings.name)
56         self.assertTrue(settings.admin_state_up)
57         self.assertIsNone(settings.shared)
58         self.assertIsNone(settings.project_name)
59         self.assertFalse(settings.external)
60         self.assertIsNone(settings.network_type)
61         self.assertEqual(0, len(settings.subnet_settings))
62
63     def test_all(self):
64         sub_settings = SubnetSettings(name='foo-subnet', cidr='10.0.0.0/24')
65         settings = NetworkSettings(name='foo', admin_state_up=False,
66                                    shared=True, project_name='bar',
67                                    external=True,
68                                    network_type='flat', physical_network='phy',
69                                    subnet_settings=[sub_settings])
70         self.assertEqual('foo', settings.name)
71         self.assertFalse(settings.admin_state_up)
72         self.assertTrue(settings.shared)
73         self.assertEqual('bar', settings.project_name)
74         self.assertTrue(settings.external)
75         self.assertEqual('flat', settings.network_type)
76         self.assertEqual('phy', settings.physical_network)
77         self.assertEqual(1, len(settings.subnet_settings))
78         self.assertEqual('foo-subnet', settings.subnet_settings[0].name)
79
80     def test_config_all(self):
81         settings = NetworkSettings(
82             **{'name': 'foo', 'admin_state_up': False, 'shared': True,
83                'project_name': 'bar', 'external': True, 'network_type': 'flat',
84                'physical_network': 'phy',
85                'subnets':
86                    [{'subnet': {'name': 'foo-subnet',
87                                 'cidr': '10.0.0.0/24'}}]})
88         self.assertEqual('foo', settings.name)
89         self.assertFalse(settings.admin_state_up)
90         self.assertTrue(settings.shared)
91         self.assertEqual('bar', settings.project_name)
92         self.assertTrue(settings.external)
93         self.assertEqual('flat', settings.network_type)
94         self.assertEqual('phy', settings.physical_network)
95         self.assertEqual(1, len(settings.subnet_settings))
96         self.assertEqual('foo-subnet', settings.subnet_settings[0].name)
97
98
99 class SubnetSettingsUnitTests(unittest.TestCase):
100     """
101     Tests the construction of the SubnetSettings class
102     """
103
104     def test_no_params(self):
105         with self.assertRaises(Exception):
106             SubnetSettings()
107
108     def test_empty_config(self):
109         with self.assertRaises(Exception):
110             SubnetSettings(**dict())
111
112     def test_name_only(self):
113         with self.assertRaises(Exception):
114             SubnetSettings(name='foo')
115
116     def test_config_with_name_only(self):
117         with self.assertRaises(Exception):
118             SubnetSettings(**{'name': 'foo'})
119
120     def test_name_cidr_only(self):
121         settings = SubnetSettings(name='foo', cidr='10.0.0.0/24')
122         self.assertEqual('foo', settings.name)
123         self.assertEqual('10.0.0.0/24', settings.cidr)
124         self.assertEqual(4, settings.ip_version)
125         self.assertIsNone(settings.project_name)
126         self.assertIsNone(settings.start)
127         self.assertIsNone(settings.end)
128         self.assertIsNone(settings.enable_dhcp)
129         self.assertEqual(1, len(settings.dns_nameservers))
130         self.assertEqual('8.8.8.8', settings.dns_nameservers[0])
131         self.assertIsNone(settings.host_routes)
132         self.assertIsNone(settings.destination)
133         self.assertIsNone(settings.nexthop)
134         self.assertIsNone(settings.ipv6_ra_mode)
135         self.assertIsNone(settings.ipv6_address_mode)
136
137     def test_config_with_name_cidr_only(self):
138         settings = SubnetSettings(**{'name': 'foo', 'cidr': '10.0.0.0/24'})
139         self.assertEqual('foo', settings.name)
140         self.assertEqual('10.0.0.0/24', settings.cidr)
141         self.assertEqual(4, settings.ip_version)
142         self.assertIsNone(settings.project_name)
143         self.assertIsNone(settings.start)
144         self.assertIsNone(settings.end)
145         self.assertIsNone(settings.gateway_ip)
146         self.assertIsNone(settings.enable_dhcp)
147         self.assertEqual(1, len(settings.dns_nameservers))
148         self.assertEqual('8.8.8.8', settings.dns_nameservers[0])
149         self.assertIsNone(settings.host_routes)
150         self.assertIsNone(settings.destination)
151         self.assertIsNone(settings.nexthop)
152         self.assertIsNone(settings.ipv6_ra_mode)
153         self.assertIsNone(settings.ipv6_address_mode)
154
155     def test_all(self):
156         host_routes = {'destination': '0.0.0.0/0', 'nexthop': '123.456.78.9'}
157         settings = SubnetSettings(name='foo', cidr='10.0.0.0/24', ip_version=6,
158                                   project_name='bar-project',
159                                   start='10.0.0.2', end='10.0.0.101',
160                                   gateway_ip='10.0.0.1', enable_dhcp=False,
161                                   dns_nameservers=['8.8.8.8'],
162                                   host_routes=[host_routes],
163                                   destination='dest',
164                                   nexthop='hop',
165                                   ipv6_ra_mode='dhcpv6-stateful',
166                                   ipv6_address_mode='slaac')
167         self.assertEqual('foo', settings.name)
168         self.assertEqual('10.0.0.0/24', settings.cidr)
169         self.assertEqual(6, settings.ip_version)
170         self.assertEqual('bar-project', settings.project_name)
171         self.assertEqual('10.0.0.2', settings.start)
172         self.assertEqual('10.0.0.101', settings.end)
173         self.assertEqual('10.0.0.1', settings.gateway_ip)
174         self.assertEqual(False, settings.enable_dhcp)
175         self.assertEqual(1, len(settings.dns_nameservers))
176         self.assertEqual('8.8.8.8', settings.dns_nameservers[0])
177         self.assertEqual(1, len(settings.host_routes))
178         self.assertEqual(host_routes, settings.host_routes[0])
179         self.assertEqual('dest', settings.destination)
180         self.assertEqual('hop', settings.nexthop)
181         self.assertEqual('dhcpv6-stateful', settings.ipv6_ra_mode)
182         self.assertEqual('slaac', settings.ipv6_address_mode)
183
184     def test_config_all(self):
185         host_routes = {'destination': '0.0.0.0/0', 'nexthop': '123.456.78.9'}
186         settings = SubnetSettings(
187             **{'name': 'foo', 'cidr': '10.0.0.0/24', 'ip_version': 6,
188                'project_name': 'bar-project', 'start': '10.0.0.2',
189                'end': '10.0.0.101',
190                'gateway_ip': '10.0.0.1', 'enable_dhcp': False,
191                'dns_nameservers': ['8.8.8.8'], 'host_routes': [host_routes],
192                'destination': 'dest', 'nexthop': 'hop',
193                'ipv6_ra_mode': 'dhcpv6-stateful',
194                'ipv6_address_mode': 'slaac'})
195         self.assertEqual('foo', settings.name)
196         self.assertEqual('10.0.0.0/24', settings.cidr)
197         self.assertEqual(6, settings.ip_version)
198         self.assertEqual('bar-project', settings.project_name)
199         self.assertEqual('10.0.0.2', settings.start)
200         self.assertEqual('10.0.0.101', settings.end)
201         self.assertEqual('10.0.0.1', settings.gateway_ip)
202         self.assertEqual(False, settings.enable_dhcp)
203         self.assertEqual(1, len(settings.dns_nameservers))
204         self.assertEqual('8.8.8.8', settings.dns_nameservers[0])
205         self.assertEqual(1, len(settings.host_routes))
206         self.assertEqual(host_routes, settings.host_routes[0])
207         self.assertEqual('dest', settings.destination)
208         self.assertEqual('hop', settings.nexthop)
209         self.assertEqual('dhcpv6-stateful', settings.ipv6_ra_mode)
210         self.assertEqual('slaac', settings.ipv6_address_mode)
211
212
213 class PortSettingsUnitTests(unittest.TestCase):
214     """
215     Tests the construction of the PortSettings class
216     """
217
218     def test_no_params(self):
219         with self.assertRaises(Exception):
220             PortSettings()
221
222     def test_empty_config(self):
223         with self.assertRaises(Exception):
224             PortSettings(**dict())
225
226     def test_name_only(self):
227         with self.assertRaises(Exception):
228             PortSettings(name='foo')
229
230     def test_config_name_only(self):
231         with self.assertRaises(Exception):
232             PortSettings(**{'name': 'foo'})
233
234     def test_name_netname_only(self):
235         settings = PortSettings(name='foo', network_name='bar')
236         self.assertEqual('foo', settings.name)
237         self.assertEqual('bar', settings.network_name)
238         self.assertTrue(settings.admin_state_up)
239         self.assertIsNone(settings.project_name)
240         self.assertIsNone(settings.mac_address)
241         self.assertIsNone(settings.ip_addrs)
242         self.assertIsNone(settings.fixed_ips)
243         self.assertIsNone(settings.security_groups)
244         self.assertIsNone(settings.allowed_address_pairs)
245         self.assertIsNone(settings.opt_value)
246         self.assertIsNone(settings.opt_name)
247         self.assertIsNone(settings.device_owner)
248         self.assertIsNone(settings.device_id)
249
250     def test_config_with_name_netname_only(self):
251         settings = PortSettings(**{'name': 'foo', 'network_name': 'bar'})
252         self.assertEqual('foo', settings.name)
253         self.assertEqual('bar', settings.network_name)
254         self.assertTrue(settings.admin_state_up)
255         self.assertIsNone(settings.project_name)
256         self.assertIsNone(settings.mac_address)
257         self.assertIsNone(settings.ip_addrs)
258         self.assertIsNone(settings.fixed_ips)
259         self.assertIsNone(settings.security_groups)
260         self.assertIsNone(settings.allowed_address_pairs)
261         self.assertIsNone(settings.opt_value)
262         self.assertIsNone(settings.opt_name)
263         self.assertIsNone(settings.device_owner)
264         self.assertIsNone(settings.device_id)
265
266     def test_all(self):
267         ip_addrs = [{'subnet_name', 'foo-sub', 'ip', '10.0.0.10'}]
268         fixed_ips = {'sub_id', '10.0.0.10'}
269         allowed_address_pairs = {'10.0.0.101', '1234.5678'}
270
271         settings = PortSettings(name='foo', network_name='bar',
272                                 admin_state_up=False,
273                                 project_name='foo-project',
274                                 mac_address='1234', ip_addrs=ip_addrs,
275                                 fixed_ips=fixed_ips,
276                                 security_groups=['foo_grp_id'],
277                                 allowed_address_pairs=allowed_address_pairs,
278                                 opt_value='opt value', opt_name='opt name',
279                                 device_owner='owner',
280                                 device_id='device number')
281         self.assertEqual('foo', settings.name)
282         self.assertEqual('bar', settings.network_name)
283         self.assertFalse(settings.admin_state_up)
284         self.assertEqual('foo-project', settings.project_name)
285         self.assertEqual('1234', settings.mac_address)
286         self.assertEqual(ip_addrs, settings.ip_addrs)
287         self.assertEqual(fixed_ips, settings.fixed_ips)
288         self.assertEqual(1, len(settings.security_groups))
289         self.assertEqual('foo_grp_id', settings.security_groups[0])
290         self.assertEqual(allowed_address_pairs, settings.allowed_address_pairs)
291         self.assertEqual('opt value', settings.opt_value)
292         self.assertEqual('opt name', settings.opt_name)
293         self.assertEqual('owner', settings.device_owner)
294         self.assertEqual('device number', settings.device_id)
295
296     def test_config_all(self):
297         ip_addrs = [{'subnet_name', 'foo-sub', 'ip', '10.0.0.10'}]
298         fixed_ips = {'sub_id', '10.0.0.10'}
299         allowed_address_pairs = {'10.0.0.101', '1234.5678'}
300
301         settings = PortSettings(
302             **{'name': 'foo', 'network_name': 'bar', 'admin_state_up': False,
303                'project_name': 'foo-project', 'mac_address': '1234',
304                'ip_addrs': ip_addrs,
305                'fixed_ips': fixed_ips, 'security_groups': ['foo_grp_id'],
306                'allowed_address_pairs': allowed_address_pairs,
307                'opt_value': 'opt value',
308                'opt_name': 'opt name', 'device_owner': 'owner',
309                'device_id': 'device number'})
310         self.assertEqual('foo', settings.name)
311         self.assertEqual('bar', settings.network_name)
312         self.assertFalse(settings.admin_state_up)
313         self.assertEqual('foo-project', settings.project_name)
314         self.assertEqual('1234', settings.mac_address)
315         self.assertEqual(ip_addrs, settings.ip_addrs)
316         self.assertEqual(fixed_ips, settings.fixed_ips)
317         self.assertEqual(1, len(settings.security_groups))
318         self.assertEqual('foo_grp_id', settings.security_groups[0])
319         self.assertEqual(allowed_address_pairs, settings.allowed_address_pairs)
320         self.assertEqual('opt value', settings.opt_value)
321         self.assertEqual('opt name', settings.opt_name)
322         self.assertEqual('owner', settings.device_owner)
323         self.assertEqual('device number', settings.device_id)
324
325
326 class CreateNetworkSuccessTests(OSIntegrationTestCase):
327     """
328     Test for the CreateNework class defined in create_nework.py
329     """
330
331     def setUp(self):
332         """
333         Sets up object for test
334         """
335         super(self.__class__, self).__start__()
336
337         guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
338         self.net_config = openstack_tests.get_pub_net_config(
339             net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet',
340             router_name=guid + '-pub-router', external_net=self.ext_net_name)
341
342         self.neutron = neutron_utils.neutron_client(self.os_creds)
343
344         # Initialize for cleanup
345         self.net_creator = None
346         self.router_creator = None
347         self.neutron = neutron_utils.neutron_client(self.os_creds)
348
349     def tearDown(self):
350         """
351         Cleans the network
352         """
353         if self.router_creator:
354             self.router_creator.clean()
355
356         if self.net_creator:
357             if len(self.net_creator.get_subnets()) > 0:
358                 # Validate subnet has been deleted
359                 neutron_utils_tests.validate_subnet(
360                     self.neutron,
361                     self.net_creator.network_settings.subnet_settings[0].name,
362                     self.net_creator.network_settings.subnet_settings[0].cidr,
363                     False)
364
365             if self.net_creator.get_network():
366                 # Validate network has been deleted
367                 neutron_utils_tests.validate_network(
368                     self.neutron, self.net_creator.network_settings.name,
369                     False)
370             self.net_creator.clean()
371
372         super(self.__class__, self).__clean__()
373
374     def test_create_network_without_router(self):
375         """
376         Tests the creation of an OpenStack network without a router.
377         """
378         # Create Nework
379         self.net_creator = OpenStackNetwork(self.os_creds,
380                                             self.net_config.network_settings)
381         self.net_creator.create()
382
383         # Validate network was created
384         neutron_utils_tests.validate_network(
385             self.neutron, self.net_creator.network_settings.name, True)
386
387         # Validate subnets
388         neutron_utils_tests.validate_subnet(
389             self.neutron,
390             self.net_creator.network_settings.subnet_settings[0].name,
391             self.net_creator.network_settings.subnet_settings[0].cidr, True)
392
393     def test_create_delete_network(self):
394         """
395         Tests the creation of an OpenStack network, it's deletion, then cleanup
396         """
397         # Create Nework
398         self.net_creator = OpenStackNetwork(self.os_creds,
399                                             self.net_config.network_settings)
400         self.net_creator.create()
401
402         # Validate network was created
403         neutron_utils_tests.validate_network(
404             self.neutron, self.net_creator.network_settings.name, True)
405
406         neutron_utils.delete_network(self.neutron,
407                                      self.net_creator.get_network())
408         self.assertIsNone(neutron_utils.get_network(
409             self.neutron, self.net_creator.network_settings.name))
410
411         # This shall not throw an exception here
412         self.net_creator.clean()
413
414     def test_create_network_with_router(self):
415         """
416         Tests the creation of an OpenStack network with a router.
417         """
418         # Create Network
419         self.net_creator = OpenStackNetwork(self.os_creds,
420                                             self.net_config.network_settings)
421         self.net_creator.create()
422
423         # Create Router
424         self.router_creator = create_router.OpenStackRouter(
425             self.os_creds, self.net_config.router_settings)
426         self.router_creator.create()
427
428         # Validate network was created
429         neutron_utils_tests.validate_network(
430             self.neutron, self.net_creator.network_settings.name, True)
431
432         # Validate subnets
433         neutron_utils_tests.validate_subnet(
434             self.neutron,
435             self.net_creator.network_settings.subnet_settings[0].name,
436             self.net_creator.network_settings.subnet_settings[0].cidr, True)
437
438         # Validate routers
439         neutron_utils_tests.validate_router(
440             self.neutron, self.router_creator.router_settings.name, True)
441
442         neutron_utils_tests.validate_interface_router(
443             self.router_creator.get_internal_router_interface(),
444             self.router_creator.get_router(),
445             self.net_creator.get_subnets()[0])
446
447     def test_create_networks_same_name(self):
448         """
449         Tests the creation of an OpenStack network and ensures that the
450         OpenStackNetwork object will not create a second.
451         """
452         # Create Nework
453         self.net_creator = OpenStackNetwork(self.os_creds,
454                                             self.net_config.network_settings)
455         self.net_creator.create()
456
457         self.net_creator2 = OpenStackNetwork(self.os_creds,
458                                              self.net_config.network_settings)
459         self.net_creator2.create()
460
461         self.assertEqual(self.net_creator.get_network()['network']['id'],
462                          self.net_creator2.get_network()['network']['id'])
463
464
465 class CreateNetworkTypeTests(OSComponentTestCase):
466     """
467     Test for the CreateNework class defined in create_nework.py for testing
468     creating networks of different types
469     """
470
471     def setUp(self):
472         """
473         Sets up object for test
474         """
475         guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
476         self.net_config = openstack_tests.get_pub_net_config(
477             net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet')
478
479         self.neutron = neutron_utils.neutron_client(self.os_creds)
480
481         # Initialize for cleanup
482         self.net_creator = None
483         self.neutron = neutron_utils.neutron_client(self.os_creds)
484
485     def tearDown(self):
486         """
487         Cleans the network
488         """
489         if self.net_creator:
490             if len(self.net_creator.get_subnets()) > 0:
491                 # Validate subnet has been deleted
492                 neutron_utils_tests.validate_subnet(
493                     self.neutron,
494                     self.net_creator.network_settings.subnet_settings[0].name,
495                     self.net_creator.network_settings.subnet_settings[0].cidr,
496                     False)
497
498             if self.net_creator.get_network():
499                 # Validate network has been deleted
500                 neutron_utils_tests.validate_network(
501                     self.neutron, self.net_creator.network_settings.name,
502                     False)
503             self.net_creator.clean()
504
505     # TODO - determine why this is not working on Newton
506     #        - Unable to create the network. No tenant network is available for allocation.
507     # def test_create_network_type_vlan(self):
508     #     """
509     #     Tests the creation of an OpenStack network of type vlan.
510     #     """
511     #     # Create Network
512     #     network_type = 'vlan'
513     #     net_settings = NetworkSettings(name=self.net_config.network_settings.name,
514     #                                    subnet_settings=self.net_config.network_settings.subnet_settings,
515     #                                    network_type=network_type)
516     #
517     #     # When setting the network_type, creds must be admin
518     #     self.net_creator = OpenStackNetwork(self.os_creds, net_settings)
519     #     network = self.net_creator.create()
520     #
521     #     # Validate network was created
522     #     neutron_utils_tests.validate_network(self.neutron, net_settings.name, True)
523     #
524     #     self.assertEquals(network_type, network['network']['provider:network_type'])
525
526     def test_create_network_type_vxlan(self):
527         """
528         Tests the creation of an OpenStack network of type vxlan.
529         """
530         # Create Network
531         network_type = 'vxlan'
532         net_settings = NetworkSettings(
533             name=self.net_config.network_settings.name,
534             subnet_settings=self.net_config.network_settings.subnet_settings,
535             network_type=network_type)
536
537         # When setting the network_type, creds must be admin
538         self.net_creator = OpenStackNetwork(self.os_creds, net_settings)
539         network = self.net_creator.create()
540
541         # Validate network was created
542         neutron_utils_tests.validate_network(self.neutron, net_settings.name,
543                                              True)
544
545         self.assertEqual(network_type,
546                          network['network']['provider:network_type'])
547
548     # TODO - determine what value we need to place into physical_network
549     #        - Do not know what vaule to place into the 'physical_network' setting.
550     # def test_create_network_type_flat(self):
551     #     """
552     #     Tests the creation of an OpenStack network of type flat.
553     #     """
554     #     # Create Network
555     #     network_type = 'flat'
556     #
557     #     # Unable to find documentation on how to find a value that will work here.
558     #     # https://visibilityspots.org/vlan-flat-neutron-provider.html
559     #     # https://community.rackspace.com/products/f/45/t/4225
560     #     # It appears that this may be due to how OPNFV is configuring OpenStack.
561     #     physical_network = '???'
562     #     net_settings = NetworkSettings(name=self.net_config.network_settings.name,
563     #                                    subnet_settings=self.net_config.network_settings.subnet_settings,
564     #                                    network_type=network_type, physical_network=physical_network)
565     #     self.net_creator = OpenStackNetwork(self.os_creds, net_settings)
566     #     network = self.net_creator.create()
567     #
568     #     # Validate network was created
569     #     neutron_utils_tests.validate_network(self.neutron, net_settings.name, True)
570     #
571     #     self.assertEquals(network_type, network['network']['provider:network_type'])
572
573     def test_create_network_type_foo(self):
574         """
575         Tests the creation of an OpenStack network of type foo which should
576         raise an exception.
577         """
578         # Create Network
579         network_type = 'foo'
580         net_settings = NetworkSettings(
581             name=self.net_config.network_settings.name,
582             subnet_settings=self.net_config.network_settings.subnet_settings,
583             network_type=network_type)
584         self.net_creator = OpenStackNetwork(self.os_creds, net_settings)
585         with self.assertRaises(Exception):
586             self.net_creator.create()