diff --git a/tobiko/__init__.py b/tobiko/__init__.py index 88867057e..563d83056 100644 --- a/tobiko/__init__.py +++ b/tobiko/__init__.py @@ -15,7 +15,6 @@ from __future__ import absolute_import from tobiko.common import _asserts from tobiko.common import _exception -from tobiko.common import _find from tobiko.common import _fixture from tobiko.common.managers import testcase as testcase_manager from tobiko.common.managers import loader as loader_manager @@ -30,9 +29,6 @@ TobikoException = _exception.TobikoException check_valid_type = _exception.check_valid_type exc_info = _exception.exc_info -find_by_attributes = _find.find_by_attributes -find_by_items = _find.find_by_items - is_fixture = _fixture.is_fixture get_fixture = _fixture.get_fixture fixture_property = _fixture.fixture_property diff --git a/tobiko/common/_find.py b/tobiko/common/_find.py deleted file mode 100644 index b829e1eec..000000000 --- a/tobiko/common/_find.py +++ /dev/null @@ -1,44 +0,0 @@ -# Copyright 2019 Red Hat -# -# Licensed under the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. You may obtain -# a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. -from __future__ import absolute_import - - -def find_by_attributes(objects, exclude=False, **attributes): - exclude = bool(exclude) - if attributes: - selection = [] - for obj in objects: - for key, value in attributes.items(): - matching = value == getattr(obj, key) - if matching is exclude: - break - else: - selection.append(obj) - objects = selection - return objects - - -def find_by_items(mappings, exclude=False, **items): - exclude = bool(exclude) - if items: - selection = [] - for mapping in mappings: - for key, value in items.items(): - matching = value == mapping[key] - if matching is exclude: - break - else: - selection.append(mapping) - mappings = selection - return mappings diff --git a/tobiko/openstack/neutron/__init__.py b/tobiko/openstack/neutron/__init__.py index be920da81..71816694b 100644 --- a/tobiko/openstack/neutron/__init__.py +++ b/tobiko/openstack/neutron/__init__.py @@ -30,10 +30,10 @@ list_ports = _client.list_ports list_subnets = _client.list_subnets list_subnet_cidrs = _client.list_subnet_cidrs list_agents = _client.list_agents -show_network = _client.show_network -show_router = _client.show_router -show_port = _client.show_port -show_subnet = _client.show_subnet +get_network = _client.get_network +get_router = _client.get_router +get_port = _client.get_port +get_subnet = _client.get_subnet new_ipv4_cidr = _cidr.new_ipv4_cidr new_ipv6_cidr = _cidr.new_ipv6_cidr diff --git a/tobiko/openstack/neutron/_agent.py b/tobiko/openstack/neutron/_agent.py index 09bab8aa5..c0fc4c094 100644 --- a/tobiko/openstack/neutron/_agent.py +++ b/tobiko/openstack/neutron/_agent.py @@ -34,9 +34,9 @@ class NetworkingAgentFixture(tobiko.SharedFixture): self.agents = _client.list_agents(client=self.client) -def get_networking_agents(**params): +def get_networking_agents(**attributes): agents = tobiko.setup_fixture(NetworkingAgentFixture).agents - return tobiko.find_by_items(agents, **params) + return agents.with_items(**attributes) def missing_networking_agents(count=1, **params): diff --git a/tobiko/openstack/neutron/_client.py b/tobiko/openstack/neutron/_client.py index 57a88b4df..3435f9c53 100644 --- a/tobiko/openstack/neutron/_client.py +++ b/tobiko/openstack/neutron/_client.py @@ -20,7 +20,6 @@ from neutronclient.v2_0 import client as neutronclient import tobiko from tobiko.openstack import _client -from tobiko.openstack import _find class NeutronClientFixture(_client.OpenstackClientFixture): @@ -62,85 +61,86 @@ def get_neutron_client(session=None, shared=True, init_client=None, return client.client -def find_network(obj, properties=None, client=None, check_found=True, - check_unique=True, **params): +_RAISE_ERROR = object() + + +def find_network(client=None, unique=False, default=_RAISE_ERROR, + **attributes): """Look for a network matching some property values""" - resources = list_networks(client=client, **params) - return _find.find_resource(obj=obj, - resources=resources, - properties=properties, - check_found=check_found, - check_unique=check_unique) + networks = list_networks(client=client, **attributes) + if default is _RAISE_ERROR or networks: + if unique: + return networks.first + else: + return networks.unique + else: + return default -def find_port(obj, properties=None, client=None, check_found=True, - check_unique=True, **params): +def find_port(client=None, unique=False, default=_RAISE_ERROR, **attributes): """Look for a port matching some property values""" - resources = list_ports(client=client, **params) - return _find.find_resource(obj=obj, - resources=resources, - properties=properties, - check_found=check_found, - check_unique=check_unique) + ports = list_ports(client=client, **attributes) + if default is _RAISE_ERROR or ports: + if unique: + return ports.first + else: + return ports.unique + else: + return default -def find_subnet(obj, properties=None, client=None, check_found=True, - check_unique=False, **params): +def find_subnet(client=None, unique=False, default=_RAISE_ERROR, **attributes): """Look for a subnet matching some property values""" - resources = list_subnets(client=client, **params) - return _find.find_resource(obj=obj, - resources=resources, - properties=properties, - check_found=check_found, - check_unique=check_unique) + subnets = list_subnets(client=client, **attributes) + if default is _RAISE_ERROR or subnets: + if unique: + return subnets.first + else: + return subnets.unique + else: + return default -def list_networks(show=False, client=None, **params): +def list_networks(client=None, **params): networks = neutron_client(client).list_networks(**params)['networks'] - if show: - networks = [show_network(n['id'], client=client) for n in networks] - return networks + return tobiko.select(networks) -def list_ports(show=False, client=None, **params): +def list_ports(client=None, **params): ports = neutron_client(client).list_ports(**params)['ports'] - if show: - ports = [show_port(p['id'], client=client) for p in ports] - return ports + return tobiko.select(ports) -def list_subnets(show=False, client=None, **params): +def list_subnets(client=None, **params): subnets = neutron_client(client).list_subnets(**params) if isinstance(subnets, collections.Mapping): subnets = subnets['subnets'] - if show: - subnets = [show_subnet(s['id'], client=client) for s in subnets] - return subnets + return tobiko.select(subnets) def list_agents(client=None, **params): agents = neutron_client(client).list_agents(**params) if isinstance(agents, collections.Mapping): agents = agents['agents'] - return agents + return tobiko.select(agents) def list_subnet_cidrs(client=None, **params): - return [netaddr.IPNetwork(subnet['cidr']) - for subnet in list_subnets(client=client, **params)] + return tobiko.select(netaddr.IPNetwork(subnet['cidr']) + for subnet in list_subnets(client=client, **params)) -def show_network(network, client=None, **params): +def get_network(network, client=None, **params): return neutron_client(client).show_network(network, **params)['network'] -def show_port(port, client=None, **params): +def get_port(port, client=None, **params): return neutron_client(client).show_port(port, **params)['port'] -def show_router(router, client=None, **params): +def get_router(router, client=None, **params): return neutron_client(client).show_router(router, **params)['router'] -def show_subnet(subnet, client=None, **params): +def get_subnet(subnet, client=None, **params): return neutron_client(client).show_subnet(subnet, **params)['subnet'] diff --git a/tobiko/openstack/stacks/_neutron.py b/tobiko/openstack/stacks/_neutron.py index c935be9e2..577f7acb0 100644 --- a/tobiko/openstack/stacks/_neutron.py +++ b/tobiko/openstack/stacks/_neutron.py @@ -98,11 +98,11 @@ class NetworkStackFixture(heat.HeatStackFixture): @property def network_details(self): - return neutron.show_network(self.network_id) + return neutron.get_network(self.network_id) @property def ipv4_subnet_details(self): - return neutron.show_subnet(self.ipv4_subnet_id) + return neutron.get_subnet(self.ipv4_subnet_id) @property def ipv4_subnet_cidr(self): @@ -110,7 +110,7 @@ class NetworkStackFixture(heat.HeatStackFixture): @property def ipv6_subnet_details(self): - return neutron.show_subnet(self.ipv6_subnet_id) + return neutron.get_subnet(self.ipv6_subnet_id) @property def ipv6_subnet_cidr(self): @@ -118,29 +118,25 @@ class NetworkStackFixture(heat.HeatStackFixture): @property def gateway_details(self): - return neutron.show_router(self.gateway_id) + return neutron.get_router(self.gateway_id) @property def ipv4_gateway_port_details(self): return neutron.find_port( - [{'subnet_id': self.ipv4_subnet_id, - 'ip_address': self.ipv4_subnet_details['gateway_ip']}], - properties=['fixed_ips'], + fixed_ips='subnet_id=' + self.ipv4_subnet_id, device_id=self.gateway_id, network_id=self.network_id) @property def ipv6_gateway_port_details(self): return neutron.find_port( - [{'subnet_id': self.ipv6_subnet_id, - 'ip_address': self.ipv6_subnet_details['gateway_ip']}], - properties=['fixed_ips'], + fixed_ips='subnet_id=' + self.ipv6_subnet_id, device_id=self.gateway_id, network_id=self.network_id) @property def gateway_network_details(self): - return neutron.show_network(self.gateway_network_id) + return neutron.get_network(self.gateway_network_id) @neutron.skip_if_missing_networking_extensions('net-mtu-writable') diff --git a/tobiko/tests/functional/openstack/stacks/test_neutron.py b/tobiko/tests/functional/openstack/stacks/test_neutron.py index ce04ed21f..e02071fc1 100644 --- a/tobiko/tests/functional/openstack/stacks/test_neutron.py +++ b/tobiko/tests/functional/openstack/stacks/test_neutron.py @@ -44,18 +44,14 @@ class NetworkTestCase(testtools.TestCase): if not self.stack.has_ipv4: tobiko.skip('Stack {!s} has no ipv4 subnet', self.stack.stack_name) - subnet = neutron.find_subnet(str(self.stack.ipv4_subnet_cidr), - properties=['cidr']) - self.assertEqual(neutron.show_subnet(self.stack.ipv4_subnet_id), - subnet) + subnet = neutron.find_subnet(cidr=str(self.stack.ipv4_subnet_cidr)) + self.assertEqual(neutron.get_subnet(self.stack.ipv4_subnet_id), subnet) def test_ipv6_subnet_cidr(self): if not self.stack.has_ipv6: tobiko.skip('Stack {!s} has no ipv6 subnet', self.stack.stack_name) - subnet = neutron.find_subnet(str(self.stack.ipv6_subnet_cidr), - properties=['cidr']) - self.assertEqual(neutron.show_subnet(self.stack.ipv6_subnet_id), - subnet) + subnet = neutron.find_subnet(cidr=str(self.stack.ipv6_subnet_cidr)) + self.assertEqual(neutron.get_subnet(self.stack.ipv6_subnet_id), subnet) def test_gateway_network(self): if not self.stack.has_gateway: diff --git a/tobiko/tests/functional/openstack/test_neutron.py b/tobiko/tests/functional/openstack/test_neutron.py index 063711eed..963443e92 100644 --- a/tobiko/tests/functional/openstack/test_neutron.py +++ b/tobiko/tests/functional/openstack/test_neutron.py @@ -34,14 +34,14 @@ class NeutronApiTestCase(testtools.TestCase): stack = tobiko.required_setup_fixture(stacks.NetworkStackFixture) def test_find_network_with_id(self): - network = neutron.find_network(self.stack.network_id) + network = neutron.find_network(id=self.stack.network_id) self.assertEqual(self.stack.network_id, network['id']) def test_find_floating_network(self): floating_network = CONF.tobiko.neutron.floating_network if not floating_network: tobiko.skip('floating_network not configured') - network = neutron.find_network(floating_network) + network = neutron.find_network(name=floating_network) self.assertIn(floating_network, [network['name'], network['id']]) self.assertEqual(self.stack.gateway_network_id, network['id']) @@ -67,8 +67,8 @@ class NeutronApiTestCase(testtools.TestCase): cidr = netaddr.IPNetwork(self.stack.ipv6_subnet_details['cidr']) self.assertIn(cidr, subnets_cidrs) - def test_show_network(self): - network = neutron.show_network(self.stack.network_id) + def test_get_network(self): + network = neutron.get_network(self.stack.network_id) self.assertEqual(self.stack.network_id, network['id']) self.assertEqual(self.stack.port_security_enabled, network['port_security_enabled']) @@ -81,26 +81,26 @@ class NeutronApiTestCase(testtools.TestCase): else: self.assertNotIn(self.stack.ipv6_subnet_id, network['subnets']) - def test_show_router(self): + def test_get_router(self): if not self.stack.has_gateway: tobiko.skip("Stack {stack} has no gateway router", stack=self.stack.stack_name) - router = neutron.show_router(self.stack.gateway_id) + router = neutron.get_router(self.stack.gateway_id) self.assertEqual(self.stack.gateway_id, router['id']) - def test_show_ipv4_subnet(self): + def test_get_ipv4_subnet(self): if not self.stack.has_ipv4: tobiko.skip("Stack {stack} has no IPv4 subnet", stack=self.stack.stack_name) - subnet = neutron.show_subnet(self.stack.ipv4_subnet_id) + subnet = neutron.get_subnet(self.stack.ipv4_subnet_id) self.assertEqual(self.stack.ipv4_subnet_id, subnet['id']) self.assertEqual(self.stack.ipv4_subnet_details, subnet) - def test_show_ipv6_subnet(self): + def test_get_ipv6_subnet(self): if not self.stack.has_ipv6: tobiko.skip("Stack {stack} has no IPv6 subnet", stack=self.stack.stack_name) - subnet = neutron.show_subnet(self.stack.ipv6_subnet_id) + subnet = neutron.get_subnet(self.stack.ipv6_subnet_id) self.assertEqual(self.stack.ipv6_subnet_id, subnet['id']) self.assertEqual(self.stack.ipv6_subnet_details, subnet) diff --git a/tobiko/tests/scenario/nova/test_server.py b/tobiko/tests/scenario/nova/test_server.py index 6748f37d6..5500b37b3 100644 --- a/tobiko/tests/scenario/nova/test_server.py +++ b/tobiko/tests/scenario/nova/test_server.py @@ -39,5 +39,5 @@ class ServerStackResourcesTest(testtools.TestCase): port = self.stack.resources.port self.assertEqual('OS::Neutron::Port', port.resource_type) # Verify actual port status (is alive, etc.) - port_data = neutron.show_port(port.physical_resource_id) + port_data = neutron.get_port(port.physical_resource_id) self.assertEqual(port.physical_resource_id, port_data['id'])