1 # Copyright (c) 2017 Cable Television Laboratories, Inc. ("CableLabs")
2 # and others. All rights reserved.
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:
8 # http://www.apache.org/licenses/LICENSE-2.0
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.
18 from snaps.openstack import create_router
19 from snaps.openstack.create_network import (OpenStackNetwork, NetworkSettings,
20 SubnetSettings, PortSettings,
24 from snaps.openstack.tests import openstack_tests
25 from snaps.openstack.tests.os_source_file_test import (OSIntegrationTestCase,
27 from snaps.openstack.utils import neutron_utils
28 from snaps.openstack.utils.tests import neutron_utils_tests
30 __author__ = 'spisarski'
33 class NetworkSettingsUnitTests(unittest.TestCase):
35 Tests the construction of the NetworkSettings class
38 def test_no_params(self):
39 with self.assertRaises(NetworkSettingsError):
42 def test_empty_config(self):
43 with self.assertRaises(NetworkSettingsError):
44 NetworkSettings(**dict())
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.assertIsNone(settings.segmentation_id)
55 self.assertEqual(0, len(settings.subnet_settings))
57 def test_config_with_name_only(self):
58 settings = NetworkSettings(**{'name': 'foo'})
59 self.assertEqual('foo', settings.name)
60 self.assertTrue(settings.admin_state_up)
61 self.assertIsNone(settings.shared)
62 self.assertIsNone(settings.project_name)
63 self.assertFalse(settings.external)
64 self.assertIsNone(settings.network_type)
65 self.assertIsNone(settings.segmentation_id)
66 self.assertEqual(0, len(settings.subnet_settings))
69 sub_settings = SubnetSettings(name='foo-subnet', cidr='10.0.0.0/24')
70 settings = NetworkSettings(name='foo', admin_state_up=False,
71 shared=True, project_name='bar',
73 network_type='vlan', physical_network='phy',
75 subnet_settings=[sub_settings])
76 self.assertEqual('foo', settings.name)
77 self.assertFalse(settings.admin_state_up)
78 self.assertTrue(settings.shared)
79 self.assertEqual('bar', settings.project_name)
80 self.assertTrue(settings.external)
81 self.assertEqual('vlan', settings.network_type)
82 self.assertEqual('phy', settings.physical_network)
83 self.assertEqual(2366, settings.segmentation_id)
84 self.assertEqual(1, len(settings.subnet_settings))
85 self.assertEqual('foo-subnet', settings.subnet_settings[0].name)
87 def test_config_all(self):
88 settings = NetworkSettings(
89 **{'name': 'foo', 'admin_state_up': False, 'shared': True,
90 'project_name': 'bar', 'external': True, 'network_type': 'vlan',
91 'physical_network': 'phy',
92 'segmentation_id': 2366,
94 [{'subnet': {'name': 'foo-subnet',
95 'cidr': '10.0.0.0/24'}}]})
96 self.assertEqual('foo', settings.name)
97 self.assertFalse(settings.admin_state_up)
98 self.assertTrue(settings.shared)
99 self.assertEqual('bar', settings.project_name)
100 self.assertTrue(settings.external)
101 self.assertEqual('vlan', settings.network_type)
102 self.assertEqual('phy', settings.physical_network)
103 self.assertEqual(2366, settings.segmentation_id)
104 self.assertEqual(1, len(settings.subnet_settings))
105 self.assertEqual('foo-subnet', settings.subnet_settings[0].name)
108 class SubnetSettingsUnitTests(unittest.TestCase):
110 Tests the construction of the SubnetSettings class
113 def test_no_params(self):
114 with self.assertRaises(SubnetSettingsError):
117 def test_empty_config(self):
118 with self.assertRaises(SubnetSettingsError):
119 SubnetSettings(**dict())
121 def test_name_only(self):
122 with self.assertRaises(SubnetSettingsError):
123 SubnetSettings(name='foo')
125 def test_config_with_name_only(self):
126 with self.assertRaises(SubnetSettingsError):
127 SubnetSettings(**{'name': 'foo'})
129 def test_name_cidr_only(self):
130 settings = SubnetSettings(name='foo', cidr='10.0.0.0/24')
131 self.assertEqual('foo', settings.name)
132 self.assertEqual('10.0.0.0/24', settings.cidr)
133 self.assertEqual(4, settings.ip_version)
134 self.assertIsNone(settings.project_name)
135 self.assertIsNone(settings.start)
136 self.assertIsNone(settings.end)
137 self.assertIsNone(settings.enable_dhcp)
138 self.assertEqual(1, len(settings.dns_nameservers))
139 self.assertEqual('8.8.8.8', settings.dns_nameservers[0])
140 self.assertIsNone(settings.host_routes)
141 self.assertIsNone(settings.destination)
142 self.assertIsNone(settings.nexthop)
143 self.assertIsNone(settings.ipv6_ra_mode)
144 self.assertIsNone(settings.ipv6_address_mode)
146 def test_config_with_name_cidr_only(self):
147 settings = SubnetSettings(**{'name': 'foo', 'cidr': '10.0.0.0/24'})
148 self.assertEqual('foo', settings.name)
149 self.assertEqual('10.0.0.0/24', settings.cidr)
150 self.assertEqual(4, settings.ip_version)
151 self.assertIsNone(settings.project_name)
152 self.assertIsNone(settings.start)
153 self.assertIsNone(settings.end)
154 self.assertIsNone(settings.gateway_ip)
155 self.assertIsNone(settings.enable_dhcp)
156 self.assertEqual(1, len(settings.dns_nameservers))
157 self.assertEqual('8.8.8.8', settings.dns_nameservers[0])
158 self.assertIsNone(settings.host_routes)
159 self.assertIsNone(settings.destination)
160 self.assertIsNone(settings.nexthop)
161 self.assertIsNone(settings.ipv6_ra_mode)
162 self.assertIsNone(settings.ipv6_address_mode)
165 host_routes = {'destination': '0.0.0.0/0', 'nexthop': '123.456.78.9'}
166 settings = SubnetSettings(name='foo', cidr='10.0.0.0/24', ip_version=6,
167 project_name='bar-project',
168 start='10.0.0.2', end='10.0.0.101',
169 gateway_ip='10.0.0.1', enable_dhcp=False,
170 dns_nameservers=['8.8.8.8'],
171 host_routes=[host_routes],
174 ipv6_ra_mode='dhcpv6-stateful',
175 ipv6_address_mode='slaac')
176 self.assertEqual('foo', settings.name)
177 self.assertEqual('10.0.0.0/24', settings.cidr)
178 self.assertEqual(6, settings.ip_version)
179 self.assertEqual('bar-project', settings.project_name)
180 self.assertEqual('10.0.0.2', settings.start)
181 self.assertEqual('10.0.0.101', settings.end)
182 self.assertEqual('10.0.0.1', settings.gateway_ip)
183 self.assertEqual(False, settings.enable_dhcp)
184 self.assertEqual(1, len(settings.dns_nameservers))
185 self.assertEqual('8.8.8.8', settings.dns_nameservers[0])
186 self.assertEqual(1, len(settings.host_routes))
187 self.assertEqual(host_routes, settings.host_routes[0])
188 self.assertEqual('dest', settings.destination)
189 self.assertEqual('hop', settings.nexthop)
190 self.assertEqual('dhcpv6-stateful', settings.ipv6_ra_mode)
191 self.assertEqual('slaac', settings.ipv6_address_mode)
193 def test_config_all(self):
194 host_routes = {'destination': '0.0.0.0/0', 'nexthop': '123.456.78.9'}
195 settings = SubnetSettings(
196 **{'name': 'foo', 'cidr': '10.0.0.0/24', 'ip_version': 6,
197 'project_name': 'bar-project', 'start': '10.0.0.2',
199 'gateway_ip': '10.0.0.1', 'enable_dhcp': False,
200 'dns_nameservers': ['8.8.8.8'], 'host_routes': [host_routes],
201 'destination': 'dest', 'nexthop': 'hop',
202 'ipv6_ra_mode': 'dhcpv6-stateful',
203 'ipv6_address_mode': 'slaac'})
204 self.assertEqual('foo', settings.name)
205 self.assertEqual('10.0.0.0/24', settings.cidr)
206 self.assertEqual(6, settings.ip_version)
207 self.assertEqual('bar-project', settings.project_name)
208 self.assertEqual('10.0.0.2', settings.start)
209 self.assertEqual('10.0.0.101', settings.end)
210 self.assertEqual('10.0.0.1', settings.gateway_ip)
211 self.assertEqual(False, settings.enable_dhcp)
212 self.assertEqual(1, len(settings.dns_nameservers))
213 self.assertEqual('8.8.8.8', settings.dns_nameservers[0])
214 self.assertEqual(1, len(settings.host_routes))
215 self.assertEqual(host_routes, settings.host_routes[0])
216 self.assertEqual('dest', settings.destination)
217 self.assertEqual('hop', settings.nexthop)
218 self.assertEqual('dhcpv6-stateful', settings.ipv6_ra_mode)
219 self.assertEqual('slaac', settings.ipv6_address_mode)
222 class PortSettingsUnitTests(unittest.TestCase):
224 Tests the construction of the PortSettings class
227 def test_no_params(self):
228 with self.assertRaises(PortSettingsError):
231 def test_empty_config(self):
232 with self.assertRaises(PortSettingsError):
233 PortSettings(**dict())
235 def test_name_only(self):
236 with self.assertRaises(PortSettingsError):
237 PortSettings(name='foo')
239 def test_config_name_only(self):
240 with self.assertRaises(PortSettingsError):
241 PortSettings(**{'name': 'foo'})
243 def test_name_netname_only(self):
244 settings = PortSettings(name='foo', network_name='bar')
245 self.assertEqual('foo', settings.name)
246 self.assertEqual('bar', settings.network_name)
247 self.assertTrue(settings.admin_state_up)
248 self.assertIsNone(settings.project_name)
249 self.assertIsNone(settings.mac_address)
250 self.assertIsNone(settings.ip_addrs)
251 self.assertIsNone(settings.fixed_ips)
252 self.assertIsNone(settings.security_groups)
253 self.assertIsNone(settings.allowed_address_pairs)
254 self.assertIsNone(settings.opt_value)
255 self.assertIsNone(settings.opt_name)
256 self.assertIsNone(settings.device_owner)
257 self.assertIsNone(settings.device_id)
259 def test_config_with_name_netname_only(self):
260 settings = PortSettings(**{'name': 'foo', 'network_name': 'bar'})
261 self.assertEqual('foo', settings.name)
262 self.assertEqual('bar', settings.network_name)
263 self.assertTrue(settings.admin_state_up)
264 self.assertIsNone(settings.project_name)
265 self.assertIsNone(settings.mac_address)
266 self.assertIsNone(settings.ip_addrs)
267 self.assertIsNone(settings.fixed_ips)
268 self.assertIsNone(settings.security_groups)
269 self.assertIsNone(settings.allowed_address_pairs)
270 self.assertIsNone(settings.opt_value)
271 self.assertIsNone(settings.opt_name)
272 self.assertIsNone(settings.device_owner)
273 self.assertIsNone(settings.device_id)
276 ip_addrs = [{'subnet_name', 'foo-sub', 'ip', '10.0.0.10'}]
277 fixed_ips = {'sub_id', '10.0.0.10'}
278 allowed_address_pairs = {'10.0.0.101', '1234.5678'}
280 settings = PortSettings(name='foo', network_name='bar',
281 admin_state_up=False,
282 project_name='foo-project',
283 mac_address='1234', ip_addrs=ip_addrs,
285 security_groups=['foo_grp_id'],
286 allowed_address_pairs=allowed_address_pairs,
287 opt_value='opt value', opt_name='opt name',
288 device_owner='owner',
289 device_id='device number')
290 self.assertEqual('foo', settings.name)
291 self.assertEqual('bar', settings.network_name)
292 self.assertFalse(settings.admin_state_up)
293 self.assertEqual('foo-project', settings.project_name)
294 self.assertEqual('1234', settings.mac_address)
295 self.assertEqual(ip_addrs, settings.ip_addrs)
296 self.assertEqual(fixed_ips, settings.fixed_ips)
297 self.assertEqual(1, len(settings.security_groups))
298 self.assertEqual('foo_grp_id', settings.security_groups[0])
299 self.assertEqual(allowed_address_pairs, settings.allowed_address_pairs)
300 self.assertEqual('opt value', settings.opt_value)
301 self.assertEqual('opt name', settings.opt_name)
302 self.assertEqual('owner', settings.device_owner)
303 self.assertEqual('device number', settings.device_id)
305 def test_config_all(self):
306 ip_addrs = [{'subnet_name', 'foo-sub', 'ip', '10.0.0.10'}]
307 fixed_ips = {'sub_id', '10.0.0.10'}
308 allowed_address_pairs = {'10.0.0.101', '1234.5678'}
310 settings = PortSettings(
311 **{'name': 'foo', 'network_name': 'bar', 'admin_state_up': False,
312 'project_name': 'foo-project', 'mac_address': '1234',
313 'ip_addrs': ip_addrs,
314 'fixed_ips': fixed_ips, 'security_groups': ['foo_grp_id'],
315 'allowed_address_pairs': allowed_address_pairs,
316 'opt_value': 'opt value',
317 'opt_name': 'opt name', 'device_owner': 'owner',
318 'device_id': 'device number'})
319 self.assertEqual('foo', settings.name)
320 self.assertEqual('bar', settings.network_name)
321 self.assertFalse(settings.admin_state_up)
322 self.assertEqual('foo-project', settings.project_name)
323 self.assertEqual('1234', settings.mac_address)
324 self.assertEqual(ip_addrs, settings.ip_addrs)
325 self.assertEqual(fixed_ips, settings.fixed_ips)
326 self.assertEqual(1, len(settings.security_groups))
327 self.assertEqual('foo_grp_id', settings.security_groups[0])
328 self.assertEqual(allowed_address_pairs, settings.allowed_address_pairs)
329 self.assertEqual('opt value', settings.opt_value)
330 self.assertEqual('opt name', settings.opt_name)
331 self.assertEqual('owner', settings.device_owner)
332 self.assertEqual('device number', settings.device_id)
335 class CreateNetworkSuccessTests(OSIntegrationTestCase):
337 Test for the CreateNework class defined in create_nework.py
342 Sets up object for test
344 super(self.__class__, self).__start__()
346 guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
347 self.net_config = openstack_tests.get_pub_net_config(
348 net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet',
349 router_name=guid + '-pub-router', external_net=self.ext_net_name)
351 self.neutron = neutron_utils.neutron_client(self.os_creds)
353 # Initialize for cleanup
354 self.net_creator = None
355 self.router_creator = None
356 self.neutron = neutron_utils.neutron_client(self.os_creds)
362 if self.router_creator:
363 self.router_creator.clean()
366 self.net_creator.clean()
368 super(self.__class__, self).__clean__()
370 def test_create_network_without_router(self):
372 Tests the creation of an OpenStack network without a router.
375 self.net_creator = OpenStackNetwork(self.os_creds,
376 self.net_config.network_settings)
377 self.net_creator.create()
379 # Validate network was created
380 self.assertTrue(neutron_utils_tests.validate_network(
381 self.neutron, self.net_creator.network_settings.name, True))
384 self.assertTrue(neutron_utils_tests.validate_subnet(
386 self.net_creator.network_settings.subnet_settings[0].name,
387 self.net_creator.network_settings.subnet_settings[0].cidr, True))
389 def test_create_delete_network(self):
391 Tests the creation of an OpenStack network, it's deletion, then cleanup
394 self.net_creator = OpenStackNetwork(self.os_creds,
395 self.net_config.network_settings)
396 self.net_creator.create()
398 # Validate network was created
399 self.assertTrue(neutron_utils_tests.validate_network(
400 self.neutron, self.net_creator.network_settings.name, True))
402 neutron_utils.delete_network(self.neutron,
403 self.net_creator.get_network())
404 self.assertIsNone(neutron_utils.get_network(
405 self.neutron, network_settings=self.net_creator.network_settings))
407 # This shall not throw an exception here
408 self.net_creator.clean()
410 def test_create_network_with_router(self):
412 Tests the creation of an OpenStack network with a router.
415 self.net_creator = OpenStackNetwork(self.os_creds,
416 self.net_config.network_settings)
417 self.net_creator.create()
420 self.router_creator = create_router.OpenStackRouter(
421 self.os_creds, self.net_config.router_settings)
422 self.router_creator.create()
424 # Validate network was created
425 self.assertTrue(neutron_utils_tests.validate_network(
426 self.neutron, self.net_creator.network_settings.name, True))
429 self.assertTrue(neutron_utils_tests.validate_subnet(
431 self.net_creator.network_settings.subnet_settings[0].name,
432 self.net_creator.network_settings.subnet_settings[0].cidr, True))
435 neutron_utils_tests.validate_router(
436 self.neutron, self.router_creator.router_settings.name, True)
438 neutron_utils_tests.validate_interface_router(
439 self.router_creator.get_internal_router_interface(),
440 self.router_creator.get_router(),
441 self.net_creator.get_subnets()[0])
443 def test_create_networks_same_name(self):
445 Tests the creation of an OpenStack network and ensures that the
446 OpenStackNetwork object will not create a second.
449 self.net_creator = OpenStackNetwork(self.os_creds,
450 self.net_config.network_settings)
451 self.net_creator.create()
453 self.net_creator2 = OpenStackNetwork(self.os_creds,
454 self.net_config.network_settings)
455 self.net_creator2.create()
457 self.assertEqual(self.net_creator.get_network().id,
458 self.net_creator2.get_network().id)
460 def test_create_network_router_admin_user_to_new_project(self):
462 Tests the creation of an OpenStack network and router with the current
463 user to the admin project.
465 # Create Network/Subnet where the project names have been changed
466 admin_project_name = self.admin_os_creds.project_name
467 self.net_config.network_settings.project_name = admin_project_name
468 self.net_config.network_settings.subnet_settings[0].project_name = \
470 self.net_creator = OpenStackNetwork(self.os_creds,
471 self.net_config.network_settings)
472 self.net_creator.create()
474 retrieved_net = neutron_utils.get_network(
475 self.neutron, network_settings=self.net_config.network_settings)
477 self.assertEqual(self.net_creator.get_network().id, retrieved_net.id)
480 self.net_config.router_settings.project_name = admin_project_name
481 self.router_creator = create_router.OpenStackRouter(
482 self.os_creds, self.net_config.router_settings)
483 self.router_creator.create()
485 retrieved_router = neutron_utils.get_router(
486 self.neutron, router_settings=self.router_creator.router_settings)
488 self.router_creator.get_router().id, retrieved_router.id)
490 def test_create_network_router_new_user_to_admin_project(self):
492 Tests the creation of an OpenStack network and router with the admin
493 user to the new project.
495 # Create Network/Subnet where the project names have been changed
496 new_project_name = self.os_creds.project_name
497 self.net_config.network_settings.project_name = new_project_name
498 self.net_config.network_settings.subnet_settings[0].project_name = \
500 self.net_creator = OpenStackNetwork(self.admin_os_creds,
501 self.net_config.network_settings)
502 self.net_creator.create()
504 retrieved_net = neutron_utils.get_network(
505 self.neutron, network_settings=self.net_config.network_settings)
507 self.assertEqual(self.net_creator.get_network().id, retrieved_net.id)
510 self.net_config.router_settings.project_name = new_project_name
511 self.router_creator = create_router.OpenStackRouter(
512 self.admin_os_creds, self.net_config.router_settings)
513 self.router_creator.create()
515 retrieved_router = neutron_utils.get_router(
516 self.neutron, router_settings=self.router_creator.router_settings)
518 self.router_creator.get_router().id, retrieved_router.id)
521 class CreateNetworkTypeTests(OSComponentTestCase):
523 Test for the CreateNework class defined in create_nework.py for testing
524 creating networks of different types
529 Sets up object for test
531 guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
532 self.net_config = openstack_tests.get_pub_net_config(
533 net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet')
535 self.neutron = neutron_utils.neutron_client(self.os_creds)
537 # Initialize for cleanup
538 self.net_creator = None
539 self.neutron = neutron_utils.neutron_client(self.os_creds)
546 self.net_creator.clean()
548 def test_create_network_type_vlan(self):
550 Tests the creation of an OpenStack network of type vlan.
553 network_type = 'vlan'
554 net_settings = NetworkSettings(
555 name=self.net_config.network_settings.name,
556 subnet_settings=self.net_config.network_settings.subnet_settings,
557 network_type=network_type)
559 # When setting the network_type, creds must be admin
560 self.net_creator = OpenStackNetwork(self.os_creds, net_settings)
561 network = self.net_creator.create()
563 # Validate network was created
564 self.assertTrue(neutron_utils_tests.validate_network(
565 self.neutron, net_settings.name, True))
567 self.assertEquals(network_type, network.type)
569 def test_create_network_type_vlan_with_physical_net_and_seg_id(self):
571 Tests the creation of an OpenStack network of type vlan with
572 specified physical network and segmentation id.
575 network_type = 'vlan'
577 # The two values must be variable to work on all OpenStack pods
578 physical_network = 'datacentre'
579 segmentation_id = 2366
581 net_settings = NetworkSettings(
582 name=self.net_config.network_settings.name,
583 subnet_settings=self.net_config.network_settings.subnet_settings,
584 network_type=network_type,
585 physical_network=physical_network,
586 segmentation_id=segmentation_id)
588 # When setting the network_type, creds must be admin
589 self.net_creator = OpenStackNetwork(self.os_creds, net_settings)
590 network = self.net_creator.create()
592 # Validate network was created
593 self.assertTrue(neutron_utils_tests.validate_network(
594 self.neutron, net_settings.name, True))
596 self.assertEquals(network_type, network.type)
598 def test_create_network_type_vxlan(self):
600 Tests the creation of an OpenStack network of type vxlan.
603 network_type = 'vxlan'
604 net_settings = NetworkSettings(
605 name=self.net_config.network_settings.name,
606 subnet_settings=self.net_config.network_settings.subnet_settings,
607 network_type=network_type)
609 # When setting the network_type, creds must be admin
610 self.net_creator = OpenStackNetwork(self.os_creds, net_settings)
611 network = self.net_creator.create()
613 # Validate network was created
614 self.assertTrue(neutron_utils_tests.validate_network(
615 self.neutron, net_settings.name, True))
617 self.assertEqual(network_type, network.type)
619 def test_create_network_type_flat(self):
621 Tests the creation of an OpenStack network of type flat.
624 network_type = 'flat'
626 # This value must be variable to work on all OpenStack pods
627 physical_network = 'datacentre'
629 net_settings = NetworkSettings(
630 name=self.net_config.network_settings.name,
631 subnet_settings=self.net_config.network_settings.subnet_settings,
632 network_type=network_type, physical_network=physical_network)
633 self.net_creator = OpenStackNetwork(self.os_creds, net_settings)
634 network = self.net_creator.create()
636 # Validate network was created
637 self.assertTrue(neutron_utils_tests.validate_network(
638 self.neutron, net_settings.name, True))
640 self.assertEquals(network_type, network.type)
642 def test_create_network_type_foo(self):
644 Tests the creation of an OpenStack network of type foo which should
649 net_settings = NetworkSettings(
650 name=self.net_config.network_settings.name,
651 subnet_settings=self.net_config.network_settings.subnet_settings,
652 network_type=network_type)
653 self.net_creator = OpenStackNetwork(self.os_creds, net_settings)
654 with self.assertRaises(Exception):
655 self.net_creator.create()