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.assertEqual(0, len(settings.subnet_settings))
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))
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',
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)
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',
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)
102 class SubnetSettingsUnitTests(unittest.TestCase):
104 Tests the construction of the SubnetSettings class
107 def test_no_params(self):
108 with self.assertRaises(SubnetSettingsError):
111 def test_empty_config(self):
112 with self.assertRaises(SubnetSettingsError):
113 SubnetSettings(**dict())
115 def test_name_only(self):
116 with self.assertRaises(SubnetSettingsError):
117 SubnetSettings(name='foo')
119 def test_config_with_name_only(self):
120 with self.assertRaises(SubnetSettingsError):
121 SubnetSettings(**{'name': 'foo'})
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)
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)
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],
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)
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',
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)
216 class PortSettingsUnitTests(unittest.TestCase):
218 Tests the construction of the PortSettings class
221 def test_no_params(self):
222 with self.assertRaises(PortSettingsError):
225 def test_empty_config(self):
226 with self.assertRaises(PortSettingsError):
227 PortSettings(**dict())
229 def test_name_only(self):
230 with self.assertRaises(PortSettingsError):
231 PortSettings(name='foo')
233 def test_config_name_only(self):
234 with self.assertRaises(PortSettingsError):
235 PortSettings(**{'name': 'foo'})
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)
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)
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'}
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,
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)
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'}
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)
329 class CreateNetworkSuccessTests(OSIntegrationTestCase):
331 Test for the CreateNework class defined in create_nework.py
336 Sets up object for test
338 super(self.__class__, self).__start__()
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)
345 self.neutron = neutron_utils.neutron_client(self.os_creds)
347 # Initialize for cleanup
348 self.net_creator = None
349 self.router_creator = None
350 self.neutron = neutron_utils.neutron_client(self.os_creds)
356 if self.router_creator:
357 self.router_creator.clean()
360 self.net_creator.clean()
362 super(self.__class__, self).__clean__()
364 def test_create_network_without_router(self):
366 Tests the creation of an OpenStack network without a router.
369 self.net_creator = OpenStackNetwork(self.os_creds,
370 self.net_config.network_settings)
371 self.net_creator.create()
373 # Validate network was created
374 self.assertTrue(neutron_utils_tests.validate_network(
375 self.neutron, self.net_creator.network_settings.name, True))
378 self.assertTrue(neutron_utils_tests.validate_subnet(
380 self.net_creator.network_settings.subnet_settings[0].name,
381 self.net_creator.network_settings.subnet_settings[0].cidr, True))
383 def test_create_delete_network(self):
385 Tests the creation of an OpenStack network, it's deletion, then cleanup
388 self.net_creator = OpenStackNetwork(self.os_creds,
389 self.net_config.network_settings)
390 self.net_creator.create()
392 # Validate network was created
393 self.assertTrue(neutron_utils_tests.validate_network(
394 self.neutron, self.net_creator.network_settings.name, True))
396 neutron_utils.delete_network(self.neutron,
397 self.net_creator.get_network())
398 self.assertIsNone(neutron_utils.get_network(
399 self.neutron, network_settings=self.net_creator.network_settings))
401 # This shall not throw an exception here
402 self.net_creator.clean()
404 def test_create_network_with_router(self):
406 Tests the creation of an OpenStack network with a router.
409 self.net_creator = OpenStackNetwork(self.os_creds,
410 self.net_config.network_settings)
411 self.net_creator.create()
414 self.router_creator = create_router.OpenStackRouter(
415 self.os_creds, self.net_config.router_settings)
416 self.router_creator.create()
418 # Validate network was created
419 self.assertTrue(neutron_utils_tests.validate_network(
420 self.neutron, self.net_creator.network_settings.name, True))
423 self.assertTrue(neutron_utils_tests.validate_subnet(
425 self.net_creator.network_settings.subnet_settings[0].name,
426 self.net_creator.network_settings.subnet_settings[0].cidr, True))
429 neutron_utils_tests.validate_router(
430 self.neutron, self.router_creator.router_settings.name, True)
432 neutron_utils_tests.validate_interface_router(
433 self.router_creator.get_internal_router_interface(),
434 self.router_creator.get_router(),
435 self.net_creator.get_subnets()[0])
437 def test_create_networks_same_name(self):
439 Tests the creation of an OpenStack network and ensures that the
440 OpenStackNetwork object will not create a second.
443 self.net_creator = OpenStackNetwork(self.os_creds,
444 self.net_config.network_settings)
445 self.net_creator.create()
447 self.net_creator2 = OpenStackNetwork(self.os_creds,
448 self.net_config.network_settings)
449 self.net_creator2.create()
451 self.assertEqual(self.net_creator.get_network().id,
452 self.net_creator2.get_network().id)
454 def test_create_network_router_admin_user_to_new_project(self):
456 Tests the creation of an OpenStack network and router with the current
457 user to the admin project.
459 # Create Network/Subnet where the project names have been changed
460 admin_project_name = self.admin_os_creds.project_name
461 self.net_config.network_settings.project_name = admin_project_name
462 self.net_config.network_settings.subnet_settings[0].project_name = \
464 self.net_creator = OpenStackNetwork(self.os_creds,
465 self.net_config.network_settings)
466 self.net_creator.create()
468 retrieved_net = neutron_utils.get_network(
469 self.neutron, network_settings=self.net_config.network_settings)
471 self.assertEqual(self.net_creator.get_network().id, retrieved_net.id)
474 self.net_config.router_settings.project_name = admin_project_name
475 self.router_creator = create_router.OpenStackRouter(
476 self.os_creds, self.net_config.router_settings)
477 self.router_creator.create()
479 retrieved_router = neutron_utils.get_router(
480 self.neutron, router_settings=self.router_creator.router_settings)
482 self.router_creator.get_router().id, retrieved_router.id)
484 def test_create_network_router_new_user_to_admin_project(self):
486 Tests the creation of an OpenStack network and router with the admin
487 user to the new project.
489 # Create Network/Subnet where the project names have been changed
490 new_project_name = self.os_creds.project_name
491 self.net_config.network_settings.project_name = new_project_name
492 self.net_config.network_settings.subnet_settings[0].project_name = \
494 self.net_creator = OpenStackNetwork(self.admin_os_creds,
495 self.net_config.network_settings)
496 self.net_creator.create()
498 retrieved_net = neutron_utils.get_network(
499 self.neutron, network_settings=self.net_config.network_settings)
501 self.assertEqual(self.net_creator.get_network().id, retrieved_net.id)
504 self.net_config.router_settings.project_name = new_project_name
505 self.router_creator = create_router.OpenStackRouter(
506 self.admin_os_creds, self.net_config.router_settings)
507 self.router_creator.create()
509 retrieved_router = neutron_utils.get_router(
510 self.neutron, router_settings=self.router_creator.router_settings)
512 self.router_creator.get_router().id, retrieved_router.id)
515 class CreateNetworkTypeTests(OSComponentTestCase):
517 Test for the CreateNework class defined in create_nework.py for testing
518 creating networks of different types
523 Sets up object for test
525 guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
526 self.net_config = openstack_tests.get_pub_net_config(
527 net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet')
529 self.neutron = neutron_utils.neutron_client(self.os_creds)
531 # Initialize for cleanup
532 self.net_creator = None
533 self.neutron = neutron_utils.neutron_client(self.os_creds)
540 self.net_creator.clean()
542 def test_create_network_type_vlan(self):
544 Tests the creation of an OpenStack network of type vlan.
547 network_type = 'vlan'
548 net_settings = NetworkSettings(
549 name=self.net_config.network_settings.name,
550 subnet_settings=self.net_config.network_settings.subnet_settings,
551 network_type=network_type)
553 # When setting the network_type, creds must be admin
554 self.net_creator = OpenStackNetwork(self.os_creds, net_settings)
555 network = self.net_creator.create()
557 # Validate network was created
558 self.assertTrue(neutron_utils_tests.validate_network(
559 self.neutron, net_settings.name, True))
561 self.assertEquals(network_type, network.type)
563 def test_create_network_type_vxlan(self):
565 Tests the creation of an OpenStack network of type vxlan.
568 network_type = 'vxlan'
569 net_settings = NetworkSettings(
570 name=self.net_config.network_settings.name,
571 subnet_settings=self.net_config.network_settings.subnet_settings,
572 network_type=network_type)
574 # When setting the network_type, creds must be admin
575 self.net_creator = OpenStackNetwork(self.os_creds, net_settings)
576 network = self.net_creator.create()
578 # Validate network was created
579 self.assertTrue(neutron_utils_tests.validate_network(
580 self.neutron, net_settings.name, True))
582 self.assertEqual(network_type, network.type)
584 def test_create_network_type_flat(self):
586 Tests the creation of an OpenStack network of type flat.
589 network_type = 'flat'
591 # This value must be variable to work on all OpenStack pods
592 physical_network = 'datacentre'
594 net_settings = NetworkSettings(
595 name=self.net_config.network_settings.name,
596 subnet_settings=self.net_config.network_settings.subnet_settings,
597 network_type=network_type, physical_network=physical_network)
598 self.net_creator = OpenStackNetwork(self.os_creds, net_settings)
599 network = self.net_creator.create()
601 # Validate network was created
602 self.assertTrue(neutron_utils_tests.validate_network(
603 self.neutron, net_settings.name, True))
605 self.assertEquals(network_type, network.type)
607 def test_create_network_type_foo(self):
609 Tests the creation of an OpenStack network of type foo which should
614 net_settings = NetworkSettings(
615 name=self.net_config.network_settings.name,
616 subnet_settings=self.net_config.network_settings.subnet_settings,
617 network_type=network_type)
618 self.net_creator = OpenStackNetwork(self.os_creds, net_settings)
619 with self.assertRaises(Exception):
620 self.net_creator.create()