openstack/openstacksdk
Code Issues Proposed changes

Merge "tests: Sanity check 'test_proxy_base'"

Browse Source
This commit is contained in:
Zuul
2021-05-19 20:57:30 +00:00
committed by Gerrit Code Review
parent cfafdc9cec 2e726ef772
commit 32d2a5522c
24 changed files with 427 additions and 300 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
openstack/baremetal/v1/_proxy.py
View File

@@ -168,19 +168,20 @@ class Proxy(proxy.Proxy):
return self._delete(_chassis.Chassis, chassis,
ignore_missing=ignore_missing)
def drivers(self, details=False):
def drivers(self, details=False, **query):
"""Retrieve a generator of drivers.
:param bool details: A boolean indicating whether the detailed
information for every driver should be returned.
:param kwargs query: Optional query parameters to be sent to limit
the resources being returned.
:returns: A generator of driver instances.
"""
kwargs = {}
# NOTE(dtantsur): details are available starting with API microversion
# 1.30. Thus we do not send any value if not needed.
if details:
kwargs['details'] = True
return self._list(_driver.Driver, **kwargs)
query['details'] = True
return self._list(_driver.Driver, **query)
def get_driver(self, driver):
"""Get a specific driver.

7
openstack/block_storage/v2/_proxy.py
View File

@@ -201,12 +201,15 @@ class Proxy(_base_proxy.BaseBlockStorageProxy):
volume = self._get_resource(_volume.Volume, volume)
volume.extend(self, size)
def backend_pools(self):
def backend_pools(self, **query):
"""Returns a generator of cinder Back-end storage pools
:param kwargs query: Optional query parameters to be sent to limit
the resources being returned.
:returns A generator of cinder Back-end storage pools objects
"""
return self._list(_stats.Pools)
return self._list(_stats.Pools, **query)
def backups(self, details=True, **query):
"""Retrieve a generator of backups

7
openstack/block_storage/v3/_proxy.py
View File

@@ -417,12 +417,15 @@ class Proxy(_base_proxy.BaseBlockStorageProxy):
volume = self._get_resource(_volume.Volume, volume)
volume.retype(self, new_type, migration_policy)
def backend_pools(self):
def backend_pools(self, **query):
"""Returns a generator of cinder Back-end storage pools
:param kwargs query: Optional query parameters to be sent to limit
the resources being returned.
:returns A generator of cinder Back-end storage pools objects
"""
return self._list(_stats.Pools)
return self._list(_stats.Pools, **query)
def backups(self, details=True, **query):
"""Retrieve a generator of backups

4
openstack/clustering/v1/_proxy.py
View File

@@ -420,12 +420,14 @@ class Proxy(proxy.Proxy):
obj = self._find(_cluster.Cluster, cluster, ignore_missing=False)
return obj.policy_update(self, policy, **params)
def collect_cluster_attrs(self, cluster, path):
def collect_cluster_attrs(self, cluster, path, **query):
"""Collect attribute values across a cluster.
:param cluster: The value can be either the ID of a cluster or a
:class:`~openstack.clustering.v1.cluster.Cluster` instance.
:param path: A Json path string specifying the attribute to collect.
:param query: Optional query parameters to be sent to limit the
resources being returned.
:returns: A dictionary containing the list of attribute values.
"""

12
openstack/compute/v2/_proxy.py
View File

@@ -1144,6 +1144,8 @@ class Proxy(proxy.Proxy):
return self._create(_server_interface.ServerInterface,
server_id=server_id, **attrs)
# TODO(stephenfin): Does this work? There's no 'value' parameter for the
# call to '_delete'
def delete_server_interface(self, server_interface, server=None,
ignore_missing=True):
"""Delete a server interface
@@ -1169,7 +1171,7 @@ class Proxy(proxy.Proxy):
server_interface = resource.Resource._get_id(server_interface)
self._delete(_server_interface.ServerInterface,
port_id=server_interface,
server_interface,
server_id=server_id,
ignore_missing=ignore_missing)
@@ -1197,18 +1199,20 @@ class Proxy(proxy.Proxy):
return self._get(_server_interface.ServerInterface,
server_id=server_id, port_id=server_interface)
def server_interfaces(self, server):
def server_interfaces(self, server, **query):
"""Return a generator of server interfaces
:param server: The server can be either the ID of a server or a
:class:`~openstack.compute.v2.server.Server`.
:class:`~openstack.compute.v2.server.Server`.
:param query: Optional query parameters to be sent to limit the
resources being returned.
:returns: A generator of ServerInterface objects
:rtype: :class:`~openstack.compute.v2.server_interface.ServerInterface`
"""
server_id = resource.Resource._get_id(server)
return self._list(_server_interface.ServerInterface,
server_id=server_id)
server_id=server_id, **query)
def server_ips(self, server, network_label=None):
"""Return a generator of server IPs

22
openstack/image/v2/_proxy.py
View File

@@ -579,16 +579,18 @@ class Proxy(_base_proxy.BaseImageProxy):
image_id = resource.Resource._get_id(image)
return self._create(_member.Member, image_id=image_id, **attrs)
def remove_member(self, member, image, ignore_missing=True):
def remove_member(self, member, image=None, ignore_missing=True):
"""Delete a member
:param member: The value can be either the ID of a member or a
:class:`~openstack.image.v2.member.Member` instance.
:class:`~openstack.image.v2.member.Member` instance.
:param image: The value can be either the ID of an image or a
:class:`~openstack.image.v2.image.Image` instance that the member
is part of. This is required if ``member`` is an ID.
:param bool ignore_missing: When set to ``False``
:class:`~openstack.exceptions.ResourceNotFound` will be
raised when the member does not exist.
When set to ``True``, no exception will be set when
attempting to delete a nonexistent member.
:class:`~openstack.exceptions.ResourceNotFound` will be raised when
the member does not exist. When set to ``True``, no exception will
be set when attempting to delete a nonexistent member.
:returns: ``None``
"""
@@ -632,12 +634,14 @@ class Proxy(_base_proxy.BaseImageProxy):
return self._get(_member.Member, member_id=member_id,
image_id=image_id)
def members(self, image):
def members(self, image, **query):
"""Return a generator of members
:param image: This is the image that the member belongs to,
the value can be the ID of a image or a
:class:`~openstack.image.v2.image.Image` instance.
the value can be the ID of a image or a
:class:`~openstack.image.v2.image.Image` instance.
:param kwargs query: Optional query parameters to be sent to limit
the resources being returned.
:returns: A generator of member objects
:rtype: :class:`~openstack.image.v2.member.Member`

19
openstack/proxy.py
View File

@@ -310,14 +310,13 @@ class Proxy(adapter.Adapter):
"""Get a resource object to work on
:param resource_type: The type of resource to operate on. This should
be a subclass of
:class:`~openstack.resource.Resource` with a
``from_id`` method.
be a subclass of :class:`~openstack.resource.Resource` with a
``from_id`` method.
:param value: The ID of a resource or an object of ``resource_type``
class if using an existing instance, or ``munch.Munch``,
or None to create a new instance.
:param path_args: A dict containing arguments for forming the request
URL, if needed.
class if using an existing instance, or ``munch.Munch``,
or None to create a new instance.
:param attrs: A dict containing arguments for forming the request
URL, if needed.
"""
conn = self._get_connection()
if value is None:
@@ -374,6 +373,7 @@ class Proxy(adapter.Adapter):
ignore_missing=ignore_missing,
**attrs)
# TODO(stephenfin): Update docstring for attrs since it's a lie
@_check_resource(strict=False)
def _delete(self, resource_type, value, ignore_missing=True, **attrs):
"""Delete a resource
@@ -512,16 +512,13 @@ class Proxy(adapter.Adapter):
error_message="No {resource_type} found for {value}".format(
resource_type=resource_type.__name__, value=value))
def _list(self, resource_type, value=None,
def _list(self, resource_type,
paginated=True, base_path=None, **attrs):
"""List a resource
:param resource_type: The type of resource to list. This should
be a :class:`~openstack.resource.Resource`
subclass with a ``from_id`` method.
:param value: The resource to list. It can be the ID of a resource, or
a :class:`~openstack.resource.Resource` object. When set
to None, a new bare resource is created.
:param bool paginated: When set to ``False``, expect all of the data
to be returned in one response. When set to
``True``, the resource supports data being

7
openstack/tests/unit/block_storage/v3/test_proxy.py
View File

@@ -104,9 +104,9 @@ class TestVolumeProxy(test_proxy_base.TestProxyBase):
def test_type_encryption_get(self):
self.verify_get(self.proxy.get_type_encryption,
type.TypeEncryption,
expected_args=[type.TypeEncryption],
expected_args=[],
expected_kwargs={
'volume_type_id': 'value',
'volume_type_id': 'resource_id',
'requires_id': False
})
@@ -250,7 +250,8 @@ class TestVolumeProxy(test_proxy_base.TestProxyBase):
def test_limits_get(self):
self.verify_get(
self.proxy.get_limits, limits.Limit, ignore_value=True,
self.proxy.get_limits, limits.Limit,
method_args=[],
expected_kwargs={'requires_id': False})
def test_capabilites_get(self):

4
openstack/tests/unit/clustering/v1/test_proxy.py
View File

@@ -37,7 +37,7 @@ class TestClusterProxy(test_proxy_base.TestProxyBase):
def test_build_info_get(self):
self.verify_get(self.proxy.get_build_info, build_info.BuildInfo,
ignore_value=True,
method_args=[],
expected_kwargs={'requires_id': False})
def test_profile_types(self):
@@ -137,6 +137,7 @@ class TestClusterProxy(test_proxy_base.TestProxyBase):
self.verify_list(self.proxy.collect_cluster_attrs,
cluster_attr.ClusterAttr,
method_args=['FAKE_ID', 'path.to.attr'],
expected_args=[],
expected_kwargs={'cluster_id': 'FAKE_ID',
'path': 'path.to.attr'})
@@ -265,6 +266,7 @@ class TestClusterProxy(test_proxy_base.TestProxyBase):
self.verify_list(self.proxy.cluster_policies,
cluster_policy.ClusterPolicy,
method_args=["FAKE_CLUSTER"],
expected_args=[],
expected_kwargs={"cluster_id": "FAKE_CLUSTER"})
def test_get_cluster_policy(self):

51
openstack/tests/unit/compute/v2/test_proxy.py
View File

@@ -56,7 +56,7 @@ class TestFlavor(TestComputeProxy):
self.verify_find(
self.proxy.find_flavor, flavor.Flavor,
method_kwargs={"a": "b"},
expected_kwargs={"a": "b", "ignore_missing": False}
expected_kwargs={"a": "b", "ignore_missing": True}
)
def test_flavor_find_fetch_extra(self):
@@ -286,7 +286,7 @@ class TestKeyPair(TestComputeProxy):
)
def test_keypairs(self):
self.verify_list_no_kwargs(self.proxy.keypairs, keypair.Keypair)
self.verify_list(self.proxy.keypairs, keypair.Keypair)
def test_keypairs_user_id(self):
self.verify_list(
@@ -312,7 +312,7 @@ class TestAggregate(TestComputeProxy):
self.verify_find(self.proxy.find_aggregate, aggregate.Aggregate)
def test_aggregates(self):
self.verify_list_no_kwargs(self.proxy.aggregates, aggregate.Aggregate)
self.verify_list(self.proxy.aggregates, aggregate.Aggregate)
def test_aggregate_get(self):
self.verify_get(self.proxy.get_aggregate, aggregate.Aggregate)
@@ -355,8 +355,7 @@ class TestAggregate(TestComputeProxy):
class TestService(TestComputeProxy):
def test_services(self):
self.verify_list_no_kwargs(
self.proxy.services, service.Service)
self.verify_list(self.proxy.services, service.Service)
@mock.patch('openstack.utils.supports_microversion', autospec=True,
return_value=False)
@@ -450,11 +449,13 @@ class TestHypervisor(TestComputeProxy):
def test_hypervisors_not_detailed(self):
self.verify_list(self.proxy.hypervisors, hypervisor.Hypervisor,
method_kwargs={"details": False})
method_kwargs={"details": False},
expected_kwargs={})
def test_hypervisors_detailed(self):
self.verify_list(self.proxy.hypervisors, hypervisor.HypervisorDetail,
method_kwargs={"details": True})
method_kwargs={"details": True},
expected_kwargs={})
@mock.patch('openstack.utils.supports_microversion', autospec=True,
return_value=False)
@@ -462,8 +463,9 @@ class TestHypervisor(TestComputeProxy):
self.verify_list(
self.proxy.hypervisors,
hypervisor.Hypervisor,
base_path='/os-hypervisors/detail',
method_kwargs={'details': True},
base_path='/os-hypervisors/detail'
expected_kwargs={},
)
@mock.patch('openstack.utils.supports_microversion', autospec=True,
@@ -472,8 +474,9 @@ class TestHypervisor(TestComputeProxy):
self.verify_list(
self.proxy.hypervisors,
hypervisor.Hypervisor,
base_path='/os-hypervisors/substring/search',
method_kwargs={'hypervisor_hostname_pattern': 'substring'},
base_path='/os-hypervisors/substring/search'
expected_kwargs={},
)
@mock.patch('openstack.utils.supports_microversion', autospec=True,
@@ -492,7 +495,7 @@ class TestHypervisor(TestComputeProxy):
hypervisor.Hypervisor,
expected_kwargs={
'list_base_path': '/os-hypervisors/detail',
'ignore_missing': False})
'ignore_missing': True})
def test_find_hypervisor_no_detail(self):
self.verify_find(self.proxy.find_hypervisor,
@@ -500,7 +503,7 @@ class TestHypervisor(TestComputeProxy):
method_kwargs={'details': False},
expected_kwargs={
'list_base_path': None,
'ignore_missing': False})
'ignore_missing': True})
def test_get_hypervisor(self):
self.verify_get(self.proxy.get_hypervisor,
@@ -519,7 +522,7 @@ class TestCompute(TestComputeProxy):
self.verify_find(self.proxy.find_extension, extension.Extension)
def test_extensions(self):
self.verify_list_no_kwargs(self.proxy.extensions, extension.Extension)
self.verify_list(self.proxy.extensions, extension.Extension)
def test_image_delete(self):
self.verify_delete(self.proxy.delete_image, image.Image, False)
@@ -544,7 +547,7 @@ class TestCompute(TestComputeProxy):
expected_kwargs={"query": 1})
def test_limits_get(self):
self.verify_get(self.proxy.get_limits, limits.Limits, value=[])
self.verify_get(self.proxy.get_limits, limits.Limits, method_args=[])
def test_server_interface_create(self):
self.verify_create(self.proxy.create_server_interface,
@@ -565,9 +568,9 @@ class TestCompute(TestComputeProxy):
self.proxy.delete_server_interface,
method_args=[test_interface],
method_kwargs={"server": server_id},
expected_args=[server_interface.ServerInterface],
expected_args=[
server_interface.ServerInterface, interface_id],
expected_kwargs={"server_id": server_id,
"port_id": interface_id,
"ignore_missing": True})
# Case2: ServerInterface ID is provided as value
@@ -575,9 +578,9 @@ class TestCompute(TestComputeProxy):
self.proxy.delete_server_interface,
method_args=[interface_id],
method_kwargs={"server": server_id},
expected_args=[server_interface.ServerInterface],
expected_args=[
server_interface.ServerInterface, interface_id],
expected_kwargs={"server_id": server_id,
"port_id": interface_id,
"ignore_missing": True})
def test_server_interface_delete_ignore(self):
@@ -585,9 +588,8 @@ class TestCompute(TestComputeProxy):
self.verify_delete(self.proxy.delete_server_interface,
server_interface.ServerInterface, True,
method_kwargs={"server": "test_id"},
expected_args=[server_interface.ServerInterface],
expected_kwargs={"server_id": "test_id",
"port_id": "resource_or_id"})
expected_args=[],
expected_kwargs={"server_id": "test_id"})
def test_server_interface_get(self):
self.proxy._get_uri_attribute = lambda *args: args[1]
@@ -619,18 +621,21 @@ class TestCompute(TestComputeProxy):
self.verify_list(self.proxy.server_interfaces,
server_interface.ServerInterface,
method_args=["test_id"],
expected_args=[],
expected_kwargs={"server_id": "test_id"})
def test_server_ips_with_network_label(self):
self.verify_list(self.proxy.server_ips, server_ip.ServerIP,
method_args=["test_id"],
method_kwargs={"network_label": "test_label"},
expected_args=[],
expected_kwargs={"server_id": "test_id",
"network_label": "test_label"})
def test_server_ips_without_network_label(self):
self.verify_list(self.proxy.server_ips, server_ip.ServerIP,
method_args=["test_id"],
expected_args=[],
expected_kwargs={"server_id": "test_id",
"network_label": None})
@@ -854,12 +859,14 @@ class TestCompute(TestComputeProxy):
def test_availability_zones_not_detailed(self):
self.verify_list(self.proxy.availability_zones,
az.AvailabilityZone,
method_kwargs={"details": False})
method_kwargs={"details": False},
expected_kwargs={})
def test_availability_zones_detailed(self):
self.verify_list(self.proxy.availability_zones,
az.AvailabilityZoneDetail,
method_kwargs={"details": True})
method_kwargs={"details": True},
expected_kwargs={})
def test_get_all_server_metadata(self):
self._verify2("openstack.compute.v2.server.Server.get_metadata",

27
openstack/tests/unit/database/v1/test_proxy.py
View File

@@ -24,21 +24,24 @@ class TestDatabaseProxy(test_proxy_base.TestProxyBase):
self.proxy = _proxy.Proxy(self.session)
def test_database_create_attrs(self):
self.verify_create(self.proxy.create_database, database.Database,
self.verify_create(self.proxy.create_database,
database.Database,
method_kwargs={"instance": "id"},
expected_kwargs={"instance_id": "id"})
def test_database_delete(self):
self.verify_delete(self.proxy.delete_database,
database.Database, False,
input_path_args={"instance": "test_id"},
expected_path_args={"instance_id": "test_id"})
database.Database,
ignore_missing=False,
method_kwargs={"instance": "test_id"},
expected_kwargs={"instance_id": "test_id"})
def test_database_delete_ignore(self):
self.verify_delete(self.proxy.delete_database,
database.Database, True,
input_path_args={"instance": "test_id"},
expected_path_args={"instance_id": "test_id"})
database.Database,
ignore_missing=True,
method_kwargs={"instance": "test_id"},
expected_kwargs={"instance_id": "test_id"})
def test_database_find(self):
self._verify2('openstack.proxy.Proxy._find',
@@ -51,6 +54,7 @@ class TestDatabaseProxy(test_proxy_base.TestProxyBase):
def test_databases(self):
self.verify_list(self.proxy.databases, database.Database,
method_args=["id"],
expected_args=[],
expected_kwargs={"instance_id": "id"})
def test_database_get(self):
@@ -95,13 +99,13 @@ class TestDatabaseProxy(test_proxy_base.TestProxyBase):
def test_user_delete(self):
self.verify_delete(self.proxy.delete_user, user.User, False,
input_path_args={"instance": "id"},
expected_path_args={"instance_id": "id"})
method_kwargs={"instance": "id"},
expected_kwargs={"instance_id": "id"})
def test_user_delete_ignore(self):
self.verify_delete(self.proxy.delete_user, user.User, True,
input_path_args={"instance": "id"},
expected_path_args={"instance_id": "id"})
method_kwargs={"instance": "id"},
expected_kwargs={"instance_id": "id"})
def test_user_find(self):
self._verify2('openstack.proxy.Proxy._find',
@@ -114,6 +118,7 @@ class TestDatabaseProxy(test_proxy_base.TestProxyBase):
def test_users(self):
self.verify_list(self.proxy.users, user.User,
method_args=["test_instance"],
expected_args=[],
expected_kwargs={"instance_id": "test_instance"})
def test_user_get(self):

6
openstack/tests/unit/dns/v2/test_proxy.py
View File

@@ -84,7 +84,7 @@ class TestDnsRecordset(TestDnsProxy):
def test_recordsets(self):
self.verify_list(self.proxy.recordsets, recordset.Recordset,
base_path='/recordsets')
expected_kwargs={'base_path': '/recordsets'})
def test_recordsets_zone(self):
self.verify_list(self.proxy.recordsets, recordset.Recordset,
@@ -143,7 +143,7 @@ class TestDnsZoneExport(TestDnsProxy):
def test_zone_export_get_text(self):
self.verify_get(self.proxy.get_zone_export_text,
zone_export.ZoneExport,
value=[{'id': 'zone_export_id_value'}],
method_args=[{'id': 'zone_export_id_value'}],
expected_kwargs={
'base_path': '/zones/tasks/export/%(id)s/export'
})
@@ -156,6 +156,7 @@ class TestDnsZoneExport(TestDnsProxy):
zone_export.ZoneExport,
method_args=[{'id': 'zone_id_value'}],
method_kwargs={'name': 'id'},
expected_args=[],
expected_kwargs={'name': 'id',
'zone_id': 'zone_id_value',
'prepend_key': False})
@@ -179,6 +180,7 @@ class TestDnsZoneTransferRequest(TestDnsProxy):
zone_transfer.ZoneTransferRequest,
method_args=[{'id': 'zone_id_value'}],
method_kwargs={'name': 'id'},
expected_args=[],
expected_kwargs={'name': 'id',
'zone_id': 'zone_id_value',
'prepend_key': False})

2
openstack/tests/unit/image/v2/test_proxy.py
View File

@@ -364,7 +364,7 @@ class TestImageProxy(test_proxy_base.TestProxyBase):
def test_members(self):
self.verify_list(self.proxy.members, member.Member,
method_args=('image_1',),
method_kwargs={'image': 'image_1'},
expected_kwargs={'image_id': 'image_1'})
def test_images_schema_get(self):

4
openstack/tests/unit/instance_ha/v1/test_proxy.py
View File

@@ -31,12 +31,13 @@ class TestInstanceHaProxy(test_proxy_base.TestProxyBase):
self.verify_list(self.proxy.hosts,
host.Host,
method_args=[SEGMENT_ID],
expected_args=[],
expected_kwargs={"segment_id": SEGMENT_ID})
def test_host_get(self):
self.verify_get(self.proxy.get_host,
host.Host,
value=[HOST_ID],
method_args=[HOST_ID],
method_kwargs={"segment_id": SEGMENT_ID},
expected_kwargs={"segment_id": SEGMENT_ID})
@@ -45,6 +46,7 @@ class TestInstanceHaProxy(test_proxy_base.TestProxyBase):
host.Host,
method_args=[SEGMENT_ID],
method_kwargs={},
expected_args=[],
expected_kwargs={"segment_id": SEGMENT_ID})
def test_host_update(self):

0
openstack/tests/unit/load_balancer/v2/__init__.py Normal file
View File

23
openstack/tests/unit/load_balancer/test_proxy.py → openstack/tests/unit/load_balancer/v2/test_proxy.py
View File

@@ -56,8 +56,8 @@ class TestLoadBalancerProxy(test_proxy_base.TestProxyBase):
def test_load_balancer_stats_get(self):
self.verify_get(self.proxy.get_load_balancer_statistics,
lb.LoadBalancerStats,
value=[self.LB_ID],
expected_args=[lb.LoadBalancerStats],
method_args=[self.LB_ID],
expected_args=[],
expected_kwargs={'lb_id': self.LB_ID,
'requires_id': False})
@@ -108,7 +108,7 @@ class TestLoadBalancerProxy(test_proxy_base.TestProxyBase):
def test_load_balancer_failover(self):
self.verify_update(self.proxy.failover_load_balancer,
lb.LoadBalancerFailover,
value=[self.LB_ID],
method_args=[self.LB_ID],
expected_args=[],
expected_kwargs={'lb_id': self.LB_ID})
@@ -123,8 +123,8 @@ class TestLoadBalancerProxy(test_proxy_base.TestProxyBase):
def test_listener_stats_get(self):
self.verify_get(self.proxy.get_listener_statistics,
listener.ListenerStats,
value=[self.LISTENER_ID],
expected_args=[listener.ListenerStats],
method_args=[self.LISTENER_ID],
expected_args=[],
expected_kwargs={'listener_id': self.LISTENER_ID,
'requires_id': False})
@@ -189,9 +189,11 @@ class TestLoadBalancerProxy(test_proxy_base.TestProxyBase):
def test_member_delete(self):
self.verify_delete(self.proxy.delete_member,
member.Member,
True,
ignore_missing=True,
method_kwargs={'pool': self.POOL_ID},
expected_kwargs={'pool_id': self.POOL_ID})
expected_kwargs={
'pool_id': self.POOL_ID,
'ignore_missing': True})
def test_member_find(self):
self._verify2('openstack.proxy.Proxy._find',
@@ -277,7 +279,7 @@ class TestLoadBalancerProxy(test_proxy_base.TestProxyBase):
def test_l7_rule_delete(self):
self.verify_delete(self.proxy.delete_l7_rule,
l7_rule.L7Rule,
True,
ignore_missing=True,
method_kwargs={'l7_policy': self.L7_POLICY_ID},
expected_kwargs={'l7policy_id': self.L7_POLICY_ID})
@@ -324,6 +326,7 @@ class TestLoadBalancerProxy(test_proxy_base.TestProxyBase):
self.verify_list(self.proxy.provider_flavor_capabilities,
provider.ProviderFlavorCapabilities,
method_args=[self.AMPHORA],
expected_args=[],
expected_kwargs={'provider': self.AMPHORA})
def test_flavor_profiles(self):
@@ -380,14 +383,14 @@ class TestLoadBalancerProxy(test_proxy_base.TestProxyBase):
def test_amphora_configure(self):
self.verify_update(self.proxy.configure_amphora,
amphora.AmphoraConfig,
value=[self.AMPHORA_ID],
method_args=[self.AMPHORA_ID],
expected_args=[],
expected_kwargs={'amphora_id': self.AMPHORA_ID})
def test_amphora_failover(self):
self.verify_update(self.proxy.failover_amphora,
amphora.AmphoraFailover,
value=[self.AMPHORA_ID],
method_args=[self.AMPHORA_ID],
expected_args=[],
expected_kwargs={'amphora_id': self.AMPHORA_ID})

43
openstack/tests/unit/message/v2/test_proxy.py
View File

@@ -73,7 +73,7 @@ class TestMessageProxy(test_proxy_base.TestProxyBase):
def test_messages(self):
self.verify_list(self.proxy.messages, message.Message,
method_args=["test_queue"],
method_kwargs={"queue_name": "test_queue"},
expected_kwargs={"queue_name": "test_queue"})
@mock.patch.object(proxy_base.Proxy, '_get_resource')
@@ -150,25 +150,29 @@ class TestMessageProxy(test_proxy_base.TestProxyBase):
def test_subscriptions(self):
self.verify_list(self.proxy.subscriptions, subscription.Subscription,
method_args=["test_queue"],
method_kwargs={"queue_name": "test_queue"},
expected_kwargs={"queue_name": "test_queue"})
@mock.patch.object(proxy_base.Proxy, '_get_resource')
def test_subscription_delete(self, mock_get_resource):
mock_get_resource.return_value = "resource_or_id"
mock_get_resource.return_value = "test_subscription"
self.verify_delete(self.proxy.delete_subscription,
subscription.Subscription, False,
["test_queue", "resource_or_id"])
subscription.Subscription,
ignore_missing=False,
method_args=["test_queue", "resource_or_id"],
expected_args=["test_subscription"])
mock_get_resource.assert_called_once_with(
subscription.Subscription, "resource_or_id",
queue_name="test_queue")
@mock.patch.object(proxy_base.Proxy, '_get_resource')
def test_subscription_delete_ignore(self, mock_get_resource):
mock_get_resource.return_value = "resource_or_id"
mock_get_resource.return_value = "test_subscription"
self.verify_delete(self.proxy.delete_subscription,
subscription.Subscription, True,
["test_queue", "resource_or_id"])
subscription.Subscription,
ignore_missing=True,
method_args=["test_queue", "resource_or_id"],
expected_args=["test_subscription"])
mock_get_resource.assert_called_once_with(
subscription.Subscription, "resource_or_id",
queue_name="test_queue")
@@ -202,12 +206,21 @@ class TestMessageProxy(test_proxy_base.TestProxyBase):
def test_claim_delete(self):
self.verify_delete(self.proxy.delete_claim,
claim.Claim, False,
["test_queue", "resource_or_id"],
expected_kwargs={"queue_name": "test_queue"})
claim.Claim,
ignore_missing=False,
method_args=["test_queue", "test_claim"],
expected_args=["test_claim"],
expected_kwargs={
"queue_name": "test_queue",
"ignore_missing": False})
def test_claim_delete_ignore(self):
self.verify_delete(self.proxy.delete_claim,
claim.Claim, True,
["test_queue", "resource_or_id"],
expected_kwargs={"queue_name": "test_queue"})
self.verify_delete(
self.proxy.delete_claim,
claim.Claim,
ignore_missing=True,
method_args=["test_queue", "test_claim"],
expected_args=["test_claim"],
expected_kwargs={
"queue_name": "test_queue", "ignore_missing": True,
})

139
openstack/tests/unit/network/v2/test_proxy.py
View File

@@ -72,23 +72,37 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
super(TestNetworkProxy, self).setUp()
self.proxy = _proxy.Proxy(self.session)
def verify_update(self, test_method, resource_type, value=None,
mock_method="openstack.network.v2._proxy.Proxy._update",
expected_result="result", path_args=None, **kwargs):
super(TestNetworkProxy, self).verify_update(
test_method, resource_type, value=value, mock_method=mock_method,
expected_result=expected_result, path_args=path_args, **kwargs)
def verify_update(
self, test_method, resource_type, base_path=None, *,
method_args=None, method_kwargs=None,
expected_args=None, expected_kwargs=None, expected_result="result",
mock_method="openstack.network.v2._proxy.Proxy._update",
):
super().verify_update(
test_method,
resource_type,
base_path=base_path,
method_args=method_args,
method_kwargs=method_kwargs,
expected_args=expected_args,
expected_kwargs=expected_kwargs,
expected_result=expected_result,
mock_method=mock_method)
def verify_delete(self, test_method, resource_type, ignore,
input_path_args=None, expected_path_args=None,
method_kwargs=None, expected_args=None,
expected_kwargs=None,
mock_method="openstack.network.v2._proxy.Proxy._delete"):
super(TestNetworkProxy, self).verify_delete(
test_method, resource_type, ignore,
input_path_args=input_path_args,
expected_path_args=expected_path_args, method_kwargs=method_kwargs,
expected_args=expected_args, expected_kwargs=expected_kwargs,
def verify_delete(
self, test_method, resource_type, ignore_missing=True, *,
method_args=None, method_kwargs=None,
expected_args=None, expected_kwargs=None,
mock_method="openstack.network.v2._proxy.Proxy._delete",
):
super().verify_delete(
test_method,
resource_type,
ignore_missing=ignore_missing,
method_args=method_args,
method_kwargs=method_kwargs,
expected_args=expected_args,
expected_kwargs=expected_kwargs,
mock_method=mock_method)
def test_address_scope_create_attrs(self):
@@ -134,8 +148,9 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
self.verify_update(self.proxy.update_agent, agent.Agent)
def test_availability_zones(self):
self.verify_list_no_kwargs(self.proxy.availability_zones,
availability_zone.AvailabilityZone)
self.verify_list(
self.proxy.availability_zones,
availability_zone.AvailabilityZone)
def test_dhcp_agent_hosting_networks(self):
self.verify_list(
@@ -438,14 +453,20 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
expected_kwargs={"pool_id": "test_id"})
def test_pool_member_delete(self):
self.verify_delete(self.proxy.delete_pool_member,
pool_member.PoolMember, False,
{"pool": "test_id"}, {"pool_id": "test_id"})
self.verify_delete(
self.proxy.delete_pool_member,
pool_member.PoolMember,
ignore_missing=False,
method_kwargs={"pool": "test_id"},
expected_kwargs={"pool_id": "test_id"})
def test_pool_member_delete_ignore(self):
self.verify_delete(self.proxy.delete_pool_member,
pool_member.PoolMember, True,
{"pool": "test_id"}, {"pool_id": "test_id"})
self.verify_delete(
self.proxy.delete_pool_member,
pool_member.PoolMember,
ignore_missing=True,
method_kwargs={"pool": "test_id"},
expected_kwargs={"pool_id": "test_id"})
def test_pool_member_find(self):
self._verify2('openstack.proxy.Proxy._find',
@@ -463,9 +484,11 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
expected_kwargs={"pool_id": "POOL"})
def test_pool_members(self):
self.verify_list(self.proxy.pool_members, pool_member.PoolMember,
method_args=["test_id"],
expected_kwargs={"pool_id": "test_id"})
self.verify_list(
self.proxy.pool_members, pool_member.PoolMember,
method_args=["test_id"],
expected_args=[],
expected_kwargs={"pool_id": "test_id"})
def test_pool_member_update(self):
self._verify2("openstack.network.v2._proxy.Proxy._update",
@@ -551,14 +574,18 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
self.verify_delete(
self.proxy.delete_qos_bandwidth_limit_rule,
qos_bandwidth_limit_rule.QoSBandwidthLimitRule,
False, input_path_args=["resource_or_id", QOS_POLICY_ID],
ignore_missing=False,
method_args=["resource_or_id", QOS_POLICY_ID],
expected_args=["resource_or_id"],
expected_kwargs={'qos_policy_id': QOS_POLICY_ID})
def test_qos_bandwidth_limit_rule_delete_ignore(self):
self.verify_delete(
self.proxy.delete_qos_bandwidth_limit_rule,
qos_bandwidth_limit_rule.QoSBandwidthLimitRule,
True, input_path_args=["resource_or_id", QOS_POLICY_ID],
ignore_missing=True,
method_args=["resource_or_id", QOS_POLICY_ID],
expected_args=["resource_or_id"],
expected_kwargs={'qos_policy_id': QOS_POLICY_ID})
def test_qos_bandwidth_limit_rule_find(self):
@@ -609,15 +636,19 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
self.verify_delete(
self.proxy.delete_qos_dscp_marking_rule,
qos_dscp_marking_rule.QoSDSCPMarkingRule,
False, input_path_args=["resource_or_id", QOS_POLICY_ID],
expected_path_args={'qos_policy_id': QOS_POLICY_ID},)
ignore_missing=False,
method_args=["resource_or_id", QOS_POLICY_ID],
expected_args=["resource_or_id"],
expected_kwargs={'qos_policy_id': QOS_POLICY_ID})
def test_qos_dscp_marking_rule_delete_ignore(self):
self.verify_delete(
self.proxy.delete_qos_dscp_marking_rule,
qos_dscp_marking_rule.QoSDSCPMarkingRule,
True, input_path_args=["resource_or_id", QOS_POLICY_ID],
expected_path_args={'qos_policy_id': QOS_POLICY_ID}, )
ignore_missing=True,
method_args=["resource_or_id", QOS_POLICY_ID],
expected_args=["resource_or_id"],
expected_kwargs={'qos_policy_id': QOS_POLICY_ID}, )
def test_qos_dscp_marking_rule_find(self):
policy = qos_policy.QoSPolicy.new(id=QOS_POLICY_ID)
@@ -666,15 +697,19 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
self.verify_delete(
self.proxy.delete_qos_minimum_bandwidth_rule,
qos_minimum_bandwidth_rule.QoSMinimumBandwidthRule,
False, input_path_args=["resource_or_id", QOS_POLICY_ID],
expected_path_args={'qos_policy_id': QOS_POLICY_ID},)
ignore_missing=False,
method_args=["resource_or_id", QOS_POLICY_ID],
expected_args=["resource_or_id"],
expected_kwargs={'qos_policy_id': QOS_POLICY_ID})
def test_qos_minimum_bandwidth_rule_delete_ignore(self):
self.verify_delete(
self.proxy.delete_qos_minimum_bandwidth_rule,
qos_minimum_bandwidth_rule.QoSMinimumBandwidthRule,
True, input_path_args=["resource_or_id", QOS_POLICY_ID],
expected_path_args={'qos_policy_id': QOS_POLICY_ID}, )
ignore_missing=True,
method_args=["resource_or_id", QOS_POLICY_ID],
expected_args=["resource_or_id"],
expected_kwargs={'qos_policy_id': QOS_POLICY_ID})
def test_qos_minimum_bandwidth_rule_find(self):
policy = qos_policy.QoSPolicy.new(id=QOS_POLICY_ID)
@@ -1299,9 +1334,8 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
def test_validate_topology(self):
self.verify_get(self.proxy.validate_auto_allocated_topology,
auto_allocated_topology.ValidateTopology,
value=[mock.sentinel.project_id],
expected_args=[
auto_allocated_topology.ValidateTopology],
method_args=[mock.sentinel.project_id],
expected_args=[],
expected_kwargs={"project": mock.sentinel.project_id,
"requires_id": False})
@@ -1331,15 +1365,19 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
self.verify_delete(
self.proxy.delete_floating_ip_port_forwarding,
port_forwarding.PortForwarding,
False, input_path_args=[FIP_ID, "resource_or_id"],
expected_path_args={'floatingip_id': FIP_ID},)
ignore_missing=False,
method_args=[FIP_ID, "resource_or_id"],
expected_args=["resource_or_id"],
expected_kwargs={'floatingip_id': FIP_ID})
def test_delete_floating_ip_port_forwarding_ignore(self):
self.verify_delete(
self.proxy.delete_floating_ip_port_forwarding,
port_forwarding.PortForwarding,
True, input_path_args=[FIP_ID, "resource_or_id"],
expected_path_args={'floatingip_id': FIP_ID}, )
ignore_missing=True,
method_args=[FIP_ID, "resource_or_id"],
expected_args=["resource_or_id"],
expected_kwargs={'floatingip_id': FIP_ID})
def test_find_floating_ip_port_forwarding(self):
fip = floating_ip.FloatingIP.new(id=FIP_ID)
@@ -1391,16 +1429,20 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
self.verify_delete(
self.proxy.delete_conntrack_helper,
l3_conntrack_helper.ConntrackHelper,
False, input_path_args=['resource_or_id', r],
expected_path_args={'router_id': ROUTER_ID},)
ignore_missing=False,
method_args=['resource_or_id', r],
expected_args=['resource_or_id'],
expected_kwargs={'router_id': ROUTER_ID},)
def test_delete_l3_conntrack_helper_ignore(self):
r = router.Router.new(id=ROUTER_ID)
self.verify_delete(
self.proxy.delete_conntrack_helper,
l3_conntrack_helper.ConntrackHelper,
True, input_path_args=['resource_or_id', r],
expected_path_args={'router_id': ROUTER_ID}, )
ignore_missing=True,
method_args=['resource_or_id', r],
expected_args=['resource_or_id'],
expected_kwargs={'router_id': ROUTER_ID},)
def test_get_l3_conntrack_helper(self):
r = router.Router.new(id=ROUTER_ID)
@@ -1416,6 +1458,7 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
self.verify_list(self.proxy.conntrack_helpers,
l3_conntrack_helper.ConntrackHelper,
method_args=[ROUTER_ID],
expected_args=[],
expected_kwargs={'router_id': ROUTER_ID})
def test_update_l3_conntrack_helper(self):

9
openstack/tests/unit/object_store/v1/test_proxy.py
View File

@@ -39,11 +39,13 @@ class TestObjectStoreProxy(test_proxy_base.TestProxyBase):
endpoint=self.endpoint, container=self.container)
def test_account_metadata_get(self):
self.verify_head(self.proxy.get_account_metadata, account.Account)
self.verify_head(
self.proxy.get_account_metadata, account.Account,
method_args=[])
def test_container_metadata_get(self):
self.verify_head(self.proxy.get_container_metadata,
container.Container, value="container")
container.Container, method_args=["container"])
def test_container_delete(self):
self.verify_delete(self.proxy.delete_container,
@@ -58,6 +60,7 @@ class TestObjectStoreProxy(test_proxy_base.TestProxyBase):
self.proxy.create_container,
container.Container,
method_args=['container_name'],
expected_args=[],
expected_kwargs={'name': 'container_name', "x": 1, "y": 2, "z": 3})
def test_object_metadata_get(self):
@@ -103,7 +106,7 @@ class TestObjectStoreProxy(test_proxy_base.TestProxyBase):
kwargs = dict(container="container")
self.verify_get(
self.proxy.get_object, obj.Object,
value=["object"],
method_args=["object"],
method_kwargs=kwargs,
expected_kwargs=kwargs)

10
openstack/tests/unit/orchestration/v1/test_proxy.py
View File

@@ -35,9 +35,11 @@ class TestOrchestrationProxy(test_proxy_base.TestProxyBase):
self.verify_create(self.proxy.create_stack, stack.Stack)
def test_create_stack_preview(self):
method_kwargs = {"preview": True, "x": 1, "y": 2, "z": 3}
self.verify_create(self.proxy.create_stack, stack.Stack,
method_kwargs=method_kwargs)
self.verify_create(
self.proxy.create_stack,
stack.Stack,
method_kwargs={"preview": True, "x": 1, "y": 2, "z": 3},
expected_kwargs={"x": 1, "y": 2, "z": 3})
def test_find_stack(self):
self.verify_find(self.proxy.find_stack, stack.Stack,
@@ -223,6 +225,7 @@ class TestOrchestrationProxy(test_proxy_base.TestProxyBase):
self.verify_list(self.proxy.resources, resource.Resource,
method_args=[stk],
expected_args=[],
expected_kwargs={'stack_name': stack_name,
'stack_id': stack_id})
@@ -237,6 +240,7 @@ class TestOrchestrationProxy(test_proxy_base.TestProxyBase):
self.verify_list(self.proxy.resources, resource.Resource,
method_args=[stack_id],
expected_args=[],
expected_kwargs={'stack_name': stack_name,
'stack_id': stack_id})

4
openstack/tests/unit/placement/v1/test_proxy.py
View File

@@ -51,7 +51,7 @@ class TestPlacementProxy(test_proxy_base.TestProxyBase):
)
def test_resource_classes(self):
self.verify_list_no_kwargs(
self.verify_list(
self.proxy.resource_classes,
resource_class.ResourceClass,
)
@@ -85,7 +85,7 @@ class TestPlacementProxy(test_proxy_base.TestProxyBase):
)
def test_resource_providers(self):
self.verify_list_no_kwargs(
self.verify_list(
self.proxy.resource_providers,
resource_provider.ResourceProvider,
)

318
openstack/tests/unit/test_proxy_base.py
View File

@@ -20,14 +20,6 @@ class TestProxyBase(base.TestCase):
super(TestProxyBase, self).setUp()
self.session = mock.Mock()
def _add_path_args_for_verify(self, path_args, method_args,
expected_kwargs, value=None):
if path_args is not None:
if value is None:
for key in path_args:
method_args.append(path_args[key])
expected_kwargs['path_args'] = path_args
def _verify(self, mock_method, test_method,
method_args=None, method_kwargs=None,
expected_args=None, expected_kwargs=None,
@@ -78,80 +70,99 @@ class TestProxyBase(base.TestCase):
**method_kwargs))
# Check how the mock was called in detail
(called_args, called_kwargs) = mocked.call_args
self.assertEqual(list(called_args), expected_args)
base_path = expected_kwargs.get('base_path', None)
self.assertEqual(expected_args, list(called_args))
# NOTE(gtema): if base_path is not in expected_kwargs or empty
# exclude it from the comparison, since some methods might
# still invoke method with None value
if not base_path:
base_path = expected_kwargs.get('base_path', None)
if base_path is None:
expected_kwargs.pop('base_path', None)
called_kwargs.pop('base_path', None)
self.assertDictEqual(called_kwargs, expected_kwargs)
# ditto for paginated
paginated = expected_kwargs.get('paginated', None)
if paginated is None:
expected_kwargs.pop('paginated', None)
called_kwargs.pop('paginated', None)
# and ignore_missing
ignore_missing = expected_kwargs.get('ignore_missing', None)
if ignore_missing is None:
expected_kwargs.pop('ignore_missing', None)
called_kwargs.pop('ignore_missing', None)
self.assertDictEqual(expected_kwargs, called_kwargs)
else:
self.assertEqual(expected_result, test_method())
mocked.assert_called_with(test_method.__self__)
def verify_create(self, test_method, resource_type,
mock_method="openstack.proxy.Proxy._create",
expected_result="result", **kwargs):
the_kwargs = {"x": 1, "y": 2, "z": 3}
method_kwargs = kwargs.pop("method_kwargs", the_kwargs)
expected_args = kwargs.pop('expected_args', [resource_type])
# Default the_kwargs should be copied, since we might need to extend it
expected_kwargs = kwargs.pop("expected_kwargs", the_kwargs.copy())
expected_kwargs["base_path"] = kwargs.pop("base_path", None)
def verify_create(
self, test_method, resource_type, base_path=None, *,
method_args=None, method_kwargs=None,
expected_args=None, expected_kwargs=None, expected_result="result",
mock_method="openstack.proxy.Proxy._create",
):
if method_args is None:
method_args = []
if method_kwargs is None:
method_kwargs = {"x": 1, "y": 2, "z": 3}
if expected_args is None:
expected_args = method_args.copy()
if expected_kwargs is None:
expected_kwargs = method_kwargs.copy()
expected_kwargs["base_path"] = base_path
self._verify2(mock_method, test_method,
expected_result=expected_result,
method_kwargs=method_kwargs,
expected_args=expected_args,
expected_kwargs=expected_kwargs,
**kwargs)
self._verify2(
mock_method,
test_method,
method_args=method_args,
method_kwargs=method_kwargs,
expected_args=[resource_type] + expected_args,
expected_kwargs=expected_kwargs,
expected_result=expected_result,
)
def verify_delete(self, test_method, resource_type, ignore,
input_path_args=None, expected_path_args=None,
method_kwargs=None, expected_args=None,
expected_kwargs=None,
mock_method="openstack.proxy.Proxy._delete"):
method_args = ["resource_or_id"]
def verify_delete(
self, test_method, resource_type, ignore_missing=True, *,
method_args=None, method_kwargs=None,
expected_args=None, expected_kwargs=None,
mock_method="openstack.proxy.Proxy._delete",
):
method_args = method_args or ['resource_id']
method_kwargs = method_kwargs or {}
method_kwargs["ignore_missing"] = ignore
if isinstance(input_path_args, dict):
for key in input_path_args:
method_kwargs[key] = input_path_args[key]
elif isinstance(input_path_args, list):
method_args = input_path_args
expected_kwargs = expected_kwargs or {}
expected_kwargs["ignore_missing"] = ignore
if expected_path_args:
expected_kwargs.update(expected_path_args)
expected_args = expected_args or [resource_type, "resource_or_id"]
self._verify2(mock_method, test_method,
method_args=method_args,
method_kwargs=method_kwargs,
expected_args=expected_args,
expected_kwargs=expected_kwargs)
method_kwargs["ignore_missing"] = ignore_missing
expected_args = expected_args or method_args.copy()
expected_kwargs = expected_kwargs or method_kwargs.copy()
def verify_get(self, test_method, resource_type, value=None, args=None,
mock_method="openstack.proxy.Proxy._get",
ignore_value=False, **kwargs):
the_value = value
if value is None:
the_value = [] if ignore_value else ["value"]
expected_args = kwargs.pop("expected_args", [])
expected_kwargs = kwargs.pop("expected_kwargs", {})
method_kwargs = kwargs.pop("method_kwargs", kwargs)
if args:
expected_kwargs["args"] = args
if kwargs:
expected_kwargs["path_args"] = kwargs
if not expected_args:
expected_args = [resource_type] + the_value
self._verify2(mock_method, test_method,
method_args=the_value,
method_kwargs=method_kwargs or {},
expected_args=expected_args,
expected_kwargs=expected_kwargs)
self._verify2(
mock_method,
test_method,
method_args=method_args,
method_kwargs=method_kwargs,
expected_args=[resource_type] + expected_args,
expected_kwargs=expected_kwargs,
)
def verify_get(
self, test_method, resource_type, requires_id=False, base_path=None, *,
method_args=None, method_kwargs=None,
expected_args=None, expected_kwargs=None,
mock_method="openstack.proxy.Proxy._get",
):
if method_args is None:
method_args = ['resource_id']
if method_kwargs is None:
method_kwargs = {}
if expected_args is None:
expected_args = method_args.copy()
if expected_kwargs is None:
expected_kwargs = method_kwargs.copy()
self._verify2(
mock_method,
test_method,
method_args=method_args,
method_kwargs=method_kwargs,
expected_args=[resource_type] + expected_args,
expected_kwargs=expected_kwargs,
)
def verify_get_overrided(self, proxy, resource_type, patch_target):
with mock.patch(patch_target, autospec=True) as res:
@@ -162,89 +173,106 @@ class TestProxyBase(base.TestCase):
base_path=None,
error_message=mock.ANY)
def verify_head(self, test_method, resource_type,
mock_method="openstack.proxy.Proxy._head",
value=None, **kwargs):
the_value = [value] if value is not None else []
expected_kwargs = {"path_args": kwargs} if kwargs else {}
self._verify2(mock_method, test_method,
method_args=the_value,
method_kwargs=kwargs,
expected_args=[resource_type] + the_value,
expected_kwargs=expected_kwargs)
def verify_head(
self, test_method, resource_type, base_path=None, *,
method_args=None, method_kwargs=None,
expected_args=None, expected_kwargs=None,
mock_method="openstack.proxy.Proxy._head",
):
if method_args is None:
method_args = ['resource_id']
if method_kwargs is None:
method_kwargs = {}
expected_args = expected_args or method_args.copy()
expected_kwargs = expected_kwargs or method_kwargs.copy()
def verify_find(self, test_method, resource_type, value=None,
mock_method="openstack.proxy.Proxy._find",
path_args=None, **kwargs):
method_args = value or ["name_or_id"]
expected_kwargs = kwargs.pop('expected_kwargs', {})
self._verify2(
mock_method,
test_method,
method_args=method_args,
method_kwargs=method_kwargs,
expected_args=[resource_type] + expected_args,
expected_kwargs=expected_kwargs,
)
self._add_path_args_for_verify(path_args, method_args, expected_kwargs,
value=value)
def verify_find(
self, test_method, resource_type, name_or_id='resource_name',
ignore_missing=True, *,
method_args=None, method_kwargs=None,
expected_args=None, expected_kwargs=None,
mock_method="openstack.proxy.Proxy._find",
):
method_args = [name_or_id] + (method_args or [])
method_kwargs = method_kwargs or {}
method_kwargs["ignore_missing"] = ignore_missing
expected_args = expected_args or method_args.copy()
expected_kwargs = expected_kwargs or method_kwargs.copy()
# TODO(briancurtin): if sub-tests worked in this mess of
# test dependencies, the following would be a lot easier to work with.
expected_kwargs["ignore_missing"] = False
self._verify2(mock_method, test_method,
method_args=method_args + [False],
expected_args=[resource_type, "name_or_id"],
expected_kwargs=expected_kwargs,
expected_result="result",
**kwargs)
self._verify2(
mock_method,
test_method,
method_args=method_args,
method_kwargs=method_kwargs,
expected_args=[resource_type] + expected_args,
expected_kwargs=expected_kwargs,
)
expected_kwargs["ignore_missing"] = True
self._verify2(mock_method, test_method,
method_args=method_args + [True],
expected_args=[resource_type, "name_or_id"],
expected_kwargs=expected_kwargs,
expected_result="result",
**kwargs)
def verify_list(
self, test_method, resource_type, paginated=None, base_path=None, *,
method_args=None, method_kwargs=None,
expected_args=None, expected_kwargs=None,
mock_method="openstack.proxy.Proxy._list",
):
if method_args is None:
method_args = []
if method_kwargs is None:
method_kwargs = {}
if paginated is not None:
method_kwargs["paginated"] = paginated
if expected_args is None:
expected_args = method_args.copy()
if expected_kwargs is None:
expected_kwargs = method_kwargs.copy()
if base_path is not None:
expected_kwargs["base_path"] = base_path
def verify_list(self, test_method, resource_type,
mock_method="openstack.proxy.Proxy._list",
**kwargs):
expected_kwargs = kwargs.pop("expected_kwargs", {})
if 'paginated' in kwargs:
expected_kwargs.update({"paginated": kwargs.pop('paginated')})
method_kwargs = kwargs.pop("method_kwargs", {})
expected_kwargs["base_path"] = kwargs.pop("base_path", None)
self._verify2(mock_method, test_method,
method_kwargs=method_kwargs,
expected_args=[resource_type],
expected_kwargs=expected_kwargs,
expected_result=["result"],
**kwargs)
self._verify2(
mock_method,
test_method,
method_args=method_args,
method_kwargs=method_kwargs,
expected_args=[resource_type] + expected_args,
expected_kwargs=expected_kwargs,
)
def verify_list_no_kwargs(self, test_method, resource_type,
mock_method="openstack.proxy.Proxy._list"):
self._verify2(mock_method, test_method,
method_kwargs={},
expected_args=[resource_type],
expected_kwargs={},
expected_result=["result"])
def verify_update(
self, test_method, resource_type, base_path=None, *,
method_args=None, method_kwargs=None,
expected_args=None, expected_kwargs=None, expected_result="result",
mock_method="openstack.proxy.Proxy._update",
):
if method_args is None:
method_args = ['resource_id']
if method_kwargs is None:
method_kwargs = {"x": 1, "y": 2, "z": 3}
method_kwargs["base_path"] = base_path
if expected_args is None:
expected_args = method_args.copy()
if expected_kwargs is None:
expected_kwargs = method_kwargs.copy()
def verify_update(self, test_method, resource_type, value=None,
mock_method="openstack.proxy.Proxy._update",
expected_result="result", path_args=None, **kwargs):
method_args = value or ["resource_or_id"]
method_kwargs = kwargs.pop("method_kwargs", {})
method_kwargs.update({"x": 1, "y": 2, "z": 3})
expected_args = kwargs.pop("expected_args", ["resource_or_id"])
expected_kwargs = kwargs.pop("expected_kwargs", method_kwargs.copy())
expected_kwargs["base_path"] = kwargs.pop("base_path", None)
self._add_path_args_for_verify(path_args, method_args, expected_kwargs,
value=value)
self._verify2(mock_method, test_method,
expected_result=expected_result,
method_args=method_args,
method_kwargs=method_kwargs,
expected_args=[resource_type] + expected_args,
expected_kwargs=expected_kwargs,
**kwargs)
self._verify2(
mock_method,
test_method,
method_args=method_args,
method_kwargs=method_kwargs,
expected_args=[resource_type] + expected_args,
expected_kwargs=expected_kwargs,
)
def verify_wait_for_status(
self, test_method,
mock_method="openstack.resource.wait_for_status", **kwargs):
self, test_method,
mock_method="openstack.resource.wait_for_status",
**kwargs,
):
self._verify(mock_method, test_method, **kwargs)

0
openstack/tests/unit/workflow/v2/__init__.py Normal file
View File

0
openstack/tests/unit/workflow/test_proxy.py → openstack/tests/unit/workflow/v2/test_proxy.py
View File