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.
17 from snaps.openstack import create_router
18 from snaps.openstack.create_network import NetworkSettings, SubnetSettings, \
20 from snaps.openstack.create_security_group import SecurityGroupSettings, \
21 SecurityGroupRuleSettings, Direction
22 from snaps.openstack.tests import openstack_tests
23 from snaps.openstack.tests import validation_utils
24 from snaps.openstack.tests.os_source_file_test import OSComponentTestCase
25 from snaps.openstack.utils import keystone_utils
26 from snaps.openstack.utils import neutron_utils
27 from snaps.openstack.utils.neutron_utils import NeutronException
29 __author__ = 'spisarski'
35 class NeutronSmokeTests(OSComponentTestCase):
37 Tests to ensure that the neutron client can communicate with the cloud
40 def test_neutron_connect_success(self):
42 Tests to ensure that the proper credentials can connect.
44 neutron = neutron_utils.neutron_client(self.os_creds)
46 networks = neutron.list_networks()
49 networks = networks.get('networks')
50 for network in networks:
51 if network.get('name') == self.ext_net_name:
53 self.assertTrue(found)
55 def test_neutron_connect_fail(self):
57 Tests to ensure that the improper credentials cannot connect.
59 from snaps.openstack.os_credentials import OSCreds
61 with self.assertRaises(Exception):
62 neutron = neutron_utils.neutron_client(
63 OSCreds(username='user', password='pass', auth_url='url',
64 project_name='project'))
65 neutron.list_networks()
67 def test_retrieve_ext_network_name(self):
69 Tests the neutron_utils.get_external_network_names to ensure the
70 configured self.ext_net_name is contained within the returned list
73 neutron = neutron_utils.neutron_client(self.os_creds)
74 ext_networks = neutron_utils.get_external_networks(neutron)
76 for network in ext_networks:
77 if network.name == self.ext_net_name:
80 self.assertTrue(found)
83 class NeutronUtilsNetworkTests(OSComponentTestCase):
85 Test for creating networks via neutron_utils.py
89 guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
90 self.port_name = str(guid) + '-port'
91 self.neutron = neutron_utils.neutron_client(self.os_creds)
93 self.net_config = openstack_tests.get_pub_net_config(
94 net_name=guid + '-pub-net')
98 Cleans the remote OpenStack objects
101 neutron_utils.delete_network(self.neutron, self.network)
102 validate_network(self.neutron, self.network.name, False)
104 def test_create_network(self):
106 Tests the neutron_utils.create_neutron_net() function
108 self.network = neutron_utils.create_network(
109 self.neutron, self.os_creds, self.net_config.network_settings)
110 self.assertEqual(self.net_config.network_settings.name,
112 self.assertTrue(validate_network(
113 self.neutron, self.net_config.network_settings.name, True))
115 def test_create_network_empty_name(self):
117 Tests the neutron_utils.create_neutron_net() function with an empty
120 with self.assertRaises(Exception):
121 self.network = neutron_utils.create_network(
122 self.neutron, self.os_creds,
123 network_settings=NetworkSettings(name=''))
125 def test_create_network_null_name(self):
127 Tests the neutron_utils.create_neutron_net() function when the network
130 with self.assertRaises(Exception):
131 self.network = neutron_utils.create_network(
132 self.neutron, self.os_creds,
133 network_settings=NetworkSettings())
136 class NeutronUtilsSubnetTests(OSComponentTestCase):
138 Test for creating networks with subnets via neutron_utils.py
142 guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
143 self.port_name = str(guid) + '-port'
144 self.neutron = neutron_utils.neutron_client(self.os_creds)
147 self.net_config = openstack_tests.get_pub_net_config(
148 net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet',
149 external_net=self.ext_net_name)
153 Cleans the remote OpenStack objects
156 neutron_utils.delete_subnet(self.neutron, self.subnet)
157 validate_subnet(self.neutron, self.subnet.name,
158 self.net_config.network_settings.subnet_settings[
162 neutron_utils.delete_network(self.neutron, self.network)
163 validate_network(self.neutron, self.network.name, False)
165 def test_create_subnet(self):
167 Tests the neutron_utils.create_neutron_net() function
169 self.network = neutron_utils.create_network(
170 self.neutron, self.os_creds, self.net_config.network_settings)
171 self.assertEqual(self.net_config.network_settings.name,
173 self.assertTrue(validate_network(
174 self.neutron, self.net_config.network_settings.name, True))
176 subnet_setting = self.net_config.network_settings.subnet_settings[0]
177 self.subnet = neutron_utils.create_subnet(
178 self.neutron, subnet_setting, self.os_creds, network=self.network)
180 self.neutron, subnet_setting.name, subnet_setting.cidr, True)
182 def test_create_subnet_null_name(self):
184 Tests the neutron_utils.create_neutron_subnet() function for an
185 Exception when the subnet name is None
187 self.network = neutron_utils.create_network(
188 self.neutron, self.os_creds, self.net_config.network_settings)
189 self.assertEqual(self.net_config.network_settings.name,
191 self.assertTrue(validate_network(
192 self.neutron, self.net_config.network_settings.name, True))
194 with self.assertRaises(Exception):
195 SubnetSettings(cidr=self.net_config.subnet_cidr)
197 def test_create_subnet_empty_name(self):
199 Tests the neutron_utils.create_neutron_net() function with an empty
202 self.network = neutron_utils.create_network(
203 self.neutron, self.os_creds, self.net_config.network_settings)
204 self.assertEqual(self.net_config.network_settings.name,
206 self.assertTrue(validate_network(
207 self.neutron, self.net_config.network_settings.name, True))
209 subnet_setting = self.net_config.network_settings.subnet_settings[0]
210 neutron_utils.create_subnet(
211 self.neutron, subnet_setting, self.os_creds, network=self.network)
212 validate_subnet(self.neutron, '', subnet_setting.cidr, True)
214 def test_create_subnet_null_cidr(self):
216 Tests the neutron_utils.create_neutron_subnet() function for an
217 Exception when the subnet CIDR value is None
219 self.network = neutron_utils.create_network(
220 self.neutron, self.os_creds, self.net_config.network_settings)
221 self.assertEqual(self.net_config.network_settings.name,
223 self.assertTrue(validate_network(
224 self.neutron, self.net_config.network_settings.name, True))
226 with self.assertRaises(Exception):
227 sub_sets = SubnetSettings(
228 cidr=None, name=self.net_config.subnet_name)
229 neutron_utils.create_subnet(
230 self.neutron, sub_sets, self.os_creds, network=self.network)
232 def test_create_subnet_empty_cidr(self):
234 Tests the neutron_utils.create_neutron_subnet() function for an
235 Exception when the subnet CIDR value is empty
237 self.network = neutron_utils.create_network(
238 self.neutron, self.os_creds, self.net_config.network_settings)
239 self.assertEqual(self.net_config.network_settings.name,
241 self.assertTrue(validate_network(
242 self.neutron, self.net_config.network_settings.name, True))
244 with self.assertRaises(Exception):
245 sub_sets = SubnetSettings(
246 cidr='', name=self.net_config.subnet_name)
247 neutron_utils.create_subnet(self.neutron, sub_sets, self.os_creds,
248 network=self.network)
251 class NeutronUtilsRouterTests(OSComponentTestCase):
253 Test for creating routers via neutron_utils.py
257 guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
258 self.port_name = str(guid) + '-port'
259 self.neutron = neutron_utils.neutron_client(self.os_creds)
264 self.interface_router = None
265 self.net_config = openstack_tests.get_pub_net_config(
266 net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet',
267 router_name=guid + '-pub-router', external_net=self.ext_net_name)
271 Cleans the remote OpenStack objects
273 if self.interface_router:
274 neutron_utils.remove_interface_router(self.neutron, self.router,
278 neutron_utils.delete_router(self.neutron, self.router)
279 validate_router(self.neutron, self.router.name, False)
282 neutron_utils.delete_port(self.neutron, self.port)
285 neutron_utils.delete_subnet(self.neutron, self.subnet)
287 self.neutron, self.subnet.name,
288 self.net_config.network_settings.subnet_settings[0].cidr,
292 neutron_utils.delete_network(self.neutron, self.network)
293 validate_network(self.neutron, self.network.name, False)
295 def test_create_router_simple(self):
297 Tests the neutron_utils.create_neutron_net() function when an external
300 self.router = neutron_utils.create_router(
301 self.neutron, self.os_creds, self.net_config.router_settings)
302 validate_router(self.neutron, self.net_config.router_settings.name,
305 def test_create_router_with_public_interface(self):
307 Tests the neutron_utils.create_neutron_net() function when an external
310 subnet_setting = self.net_config.network_settings.subnet_settings[0]
311 self.net_config = openstack_tests.OSNetworkConfig(
312 self.net_config.network_settings.name,
315 self.net_config.router_settings.name,
317 self.router = neutron_utils.create_router(
318 self.neutron, self.os_creds, self.net_config.router_settings)
319 validate_router(self.neutron, self.net_config.router_settings.name,
321 # TODO - Add validation that the router gatway has been set
323 def test_create_router_empty_name(self):
325 Tests the neutron_utils.create_neutron_net() function
327 with self.assertRaises(Exception):
328 this_router_settings = create_router.RouterSettings(name='')
329 self.router = neutron_utils.create_router(self.neutron,
331 this_router_settings)
333 def test_create_router_null_name(self):
335 Tests the neutron_utils.create_neutron_subnet() function when the
336 subnet CIDR value is None
338 with self.assertRaises(Exception):
339 this_router_settings = create_router.RouterSettings()
340 self.router = neutron_utils.create_router(self.neutron,
342 this_router_settings)
343 validate_router(self.neutron, None, True)
345 def test_add_interface_router(self):
347 Tests the neutron_utils.add_interface_router() function
349 self.network = neutron_utils.create_network(
350 self.neutron, self.os_creds, self.net_config.network_settings)
351 self.assertEqual(self.net_config.network_settings.name,
353 self.assertTrue(validate_network(
354 self.neutron, self.net_config.network_settings.name, True))
356 subnet_setting = self.net_config.network_settings.subnet_settings[0]
357 self.subnet = neutron_utils.create_subnet(
358 self.neutron, subnet_setting,
359 self.os_creds, self.network)
363 subnet_setting.cidr, True)
365 self.router = neutron_utils.create_router(
366 self.neutron, self.os_creds, self.net_config.router_settings)
367 validate_router(self.neutron, self.net_config.router_settings.name,
370 self.interface_router = neutron_utils.add_interface_router(
371 self.neutron, self.router, self.subnet)
372 validate_interface_router(self.interface_router, self.router,
375 def test_add_interface_router_null_router(self):
377 Tests the neutron_utils.add_interface_router() function for an
378 Exception when the router value is None
380 self.network = neutron_utils.create_network(
381 self.neutron, self.os_creds, self.net_config.network_settings)
382 self.assertEqual(self.net_config.network_settings.name,
384 self.assertTrue(validate_network(
385 self.neutron, self.net_config.network_settings.name, True))
387 subnet_setting = self.net_config.network_settings.subnet_settings[0]
388 self.subnet = neutron_utils.create_subnet(
389 self.neutron, subnet_setting,
390 self.os_creds, self.network)
392 self.neutron, subnet_setting.name, subnet_setting.cidr, True)
394 with self.assertRaises(NeutronException):
395 self.interface_router = neutron_utils.add_interface_router(
396 self.neutron, self.router, self.subnet)
398 def test_add_interface_router_null_subnet(self):
400 Tests the neutron_utils.add_interface_router() function for an
401 Exception when the subnet value is None
403 self.network = neutron_utils.create_network(
404 self.neutron, self.os_creds, self.net_config.network_settings)
405 self.assertEqual(self.net_config.network_settings.name,
407 self.assertTrue(validate_network(
408 self.neutron, self.net_config.network_settings.name, True))
410 self.router = neutron_utils.create_router(
411 self.neutron, self.os_creds, self.net_config.router_settings)
412 validate_router(self.neutron, self.net_config.router_settings.name,
415 with self.assertRaises(NeutronException):
416 self.interface_router = neutron_utils.add_interface_router(
417 self.neutron, self.router, self.subnet)
419 def test_create_port(self):
421 Tests the neutron_utils.create_port() function
423 self.network = neutron_utils.create_network(
424 self.neutron, self.os_creds, self.net_config.network_settings)
425 self.assertEqual(self.net_config.network_settings.name,
427 self.assertTrue(validate_network(
428 self.neutron, self.net_config.network_settings.name, True))
430 subnet_setting = self.net_config.network_settings.subnet_settings[0]
431 self.subnet = neutron_utils.create_subnet(
432 self.neutron, subnet_setting, self.os_creds, self.network)
433 validate_subnet(self.neutron, subnet_setting.name,
434 subnet_setting.cidr, True)
436 self.port = neutron_utils.create_port(
437 self.neutron, self.os_creds, PortSettings(
440 'subnet_name': subnet_setting.name,
442 network_name=self.net_config.network_settings.name))
443 validate_port(self.neutron, self.port, self.port_name)
445 def test_create_port_empty_name(self):
447 Tests the neutron_utils.create_port() function
449 self.network = neutron_utils.create_network(
450 self.neutron, self.os_creds, self.net_config.network_settings)
451 self.assertEqual(self.net_config.network_settings.name,
453 self.assertTrue(validate_network(
454 self.neutron, self.net_config.network_settings.name, True))
456 subnet_setting = self.net_config.network_settings.subnet_settings[0]
457 self.subnet = neutron_utils.create_subnet(
458 self.neutron, subnet_setting, self.os_creds, self.network)
459 validate_subnet(self.neutron, subnet_setting.name, subnet_setting.cidr,
462 self.port = neutron_utils.create_port(
463 self.neutron, self.os_creds, PortSettings(
465 network_name=self.net_config.network_settings.name,
467 'subnet_name': subnet_setting.name,
469 validate_port(self.neutron, self.port, self.port_name)
471 def test_create_port_null_name(self):
473 Tests the neutron_utils.create_port() function for an Exception when
474 the port name value is None
476 self.network = neutron_utils.create_network(
477 self.neutron, self.os_creds, self.net_config.network_settings)
478 self.assertEqual(self.net_config.network_settings.name,
480 self.assertTrue(validate_network(
481 self.neutron, self.net_config.network_settings.name, True))
483 subnet_setting = self.net_config.network_settings.subnet_settings[0]
484 self.subnet = neutron_utils.create_subnet(
485 self.neutron, subnet_setting,
486 self.os_creds, self.network)
490 subnet_setting.cidr, True)
492 with self.assertRaises(Exception):
493 self.port = neutron_utils.create_port(
494 self.neutron, self.os_creds,
496 network_name=self.net_config.network_settings.name,
498 'subnet_name': subnet_setting.name,
501 def test_create_port_null_network_object(self):
503 Tests the neutron_utils.create_port() function for an Exception when
504 the network object is None
506 with self.assertRaises(Exception):
507 self.port = neutron_utils.create_port(
508 self.neutron, self.os_creds,
511 network_name=self.net_config.network_settings.name,
514 self.net_config.network_settings.subnet_settings[
518 def test_create_port_null_ip(self):
520 Tests the neutron_utils.create_port() function for an Exception when
523 self.network = neutron_utils.create_network(
524 self.neutron, self.os_creds, self.net_config.network_settings)
525 self.assertEqual(self.net_config.network_settings.name,
527 self.assertTrue(validate_network(
528 self.neutron, self.net_config.network_settings.name, True))
530 subnet_setting = self.net_config.network_settings.subnet_settings[0]
531 self.subnet = neutron_utils.create_subnet(
532 self.neutron, subnet_setting,
533 self.os_creds, self.network)
537 subnet_setting.cidr, True)
539 with self.assertRaises(Exception):
540 self.port = neutron_utils.create_port(
541 self.neutron, self.os_creds,
544 network_name=self.net_config.network_settings.name,
546 'subnet_name': subnet_setting.name,
549 def test_create_port_invalid_ip(self):
551 Tests the neutron_utils.create_port() function for an Exception when
554 self.network = neutron_utils.create_network(
555 self.neutron, self.os_creds, self.net_config.network_settings)
556 self.assertEqual(self.net_config.network_settings.name,
558 self.assertTrue(validate_network(
559 self.neutron, self.net_config.network_settings.name, True))
561 subnet_setting = self.net_config.network_settings.subnet_settings[0]
562 self.subnet = neutron_utils.create_subnet(
563 self.neutron, subnet_setting, self.os_creds, self.network)
564 validate_subnet(self.neutron,
566 subnet_setting.cidr, True)
568 with self.assertRaises(Exception):
569 self.port = neutron_utils.create_port(
570 self.neutron, self.os_creds,
573 network_name=self.net_config.network_settings.name,
575 'subnet_name': subnet_setting.name,
578 def test_create_port_invalid_ip_to_subnet(self):
580 Tests the neutron_utils.create_port() function for an Exception when
583 self.network = neutron_utils.create_network(
584 self.neutron, self.os_creds, self.net_config.network_settings)
585 self.assertEqual(self.net_config.network_settings.name,
587 self.assertTrue(validate_network(
588 self.neutron, self.net_config.network_settings.name, True))
590 subnet_setting = self.net_config.network_settings.subnet_settings[0]
591 self.subnet = neutron_utils.create_subnet(
592 self.neutron, subnet_setting, self.os_creds, self.network)
594 self.neutron, subnet_setting.name, subnet_setting.cidr, True)
596 with self.assertRaises(Exception):
597 self.port = neutron_utils.create_port(
598 self.neutron, self.os_creds,
601 network_name=self.net_config.network_settings.name,
603 'subnet_name': subnet_setting.name,
604 'ip': '10.197.123.100'}]))
607 class NeutronUtilsSecurityGroupTests(OSComponentTestCase):
609 Test for creating security groups via neutron_utils.py
613 guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
614 self.sec_grp_name = guid + 'name'
616 self.security_groups = list()
617 self.security_group_rules = list()
618 self.neutron = neutron_utils.neutron_client(self.os_creds)
619 self.keystone = keystone_utils.keystone_client(self.os_creds)
623 Cleans the remote OpenStack objects
625 for rule in self.security_group_rules:
626 neutron_utils.delete_security_group_rule(self.neutron, rule)
628 for security_group in self.security_groups:
630 neutron_utils.delete_security_group(self.neutron,
635 def test_create_delete_simple_sec_grp(self):
637 Tests the neutron_utils.create_security_group() function
639 sec_grp_settings = SecurityGroupSettings(name=self.sec_grp_name)
640 security_group = neutron_utils.create_security_group(self.neutron,
644 self.assertTrue(sec_grp_settings.name, security_group.name)
646 sec_grp_get = neutron_utils.get_security_group(self.neutron,
647 sec_grp_settings.name)
648 self.assertIsNotNone(sec_grp_get)
649 self.assertTrue(validation_utils.objects_equivalent(
650 security_group, sec_grp_get))
652 neutron_utils.delete_security_group(self.neutron, security_group)
653 sec_grp_get = neutron_utils.get_security_group(self.neutron,
654 sec_grp_settings.name)
655 self.assertIsNone(sec_grp_get)
657 def test_create_sec_grp_no_name(self):
659 Tests the SecurityGroupSettings constructor and
660 neutron_utils.create_security_group() function to ensure that
661 attempting to create a security group without a name will raise an
664 with self.assertRaises(Exception):
665 sec_grp_settings = SecurityGroupSettings()
666 self.security_groups.append(
667 neutron_utils.create_security_group(self.neutron,
671 def test_create_sec_grp_no_rules(self):
673 Tests the neutron_utils.create_security_group() function
675 sec_grp_settings = SecurityGroupSettings(name=self.sec_grp_name,
676 description='hello group')
677 self.security_groups.append(
678 neutron_utils.create_security_group(self.neutron, self.keystone,
681 self.assertTrue(sec_grp_settings.name, self.security_groups[0].name)
682 self.assertEqual(sec_grp_settings.name, self.security_groups[0].name)
684 sec_grp_get = neutron_utils.get_security_group(self.neutron,
685 sec_grp_settings.name)
686 self.assertIsNotNone(sec_grp_get)
687 self.assertEqual(self.security_groups[0], sec_grp_get)
689 def test_create_sec_grp_one_rule(self):
691 Tests the neutron_utils.create_security_group() function
694 sec_grp_rule_settings = SecurityGroupRuleSettings(
695 sec_grp_name=self.sec_grp_name, direction=Direction.ingress)
696 sec_grp_settings = SecurityGroupSettings(
697 name=self.sec_grp_name, description='hello group',
698 rule_settings=[sec_grp_rule_settings])
700 self.security_groups.append(
701 neutron_utils.create_security_group(self.neutron, self.keystone,
703 free_rules = neutron_utils.get_rules_by_security_group(
704 self.neutron, self.security_groups[0])
705 for free_rule in free_rules:
706 self.security_group_rules.append(free_rule)
708 self.security_group_rules.append(
709 neutron_utils.create_security_group_rule(
710 self.neutron, sec_grp_settings.rule_settings[0]))
712 # Refresh object so it is populated with the newly added rule
713 security_group = neutron_utils.get_security_group(
714 self.neutron, sec_grp_settings.name)
716 rules = neutron_utils.get_rules_by_security_group(self.neutron,
720 validation_utils.objects_equivalent(
721 self.security_group_rules, rules))
723 self.assertTrue(sec_grp_settings.name, security_group.name)
725 sec_grp_get = neutron_utils.get_security_group(self.neutron,
726 sec_grp_settings.name)
727 self.assertIsNotNone(sec_grp_get)
728 self.assertEqual(security_group, sec_grp_get)
730 def test_get_sec_grp_by_id(self):
732 Tests the neutron_utils.create_security_group() function
735 self.security_groups.append(neutron_utils.create_security_group(
736 self.neutron, self.keystone,
737 SecurityGroupSettings(name=self.sec_grp_name + '-1',
738 description='hello group')))
739 self.security_groups.append(neutron_utils.create_security_group(
740 self.neutron, self.keystone,
741 SecurityGroupSettings(name=self.sec_grp_name + '-2',
742 description='hello group')))
744 sec_grp_1b = neutron_utils.get_security_group_by_id(
745 self.neutron, self.security_groups[0].id)
746 sec_grp_2b = neutron_utils.get_security_group_by_id(
747 self.neutron, self.security_groups[1].id)
749 self.assertEqual(self.security_groups[0].id, sec_grp_1b.id)
750 self.assertEqual(self.security_groups[1].id, sec_grp_2b.id)
753 class NeutronUtilsFloatingIpTests(OSComponentTestCase):
755 Test basic nova keypair functionality
760 Instantiates the CreateImage object that is responsible for downloading
761 and creating an OS image file within OpenStack
763 self.neutron = neutron_utils.neutron_client(self.os_creds)
764 self.floating_ip = None
768 Cleans the image and downloaded image file
771 neutron_utils.delete_floating_ip(self.neutron, self.floating_ip)
773 def test_floating_ips(self):
775 Tests the creation of a floating IP
778 initial_fips = neutron_utils.get_floating_ips(self.neutron)
780 self.floating_ip = neutron_utils.create_floating_ip(self.neutron,
782 all_fips = neutron_utils.get_floating_ips(self.neutron)
783 self.assertEqual(len(initial_fips) + 1, len(all_fips))
784 returned = neutron_utils.get_floating_ip(self.neutron,
786 self.assertEqual(self.floating_ip.id, returned.id)
787 self.assertEqual(self.floating_ip.ip, returned.ip)
795 def validate_network(neutron, name, exists):
797 Returns true if a network for a given name DOES NOT exist if the exists
798 parameter is false conversely true. Returns false if a network for a given
799 name DOES exist if the exists parameter is true conversely false.
800 :param neutron: The neutron client
801 :param name: The expected network name
802 :param exists: Whether or not the network name should exist or not
805 network = neutron_utils.get_network(neutron, name)
806 if exists and network:
808 if not exists and not network:
813 def validate_subnet(neutron, name, cidr, exists):
815 Returns true if a subnet for a given name DOES NOT exist if the exists
816 parameter is false conversely true. Returns false if a subnet for a given
817 name DOES exist if the exists parameter is true conversely false.
818 :param neutron: The neutron client
819 :param name: The expected subnet name
820 :param cidr: The expected CIDR value
821 :param exists: Whether or not the network name should exist or not
824 subnet = neutron_utils.get_subnet_by_name(neutron, name)
825 if exists and subnet:
826 return subnet.cidr == cidr
827 if not exists and not subnet:
832 def validate_router(neutron, name, exists):
834 Returns true if a router for a given name DOES NOT exist if the exists
835 parameter is false conversely true. Returns false if a router for a given
836 name DOES exist if the exists parameter is true conversely false.
837 :param neutron: The neutron client
838 :param name: The expected router name
839 :param exists: Whether or not the network name should exist or not
842 router = neutron_utils.get_router_by_name(neutron, name)
843 if exists and router:
848 def validate_interface_router(interface_router, router, subnet):
850 Returns true if the router ID & subnet ID have been properly included into
851 the interface router object
852 :param interface_router: the SNAPS-OO InterfaceRouter domain object
853 :param router: to validate against the interface_router
854 :param subnet: to validate against the interface_router
855 :return: True if both IDs match else False
857 subnet_id = interface_router.subnet_id
858 router_id = interface_router.port_id
860 return subnet.id == subnet_id and router.id == router_id
863 def validate_port(neutron, port_obj, this_port_name):
865 Returns true if a port for a given name DOES NOT exist if the exists
866 parameter is false conversely true. Returns false if a port for a given
867 name DOES exist if the exists parameter is true conversely false.
868 :param neutron: The neutron client
869 :param port_obj: The port object to lookup
870 :param this_port_name: The expected router name
873 os_ports = neutron.list_ports()
874 for os_port, os_port_insts in os_ports.items():
875 for os_inst in os_port_insts:
876 if os_inst['id'] == port_obj.id:
877 return os_inst['name'] == this_port_name