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,
322 ext_net = neutron_utils.get_network(self.neutron, self.ext_net_name)
324 self.router.external_gateway_info['network_id'], ext_net.id)
326 def test_create_router_empty_name(self):
328 Tests the neutron_utils.create_neutron_net() function
330 with self.assertRaises(Exception):
331 this_router_settings = create_router.RouterSettings(name='')
332 self.router = neutron_utils.create_router(self.neutron,
334 this_router_settings)
336 def test_create_router_null_name(self):
338 Tests the neutron_utils.create_neutron_subnet() function when the
339 subnet CIDR value is None
341 with self.assertRaises(Exception):
342 this_router_settings = create_router.RouterSettings()
343 self.router = neutron_utils.create_router(self.neutron,
345 this_router_settings)
346 validate_router(self.neutron, None, True)
348 def test_add_interface_router(self):
350 Tests the neutron_utils.add_interface_router() function
352 self.network = neutron_utils.create_network(
353 self.neutron, self.os_creds, self.net_config.network_settings)
354 self.assertEqual(self.net_config.network_settings.name,
356 self.assertTrue(validate_network(
357 self.neutron, self.net_config.network_settings.name, True))
359 subnet_setting = self.net_config.network_settings.subnet_settings[0]
360 self.subnet = neutron_utils.create_subnet(
361 self.neutron, subnet_setting,
362 self.os_creds, self.network)
366 subnet_setting.cidr, True)
368 self.router = neutron_utils.create_router(
369 self.neutron, self.os_creds, self.net_config.router_settings)
370 validate_router(self.neutron, self.net_config.router_settings.name,
373 self.interface_router = neutron_utils.add_interface_router(
374 self.neutron, self.router, self.subnet)
375 validate_interface_router(self.interface_router, self.router,
378 def test_add_interface_router_null_router(self):
380 Tests the neutron_utils.add_interface_router() function for an
381 Exception when the router value is None
383 self.network = neutron_utils.create_network(
384 self.neutron, self.os_creds, self.net_config.network_settings)
385 self.assertEqual(self.net_config.network_settings.name,
387 self.assertTrue(validate_network(
388 self.neutron, self.net_config.network_settings.name, True))
390 subnet_setting = self.net_config.network_settings.subnet_settings[0]
391 self.subnet = neutron_utils.create_subnet(
392 self.neutron, subnet_setting,
393 self.os_creds, self.network)
395 self.neutron, subnet_setting.name, subnet_setting.cidr, True)
397 with self.assertRaises(NeutronException):
398 self.interface_router = neutron_utils.add_interface_router(
399 self.neutron, self.router, self.subnet)
401 def test_add_interface_router_null_subnet(self):
403 Tests the neutron_utils.add_interface_router() function for an
404 Exception when the subnet value is None
406 self.network = neutron_utils.create_network(
407 self.neutron, self.os_creds, self.net_config.network_settings)
408 self.assertEqual(self.net_config.network_settings.name,
410 self.assertTrue(validate_network(
411 self.neutron, self.net_config.network_settings.name, True))
413 self.router = neutron_utils.create_router(
414 self.neutron, self.os_creds, self.net_config.router_settings)
415 validate_router(self.neutron, self.net_config.router_settings.name,
418 with self.assertRaises(NeutronException):
419 self.interface_router = neutron_utils.add_interface_router(
420 self.neutron, self.router, self.subnet)
422 def test_create_port(self):
424 Tests the neutron_utils.create_port() function
426 self.network = neutron_utils.create_network(
427 self.neutron, self.os_creds, self.net_config.network_settings)
428 self.assertEqual(self.net_config.network_settings.name,
430 self.assertTrue(validate_network(
431 self.neutron, self.net_config.network_settings.name, True))
433 subnet_setting = self.net_config.network_settings.subnet_settings[0]
434 self.subnet = neutron_utils.create_subnet(
435 self.neutron, subnet_setting, self.os_creds, self.network)
436 validate_subnet(self.neutron, subnet_setting.name,
437 subnet_setting.cidr, True)
439 self.port = neutron_utils.create_port(
440 self.neutron, self.os_creds, PortSettings(
443 'subnet_name': subnet_setting.name,
445 network_name=self.net_config.network_settings.name))
446 validate_port(self.neutron, self.port, self.port_name)
448 def test_create_port_empty_name(self):
450 Tests the neutron_utils.create_port() function
452 self.network = neutron_utils.create_network(
453 self.neutron, self.os_creds, self.net_config.network_settings)
454 self.assertEqual(self.net_config.network_settings.name,
456 self.assertTrue(validate_network(
457 self.neutron, self.net_config.network_settings.name, True))
459 subnet_setting = self.net_config.network_settings.subnet_settings[0]
460 self.subnet = neutron_utils.create_subnet(
461 self.neutron, subnet_setting, self.os_creds, self.network)
462 validate_subnet(self.neutron, subnet_setting.name, subnet_setting.cidr,
465 self.port = neutron_utils.create_port(
466 self.neutron, self.os_creds, PortSettings(
468 network_name=self.net_config.network_settings.name,
470 'subnet_name': subnet_setting.name,
472 validate_port(self.neutron, self.port, self.port_name)
474 def test_create_port_null_name(self):
476 Tests the neutron_utils.create_port() function for an Exception when
477 the port name value is None
479 self.network = neutron_utils.create_network(
480 self.neutron, self.os_creds, self.net_config.network_settings)
481 self.assertEqual(self.net_config.network_settings.name,
483 self.assertTrue(validate_network(
484 self.neutron, self.net_config.network_settings.name, True))
486 subnet_setting = self.net_config.network_settings.subnet_settings[0]
487 self.subnet = neutron_utils.create_subnet(
488 self.neutron, subnet_setting,
489 self.os_creds, self.network)
493 subnet_setting.cidr, True)
495 with self.assertRaises(Exception):
496 self.port = neutron_utils.create_port(
497 self.neutron, self.os_creds,
499 network_name=self.net_config.network_settings.name,
501 'subnet_name': subnet_setting.name,
504 def test_create_port_null_network_object(self):
506 Tests the neutron_utils.create_port() function for an Exception when
507 the network object is None
509 with self.assertRaises(Exception):
510 self.port = neutron_utils.create_port(
511 self.neutron, self.os_creds,
514 network_name=self.net_config.network_settings.name,
517 self.net_config.network_settings.subnet_settings[
521 def test_create_port_null_ip(self):
523 Tests the neutron_utils.create_port() function for an Exception when
526 self.network = neutron_utils.create_network(
527 self.neutron, self.os_creds, self.net_config.network_settings)
528 self.assertEqual(self.net_config.network_settings.name,
530 self.assertTrue(validate_network(
531 self.neutron, self.net_config.network_settings.name, True))
533 subnet_setting = self.net_config.network_settings.subnet_settings[0]
534 self.subnet = neutron_utils.create_subnet(
535 self.neutron, subnet_setting,
536 self.os_creds, self.network)
540 subnet_setting.cidr, True)
542 with self.assertRaises(Exception):
543 self.port = neutron_utils.create_port(
544 self.neutron, self.os_creds,
547 network_name=self.net_config.network_settings.name,
549 'subnet_name': subnet_setting.name,
552 def test_create_port_invalid_ip(self):
554 Tests the neutron_utils.create_port() function for an Exception when
557 self.network = neutron_utils.create_network(
558 self.neutron, self.os_creds, self.net_config.network_settings)
559 self.assertEqual(self.net_config.network_settings.name,
561 self.assertTrue(validate_network(
562 self.neutron, self.net_config.network_settings.name, True))
564 subnet_setting = self.net_config.network_settings.subnet_settings[0]
565 self.subnet = neutron_utils.create_subnet(
566 self.neutron, subnet_setting, self.os_creds, self.network)
567 validate_subnet(self.neutron,
569 subnet_setting.cidr, True)
571 with self.assertRaises(Exception):
572 self.port = neutron_utils.create_port(
573 self.neutron, self.os_creds,
576 network_name=self.net_config.network_settings.name,
578 'subnet_name': subnet_setting.name,
581 def test_create_port_invalid_ip_to_subnet(self):
583 Tests the neutron_utils.create_port() function for an Exception when
586 self.network = neutron_utils.create_network(
587 self.neutron, self.os_creds, self.net_config.network_settings)
588 self.assertEqual(self.net_config.network_settings.name,
590 self.assertTrue(validate_network(
591 self.neutron, self.net_config.network_settings.name, True))
593 subnet_setting = self.net_config.network_settings.subnet_settings[0]
594 self.subnet = neutron_utils.create_subnet(
595 self.neutron, subnet_setting, self.os_creds, self.network)
597 self.neutron, subnet_setting.name, subnet_setting.cidr, True)
599 with self.assertRaises(Exception):
600 self.port = neutron_utils.create_port(
601 self.neutron, self.os_creds,
604 network_name=self.net_config.network_settings.name,
606 'subnet_name': subnet_setting.name,
607 'ip': '10.197.123.100'}]))
610 class NeutronUtilsSecurityGroupTests(OSComponentTestCase):
612 Test for creating security groups via neutron_utils.py
616 guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
617 self.sec_grp_name = guid + 'name'
619 self.security_groups = list()
620 self.security_group_rules = list()
621 self.neutron = neutron_utils.neutron_client(self.os_creds)
622 self.keystone = keystone_utils.keystone_client(self.os_creds)
626 Cleans the remote OpenStack objects
628 for rule in self.security_group_rules:
629 neutron_utils.delete_security_group_rule(self.neutron, rule)
631 for security_group in self.security_groups:
633 neutron_utils.delete_security_group(self.neutron,
638 def test_create_delete_simple_sec_grp(self):
640 Tests the neutron_utils.create_security_group() function
642 sec_grp_settings = SecurityGroupSettings(name=self.sec_grp_name)
643 security_group = neutron_utils.create_security_group(self.neutron,
647 self.assertTrue(sec_grp_settings.name, security_group.name)
649 sec_grp_get = neutron_utils.get_security_group(self.neutron,
650 sec_grp_settings.name)
651 self.assertIsNotNone(sec_grp_get)
652 self.assertTrue(validation_utils.objects_equivalent(
653 security_group, sec_grp_get))
655 neutron_utils.delete_security_group(self.neutron, security_group)
656 sec_grp_get = neutron_utils.get_security_group(self.neutron,
657 sec_grp_settings.name)
658 self.assertIsNone(sec_grp_get)
660 def test_create_sec_grp_no_name(self):
662 Tests the SecurityGroupSettings constructor and
663 neutron_utils.create_security_group() function to ensure that
664 attempting to create a security group without a name will raise an
667 with self.assertRaises(Exception):
668 sec_grp_settings = SecurityGroupSettings()
669 self.security_groups.append(
670 neutron_utils.create_security_group(self.neutron,
674 def test_create_sec_grp_no_rules(self):
676 Tests the neutron_utils.create_security_group() function
678 sec_grp_settings = SecurityGroupSettings(name=self.sec_grp_name,
679 description='hello group')
680 self.security_groups.append(
681 neutron_utils.create_security_group(self.neutron, self.keystone,
684 self.assertTrue(sec_grp_settings.name, self.security_groups[0].name)
685 self.assertEqual(sec_grp_settings.name, self.security_groups[0].name)
687 sec_grp_get = neutron_utils.get_security_group(self.neutron,
688 sec_grp_settings.name)
689 self.assertIsNotNone(sec_grp_get)
690 self.assertEqual(self.security_groups[0], sec_grp_get)
692 def test_create_sec_grp_one_rule(self):
694 Tests the neutron_utils.create_security_group() function
697 sec_grp_rule_settings = SecurityGroupRuleSettings(
698 sec_grp_name=self.sec_grp_name, direction=Direction.ingress)
699 sec_grp_settings = SecurityGroupSettings(
700 name=self.sec_grp_name, description='hello group',
701 rule_settings=[sec_grp_rule_settings])
703 self.security_groups.append(
704 neutron_utils.create_security_group(self.neutron, self.keystone,
706 free_rules = neutron_utils.get_rules_by_security_group(
707 self.neutron, self.security_groups[0])
708 for free_rule in free_rules:
709 self.security_group_rules.append(free_rule)
711 self.security_group_rules.append(
712 neutron_utils.create_security_group_rule(
713 self.neutron, sec_grp_settings.rule_settings[0]))
715 # Refresh object so it is populated with the newly added rule
716 security_group = neutron_utils.get_security_group(
717 self.neutron, sec_grp_settings.name)
719 rules = neutron_utils.get_rules_by_security_group(self.neutron,
723 validation_utils.objects_equivalent(
724 self.security_group_rules, rules))
726 self.assertTrue(sec_grp_settings.name, security_group.name)
728 sec_grp_get = neutron_utils.get_security_group(self.neutron,
729 sec_grp_settings.name)
730 self.assertIsNotNone(sec_grp_get)
731 self.assertEqual(security_group, sec_grp_get)
733 def test_get_sec_grp_by_id(self):
735 Tests the neutron_utils.create_security_group() function
738 self.security_groups.append(neutron_utils.create_security_group(
739 self.neutron, self.keystone,
740 SecurityGroupSettings(name=self.sec_grp_name + '-1',
741 description='hello group')))
742 self.security_groups.append(neutron_utils.create_security_group(
743 self.neutron, self.keystone,
744 SecurityGroupSettings(name=self.sec_grp_name + '-2',
745 description='hello group')))
747 sec_grp_1b = neutron_utils.get_security_group_by_id(
748 self.neutron, self.security_groups[0].id)
749 sec_grp_2b = neutron_utils.get_security_group_by_id(
750 self.neutron, self.security_groups[1].id)
752 self.assertEqual(self.security_groups[0].id, sec_grp_1b.id)
753 self.assertEqual(self.security_groups[1].id, sec_grp_2b.id)
756 class NeutronUtilsFloatingIpTests(OSComponentTestCase):
758 Test basic nova keypair functionality
763 Instantiates the CreateImage object that is responsible for downloading
764 and creating an OS image file within OpenStack
766 self.neutron = neutron_utils.neutron_client(self.os_creds)
767 self.floating_ip = None
771 Cleans the image and downloaded image file
774 neutron_utils.delete_floating_ip(self.neutron, self.floating_ip)
776 def test_floating_ips(self):
778 Tests the creation of a floating IP
781 initial_fips = neutron_utils.get_floating_ips(self.neutron)
783 self.floating_ip = neutron_utils.create_floating_ip(self.neutron,
785 all_fips = neutron_utils.get_floating_ips(self.neutron)
786 self.assertEqual(len(initial_fips) + 1, len(all_fips))
787 returned = neutron_utils.get_floating_ip(self.neutron,
789 self.assertEqual(self.floating_ip.id, returned.id)
790 self.assertEqual(self.floating_ip.ip, returned.ip)
798 def validate_network(neutron, name, exists):
800 Returns true if a network for a given name DOES NOT exist if the exists
801 parameter is false conversely true. Returns false if a network for a given
802 name DOES exist if the exists parameter is true conversely false.
803 :param neutron: The neutron client
804 :param name: The expected network name
805 :param exists: Whether or not the network name should exist or not
808 network = neutron_utils.get_network(neutron, name)
809 if exists and network:
811 if not exists and not network:
816 def validate_subnet(neutron, name, cidr, exists):
818 Returns true if a subnet for a given name DOES NOT exist if the exists
819 parameter is false conversely true. Returns false if a subnet for a given
820 name DOES exist if the exists parameter is true conversely false.
821 :param neutron: The neutron client
822 :param name: The expected subnet name
823 :param cidr: The expected CIDR value
824 :param exists: Whether or not the network name should exist or not
827 subnet = neutron_utils.get_subnet_by_name(neutron, name)
828 if exists and subnet:
829 return subnet.cidr == cidr
830 if not exists and not subnet:
835 def validate_router(neutron, name, exists):
837 Returns true if a router for a given name DOES NOT exist if the exists
838 parameter is false conversely true. Returns false if a router for a given
839 name DOES exist if the exists parameter is true conversely false.
840 :param neutron: The neutron client
841 :param name: The expected router name
842 :param exists: Whether or not the network name should exist or not
845 router = neutron_utils.get_router_by_name(neutron, name)
846 if exists and router:
851 def validate_interface_router(interface_router, router, subnet):
853 Returns true if the router ID & subnet ID have been properly included into
854 the interface router object
855 :param interface_router: the SNAPS-OO InterfaceRouter domain object
856 :param router: to validate against the interface_router
857 :param subnet: to validate against the interface_router
858 :return: True if both IDs match else False
860 subnet_id = interface_router.subnet_id
861 router_id = interface_router.port_id
863 return subnet.id == subnet_id and router.id == router_id
866 def validate_port(neutron, port_obj, this_port_name):
868 Returns true if a port for a given name DOES NOT exist if the exists
869 parameter is false conversely true. Returns false if a port for a given
870 name DOES exist if the exists parameter is true conversely false.
871 :param neutron: The neutron client
872 :param port_obj: The port object to lookup
873 :param this_port_name: The expected router name
876 os_ports = neutron.list_ports()
877 for os_port, os_port_insts in os_ports.items():
878 for os_inst in os_port_insts:
879 if os_inst['id'] == port_obj.id:
880 return os_inst['name'] == this_port_name