X-Git-Url: https://gerrit.opnfv.org/gerrit/gitweb?a=blobdiff_plain;f=snaps%2Fopenstack%2Futils%2Ftests%2Fneutron_utils_tests.py;h=05d508d0e1c4f342bb44a13f8cb88a5cf2ffe77f;hb=1342eb17df248ec75cc57e9c380a7753fc432194;hp=f3cacd51e853da046e67bed06ee8d52d7911da95;hpb=b490e8dc9fb01c6f9c44dd9a585ca1a1ae00bf19;p=snaps.git diff --git a/snaps/openstack/utils/tests/neutron_utils_tests.py b/snaps/openstack/utils/tests/neutron_utils_tests.py index f3cacd5..05d508d 100644 --- a/snaps/openstack/utils/tests/neutron_utils_tests.py +++ b/snaps/openstack/utils/tests/neutron_utils_tests.py @@ -24,6 +24,7 @@ from snaps.openstack.tests import validation_utils from snaps.openstack.tests.os_source_file_test import OSComponentTestCase from snaps.openstack.utils import keystone_utils from snaps.openstack.utils import neutron_utils +from snaps.openstack.utils.neutron_utils import NeutronException __author__ = 'spisarski' @@ -73,7 +74,7 @@ class NeutronSmokeTests(OSComponentTestCase): ext_networks = neutron_utils.get_external_networks(neutron) found = False for network in ext_networks: - if network['network']['name'] == self.ext_net_name: + if network.name == self.ext_net_name: found = True break self.assertTrue(found) @@ -98,8 +99,6 @@ class NeutronUtilsNetworkTests(OSComponentTestCase): """ if self.network: neutron_utils.delete_network(self.neutron, self.network) - validate_network(self.neutron, self.network['network']['name'], - False) def test_create_network(self): """ @@ -108,10 +107,9 @@ class NeutronUtilsNetworkTests(OSComponentTestCase): self.network = neutron_utils.create_network( self.neutron, self.os_creds, self.net_config.network_settings) self.assertEqual(self.net_config.network_settings.name, - self.network['network']['name']) - self.assertTrue(validate_network(self.neutron, - self.net_config.network_settings.name, - True)) + self.network.name) + self.assertTrue(validate_network( + self.neutron, self.net_config.network_settings.name, True)) def test_create_network_empty_name(self): """ @@ -154,15 +152,15 @@ class NeutronUtilsSubnetTests(OSComponentTestCase): Cleans the remote OpenStack objects """ if self.subnet: - neutron_utils.delete_subnet(self.neutron, self.subnet) - validate_subnet(self.neutron, self.subnet.get('name'), - self.net_config.network_settings.subnet_settings[ - 0].cidr, False) - + try: + neutron_utils.delete_subnet(self.neutron, self.subnet) + except: + pass if self.network: - neutron_utils.delete_network(self.neutron, self.network) - validate_network(self.neutron, self.network['network']['name'], - False) + try: + neutron_utils.delete_network(self.neutron, self.network) + except: + pass def test_create_subnet(self): """ @@ -171,19 +169,25 @@ class NeutronUtilsSubnetTests(OSComponentTestCase): self.network = neutron_utils.create_network( self.neutron, self.os_creds, self.net_config.network_settings) self.assertEqual(self.net_config.network_settings.name, - self.network['network']['name']) - self.assertTrue(validate_network(self.neutron, - self.net_config.network_settings.name, - True)) + self.network.name) + self.assertTrue(validate_network( + self.neutron, self.net_config.network_settings.name, True)) subnet_setting = self.net_config.network_settings.subnet_settings[0] self.subnet = neutron_utils.create_subnet( - self.neutron, subnet_setting, - self.os_creds, network=self.network) - validate_subnet( - self.neutron, - subnet_setting.name, - subnet_setting.cidr, True) + self.neutron, subnet_setting, self.os_creds, network=self.network) + self.assertTrue(validate_subnet( + self.neutron, subnet_setting.name, subnet_setting.cidr, True)) + + subnet_query1 = neutron_utils.get_subnet( + self.neutron, subnet_name=subnet_setting.name) + self.assertEqual(self.subnet, subnet_query1) + + subnet_query2 = neutron_utils.get_subnets_by_network(self.neutron, + self.network) + self.assertIsNotNone(subnet_query2) + self.assertEqual(1, len(subnet_query2)) + self.assertEqual(self.subnet, subnet_query2[0]) def test_create_subnet_null_name(self): """ @@ -193,10 +197,9 @@ class NeutronUtilsSubnetTests(OSComponentTestCase): self.network = neutron_utils.create_network( self.neutron, self.os_creds, self.net_config.network_settings) self.assertEqual(self.net_config.network_settings.name, - self.network['network']['name']) - self.assertTrue(validate_network(self.neutron, - self.net_config.network_settings.name, - True)) + self.network.name) + self.assertTrue(validate_network( + self.neutron, self.net_config.network_settings.name, True)) with self.assertRaises(Exception): SubnetSettings(cidr=self.net_config.subnet_cidr) @@ -209,17 +212,27 @@ class NeutronUtilsSubnetTests(OSComponentTestCase): self.network = neutron_utils.create_network( self.neutron, self.os_creds, self.net_config.network_settings) self.assertEqual(self.net_config.network_settings.name, - self.network['network']['name']) - self.assertTrue(validate_network(self.neutron, - self.net_config.network_settings.name, - True)) + self.network.name) + self.assertTrue(validate_network( + self.neutron, self.net_config.network_settings.name, True)) subnet_setting = self.net_config.network_settings.subnet_settings[0] - neutron_utils.create_subnet( - self.neutron, subnet_setting, - self.os_creds, network=self.network) - validate_subnet(self.neutron, '', - subnet_setting.cidr, True) + self.subnet = neutron_utils.create_subnet( + self.neutron, subnet_setting, self.os_creds, network=self.network) + self.assertTrue(validate_subnet( + self.neutron, subnet_setting.name, subnet_setting.cidr, True)) + self.assertFalse(validate_subnet( + self.neutron, '', subnet_setting.cidr, True)) + + subnet_query1 = neutron_utils.get_subnet( + self.neutron, subnet_name=subnet_setting.name) + self.assertEqual(self.subnet, subnet_query1) + + subnet_query2 = neutron_utils.get_subnets_by_network(self.neutron, + self.network) + self.assertIsNotNone(subnet_query2) + self.assertEqual(1, len(subnet_query2)) + self.assertEqual(self.subnet, subnet_query2[0]) def test_create_subnet_null_cidr(self): """ @@ -229,16 +242,15 @@ class NeutronUtilsSubnetTests(OSComponentTestCase): self.network = neutron_utils.create_network( self.neutron, self.os_creds, self.net_config.network_settings) self.assertEqual(self.net_config.network_settings.name, - self.network['network']['name']) - self.assertTrue(validate_network(self.neutron, - self.net_config.network_settings.name, - True)) + self.network.name) + self.assertTrue(validate_network( + self.neutron, self.net_config.network_settings.name, True)) with self.assertRaises(Exception): - sub_sets = SubnetSettings(cidr=None, - name=self.net_config.subnet_name) - neutron_utils.create_subnet(self.neutron, sub_sets, self.os_creds, - network=self.network) + sub_sets = SubnetSettings( + cidr=None, name=self.net_config.subnet_name) + neutron_utils.create_subnet( + self.neutron, sub_sets, self.os_creds, network=self.network) def test_create_subnet_empty_cidr(self): """ @@ -248,14 +260,13 @@ class NeutronUtilsSubnetTests(OSComponentTestCase): self.network = neutron_utils.create_network( self.neutron, self.os_creds, self.net_config.network_settings) self.assertEqual(self.net_config.network_settings.name, - self.network['network']['name']) - self.assertTrue(validate_network(self.neutron, - self.net_config.network_settings.name, - True)) + self.network.name) + self.assertTrue(validate_network( + self.neutron, self.net_config.network_settings.name, True)) with self.assertRaises(Exception): - sub_sets = SubnetSettings(cidr='', - name=self.net_config.subnet_name) + sub_sets = SubnetSettings( + cidr='', name=self.net_config.subnet_name) neutron_utils.create_subnet(self.neutron, sub_sets, self.os_creds, network=self.network) @@ -287,22 +298,29 @@ class NeutronUtilsRouterTests(OSComponentTestCase): self.subnet) if self.router: - neutron_utils.delete_router(self.neutron, self.router) - validate_router(self.neutron, self.router.name, False) + try: + neutron_utils.delete_router(self.neutron, self.router) + validate_router(self.neutron, self.router.name, False) + except: + pass if self.port: - neutron_utils.delete_port(self.neutron, self.port) + try: + neutron_utils.delete_port(self.neutron, self.port) + except: + pass if self.subnet: - neutron_utils.delete_subnet(self.neutron, self.subnet) - validate_subnet(self.neutron, self.subnet.get('name'), - self.net_config.network_settings.subnet_settings[ - 0].cidr, False) + try: + neutron_utils.delete_subnet(self.neutron, self.subnet) + except: + pass if self.network: - neutron_utils.delete_network(self.neutron, self.network) - validate_network(self.neutron, self.network['network']['name'], - False) + try: + neutron_utils.delete_network(self.neutron, self.network) + except: + pass def test_create_router_simple(self): """ @@ -330,7 +348,11 @@ class NeutronUtilsRouterTests(OSComponentTestCase): self.neutron, self.os_creds, self.net_config.router_settings) validate_router(self.neutron, self.net_config.router_settings.name, True) - # TODO - Add validation that the router gatway has been set + + ext_net = neutron_utils.get_network( + self.neutron, network_name=self.ext_net_name) + self.assertEqual( + self.router.external_gateway_info['network_id'], ext_net.id) def test_create_router_empty_name(self): """ @@ -361,19 +383,16 @@ class NeutronUtilsRouterTests(OSComponentTestCase): self.network = neutron_utils.create_network( self.neutron, self.os_creds, self.net_config.network_settings) self.assertEqual(self.net_config.network_settings.name, - self.network['network']['name']) - self.assertTrue(validate_network(self.neutron, - self.net_config.network_settings.name, - True)) + self.network.name) + self.assertTrue(validate_network( + self.neutron, self.net_config.network_settings.name, True)) subnet_setting = self.net_config.network_settings.subnet_settings[0] self.subnet = neutron_utils.create_subnet( self.neutron, subnet_setting, self.os_creds, self.network) - validate_subnet( - self.neutron, - subnet_setting.name, - subnet_setting.cidr, True) + self.assertTrue(validate_subnet( + self.neutron, subnet_setting.name, subnet_setting.cidr, True)) self.router = neutron_utils.create_router( self.neutron, self.os_creds, self.net_config.router_settings) @@ -393,21 +412,18 @@ class NeutronUtilsRouterTests(OSComponentTestCase): self.network = neutron_utils.create_network( self.neutron, self.os_creds, self.net_config.network_settings) self.assertEqual(self.net_config.network_settings.name, - self.network['network']['name']) - self.assertTrue(validate_network(self.neutron, - self.net_config.network_settings.name, - True)) + self.network.name) + self.assertTrue(validate_network( + self.neutron, self.net_config.network_settings.name, True)) subnet_setting = self.net_config.network_settings.subnet_settings[0] self.subnet = neutron_utils.create_subnet( self.neutron, subnet_setting, self.os_creds, self.network) - validate_subnet( - self.neutron, - subnet_setting.name, - subnet_setting.cidr, True) + self.assertTrue(validate_subnet( + self.neutron, subnet_setting.name, subnet_setting.cidr, True)) - with self.assertRaises(Exception): + with self.assertRaises(NeutronException): self.interface_router = neutron_utils.add_interface_router( self.neutron, self.router, self.subnet) @@ -419,17 +435,16 @@ class NeutronUtilsRouterTests(OSComponentTestCase): self.network = neutron_utils.create_network( self.neutron, self.os_creds, self.net_config.network_settings) self.assertEqual(self.net_config.network_settings.name, - self.network['network']['name']) - self.assertTrue(validate_network(self.neutron, - self.net_config.network_settings.name, - True)) + self.network.name) + self.assertTrue(validate_network( + self.neutron, self.net_config.network_settings.name, True)) self.router = neutron_utils.create_router( self.neutron, self.os_creds, self.net_config.router_settings) validate_router(self.neutron, self.net_config.router_settings.name, True) - with self.assertRaises(Exception): + with self.assertRaises(NeutronException): self.interface_router = neutron_utils.add_interface_router( self.neutron, self.router, self.subnet) @@ -440,16 +455,15 @@ class NeutronUtilsRouterTests(OSComponentTestCase): self.network = neutron_utils.create_network( self.neutron, self.os_creds, self.net_config.network_settings) self.assertEqual(self.net_config.network_settings.name, - self.network['network']['name']) - self.assertTrue(validate_network(self.neutron, - self.net_config.network_settings.name, - True)) + self.network.name) + self.assertTrue(validate_network( + self.neutron, self.net_config.network_settings.name, True)) subnet_setting = self.net_config.network_settings.subnet_settings[0] self.subnet = neutron_utils.create_subnet( self.neutron, subnet_setting, self.os_creds, self.network) - validate_subnet(self.neutron, subnet_setting.name, - subnet_setting.cidr, True) + self.assertTrue(validate_subnet( + self.neutron, subnet_setting.name, subnet_setting.cidr, True)) self.port = neutron_utils.create_port( self.neutron, self.os_creds, PortSettings( @@ -467,16 +481,15 @@ class NeutronUtilsRouterTests(OSComponentTestCase): self.network = neutron_utils.create_network( self.neutron, self.os_creds, self.net_config.network_settings) self.assertEqual(self.net_config.network_settings.name, - self.network['network']['name']) - self.assertTrue(validate_network(self.neutron, - self.net_config.network_settings.name, - True)) + self.network.name) + self.assertTrue(validate_network( + self.neutron, self.net_config.network_settings.name, True)) subnet_setting = self.net_config.network_settings.subnet_settings[0] self.subnet = neutron_utils.create_subnet( self.neutron, subnet_setting, self.os_creds, self.network) - validate_subnet(self.neutron, subnet_setting.name, subnet_setting.cidr, - True) + self.assertTrue(validate_subnet(self.neutron, subnet_setting.name, + subnet_setting.cidr, True)) self.port = neutron_utils.create_port( self.neutron, self.os_creds, PortSettings( @@ -489,34 +502,32 @@ class NeutronUtilsRouterTests(OSComponentTestCase): def test_create_port_null_name(self): """ - Tests the neutron_utils.create_port() function for an Exception when - the port name value is None + Tests the neutron_utils.create_port() when the port name value is None """ self.network = neutron_utils.create_network( self.neutron, self.os_creds, self.net_config.network_settings) self.assertEqual(self.net_config.network_settings.name, - self.network['network']['name']) - self.assertTrue(validate_network(self.neutron, - self.net_config.network_settings.name, - True)) + self.network.name) + self.assertTrue(validate_network( + self.neutron, self.net_config.network_settings.name, True)) subnet_setting = self.net_config.network_settings.subnet_settings[0] self.subnet = neutron_utils.create_subnet( self.neutron, subnet_setting, self.os_creds, self.network) - validate_subnet( - self.neutron, - subnet_setting.name, - subnet_setting.cidr, True) + self.assertTrue(validate_subnet( + self.neutron, subnet_setting.name, subnet_setting.cidr, True)) - with self.assertRaises(Exception): - self.port = neutron_utils.create_port( - self.neutron, self.os_creds, - PortSettings( - network_name=self.net_config.network_settings.name, - ip_addrs=[{ - 'subnet_name': subnet_setting.name, - 'ip': ip_1}])) + self.port = neutron_utils.create_port( + self.neutron, self.os_creds, + PortSettings( + network_name=self.net_config.network_settings.name, + ip_addrs=[{ + 'subnet_name': subnet_setting.name, + 'ip': ip_1}])) + + port = neutron_utils.get_port_by_id(self.neutron, self.port.id) + self.assertEqual(self.port, port) def test_create_port_null_network_object(self): """ @@ -543,19 +554,16 @@ class NeutronUtilsRouterTests(OSComponentTestCase): self.network = neutron_utils.create_network( self.neutron, self.os_creds, self.net_config.network_settings) self.assertEqual(self.net_config.network_settings.name, - self.network['network']['name']) - self.assertTrue(validate_network(self.neutron, - self.net_config.network_settings.name, - True)) + self.network.name) + self.assertTrue(validate_network( + self.neutron, self.net_config.network_settings.name, True)) subnet_setting = self.net_config.network_settings.subnet_settings[0] self.subnet = neutron_utils.create_subnet( self.neutron, subnet_setting, self.os_creds, self.network) - validate_subnet( - self.neutron, - subnet_setting.name, - subnet_setting.cidr, True) + self.assertTrue(validate_subnet( + self.neutron, subnet_setting.name, subnet_setting.cidr, True)) with self.assertRaises(Exception): self.port = neutron_utils.create_port( @@ -575,19 +583,15 @@ class NeutronUtilsRouterTests(OSComponentTestCase): self.network = neutron_utils.create_network( self.neutron, self.os_creds, self.net_config.network_settings) self.assertEqual(self.net_config.network_settings.name, - self.network['network']['name']) - self.assertTrue(validate_network(self.neutron, - self.net_config.network_settings.name, - True)) + self.network.name) + self.assertTrue(validate_network( + self.neutron, self.net_config.network_settings.name, True)) subnet_setting = self.net_config.network_settings.subnet_settings[0] self.subnet = neutron_utils.create_subnet( - self.neutron, - subnet_setting, - self.os_creds, self.network) - validate_subnet(self.neutron, - subnet_setting.name, - subnet_setting.cidr, True) + self.neutron, subnet_setting, self.os_creds, self.network) + self.assertTrue(validate_subnet( + self.neutron, subnet_setting.name, subnet_setting.cidr, True)) with self.assertRaises(Exception): self.port = neutron_utils.create_port( @@ -607,19 +611,15 @@ class NeutronUtilsRouterTests(OSComponentTestCase): self.network = neutron_utils.create_network( self.neutron, self.os_creds, self.net_config.network_settings) self.assertEqual(self.net_config.network_settings.name, - self.network['network']['name']) - self.assertTrue(validate_network(self.neutron, - self.net_config.network_settings.name, - True)) + self.network.name) + self.assertTrue(validate_network( + self.neutron, self.net_config.network_settings.name, True)) subnet_setting = self.net_config.network_settings.subnet_settings[0] self.subnet = neutron_utils.create_subnet( - self.neutron, - subnet_setting, - self.os_creds, self.network) - validate_subnet(self.neutron, - subnet_setting.name, - subnet_setting.cidr, True) + self.neutron, subnet_setting, self.os_creds, self.network) + self.assertTrue(validate_subnet( + self.neutron, subnet_setting.name, subnet_setting.cidr, True)) with self.assertRaises(Exception): self.port = neutron_utils.create_port( @@ -671,15 +671,15 @@ class NeutronUtilsSecurityGroupTests(OSComponentTestCase): self.assertTrue(sec_grp_settings.name, security_group.name) - sec_grp_get = neutron_utils.get_security_group(self.neutron, - sec_grp_settings.name) + sec_grp_get = neutron_utils.get_security_group( + self.neutron, sec_grp_settings=sec_grp_settings) self.assertIsNotNone(sec_grp_get) self.assertTrue(validation_utils.objects_equivalent( security_group, sec_grp_get)) neutron_utils.delete_security_group(self.neutron, security_group) - sec_grp_get = neutron_utils.get_security_group(self.neutron, - sec_grp_settings.name) + sec_grp_get = neutron_utils.get_security_group( + self.neutron, sec_grp_settings=sec_grp_settings) self.assertIsNone(sec_grp_get) def test_create_sec_grp_no_name(self): @@ -709,8 +709,8 @@ class NeutronUtilsSecurityGroupTests(OSComponentTestCase): self.assertTrue(sec_grp_settings.name, self.security_groups[0].name) self.assertEqual(sec_grp_settings.name, self.security_groups[0].name) - sec_grp_get = neutron_utils.get_security_group(self.neutron, - sec_grp_settings.name) + sec_grp_get = neutron_utils.get_security_group( + self.neutron, sec_grp_settings=sec_grp_settings) self.assertIsNotNone(sec_grp_get) self.assertEqual(self.security_groups[0], sec_grp_get) @@ -739,7 +739,7 @@ class NeutronUtilsSecurityGroupTests(OSComponentTestCase): # Refresh object so it is populated with the newly added rule security_group = neutron_utils.get_security_group( - self.neutron, sec_grp_settings.name) + self.neutron, sec_grp_settings=sec_grp_settings) rules = neutron_utils.get_rules_by_security_group(self.neutron, security_group) @@ -750,8 +750,8 @@ class NeutronUtilsSecurityGroupTests(OSComponentTestCase): self.assertTrue(sec_grp_settings.name, security_group.name) - sec_grp_get = neutron_utils.get_security_group(self.neutron, - sec_grp_settings.name) + sec_grp_get = neutron_utils.get_security_group( + self.neutron, sec_grp_settings=sec_grp_settings) self.assertIsNotNone(sec_grp_get) self.assertEqual(security_group, sec_grp_get) @@ -796,7 +796,11 @@ class NeutronUtilsFloatingIpTests(OSComponentTestCase): Cleans the image and downloaded image file """ if self.floating_ip: - neutron_utils.delete_floating_ip(self.neutron, self.floating_ip) + try: + neutron_utils.delete_floating_ip( + self.neutron, self.floating_ip) + except: + pass def test_floating_ips(self): """ @@ -830,7 +834,7 @@ def validate_network(neutron, name, exists): :param exists: Whether or not the network name should exist or not :return: True/False """ - network = neutron_utils.get_network(neutron, name) + network = neutron_utils.get_network(neutron, network_name=name) if exists and network: return True if not exists and not network: @@ -849,9 +853,9 @@ def validate_subnet(neutron, name, cidr, exists): :param exists: Whether or not the network name should exist or not :return: True/False """ - subnet = neutron_utils.get_subnet_by_name(neutron, name) - if exists and subnet: - return subnet.get('cidr') == cidr + subnet = neutron_utils.get_subnet(neutron, subnet_name=name) + if exists and subnet and subnet.name == name: + return subnet.cidr == cidr if not exists and not subnet: return True return False @@ -867,7 +871,7 @@ def validate_router(neutron, name, exists): :param exists: Whether or not the network name should exist or not :return: True/False """ - router = neutron_utils.get_router_by_name(neutron, name) + router = neutron_utils.get_router(neutron, router_name=name) if exists and router: return True return False @@ -885,7 +889,7 @@ def validate_interface_router(interface_router, router, subnet): subnet_id = interface_router.subnet_id router_id = interface_router.port_id - return subnet.get('id') == subnet_id and router.get('id') == router_id + return subnet.id == subnet_id and router.id == router_id def validate_port(neutron, port_obj, this_port_name): @@ -898,9 +902,9 @@ def validate_port(neutron, port_obj, this_port_name): :param this_port_name: The expected router name :return: True/False """ - ports = neutron.list_ports() - for port, port_insts in ports.items(): - for inst in port_insts: - if inst['id'] == port_obj.id: - return inst['name'] == this_port_name + os_ports = neutron.list_ports() + for os_port, os_port_insts in os_ports.items(): + for os_inst in os_port_insts: + if os_inst['id'] == port_obj.id: + return os_inst['name'] == this_port_name return False