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)
21 from snaps.openstack.tests import openstack_tests
22 from snaps.openstack.tests.os_source_file_test import (OSIntegrationTestCase,
24 from snaps.openstack.utils import neutron_utils
25 from snaps.openstack.utils.tests import neutron_utils_tests
27 __author__ = 'spisarski'
30 class NetworkSettingsUnitTests(unittest.TestCase):
32 Tests the construction of the NetworkSettings class
35 def test_no_params(self):
36 with self.assertRaises(Exception):
39 def test_empty_config(self):
40 with self.assertRaises(Exception):
41 NetworkSettings(**dict())
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))
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))
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',
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)
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',
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)
99 class SubnetSettingsUnitTests(unittest.TestCase):
101 Tests the construction of the SubnetSettings class
104 def test_no_params(self):
105 with self.assertRaises(Exception):
108 def test_empty_config(self):
109 with self.assertRaises(Exception):
110 SubnetSettings(**dict())
112 def test_name_only(self):
113 with self.assertRaises(Exception):
114 SubnetSettings(name='foo')
116 def test_config_with_name_only(self):
117 with self.assertRaises(Exception):
118 SubnetSettings(**{'name': 'foo'})
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)
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)
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],
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)
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',
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)
213 class PortSettingsUnitTests(unittest.TestCase):
215 Tests the construction of the PortSettings class
218 def test_no_params(self):
219 with self.assertRaises(Exception):
222 def test_empty_config(self):
223 with self.assertRaises(Exception):
224 PortSettings(**dict())
226 def test_name_only(self):
227 with self.assertRaises(Exception):
228 PortSettings(name='foo')
230 def test_config_name_only(self):
231 with self.assertRaises(Exception):
232 PortSettings(**{'name': 'foo'})
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)
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)
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'}
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,
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)
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'}
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)
326 class CreateNetworkSuccessTests(OSIntegrationTestCase):
328 Test for the CreateNework class defined in create_nework.py
333 Sets up object for test
335 super(self.__class__, self).__start__()
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)
342 self.neutron = neutron_utils.neutron_client(self.os_creds)
344 # Initialize for cleanup
345 self.net_creator = None
346 self.router_creator = None
347 self.neutron = neutron_utils.neutron_client(self.os_creds)
353 if self.router_creator:
354 self.router_creator.clean()
357 if len(self.net_creator.get_subnets()) > 0:
358 # Validate subnet has been deleted
359 neutron_utils_tests.validate_subnet(
361 self.net_creator.network_settings.subnet_settings[0].name,
362 self.net_creator.network_settings.subnet_settings[0].cidr,
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,
370 self.net_creator.clean()
372 super(self.__class__, self).__clean__()
374 def test_create_network_without_router(self):
376 Tests the creation of an OpenStack network without a router.
379 self.net_creator = OpenStackNetwork(self.os_creds,
380 self.net_config.network_settings)
381 self.net_creator.create()
383 # Validate network was created
384 neutron_utils_tests.validate_network(
385 self.neutron, self.net_creator.network_settings.name, True)
388 neutron_utils_tests.validate_subnet(
390 self.net_creator.network_settings.subnet_settings[0].name,
391 self.net_creator.network_settings.subnet_settings[0].cidr, True)
393 def test_create_delete_network(self):
395 Tests the creation of an OpenStack network, it's deletion, then cleanup
398 self.net_creator = OpenStackNetwork(self.os_creds,
399 self.net_config.network_settings)
400 self.net_creator.create()
402 # Validate network was created
403 neutron_utils_tests.validate_network(
404 self.neutron, self.net_creator.network_settings.name, True)
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))
411 # This shall not throw an exception here
412 self.net_creator.clean()
414 def test_create_network_with_router(self):
416 Tests the creation of an OpenStack network with a router.
419 self.net_creator = OpenStackNetwork(self.os_creds,
420 self.net_config.network_settings)
421 self.net_creator.create()
424 self.router_creator = create_router.OpenStackRouter(
425 self.os_creds, self.net_config.router_settings)
426 self.router_creator.create()
428 # Validate network was created
429 neutron_utils_tests.validate_network(
430 self.neutron, self.net_creator.network_settings.name, True)
433 neutron_utils_tests.validate_subnet(
435 self.net_creator.network_settings.subnet_settings[0].name,
436 self.net_creator.network_settings.subnet_settings[0].cidr, True)
439 neutron_utils_tests.validate_router(
440 self.neutron, self.router_creator.router_settings.name, True)
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])
447 def test_create_networks_same_name(self):
449 Tests the creation of an OpenStack network and ensures that the
450 OpenStackNetwork object will not create a second.
453 self.net_creator = OpenStackNetwork(self.os_creds,
454 self.net_config.network_settings)
455 self.net_creator.create()
457 self.net_creator2 = OpenStackNetwork(self.os_creds,
458 self.net_config.network_settings)
459 self.net_creator2.create()
461 self.assertEqual(self.net_creator.get_network().id,
462 self.net_creator2.get_network().id)
465 class CreateNetworkTypeTests(OSComponentTestCase):
467 Test for the CreateNework class defined in create_nework.py for testing
468 creating networks of different types
473 Sets up object for test
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')
479 self.neutron = neutron_utils.neutron_client(self.os_creds)
481 # Initialize for cleanup
482 self.net_creator = None
483 self.neutron = neutron_utils.neutron_client(self.os_creds)
490 if len(self.net_creator.get_subnets()) > 0:
491 # Validate subnet has been deleted
492 neutron_utils_tests.validate_subnet(
494 self.net_creator.network_settings.subnet_settings[0].name,
495 self.net_creator.network_settings.subnet_settings[0].cidr,
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,
503 self.net_creator.clean()
505 def test_create_network_type_vlan(self):
507 Tests the creation of an OpenStack network of type vlan.
510 network_type = 'vlan'
511 net_settings = NetworkSettings(
512 name=self.net_config.network_settings.name,
513 subnet_settings=self.net_config.network_settings.subnet_settings,
514 network_type=network_type)
516 # When setting the network_type, creds must be admin
517 self.net_creator = OpenStackNetwork(self.os_creds, net_settings)
518 network = self.net_creator.create()
520 # Validate network was created
521 neutron_utils_tests.validate_network(
522 self.neutron, net_settings.name, True)
524 self.assertEquals(network_type, network.type)
526 def test_create_network_type_vxlan(self):
528 Tests the creation of an OpenStack network of type vxlan.
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)
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()
541 # Validate network was created
542 neutron_utils_tests.validate_network(self.neutron, net_settings.name,
545 self.assertEqual(network_type, network.type)
547 def test_create_network_type_flat(self):
549 Tests the creation of an OpenStack network of type flat.
552 network_type = 'flat'
554 # TODO - this value must be variable to work on all OpenStack pods
555 physical_network = 'datacentre'
556 net_settings = NetworkSettings(
557 name=self.net_config.network_settings.name,
558 subnet_settings=self.net_config.network_settings.subnet_settings,
559 network_type=network_type, physical_network=physical_network)
560 self.net_creator = OpenStackNetwork(self.os_creds, net_settings)
561 network = self.net_creator.create()
563 # Validate network was created
564 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_foo(self):
571 Tests the creation of an OpenStack network of type foo which should
576 net_settings = NetworkSettings(
577 name=self.net_config.network_settings.name,
578 subnet_settings=self.net_config.network_settings.subnet_settings,
579 network_type=network_type)
580 self.net_creator = OpenStackNetwork(self.os_creds, net_settings)
581 with self.assertRaises(Exception):
582 self.net_creator.create()