diff options
Diffstat (limited to 'snaps/openstack/utils')
| -rw-r--r-- | snaps/openstack/utils/neutron_utils.py | 45 | ||||
| -rw-r--r-- | snaps/openstack/utils/tests/neutron_utils_tests.py | 63 |
2 files changed, 59 insertions, 49 deletions
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 |