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