summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--snaps/openstack/create_instance.py9
-rw-r--r--snaps/openstack/create_network.py9
-rw-r--r--snaps/openstack/create_router.py4
-rw-r--r--snaps/openstack/tests/create_network_tests.py58
-rw-r--r--snaps/openstack/utils/neutron_utils.py45
-rw-r--r--snaps/openstack/utils/tests/neutron_utils_tests.py63
6 files changed, 86 insertions, 102 deletions
diff --git a/snaps/openstack/create_instance.py b/snaps/openstack/create_instance.py
index dcf78fe..2fdff25 100644
--- a/snaps/openstack/create_instance.py
+++ b/snaps/openstack/create_instance.py
@@ -164,8 +164,9 @@ class OpenStackVmInstance:
ext_gateway = self.__ext_gateway_by_router(
floating_ip_setting.router_name)
if ext_gateway:
- subnet = neutron_utils.get_subnet_by_name(
- self.__neutron, floating_ip_setting.subnet_name)
+ subnet = neutron_utils.get_subnet(
+ self.__neutron,
+ subnet_name=floating_ip_setting.subnet_name)
floating_ip = neutron_utils.create_floating_ip(
self.__neutron, ext_gateway)
self.__floating_ip_dict[floating_ip_setting.name] = floating_ip
@@ -350,8 +351,8 @@ class OpenStackVmInstance:
port = self.get_port_by_name(port_name)
if port:
if subnet_name:
- subnet = neutron_utils.get_subnet_by_name(self.__neutron,
- subnet_name)
+ subnet = neutron_utils.get_subnet(
+ self.__neutron, subnet_name=subnet_name)
if not subnet:
logger.warning('Cannot retrieve port IP as subnet could '
'not be located with name - %s',
diff --git a/snaps/openstack/create_network.py b/snaps/openstack/create_network.py
index 4caef12..711367c 100644
--- a/snaps/openstack/create_network.py
+++ b/snaps/openstack/create_network.py
@@ -71,8 +71,8 @@ class OpenStackNetwork:
logger.debug('Creating Subnets....')
for subnet_setting in self.network_settings.subnet_settings:
- sub_inst = neutron_utils.get_subnet_by_name(
- self.__neutron, subnet_setting.name)
+ sub_inst = neutron_utils.get_subnet(
+ self.__neutron, subnet_settings=subnet_setting)
if sub_inst:
self.__subnets.append(sub_inst)
logger.debug(
@@ -463,9 +463,8 @@ class PortSettings:
self.fixed_ips = list()
for ip_addr_dict in self.ip_addrs:
- subnet = neutron_utils.get_subnet_by_name(neutron,
- ip_addr_dict[
- 'subnet_name'])
+ subnet = neutron_utils.get_subnet(
+ neutron, subnet_name=ip_addr_dict['subnet_name'])
if subnet:
self.fixed_ips.append({'ip_address': ip_addr_dict['ip'],
'subnet_id': subnet.id})
diff --git a/snaps/openstack/create_router.py b/snaps/openstack/create_router.py
index 6bbfcf9..877e064 100644
--- a/snaps/openstack/create_router.py
+++ b/snaps/openstack/create_router.py
@@ -75,8 +75,8 @@ class OpenStackRouter:
self.__neutron, self.__os_creds, self.router_settings)
for internal_subnet_name in self.router_settings.internal_subnets:
- internal_subnet = neutron_utils.get_subnet_by_name(
- self.__neutron, internal_subnet_name)
+ internal_subnet = neutron_utils.get_subnet(
+ self.__neutron, subnet_name=internal_subnet_name)
if internal_subnet:
self.__internal_subnets.append(internal_subnet)
if internal_subnet and not cleanup and not existing:
diff --git a/snaps/openstack/tests/create_network_tests.py b/snaps/openstack/tests/create_network_tests.py
index 936eda9..3a7ab4a 100644
--- a/snaps/openstack/tests/create_network_tests.py
+++ b/snaps/openstack/tests/create_network_tests.py
@@ -357,19 +357,6 @@ class CreateNetworkSuccessTests(OSIntegrationTestCase):
self.router_creator.clean()
if self.net_creator:
- if len(self.net_creator.get_subnets()) > 0:
- # Validate subnet has been deleted
- neutron_utils_tests.validate_subnet(
- self.neutron,
- self.net_creator.network_settings.subnet_settings[0].name,
- self.net_creator.network_settings.subnet_settings[0].cidr,
- False)
-
- if self.net_creator.get_network():
- # Validate network has been deleted
- neutron_utils_tests.validate_network(
- self.neutron, self.net_creator.network_settings.name,
- False)
self.net_creator.clean()
super(self.__class__, self).__clean__()
@@ -384,14 +371,14 @@ class CreateNetworkSuccessTests(OSIntegrationTestCase):
self.net_creator.create()
# Validate network was created
- neutron_utils_tests.validate_network(
- self.neutron, self.net_creator.network_settings.name, True)
+ self.assertTrue(neutron_utils_tests.validate_network(
+ self.neutron, self.net_creator.network_settings.name, True))
# Validate subnets
- neutron_utils_tests.validate_subnet(
+ self.assertTrue(neutron_utils_tests.validate_subnet(
self.neutron,
self.net_creator.network_settings.subnet_settings[0].name,
- self.net_creator.network_settings.subnet_settings[0].cidr, True)
+ self.net_creator.network_settings.subnet_settings[0].cidr, True))
def test_create_delete_network(self):
"""
@@ -403,8 +390,8 @@ class CreateNetworkSuccessTests(OSIntegrationTestCase):
self.net_creator.create()
# Validate network was created
- neutron_utils_tests.validate_network(
- self.neutron, self.net_creator.network_settings.name, True)
+ self.assertTrue(neutron_utils_tests.validate_network(
+ self.neutron, self.net_creator.network_settings.name, True))
neutron_utils.delete_network(self.neutron,
self.net_creator.get_network())
@@ -429,14 +416,14 @@ class CreateNetworkSuccessTests(OSIntegrationTestCase):
self.router_creator.create()
# Validate network was created
- neutron_utils_tests.validate_network(
- self.neutron, self.net_creator.network_settings.name, True)
+ self.assertTrue(neutron_utils_tests.validate_network(
+ self.neutron, self.net_creator.network_settings.name, True))
# Validate subnets
- neutron_utils_tests.validate_subnet(
+ self.assertTrue(neutron_utils_tests.validate_subnet(
self.neutron,
self.net_creator.network_settings.subnet_settings[0].name,
- self.net_creator.network_settings.subnet_settings[0].cidr, True)
+ self.net_creator.network_settings.subnet_settings[0].cidr, True))
# Validate routers
neutron_utils_tests.validate_router(
@@ -550,19 +537,6 @@ class CreateNetworkTypeTests(OSComponentTestCase):
Cleans the network
"""
if self.net_creator:
- if len(self.net_creator.get_subnets()) > 0:
- # Validate subnet has been deleted
- neutron_utils_tests.validate_subnet(
- self.neutron,
- self.net_creator.network_settings.subnet_settings[0].name,
- self.net_creator.network_settings.subnet_settings[0].cidr,
- False)
-
- if self.net_creator.get_network():
- # Validate network has been deleted
- neutron_utils_tests.validate_network(
- self.neutron, self.net_creator.network_settings.name,
- False)
self.net_creator.clean()
def test_create_network_type_vlan(self):
@@ -581,8 +555,8 @@ class CreateNetworkTypeTests(OSComponentTestCase):
network = self.net_creator.create()
# Validate network was created
- neutron_utils_tests.validate_network(
- self.neutron, net_settings.name, True)
+ self.assertTrue(neutron_utils_tests.validate_network(
+ self.neutron, net_settings.name, True))
self.assertEquals(network_type, network.type)
@@ -602,8 +576,8 @@ class CreateNetworkTypeTests(OSComponentTestCase):
network = self.net_creator.create()
# Validate network was created
- neutron_utils_tests.validate_network(self.neutron, net_settings.name,
- True)
+ self.assertTrue(neutron_utils_tests.validate_network(
+ self.neutron, net_settings.name, True))
self.assertEqual(network_type, network.type)
@@ -625,8 +599,8 @@ class CreateNetworkTypeTests(OSComponentTestCase):
network = self.net_creator.create()
# Validate network was created
- neutron_utils_tests.validate_network(
- self.neutron, net_settings.name, True)
+ self.assertTrue(neutron_utils_tests.validate_network(
+ self.neutron, net_settings.name, True))
self.assertEquals(network_type, network.type)
diff --git a/snaps/openstack/utils/neutron_utils.py b/snaps/openstack/utils/neutron_utils.py
index d892504..01408fe 100644
--- a/snaps/openstack/utils/neutron_utils.py
+++ b/snaps/openstack/utils/neutron_utils.py
@@ -147,19 +147,44 @@ def delete_subnet(neutron, subnet):
neutron.delete_subnet(subnet.id)
-def get_subnet_by_name(neutron, subnet_name):
+def get_subnet(neutron, subnet_settings=None, subnet_name=None):
"""
- Returns the first subnet object (dictionary) found with a given name
+ Returns the first subnet object that fits the query else None including
+ if subnet_settings or subnet_name parameters are None.
:param neutron: the client
- :param subnet_name: the name of the network to retrieve
- :return: a SNAPS-OO Subnet domain object
+ :param subnet_settings: the subnet settings of the object to retrieve
+ :param subnet_name: the name of the subnet to retrieve
+ :return: a SNAPS-OO Subnet domain object or None
"""
- subnets = neutron.list_subnets(**{'name': subnet_name})
- for subnet, subnetInst in subnets.items():
- for inst in subnetInst:
- if inst['name'] == subnet_name:
- return Subnet(**inst)
- return None
+ sub_filter = dict()
+ if subnet_settings:
+ sub_filter['name'] = subnet_settings.name
+ sub_filter['cidr'] = subnet_settings.cidr
+ if subnet_settings.gateway_ip:
+ sub_filter['gateway_ip'] = subnet_settings.gateway_ip
+
+ if subnet_settings.enable_dhcp is not None:
+ sub_filter['enable_dhcp'] = subnet_settings.enable_dhcp
+
+ if subnet_settings.destination:
+ sub_filter['destination'] = subnet_settings.destination
+
+ if subnet_settings.nexthop:
+ sub_filter['nexthop'] = subnet_settings.nexthop
+
+ if subnet_settings.ipv6_ra_mode:
+ sub_filter['ipv6_ra_mode'] = subnet_settings.ipv6_ra_mode
+
+ if subnet_settings.ipv6_address_mode:
+ sub_filter['ipv6_address_mode'] = subnet_settings.ipv6_address_mode
+ elif subnet_name:
+ sub_filter['name'] = subnet_name
+ else:
+ return None
+
+ subnets = neutron.list_subnets(**sub_filter)
+ for subnet in subnets['subnets']:
+ return Subnet(**subnet)
def create_router(neutron, os_creds, router_settings):
diff --git a/snaps/openstack/utils/tests/neutron_utils_tests.py b/snaps/openstack/utils/tests/neutron_utils_tests.py
index 0b9559a..eb77fab 100644
--- a/snaps/openstack/utils/tests/neutron_utils_tests.py
+++ b/snaps/openstack/utils/tests/neutron_utils_tests.py
@@ -99,7 +99,6 @@ class NeutronUtilsNetworkTests(OSComponentTestCase):
"""
if self.network:
neutron_utils.delete_network(self.neutron, self.network)
- validate_network(self.neutron, self.network.name, False)
def test_create_network(self):
"""
@@ -154,13 +153,8 @@ class NeutronUtilsSubnetTests(OSComponentTestCase):
"""
if self.subnet:
neutron_utils.delete_subnet(self.neutron, self.subnet)
- validate_subnet(self.neutron, self.subnet.name,
- self.net_config.network_settings.subnet_settings[
- 0].cidr, False)
-
if self.network:
neutron_utils.delete_network(self.neutron, self.network)
- validate_network(self.neutron, self.network.name, False)
def test_create_subnet(self):
"""
@@ -176,8 +170,8 @@ class NeutronUtilsSubnetTests(OSComponentTestCase):
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.assertTrue(validate_subnet(
+ self.neutron, subnet_setting.name, subnet_setting.cidr, True))
def test_create_subnet_null_name(self):
"""
@@ -209,7 +203,10 @@ class NeutronUtilsSubnetTests(OSComponentTestCase):
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.assertTrue(validate_subnet(
+ self.neutron, subnet_setting.name, subnet_setting.cidr, True))
+ self.assertFalse(validate_subnet(
+ self.neutron, '', subnet_setting.cidr, True))
def test_create_subnet_null_cidr(self):
"""
@@ -283,14 +280,9 @@ class NeutronUtilsRouterTests(OSComponentTestCase):
if self.subnet:
neutron_utils.delete_subnet(self.neutron, self.subnet)
- validate_subnet(
- self.neutron, self.subnet.name,
- self.net_config.network_settings.subnet_settings[0].cidr,
- False)
if self.network:
neutron_utils.delete_network(self.neutron, self.network)
- validate_network(self.neutron, self.network.name, False)
def test_create_router_simple(self):
"""
@@ -361,10 +353,8 @@ class NeutronUtilsRouterTests(OSComponentTestCase):
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)
@@ -392,8 +382,8 @@ class NeutronUtilsRouterTests(OSComponentTestCase):
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(NeutronException):
self.interface_router = neutron_utils.add_interface_router(
@@ -434,8 +424,8 @@ class NeutronUtilsRouterTests(OSComponentTestCase):
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(
@@ -460,8 +450,8 @@ class NeutronUtilsRouterTests(OSComponentTestCase):
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(
@@ -488,10 +478,8 @@ class NeutronUtilsRouterTests(OSComponentTestCase):
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(
@@ -535,10 +523,8 @@ class NeutronUtilsRouterTests(OSComponentTestCase):
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(
@@ -565,9 +551,8 @@ class NeutronUtilsRouterTests(OSComponentTestCase):
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(
@@ -594,8 +579,8 @@ class NeutronUtilsRouterTests(OSComponentTestCase):
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(
@@ -825,8 +810,8 @@ 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:
+ 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