openstack
/
openstacksdk
Code Issues Proposed changes

tests: Rename 'TestProxyBase._verify2' to '_verify' 

Change-Id: I9d055751d631fa8c951f57d680448f9df26061a1
Signed-off-by: Stephen Finucane <stephenfin@redhat.com>
This commit is contained in:
Stephen Finucane 2021-04-26 12:19:12 +01:00 committed by Artem Goncharov
parent 4feb26d46f
commit 2e7b48ff4e
13 changed files with 610 additions and 598 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
openstack/tests/unit/block_storage/v2/test_proxy.py
View File

@ -88,7 +88,7 @@ class TestVolumeProxy(test_proxy_base.TestProxyBase):
self.verify_delete(self.proxy.delete_volume, volume.Volume, True)
def test_volume_extend(self):
self._verify2(
self._verify(
"openstack.block_storage.v2.volume.Volume.extend",
self.proxy.extend_volume,
method_args=["value", "new-size"],
@ -142,7 +142,7 @@ class TestVolumeProxy(test_proxy_base.TestProxyBase):
# NOTE: mock has_service
self.proxy._connection = mock.Mock()
self.proxy._connection.has_service = mock.Mock(return_value=True)
self._verify2(
self._verify(
'openstack.block_storage.v2.backup.Backup.restore',
self.proxy.restore_backup,
method_args=['volume_id'],

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

@ -82,7 +82,7 @@ class TestVolumeProxy(test_proxy_base.TestProxyBase):
def test_type_extra_specs_update(self):
kwargs = {"a": "1", "b": "2"}
id = "an_id"
self._verify2(
self._verify(
"openstack.block_storage.v3.type.Type.set_extra_specs",
self.proxy.update_type_extra_specs,
method_args=[id],
@ -94,7 +94,7 @@ class TestVolumeProxy(test_proxy_base.TestProxyBase):
expected_result=kwargs)
def test_type_extra_specs_delete(self):
self._verify2(
self._verify(
"openstack.block_storage.v3.type.Type.delete_extra_specs",
self.proxy.delete_type_extra_specs,
expected_result=None,
@ -156,21 +156,21 @@ class TestVolumeProxy(test_proxy_base.TestProxyBase):
self.verify_delete(self.proxy.delete_volume, volume.Volume, True)
def test_volume_extend(self):
self._verify2(
self._verify(
"openstack.block_storage.v3.volume.Volume.extend",
self.proxy.extend_volume,
method_args=["value", "new-size"],
expected_args=[self.proxy, "new-size"])
def test_volume_set_readonly_no_argument(self):
self._verify2(
self._verify(
"openstack.block_storage.v3.volume.Volume.set_readonly",
self.proxy.set_volume_readonly,
method_args=["value"],
expected_args=[self.proxy, True])
def test_volume_set_readonly_false(self):
self._verify2(
self._verify(
"openstack.block_storage.v3.volume.Volume.set_readonly",
self.proxy.set_volume_readonly,
method_args=["value", False],
@ -180,13 +180,13 @@ class TestVolumeProxy(test_proxy_base.TestProxyBase):
self._verify("openstack.block_storage.v3.volume.Volume.retype",
self.proxy.retype_volume,
method_args=["value", "rbd"],
expected_args=["rbd", "never"])
expected_args=[self.proxy, "rbd", "never"])
def test_volume_retype_with_migration_policy(self):
self._verify("openstack.block_storage.v3.volume.Volume.retype",
self.proxy.retype_volume,
method_args=["value", "rbd", "on-demand"],
expected_args=["rbd", "on-demand"])
expected_args=[self.proxy, "rbd", "on-demand"])
def test_backend_pools(self):
self.verify_list(self.proxy.backend_pools, stats.Pools)
@ -242,7 +242,7 @@ class TestVolumeProxy(test_proxy_base.TestProxyBase):
# NOTE: mock has_service
self.proxy._connection = mock.Mock()
self.proxy._connection.has_service = mock.Mock(return_value=True)
self._verify2(
self._verify(
'openstack.block_storage.v3.backup.Backup.restore',
self.proxy.restore_backup,
method_args=['volume_id'],

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

@ -91,7 +91,7 @@ class TestClusterProxy(test_proxy_base.TestProxyBase):
self.verify_delete(self.proxy.delete_cluster, cluster.Cluster, True)
def test_cluster_force_delete(self):
self._verify2(
self._verify(
"openstack.clustering.v1.cluster.Cluster.force_delete",
self.proxy.delete_cluster,
method_args=["value", False, True],
@ -119,7 +119,7 @@ class TestClusterProxy(test_proxy_base.TestProxyBase):
def test_resize_cluster(self, mock_find):
mock_cluster = cluster.Cluster.new(id='FAKE_CLUSTER')
mock_find.return_value = mock_cluster
self._verify2(
self._verify(
"openstack.clustering.v1.cluster.Cluster.resize",
self.proxy.resize_cluster,
method_args=["FAKE_CLUSTER"],
@ -131,7 +131,7 @@ class TestClusterProxy(test_proxy_base.TestProxyBase):
def test_resize_cluster_with_obj(self):
mock_cluster = cluster.Cluster.new(id='FAKE_CLUSTER')
self._verify2(
self._verify(
"openstack.clustering.v1.cluster.Cluster.resize",
self.proxy.resize_cluster,
method_args=[mock_cluster],
@ -151,7 +151,7 @@ class TestClusterProxy(test_proxy_base.TestProxyBase):
def test_cluster_check(self, mock_get):
mock_cluster = cluster.Cluster.new(id='FAKE_CLUSTER')
mock_get.return_value = mock_cluster
self._verify2(
self._verify(
"openstack.clustering.v1.cluster.Cluster.check",
self.proxy.check_cluster,
method_args=["FAKE_CLUSTER"],
@ -162,7 +162,7 @@ class TestClusterProxy(test_proxy_base.TestProxyBase):
def test_cluster_recover(self, mock_get):
mock_cluster = cluster.Cluster.new(id='FAKE_CLUSTER')
mock_get.return_value = mock_cluster
self._verify2(
self._verify(
"openstack.clustering.v1.cluster.Cluster.recover",
self.proxy.recover_cluster,
method_args=["FAKE_CLUSTER"],
@ -179,7 +179,7 @@ class TestClusterProxy(test_proxy_base.TestProxyBase):
self.verify_delete(self.proxy.delete_node, node.Node, True)
def test_node_force_delete(self):
self._verify2(
self._verify(
"openstack.clustering.v1.node.Node.force_delete",
self.proxy.delete_node,
method_args=["value", False, True],
@ -192,13 +192,13 @@ class TestClusterProxy(test_proxy_base.TestProxyBase):
self.verify_get(self.proxy.get_node, node.Node)
def test_node_get_with_details(self):
self._verify2('openstack.proxy.Proxy._get',
self.proxy.get_node,
method_args=['NODE_ID'],
method_kwargs={'details': True},
expected_args=[node.NodeDetail],
expected_kwargs={'node_id': 'NODE_ID',
'requires_id': False})
self._verify(
'openstack.proxy.Proxy._get',
self.proxy.get_node,
method_args=['NODE_ID'],
method_kwargs={'details': True},
expected_args=[node.NodeDetail],
expected_kwargs={'node_id': 'NODE_ID', 'requires_id': False})
def test_nodes(self):
self.verify_list(self.proxy.nodes, node.Node,
@ -212,7 +212,7 @@ class TestClusterProxy(test_proxy_base.TestProxyBase):
def test_node_check(self, mock_get):
mock_node = node.Node.new(id='FAKE_NODE')
mock_get.return_value = mock_node
self._verify2(
self._verify(
"openstack.clustering.v1.node.Node.check",
self.proxy.check_node,
method_args=["FAKE_NODE"],
@ -223,7 +223,7 @@ class TestClusterProxy(test_proxy_base.TestProxyBase):
def test_node_recover(self, mock_get):
mock_node = node.Node.new(id='FAKE_NODE')
mock_get.return_value = mock_node
self._verify2(
self._verify(
"openstack.clustering.v1.node.Node.recover",
self.proxy.recover_node,
method_args=["FAKE_NODE"],
@ -234,7 +234,7 @@ class TestClusterProxy(test_proxy_base.TestProxyBase):
def test_node_adopt(self, mock_get):
mock_node = node.Node.new()
mock_get.return_value = mock_node
self._verify2(
self._verify(
"openstack.clustering.v1.node.Node.adopt",
self.proxy.adopt_node,
method_kwargs={"preview": False, "foo": "bar"},
@ -247,7 +247,7 @@ class TestClusterProxy(test_proxy_base.TestProxyBase):
def test_node_adopt_preview(self, mock_get):
mock_node = node.Node.new()
mock_get.return_value = mock_node
self._verify2(
self._verify(
"openstack.clustering.v1.node.Node.adopt",
self.proxy.adopt_node,
method_kwargs={"preview": True, "foo": "bar"},
@ -294,29 +294,29 @@ class TestClusterProxy(test_proxy_base.TestProxyBase):
fake_cluster = cluster.Cluster.new(id='FAKE_CLUSTER')
# ClusterPolicy object as input
self._verify2('openstack.proxy.Proxy._get',
self.proxy.get_cluster_policy,
method_args=[fake_policy, "FAKE_CLUSTER"],
expected_args=[cluster_policy.ClusterPolicy,
fake_policy],
expected_kwargs={'cluster_id': 'FAKE_CLUSTER'},
expected_result=fake_policy)
self._verify(
'openstack.proxy.Proxy._get',
self.proxy.get_cluster_policy,
method_args=[fake_policy, "FAKE_CLUSTER"],
expected_args=[cluster_policy.ClusterPolicy, fake_policy],
expected_kwargs={'cluster_id': 'FAKE_CLUSTER'},
expected_result=fake_policy)
# Policy ID as input
self._verify2('openstack.proxy.Proxy._get',
self.proxy.get_cluster_policy,
method_args=["FAKE_POLICY", "FAKE_CLUSTER"],
expected_args=[cluster_policy.ClusterPolicy,
"FAKE_POLICY"],
expected_kwargs={"cluster_id": "FAKE_CLUSTER"})
self._verify(
'openstack.proxy.Proxy._get',
self.proxy.get_cluster_policy,
method_args=["FAKE_POLICY", "FAKE_CLUSTER"],
expected_args=[cluster_policy.ClusterPolicy, "FAKE_POLICY"],
expected_kwargs={"cluster_id": "FAKE_CLUSTER"})
# Cluster object as input
self._verify2('openstack.proxy.Proxy._get',
self.proxy.get_cluster_policy,
method_args=["FAKE_POLICY", fake_cluster],
expected_args=[cluster_policy.ClusterPolicy,
"FAKE_POLICY"],
expected_kwargs={"cluster_id": fake_cluster})
self._verify(
'openstack.proxy.Proxy._get',
self.proxy.get_cluster_policy,
method_args=["FAKE_POLICY", fake_cluster],
expected_args=[cluster_policy.ClusterPolicy, "FAKE_POLICY"],
expected_kwargs={"cluster_id": fake_cluster})
def test_receiver_create(self):
self.verify_create(self.proxy.create_receiver, receiver.Receiver)

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

@ -66,7 +66,7 @@ class TestFlavor(TestComputeProxy):
) as mocked:
res = flavor.Flavor()
mocked.return_value = res
self._verify2(
self._verify(
'openstack.proxy.Proxy._find',
self.proxy.find_flavor,
method_args=['res', True, True],
@ -83,7 +83,7 @@ class TestFlavor(TestComputeProxy):
) as mocked:
res = flavor.Flavor(extra_specs={'a': 'b'})
mocked.return_value = res
self._verify2(
self._verify(
'openstack.proxy.Proxy._find',
self.proxy.find_flavor,
method_args=['res', True],
@ -100,7 +100,7 @@ class TestFlavor(TestComputeProxy):
) as mocked:
res = flavor.Flavor()
mocked.return_value = res
self._verify2(
self._verify(
'openstack.proxy.Proxy._get',
self.proxy.get_flavor,
method_args=['res'],
@ -116,7 +116,7 @@ class TestFlavor(TestComputeProxy):
) as mocked:
res = flavor.Flavor()
mocked.return_value = res
self._verify2(
self._verify(
'openstack.proxy.Proxy._get',
self.proxy.get_flavor,
method_args=['res', True],
@ -132,7 +132,7 @@ class TestFlavor(TestComputeProxy):
) as mocked:
res = flavor.Flavor(extra_specs={'a': 'b'})
mocked.return_value = res
self._verify2(
self._verify(
'openstack.proxy.Proxy._get',
self.proxy.get_flavor,
method_args=['res', True],
@ -193,35 +193,35 @@ class TestFlavor(TestComputeProxy):
)
def test_flavor_get_access(self):
self._verify2(
self._verify(
"openstack.compute.v2.flavor.Flavor.get_access",
self.proxy.get_flavor_access,
method_args=["value"],
expected_args=[self.proxy])
def test_flavor_add_tenant_access(self):
self._verify2(
self._verify(
"openstack.compute.v2.flavor.Flavor.add_tenant_access",
self.proxy.flavor_add_tenant_access,
method_args=["value", "fake-tenant"],
expected_args=[self.proxy, "fake-tenant"])
def test_flavor_remove_tenant_access(self):
self._verify2(
self._verify(
"openstack.compute.v2.flavor.Flavor.remove_tenant_access",
self.proxy.flavor_remove_tenant_access,
method_args=["value", "fake-tenant"],
expected_args=[self.proxy, "fake-tenant"])
def test_flavor_fetch_extra_specs(self):
self._verify2(
self._verify(
"openstack.compute.v2.flavor.Flavor.fetch_extra_specs",
self.proxy.fetch_flavor_extra_specs,
method_args=["value"],
expected_args=[self.proxy])
def test_create_flavor_extra_specs(self):
self._verify2(
self._verify(
"openstack.compute.v2.flavor.Flavor.create_extra_specs",
self.proxy.create_flavor_extra_specs,
method_args=["value", {'a': 'b'}],
@ -229,21 +229,21 @@ class TestFlavor(TestComputeProxy):
expected_kwargs={"specs": {'a': 'b'}})
def test_get_flavor_extra_specs_prop(self):
self._verify2(
self._verify(
"openstack.compute.v2.flavor.Flavor.get_extra_specs_property",
self.proxy.get_flavor_extra_specs_property,
method_args=["value", "prop"],
expected_args=[self.proxy, "prop"])
def test_update_flavor_extra_specs_prop(self):
self._verify2(
self._verify(
"openstack.compute.v2.flavor.Flavor.update_extra_specs_property",
self.proxy.update_flavor_extra_specs_property,
method_args=["value", "prop", "val"],
expected_args=[self.proxy, "prop", "val"])
def test_delete_flavor_extra_specs_prop(self):
self._verify2(
self._verify(
"openstack.compute.v2.flavor.Flavor.delete_extra_specs_property",
self.proxy.delete_flavor_extra_specs_property,
method_args=["value", "prop"],
@ -324,35 +324,35 @@ class TestAggregate(TestComputeProxy):
self.verify_update(self.proxy.update_aggregate, aggregate.Aggregate)
def test_aggregate_add_host(self):
self._verify2(
self._verify(
"openstack.compute.v2.aggregate.Aggregate.add_host",
self.proxy.add_host_to_aggregate,
method_args=["value", "host"],
expected_args=[self.proxy, "host"])
def test_aggregate_remove_host(self):
self._verify2(
self._verify(
"openstack.compute.v2.aggregate.Aggregate.remove_host",
self.proxy.remove_host_from_aggregate,
method_args=["value", "host"],
expected_args=[self.proxy, "host"])
def test_aggregate_set_metadata(self):
self._verify2(
self._verify(
"openstack.compute.v2.aggregate.Aggregate.set_metadata",
self.proxy.set_aggregate_metadata,
method_args=["value", {'a': 'b'}],
expected_args=[self.proxy, {'a': 'b'}])
def test_aggregate_precache_image(self):
self._verify2(
self._verify(
"openstack.compute.v2.aggregate.Aggregate.precache_images",
self.proxy.aggregate_precache_images,
method_args=["value", '1'],
expected_args=[self.proxy, [{'id': '1'}]])
def test_aggregate_precache_images(self):
self._verify2(
self._verify(
"openstack.compute.v2.aggregate.Aggregate.precache_images",
self.proxy.aggregate_precache_images,
method_args=["value", ['1', '2']],
@ -366,7 +366,7 @@ class TestService(TestComputeProxy):
@mock.patch('openstack.utils.supports_microversion', autospec=True,
return_value=False)
def test_enable_service_252(self, mv_mock):
self._verify2(
self._verify(
'openstack.compute.v2.service.Service.enable',
self.proxy.enable_service,
method_args=["value", "host1", "nova-compute"],
@ -376,7 +376,7 @@ class TestService(TestComputeProxy):
@mock.patch('openstack.utils.supports_microversion', autospec=True,
return_value=True)
def test_enable_service_253(self, mv_mock):
self._verify2(
self._verify(
'openstack.proxy.Proxy._update',
self.proxy.enable_service,
method_args=["value"],
@ -388,7 +388,7 @@ class TestService(TestComputeProxy):
@mock.patch('openstack.utils.supports_microversion', autospec=True,
return_value=False)
def test_disable_service_252(self, mv_mock):
self._verify2(
self._verify(
'openstack.compute.v2.service.Service.disable',
self.proxy.disable_service,
method_args=["value", "host1", "nova-compute"],
@ -397,7 +397,7 @@ class TestService(TestComputeProxy):
@mock.patch('openstack.utils.supports_microversion', autospec=True,
return_value=True)
def test_disable_service_253(self, mv_mock):
self._verify2(
self._verify(
'openstack.proxy.Proxy._update',
self.proxy.disable_service,
method_args=["value"],
@ -412,7 +412,7 @@ class TestService(TestComputeProxy):
@mock.patch('openstack.utils.supports_microversion', autospec=True,
return_value=False)
def test_force_service_down_252(self, mv_mock):
self._verify2(
self._verify(
'openstack.compute.v2.service.Service.set_forced_down',
self.proxy.update_service_forced_down,
method_args=["value", "host1", "nova-compute"],
@ -430,7 +430,7 @@ class TestService(TestComputeProxy):
@mock.patch('openstack.utils.supports_microversion', autospec=True,
return_value=False)
def test_force_service_down_252_empty_vals_svc(self, mv_mock):
self._verify2(
self._verify(
'openstack.compute.v2.service.Service.set_forced_down',
self.proxy.update_service_forced_down,
method_args=[{'host': 'a', 'binary': 'b'}, None, None],
@ -516,7 +516,7 @@ class TestHypervisor(TestComputeProxy):
hypervisor.Hypervisor)
def test_get_hypervisor_uptime(self):
self._verify2(
self._verify(
"openstack.compute.v2.hypervisor.Hypervisor.get_uptime",
self.proxy.get_hypervisor_uptime,
method_args=["value"],
@ -570,24 +570,22 @@ class TestCompute(TestComputeProxy):
test_interface.server_id = server_id
# Case1: ServerInterface instance is provided as value
self._verify2("openstack.proxy.Proxy._delete",
self.proxy.delete_server_interface,
method_args=[test_interface],
method_kwargs={"server": server_id},
expected_args=[
server_interface.ServerInterface, interface_id],
expected_kwargs={"server_id": server_id,
"ignore_missing": True})
self._verify(
"openstack.proxy.Proxy._delete",
self.proxy.delete_server_interface,
method_args=[test_interface],
method_kwargs={"server": server_id},
expected_args=[server_interface.ServerInterface, interface_id],
expected_kwargs={"server_id": server_id, "ignore_missing": True})
# Case2: ServerInterface ID is provided as value
self._verify2("openstack.proxy.Proxy._delete",
self.proxy.delete_server_interface,
method_args=[interface_id],
method_kwargs={"server": server_id},
expected_args=[
server_interface.ServerInterface, interface_id],
expected_kwargs={"server_id": server_id,
"ignore_missing": True})
self._verify(
"openstack.proxy.Proxy._delete",
self.proxy.delete_server_interface,
method_args=[interface_id],
method_kwargs={"server": server_id},
expected_args=[server_interface.ServerInterface, interface_id],
expected_kwargs={"server_id": server_id, "ignore_missing": True})
def test_server_interface_delete_ignore(self):
self.proxy._get_uri_attribute = lambda *args: args[1]
@ -606,22 +604,22 @@ class TestCompute(TestComputeProxy):
test_interface.server_id = server_id
# Case1: ServerInterface instance is provided as value
self._verify2('openstack.proxy.Proxy._get',
self.proxy.get_server_interface,
method_args=[test_interface],
method_kwargs={"server": server_id},
expected_args=[server_interface.ServerInterface],
expected_kwargs={"port_id": interface_id,
"server_id": server_id})
self._verify(
'openstack.proxy.Proxy._get',
self.proxy.get_server_interface,
method_args=[test_interface],
method_kwargs={"server": server_id},
expected_args=[server_interface.ServerInterface],
expected_kwargs={"port_id": interface_id, "server_id": server_id})
# Case2: ServerInterface ID is provided as value
self._verify2('openstack.proxy.Proxy._get',
self.proxy.get_server_interface,
method_args=[interface_id],
method_kwargs={"server": server_id},
expected_args=[server_interface.ServerInterface],
expected_kwargs={"port_id": interface_id,
"server_id": server_id})
self._verify(
'openstack.proxy.Proxy._get',
self.proxy.get_server_interface,
method_args=[interface_id],
method_kwargs={"server": server_id},
expected_args=[server_interface.ServerInterface],
expected_kwargs={"port_id": interface_id, "server_id": server_id})
def test_server_interfaces(self):
self.verify_list(self.proxy.server_interfaces,
@ -655,7 +653,7 @@ class TestCompute(TestComputeProxy):
self.verify_delete(self.proxy.delete_server, server.Server, True)
def test_server_force_delete(self):
self._verify2(
self._verify(
"openstack.compute.v2.server.Server.force_delete",
self.proxy.delete_server,
method_args=["value", False, True],
@ -691,21 +689,21 @@ class TestCompute(TestComputeProxy):
expected_args=[self.proxy, value, 'ACTIVE', ['ERROR'], 2, 120])
def test_server_resize(self):
self._verify2(
self._verify(
"openstack.compute.v2.server.Server.resize",
self.proxy.resize_server,
method_args=["value", "test-flavor"],
expected_args=[self.proxy, "test-flavor"])
def test_server_confirm_resize(self):
self._verify2(
self._verify(
"openstack.compute.v2.server.Server.confirm_resize",
self.proxy.confirm_server_resize,
method_args=["value"],
expected_args=[self.proxy])
def test_server_revert_resize(self):
self._verify2(
self._verify(
"openstack.compute.v2.server.Server.revert_resize",
self.proxy.revert_server_resize,
method_args=["value"],
@ -718,7 +716,7 @@ class TestCompute(TestComputeProxy):
# Case1: image object is provided
# NOTE: Inside of Server.rebuild is where image_obj gets converted
# to an ID instead of object.
self._verify2(
self._verify(
'openstack.compute.v2.server.Server.rebuild',
self.proxy.rebuild_server,
method_args=["value", "test_server", "test_pass"],
@ -727,7 +725,7 @@ class TestCompute(TestComputeProxy):
expected_kwargs={"metadata": {"k1": "v1"}, "image": image_obj})
# Case2: image name or id is provided
self._verify2(
self._verify(
'openstack.compute.v2.server.Server.rebuild',
self.proxy.rebuild_server,
method_args=["value", "test_server", "test_pass"],
@ -736,21 +734,21 @@ class TestCompute(TestComputeProxy):
expected_kwargs={"metadata": {"k1": "v1"}, "image": id})
def test_add_fixed_ip_to_server(self):
self._verify2(
self._verify(
"openstack.compute.v2.server.Server.add_fixed_ip",
self.proxy.add_fixed_ip_to_server,
method_args=["value", "network-id"],
expected_args=[self.proxy, "network-id"])
def test_fixed_ip_from_server(self):
self._verify2(
self._verify(
"openstack.compute.v2.server.Server.remove_fixed_ip",
self.proxy.remove_fixed_ip_from_server,
method_args=["value", "address"],
expected_args=[self.proxy, "address"])
def test_floating_ip_to_server(self):
self._verify2(
self._verify(
"openstack.compute.v2.server.Server.add_floating_ip",
self.proxy.add_floating_ip_to_server,
method_args=["value", "floating-ip"],
@ -758,7 +756,7 @@ class TestCompute(TestComputeProxy):
expected_kwargs={'fixed_address': None})
def test_add_floating_ip_to_server_with_fixed_addr(self):
self._verify2(
self._verify(
"openstack.compute.v2.server.Server.add_floating_ip",
self.proxy.add_floating_ip_to_server,
method_args=["value", "floating-ip", 'fixed-addr'],
@ -766,63 +764,63 @@ class TestCompute(TestComputeProxy):
expected_kwargs={'fixed_address': 'fixed-addr'})
def test_remove_floating_ip_from_server(self):
self._verify2(
self._verify(
"openstack.compute.v2.server.Server.remove_floating_ip",
self.proxy.remove_floating_ip_from_server,
method_args=["value", "address"],
expected_args=[self.proxy, "address"])
def test_server_backup(self):
self._verify2(
self._verify(
"openstack.compute.v2.server.Server.backup",
self.proxy.backup_server,
method_args=["value", "name", "daily", 1],
expected_args=[self.proxy, "name", "daily", 1])
def test_server_pause(self):
self._verify2(
self._verify(
"openstack.compute.v2.server.Server.pause",
self.proxy.pause_server,
method_args=["value"],
expected_args=[self.proxy])
def test_server_unpause(self):
self._verify2(
self._verify(
"openstack.compute.v2.server.Server.unpause",
self.proxy.unpause_server,
method_args=["value"],
expected_args=[self.proxy])
def test_server_suspend(self):
self._verify2(
self._verify(
"openstack.compute.v2.server.Server.suspend",
self.proxy.suspend_server,
method_args=["value"],
expected_args=[self.proxy])
def test_server_resume(self):
self._verify2(
self._verify(
"openstack.compute.v2.server.Server.resume",
self.proxy.resume_server,
method_args=["value"],
expected_args=[self.proxy])
def test_server_lock(self):
self._verify2(
self._verify(
"openstack.compute.v2.server.Server.lock",
self.proxy.lock_server,
method_args=["value"],
expected_args=[self.proxy])
def test_server_unlock(self):
self._verify2(
self._verify(
"openstack.compute.v2.server.Server.unlock",
self.proxy.unlock_server,
method_args=["value"],
expected_args=[self.proxy])
def test_server_rescue(self):
self._verify2(
self._verify(
"openstack.compute.v2.server.Server.rescue",
self.proxy.rescue_server,
method_args=["value"],
@ -830,7 +828,7 @@ class TestCompute(TestComputeProxy):
expected_kwargs={"admin_pass": None, "image_ref": None})
def test_server_rescue_with_options(self):
self._verify2(
self._verify(
"openstack.compute.v2.server.Server.rescue",
self.proxy.rescue_server,
method_args=["value", 'PASS', 'IMG'],
@ -838,14 +836,14 @@ class TestCompute(TestComputeProxy):
expected_kwargs={"admin_pass": 'PASS', "image_ref": 'IMG'})
def test_server_unrescue(self):
self._verify2(
self._verify(
"openstack.compute.v2.server.Server.unrescue",
self.proxy.unrescue_server,
method_args=["value"],
expected_args=[self.proxy])
def test_server_evacuate(self):
self._verify2(
self._verify(
"openstack.compute.v2.server.Server.evacuate",
self.proxy.evacuate_server,
method_args=["value"],
@ -853,7 +851,7 @@ class TestCompute(TestComputeProxy):
expected_kwargs={"host": None, "admin_pass": None, "force": None})
def test_server_evacuate_with_options(self):
self._verify2(
self._verify(
"openstack.compute.v2.server.Server.evacuate",
self.proxy.evacuate_server,
method_args=["value", 'HOST2', 'NEW_PASS', True],
@ -862,42 +860,42 @@ class TestCompute(TestComputeProxy):
"host": "HOST2", "admin_pass": 'NEW_PASS', "force": True})
def test_server_start(self):
self._verify2(
self._verify(
"openstack.compute.v2.server.Server.start",
self.proxy.start_server,
method_args=["value"],
expected_args=[self.proxy])
def test_server_stop(self):
self._verify2(
self._verify(
"openstack.compute.v2.server.Server.stop",
self.proxy.stop_server,
method_args=["value"],
expected_args=[self.proxy])
def test_server_shelve(self):
self._verify2(
self._verify(
"openstack.compute.v2.server.Server.shelve",
self.proxy.shelve_server,
method_args=["value"],
expected_args=[self.proxy])
def test_server_unshelve(self):
self._verify2(
self._verify(
"openstack.compute.v2.server.Server.unshelve",
self.proxy.unshelve_server,
method_args=["value"],
expected_args=[self.proxy])
def test_get_server_output(self):
self._verify2(
self._verify(
"openstack.compute.v2.server.Server.get_console_output",
self.proxy.get_server_console_output,
method_args=["value"],
expected_args=[self.proxy],
expected_kwargs={"length": None})
self._verify2(
self._verify(
"openstack.compute.v2.server.Server.get_console_output",
self.proxy.get_server_console_output,
method_args=["value", 1],
@ -917,32 +915,34 @@ class TestCompute(TestComputeProxy):
expected_kwargs={})
def test_get_all_server_metadata(self):
self._verify2("openstack.compute.v2.server.Server.get_metadata",
self.proxy.get_server_metadata,
method_args=["value"],
method_result=server.Server(id="value", metadata={}),
expected_args=[self.proxy],
expected_result={})
self._verify(
"openstack.compute.v2.server.Server.get_metadata",
self.proxy.get_server_metadata,
method_args=["value"],
method_result=server.Server(id="value", metadata={}),
expected_args=[self.proxy],
expected_result={})
def test_set_server_metadata(self):
kwargs = {"a": "1", "b": "2"}
id = "an_id"
self._verify2("openstack.compute.v2.server.Server.set_metadata",
self.proxy.set_server_metadata,
method_args=[id],
method_kwargs=kwargs,
method_result=server.Server.existing(id=id,
metadata=kwargs),
expected_args=[self.proxy],
expected_kwargs=kwargs,
expected_result=kwargs)
self._verify(
"openstack.compute.v2.server.Server.set_metadata",
self.proxy.set_server_metadata,
method_args=[id],
method_kwargs=kwargs,
method_result=server.Server.existing(id=id, metadata=kwargs),
expected_args=[self.proxy],
expected_kwargs=kwargs,
expected_result=kwargs)
def test_delete_server_metadata(self):
self._verify2("openstack.compute.v2.server.Server.delete_metadata",
self.proxy.delete_server_metadata,
expected_result=None,
method_args=["value", "key"],
expected_args=[self.proxy, "key"])
self._verify(
"openstack.compute.v2.server.Server.delete_metadata",
self.proxy.delete_server_metadata,
expected_result=None,
method_args=["value", "key"],
expected_args=[self.proxy, "key"])
def test_create_image(self):
metadata = {'k1': 'v1'}
@ -995,7 +995,7 @@ class TestCompute(TestComputeProxy):
self.verify_list(self.proxy.server_groups, server_group.ServerGroup)
def test_live_migrate_server(self):
self._verify2(
self._verify(
'openstack.compute.v2.server.Server.live_migrate',
self.proxy.live_migrate_server,
method_args=["value", "host1", False],
@ -1003,21 +1003,21 @@ class TestCompute(TestComputeProxy):
expected_kwargs={'force': False, 'block_migration': None})
def test_fetch_security_groups(self):
self._verify2(
self._verify(
'openstack.compute.v2.server.Server.fetch_security_groups',
self.proxy.fetch_server_security_groups,
method_args=["value"],
expected_args=[self.proxy])
def test_add_security_groups(self):
self._verify2(
self._verify(
'openstack.compute.v2.server.Server.add_security_group',
self.proxy.add_security_group_to_server,
method_args=["value", {'id': 'id', 'name': 'sg'}],
expected_args=[self.proxy, 'sg'])
def test_remove_security_groups(self):
self._verify2(
self._verify(
'openstack.compute.v2.server.Server.remove_security_group',
self.proxy.remove_security_group_from_server,
method_args=["value", {'id': 'id', 'name': 'sg'}],
@ -1031,7 +1031,7 @@ class TestCompute(TestComputeProxy):
expected_kwargs={"server_id": "test_id", "type": "fake"})
def test_get_console_url(self):
self._verify2(
self._verify(
'openstack.compute.v2.server.Server.get_console_url',
self.proxy.get_server_console_url,
method_args=["value", "console_type"],

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

@ -44,12 +44,13 @@ class TestDatabaseProxy(test_proxy_base.TestProxyBase):
expected_kwargs={"instance_id": "test_id"})
def test_database_find(self):
self._verify2('openstack.proxy.Proxy._find',
self.proxy.find_database,
method_args=["db", "instance"],
expected_args=[database.Database, "db"],
expected_kwargs={"instance_id": "instance",
"ignore_missing": True})
self._verify(
'openstack.proxy.Proxy._find',
self.proxy.find_database,
method_args=["db", "instance"],
expected_args=[database.Database, "db"],
expected_kwargs={
"instance_id": "instance", "ignore_missing": True})
def test_databases(self):
self.verify_list(self.proxy.databases, database.Database,
@ -108,12 +109,13 @@ class TestDatabaseProxy(test_proxy_base.TestProxyBase):
expected_kwargs={"instance_id": "id"})
def test_user_find(self):
self._verify2('openstack.proxy.Proxy._find',
self.proxy.find_user,
method_args=["user", "instance"],
expected_args=[user.User, "user"],
expected_kwargs={"instance_id": "instance",
"ignore_missing": True})
self._verify(
'openstack.proxy.Proxy._find',
self.proxy.find_user,
method_args=["user", "instance"],
expected_args=[user.User, "user"],
expected_kwargs={
"instance_id": "instance", "ignore_missing": True})
def test_users(self):
self.verify_list(self.proxy.users, user.User,

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

@ -50,16 +50,18 @@ class TestDnsZone(TestDnsProxy):
self.verify_update(self.proxy.update_zone, zone.Zone)
def test_zone_abandon(self):
self._verify2("openstack.dns.v2.zone.Zone.abandon",
self.proxy.abandon_zone,
method_args=[{'zone': 'id'}],
expected_args=[self.proxy])
self._verify(
"openstack.dns.v2.zone.Zone.abandon",
self.proxy.abandon_zone,
method_args=[{'zone': 'id'}],
expected_args=[self.proxy])
def test_zone_xfr(self):
self._verify2("openstack.dns.v2.zone.Zone.xfr",
self.proxy.xfr_zone,
method_args=[{'zone': 'id'}],
expected_args=[self.proxy])
self._verify(
"openstack.dns.v2.zone.Zone.xfr",
self.proxy.xfr_zone,
method_args=[{'zone': 'id'}],
expected_args=[self.proxy])
class TestDnsRecordset(TestDnsProxy):
@ -92,13 +94,13 @@ class TestDnsRecordset(TestDnsProxy):
expected_kwargs={'zone_id': 'zid'})
def test_recordset_find(self):
self._verify2("openstack.proxy.Proxy._find",
self.proxy.find_recordset,
method_args=['zone', 'rs'],
method_kwargs={},
expected_args=[recordset.Recordset, 'rs'],
expected_kwargs={'ignore_missing': True,
'zone_id': 'zone'})
self._verify(
"openstack.proxy.Proxy._find",
self.proxy.find_recordset,
method_args=['zone', 'rs'],
method_kwargs={},
expected_args=[recordset.Recordset, 'rs'],
expected_kwargs={'ignore_missing': True, 'zone_id': 'zone'})
class TestDnsFloatIP(TestDnsProxy):

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

@ -54,7 +54,7 @@ class TestImageProxy(test_proxy_base.TestProxyBase):
def test_image_import(self):
original_image = image.Image(**EXAMPLE)
self._verify2(
self._verify(
"openstack.image.v2.image.Image.import_image",
self.proxy.import_image,
method_args=[original_image, "method", "uri"],
@ -219,7 +219,7 @@ class TestImageProxy(test_proxy_base.TestProxyBase):
def test_image_download(self):
original_image = image.Image(**EXAMPLE)
self._verify2(
self._verify(
'openstack.image.v2.image.Image.download',
self.proxy.download_image,
method_args=[original_image],
@ -294,28 +294,28 @@ class TestImageProxy(test_proxy_base.TestProxyBase):
self.verify_list(self.proxy.images, image.Image)
def test_add_tag(self):
self._verify2(
self._verify(
"openstack.image.v2.image.Image.add_tag",
self.proxy.add_tag,
method_args=["image", "tag"],
expected_args=[self.proxy, "tag"])
def test_remove_tag(self):
self._verify2(
self._verify(
"openstack.image.v2.image.Image.remove_tag",
self.proxy.remove_tag,
method_args=["image", "tag"],
expected_args=[self.proxy, "tag"])
def test_deactivate_image(self):
self._verify2(
self._verify(
"openstack.image.v2.image.Image.deactivate",
self.proxy.deactivate_image,
method_args=["image"],
expected_args=[self.proxy])
def test_reactivate_image(self):
self._verify2(
self._verify(
"openstack.image.v2.image.Image.reactivate",
self.proxy.reactivate_image,
method_args=["image"],
@ -327,51 +327,54 @@ class TestImageProxy(test_proxy_base.TestProxyBase):
expected_kwargs={"image_id": "test_id"})
def test_member_delete(self):
self._verify2("openstack.proxy.Proxy._delete",
self.proxy.remove_member,
method_args=["member_id"],
method_kwargs={"image": "image_id",
"ignore_missing": False},
expected_args=[member.Member],
expected_kwargs={"member_id": "member_id",
"image_id": "image_id",
"ignore_missing": False})
self._verify(
"openstack.proxy.Proxy._delete",
self.proxy.remove_member,
method_args=["member_id"],
method_kwargs={"image": "image_id", "ignore_missing": False},
expected_args=[member.Member],
expected_kwargs={
"member_id": "member_id",
"image_id": "image_id",
"ignore_missing": False})
def test_member_delete_ignore(self):
self._verify2("openstack.proxy.Proxy._delete",
self.proxy.remove_member,
method_args=["member_id"],
method_kwargs={"image": "image_id"},
expected_args=[member.Member],
expected_kwargs={"member_id": "member_id",
"image_id": "image_id",
"ignore_missing": True})
self._verify(
"openstack.proxy.Proxy._delete",
self.proxy.remove_member,
method_args=["member_id"],
method_kwargs={"image": "image_id"},
expected_args=[member.Member],
expected_kwargs={
"member_id": "member_id",
"image_id": "image_id",
"ignore_missing": True})
def test_member_update(self):
self._verify2("openstack.proxy.Proxy._update",
self.proxy.update_member,
method_args=['member_id', 'image_id'],
expected_args=[member.Member],
expected_kwargs={'member_id': 'member_id',
'image_id': 'image_id'})
self._verify(
"openstack.proxy.Proxy._update",
self.proxy.update_member,
method_args=['member_id', 'image_id'],
expected_args=[member.Member],
expected_kwargs={'member_id': 'member_id', 'image_id': 'image_id'})
def test_member_get(self):
self._verify2("openstack.proxy.Proxy._get",
self.proxy.get_member,
method_args=['member_id'],
method_kwargs={"image": "image_id"},
expected_args=[member.Member],
expected_kwargs={'member_id': 'member_id',
'image_id': 'image_id'})
self._verify(
"openstack.proxy.Proxy._get",
self.proxy.get_member,
method_args=['member_id'],
method_kwargs={"image": "image_id"},
expected_args=[member.Member],
expected_kwargs={'member_id': 'member_id', 'image_id': 'image_id'})
def test_member_find(self):
self._verify2("openstack.proxy.Proxy._find",
self.proxy.find_member,
method_args=['member_id'],
method_kwargs={"image": "image_id"},
expected_args=[member.Member, "member_id"],
expected_kwargs={'ignore_missing': True,
'image_id': 'image_id'})
self._verify(
"openstack.proxy.Proxy._find",
self.proxy.find_member,
method_args=['member_id'],
method_kwargs={"image": "image_id"},
expected_args=[member.Member, "member_id"],
expected_kwargs={'ignore_missing': True, 'image_id': 'image_id'})
def test_members(self):
self.verify_list(self.proxy.members, member.Member,
@ -379,32 +382,36 @@ class TestImageProxy(test_proxy_base.TestProxyBase):
expected_kwargs={'image_id': 'image_1'})
def test_images_schema_get(self):
self._verify2("openstack.proxy.Proxy._get",
self.proxy.get_images_schema,
expected_args=[schema.Schema],
expected_kwargs={'base_path': '/schemas/images',
'requires_id': False})
self._verify(
"openstack.proxy.Proxy._get",
self.proxy.get_images_schema,
expected_args=[schema.Schema],
expected_kwargs={
'base_path': '/schemas/images', 'requires_id': False})
def test_image_schema_get(self):
self._verify2("openstack.proxy.Proxy._get",
self.proxy.get_image_schema,
expected_args=[schema.Schema],
expected_kwargs={'base_path': '/schemas/image',
'requires_id': False})
self._verify(
"openstack.proxy.Proxy._get",
self.proxy.get_image_schema,
expected_args=[schema.Schema],
expected_kwargs={
'base_path': '/schemas/image', 'requires_id': False})
def test_members_schema_get(self):
self._verify2("openstack.proxy.Proxy._get",
self.proxy.get_members_schema,
expected_args=[schema.Schema],
expected_kwargs={'base_path': '/schemas/members',
'requires_id': False})
self._verify(
"openstack.proxy.Proxy._get",
self.proxy.get_members_schema,
expected_args=[schema.Schema],
expected_kwargs={
'base_path': '/schemas/members', 'requires_id': False})
def test_member_schema_get(self):
self._verify2("openstack.proxy.Proxy._get",
self.proxy.get_member_schema,
expected_args=[schema.Schema],
expected_kwargs={'base_path': '/schemas/member',
'requires_id': False})
self._verify(
"openstack.proxy.Proxy._get",
self.proxy.get_member_schema,
expected_args=[schema.Schema],
expected_kwargs={
'base_path': '/schemas/member', 'requires_id': False})
def test_task_get(self):
self.verify_get(self.proxy.get_task, task.Task)
@ -488,26 +495,29 @@ class TestImageProxy(test_proxy_base.TestProxyBase):
self.assertEqual('success', result.status)
def test_tasks_schema_get(self):
self._verify2("openstack.proxy.Proxy._get",
self.proxy.get_tasks_schema,
expected_args=[schema.Schema],
expected_kwargs={'base_path': '/schemas/tasks',
'requires_id': False})
self._verify(
"openstack.proxy.Proxy._get",
self.proxy.get_tasks_schema,
expected_args=[schema.Schema],
expected_kwargs={
'base_path': '/schemas/tasks', 'requires_id': False})
def test_task_schema_get(self):
self._verify2("openstack.proxy.Proxy._get",
self.proxy.get_task_schema,
expected_args=[schema.Schema],
expected_kwargs={'base_path': '/schemas/task',
'requires_id': False})
self._verify(
"openstack.proxy.Proxy._get",
self.proxy.get_task_schema,
expected_args=[schema.Schema],
expected_kwargs={
'base_path': '/schemas/task', 'requires_id': False})
def test_stores(self):
self.verify_list(self.proxy.stores, si.Store)
def test_import_info(self):
self._verify2("openstack.proxy.Proxy._get",
self.proxy.get_import_info,
method_args=[],
method_kwargs={},
expected_args=[si.Import],
expected_kwargs={'require_id': False})
self._verify(
"openstack.proxy.Proxy._get",
self.proxy.get_import_info,
method_args=[],
method_kwargs={},
expected_args=[si.Import],
expected_kwargs={'require_id': False})

82
openstack/tests/unit/load_balancer/v2/test_proxy.py
View File

@ -70,13 +70,12 @@ class TestLoadBalancerProxy(test_proxy_base.TestProxyBase):
fake_load_balancer = mock.Mock()
fake_load_balancer.id = "load_balancer_id"
mock_get_resource.return_value = fake_load_balancer
self._verify2("openstack.proxy.Proxy._delete",
self.proxy.delete_load_balancer,
method_args=["resource_or_id", True,
False],
expected_args=[lb.LoadBalancer,
fake_load_balancer],
expected_kwargs={"ignore_missing": True})
self._verify(
"openstack.proxy.Proxy._delete",
self.proxy.delete_load_balancer,
method_args=["resource_or_id", True, False],
expected_args=[lb.LoadBalancer, fake_load_balancer],
expected_kwargs={"ignore_missing": True})
self.assertFalse(fake_load_balancer.cascade)
mock_get_resource.assert_called_once_with(lb.LoadBalancer,
"resource_or_id")
@ -86,13 +85,12 @@ class TestLoadBalancerProxy(test_proxy_base.TestProxyBase):
fake_load_balancer = mock.Mock()
fake_load_balancer.id = "load_balancer_id"
mock_get_resource.return_value = fake_load_balancer
self._verify2("openstack.proxy.Proxy._delete",
self.proxy.delete_load_balancer,
method_args=["resource_or_id", True,
True],
expected_args=[lb.LoadBalancer,
fake_load_balancer],
expected_kwargs={"ignore_missing": True})
self._verify(
"openstack.proxy.Proxy._delete",
self.proxy.delete_load_balancer,
method_args=["resource_or_id", True, True],
expected_args=[lb.LoadBalancer, fake_load_balancer],
expected_kwargs={"ignore_missing": True})
self.assertTrue(fake_load_balancer.cascade)
mock_get_resource.assert_called_once_with(lb.LoadBalancer,
"resource_or_id")
@ -196,19 +194,20 @@ class TestLoadBalancerProxy(test_proxy_base.TestProxyBase):
'ignore_missing': True})
def test_member_find(self):
self._verify2('openstack.proxy.Proxy._find',
self.proxy.find_member,
method_args=["MEMBER", self.POOL_ID],
expected_args=[member.Member, "MEMBER"],
expected_kwargs={"pool_id": self.POOL_ID,
"ignore_missing": True})
self._verify(
'openstack.proxy.Proxy._find',
self.proxy.find_member,
method_args=["MEMBER", self.POOL_ID],
expected_args=[member.Member, "MEMBER"],
expected_kwargs={"pool_id": self.POOL_ID, "ignore_missing": True})
def test_member_update(self):
self._verify2('openstack.proxy.Proxy._update',
self.proxy.update_member,
method_args=["MEMBER", self.POOL_ID],
expected_args=[member.Member, "MEMBER"],
expected_kwargs={"pool_id": self.POOL_ID})
self._verify(
'openstack.proxy.Proxy._update',
self.proxy.update_member,
method_args=["MEMBER", self.POOL_ID],
expected_args=[member.Member, "MEMBER"],
expected_kwargs={"pool_id": self.POOL_ID})
def test_health_monitors(self):
self.verify_list(self.proxy.health_monitors,
@ -284,19 +283,21 @@ class TestLoadBalancerProxy(test_proxy_base.TestProxyBase):
expected_kwargs={'l7policy_id': self.L7_POLICY_ID})
def test_l7_rule_find(self):
self._verify2('openstack.proxy.Proxy._find',
self.proxy.find_l7_rule,
method_args=["RULE", self.L7_POLICY_ID],
expected_args=[l7_rule.L7Rule, "RULE"],
expected_kwargs={"l7policy_id": self.L7_POLICY_ID,
"ignore_missing": True})
self._verify(
'openstack.proxy.Proxy._find',
self.proxy.find_l7_rule,
method_args=["RULE", self.L7_POLICY_ID],
expected_args=[l7_rule.L7Rule, "RULE"],
expected_kwargs={
"l7policy_id": self.L7_POLICY_ID, "ignore_missing": True})
def test_l7_rule_update(self):
self._verify2('openstack.proxy.Proxy._update',
self.proxy.update_l7_rule,
method_args=["RULE", self.L7_POLICY_ID],
expected_args=[l7_rule.L7Rule, "RULE"],
expected_kwargs={"l7policy_id": self.L7_POLICY_ID})
self._verify(
'openstack.proxy.Proxy._update',
self.proxy.update_l7_rule,
method_args=["RULE", self.L7_POLICY_ID],
expected_args=[l7_rule.L7Rule, "RULE"],
expected_kwargs={"l7policy_id": self.L7_POLICY_ID})
def test_quotas(self):
self.verify_list(self.proxy.quotas, quota.Quota)
@ -308,10 +309,11 @@ class TestLoadBalancerProxy(test_proxy_base.TestProxyBase):
self.verify_update(self.proxy.update_quota, quota.Quota)
def test_quota_default_get(self):
self._verify2("openstack.proxy.Proxy._get",
self.proxy.get_quota_default,
expected_args=[quota.QuotaDefault],
expected_kwargs={'requires_id': False})
self._verify(
"openstack.proxy.Proxy._get",
self.proxy.get_quota_default,
expected_args=[quota.QuotaDefault],
expected_kwargs={'requires_id': False})
def test_quota_delete(self):
self.verify_delete(self.proxy.delete_quota, quota.Quota, False)

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

@ -50,7 +50,7 @@ class TestMessageProxy(test_proxy_base.TestProxyBase):
def test_message_post(self, mock_get_resource):
message_obj = message.Message(queue_name="test_queue")
mock_get_resource.return_value = message_obj
self._verify2(
self._verify(
"openstack.message.v2.message.Message.post",
self.proxy.post_message,
method_args=["test_queue", ["msg1", "msg2"]],
@ -61,10 +61,11 @@ class TestMessageProxy(test_proxy_base.TestProxyBase):
@mock.patch.object(proxy_base.Proxy, '_get_resource')
def test_message_get(self, mock_get_resource):
mock_get_resource.return_value = "resource_or_id"
self._verify2("openstack.proxy.Proxy._get",
self.proxy.get_message,
method_args=["test_queue", "resource_or_id"],
expected_args=[message.Message, "resource_or_id"])
self._verify(
"openstack.proxy.Proxy._get",
self.proxy.get_message,
method_args=["test_queue", "resource_or_id"],
expected_args=[message.Message, "resource_or_id"])
mock_get_resource.assert_called_once_with(message.Message,
"resource_or_id",
queue_name="test_queue")
@ -82,13 +83,12 @@ class TestMessageProxy(test_proxy_base.TestProxyBase):
fake_message = mock.Mock()
fake_message.id = "message_id"
mock_get_resource.return_value = fake_message
self._verify2("openstack.proxy.Proxy._delete",
self.proxy.delete_message,
method_args=["test_queue", "resource_or_id", None,
False],
expected_args=[message.Message,
fake_message],
expected_kwargs={"ignore_missing": False})
self._verify(
"openstack.proxy.Proxy._delete",
self.proxy.delete_message,
method_args=["test_queue", "resource_or_id", None, False],
expected_args=[message.Message, fake_message],
expected_kwargs={"ignore_missing": False})
self.assertIsNone(fake_message.claim_id)
mock_get_resource.assert_called_once_with(message.Message,
"resource_or_id",
@ -99,13 +99,12 @@ class TestMessageProxy(test_proxy_base.TestProxyBase):
fake_message = mock.Mock()
fake_message.id = "message_id"
mock_get_resource.return_value = fake_message
self._verify2("openstack.proxy.Proxy._delete",
self.proxy.delete_message,
method_args=["test_queue", "resource_or_id", "claim_id",
False],
expected_args=[message.Message,
fake_message],
expected_kwargs={"ignore_missing": False})
self._verify(
"openstack.proxy.Proxy._delete",
self.proxy.delete_message,
method_args=["test_queue", "resource_or_id", "claim_id", False],
expected_args=[message.Message, fake_message],
expected_kwargs={"ignore_missing": False})
self.assertEqual("claim_id", fake_message.claim_id)
mock_get_resource.assert_called_once_with(message.Message,
"resource_or_id",
@ -116,20 +115,19 @@ class TestMessageProxy(test_proxy_base.TestProxyBase):
fake_message = mock.Mock()
fake_message.id = "message_id"
mock_get_resource.return_value = fake_message
self._verify2("openstack.proxy.Proxy._delete",
self.proxy.delete_message,
method_args=["test_queue", "resource_or_id", None,
True],
expected_args=[message.Message,
fake_message],
expected_kwargs={"ignore_missing": True})
self._verify(
"openstack.proxy.Proxy._delete",
self.proxy.delete_message,
method_args=["test_queue", "resource_or_id", None, True],
expected_args=[message.Message, fake_message],
expected_kwargs={"ignore_missing": True})
self.assertIsNone(fake_message.claim_id)
mock_get_resource.assert_called_once_with(message.Message,
"resource_or_id",
queue_name="test_queue")
def test_subscription_create(self):
self._verify2(
self._verify(
"openstack.message.v2.subscription.Subscription.create",
self.proxy.create_subscription,
method_args=["test_queue"],
@ -139,11 +137,11 @@ class TestMessageProxy(test_proxy_base.TestProxyBase):
@mock.patch.object(proxy_base.Proxy, '_get_resource')
def test_subscription_get(self, mock_get_resource):
mock_get_resource.return_value = "resource_or_id"
self._verify2("openstack.proxy.Proxy._get",
self.proxy.get_subscription,
method_args=["test_queue", "resource_or_id"],
expected_args=[subscription.Subscription,
"resource_or_id"])
self._verify(
"openstack.proxy.Proxy._get",
self.proxy.get_subscription,
method_args=["test_queue", "resource_or_id"],
expected_args=[subscription.Subscription, "resource_or_id"])
mock_get_resource.assert_called_once_with(
subscription.Subscription, "resource_or_id",
queue_name="test_queue")
@ -181,7 +179,7 @@ class TestMessageProxy(test_proxy_base.TestProxyBase):
queue_name="test_queue")
def test_claim_create(self):
self._verify2(
self._verify(
"openstack.message.v2.claim.Claim.create",
self.proxy.create_claim,
method_args=["test_queue"],
@ -189,25 +187,24 @@ class TestMessageProxy(test_proxy_base.TestProxyBase):
expected_kwargs={"base_path": None})
def test_claim_get(self):
self._verify2("openstack.proxy.Proxy._get",
self.proxy.get_claim,
method_args=["test_queue", "resource_or_id"],
expected_args=[claim.Claim,
"resource_or_id"],
expected_kwargs={"queue_name": "test_queue"})
self._verify(
"openstack.proxy.Proxy._get",
self.proxy.get_claim,
method_args=["test_queue", "resource_or_id"],
expected_args=[claim.Claim, "resource_or_id"],
expected_kwargs={"queue_name": "test_queue"})
self.verify_get_overrided(
self.proxy, claim.Claim,
'openstack.message.v2.claim.Claim')
def test_claim_update(self):
self._verify2("openstack.proxy.Proxy._update",
self.proxy.update_claim,
method_args=["test_queue", "resource_or_id"],
method_kwargs={"k1": "v1"},
expected_args=[claim.Claim,
"resource_or_id"],
expected_kwargs={"queue_name": "test_queue",
"k1": "v1"})
self._verify(
"openstack.proxy.Proxy._update",
self.proxy.update_claim,
method_args=["test_queue", "resource_or_id"],
method_kwargs={"k1": "v1"},
expected_args=[claim.Claim, "resource_or_id"],
expected_kwargs={"queue_name": "test_queue", "k1": "v1"})
def test_claim_delete(self):
self.verify_delete(self.proxy.delete_claim,

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

@ -366,13 +366,13 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
self.verify_find(self.proxy.find_network, network.Network)
def test_network_find_with_filter(self):
self._verify2('openstack.proxy.Proxy._find',
self.proxy.find_network,
method_args=["net1"],
method_kwargs={"project_id": "1"},
expected_args=[network.Network, "net1"],
expected_kwargs={"project_id": "1",
"ignore_missing": True})
self._verify(
'openstack.proxy.Proxy._find',
self.proxy.find_network,
method_args=["net1"],
method_kwargs={"project_id": "1"},
expected_args=[network.Network, "net1"],
expected_kwargs={"project_id": "1", "ignore_missing": True})
def test_network_get(self):
self.verify_get(self.proxy.get_network, network.Network)
@ -469,19 +469,20 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
expected_kwargs={"pool_id": "test_id"})
def test_pool_member_find(self):
self._verify2('openstack.proxy.Proxy._find',
self.proxy.find_pool_member,
method_args=["MEMBER", "POOL"],
expected_args=[pool_member.PoolMember, "MEMBER"],
expected_kwargs={"pool_id": "POOL",
"ignore_missing": True})
self._verify(
'openstack.proxy.Proxy._find',
self.proxy.find_pool_member,
method_args=["MEMBER", "POOL"],
expected_args=[pool_member.PoolMember, "MEMBER"],
expected_kwargs={"pool_id": "POOL", "ignore_missing": True})
def test_pool_member_get(self):
self._verify2('openstack.proxy.Proxy._get',
self.proxy.get_pool_member,
method_args=["MEMBER", "POOL"],
expected_args=[pool_member.PoolMember, "MEMBER"],
expected_kwargs={"pool_id": "POOL"})
self._verify(
'openstack.proxy.Proxy._get',
self.proxy.get_pool_member,
method_args=["MEMBER", "POOL"],
expected_args=[pool_member.PoolMember, "MEMBER"],
expected_kwargs={"pool_id": "POOL"})
def test_pool_members(self):
self.verify_list(
@ -491,11 +492,12 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
expected_kwargs={"pool_id": "test_id"})
def test_pool_member_update(self):
self._verify2("openstack.network.v2._proxy.Proxy._update",
self.proxy.update_pool_member,
method_args=["MEMBER", "POOL"],
expected_args=[pool_member.PoolMember, "MEMBER"],
expected_kwargs={"pool_id": "POOL"})
self._verify(
"openstack.network.v2._proxy.Proxy._update",
self.proxy.update_pool_member,
method_args=["MEMBER", "POOL"],
expected_args=[pool_member.PoolMember, "MEMBER"],
expected_kwargs={"pool_id": "POOL"})
def test_pool_create_attrs(self):
self.verify_create(self.proxy.create_pool, pool.Pool)
@ -590,14 +592,15 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
def test_qos_bandwidth_limit_rule_find(self):
policy = qos_policy.QoSPolicy.new(id=QOS_POLICY_ID)
self._verify2('openstack.proxy.Proxy._find',
self.proxy.find_qos_bandwidth_limit_rule,
method_args=['rule_id', policy],
expected_args=[
qos_bandwidth_limit_rule.QoSBandwidthLimitRule,
'rule_id'],
expected_kwargs={'ignore_missing': True,
'qos_policy_id': QOS_POLICY_ID})
self._verify(
'openstack.proxy.Proxy._find',
self.proxy.find_qos_bandwidth_limit_rule,
method_args=['rule_id', policy],
expected_args=[
qos_bandwidth_limit_rule.QoSBandwidthLimitRule,
'rule_id'],
expected_kwargs={
'ignore_missing': True, 'qos_policy_id': QOS_POLICY_ID})
def test_qos_bandwidth_limit_rule_get(self):
self.verify_get(
@ -615,15 +618,15 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
def test_qos_bandwidth_limit_rule_update(self):
policy = qos_policy.QoSPolicy.new(id=QOS_POLICY_ID)
self._verify2('openstack.network.v2._proxy.Proxy._update',
self.proxy.update_qos_bandwidth_limit_rule,
method_args=['rule_id', policy],
method_kwargs={'foo': 'bar'},
expected_args=[
qos_bandwidth_limit_rule.QoSBandwidthLimitRule,
'rule_id'],
expected_kwargs={'qos_policy_id': QOS_POLICY_ID,
'foo': 'bar'})
self._verify(
'openstack.network.v2._proxy.Proxy._update',
self.proxy.update_qos_bandwidth_limit_rule,
method_args=['rule_id', policy],
method_kwargs={'foo': 'bar'},
expected_args=[
qos_bandwidth_limit_rule.QoSBandwidthLimitRule,
'rule_id'],
expected_kwargs={'qos_policy_id': QOS_POLICY_ID, 'foo': 'bar'})
def test_qos_dscp_marking_rule_create_attrs(self):
self.verify_create(
@ -652,13 +655,14 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
def test_qos_dscp_marking_rule_find(self):
policy = qos_policy.QoSPolicy.new(id=QOS_POLICY_ID)
self._verify2('openstack.proxy.Proxy._find',
self.proxy.find_qos_dscp_marking_rule,
method_args=['rule_id', policy],
expected_args=[qos_dscp_marking_rule.QoSDSCPMarkingRule,
'rule_id'],
expected_kwargs={'ignore_missing': True,
'qos_policy_id': QOS_POLICY_ID})
self._verify(
'openstack.proxy.Proxy._find',
self.proxy.find_qos_dscp_marking_rule,
method_args=['rule_id', policy],
expected_args=[
qos_dscp_marking_rule.QoSDSCPMarkingRule, 'rule_id'],
expected_kwargs={
'ignore_missing': True, 'qos_policy_id': QOS_POLICY_ID})
def test_qos_dscp_marking_rule_get(self):
self.verify_get(
@ -676,15 +680,15 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
def test_qos_dscp_marking_rule_update(self):
policy = qos_policy.QoSPolicy.new(id=QOS_POLICY_ID)
self._verify2('openstack.network.v2._proxy.Proxy._update',
self.proxy.update_qos_dscp_marking_rule,
method_args=['rule_id', policy],
method_kwargs={'foo': 'bar'},
expected_args=[
qos_dscp_marking_rule.QoSDSCPMarkingRule,
'rule_id'],
expected_kwargs={'qos_policy_id': QOS_POLICY_ID,
'foo': 'bar'})
self._verify(
'openstack.network.v2._proxy.Proxy._update',
self.proxy.update_qos_dscp_marking_rule,
method_args=['rule_id', policy],
method_kwargs={'foo': 'bar'},
expected_args=[
qos_dscp_marking_rule.QoSDSCPMarkingRule,
'rule_id'],
expected_kwargs={'qos_policy_id': QOS_POLICY_ID, 'foo': 'bar'})
def test_qos_minimum_bandwidth_rule_create_attrs(self):
self.verify_create(
@ -713,14 +717,15 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
def test_qos_minimum_bandwidth_rule_find(self):
policy = qos_policy.QoSPolicy.new(id=QOS_POLICY_ID)
self._verify2('openstack.proxy.Proxy._find',
self.proxy.find_qos_minimum_bandwidth_rule,
method_args=['rule_id', policy],
expected_args=[
qos_minimum_bandwidth_rule.QoSMinimumBandwidthRule,
'rule_id'],
expected_kwargs={'ignore_missing': True,
'qos_policy_id': QOS_POLICY_ID})
self._verify(
'openstack.proxy.Proxy._find',
self.proxy.find_qos_minimum_bandwidth_rule,
method_args=['rule_id', policy],
expected_args=[
qos_minimum_bandwidth_rule.QoSMinimumBandwidthRule,
'rule_id'],
expected_kwargs={
'ignore_missing': True, 'qos_policy_id': QOS_POLICY_ID})
def test_qos_minimum_bandwidth_rule_get(self):
self.verify_get(
@ -738,15 +743,16 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
def test_qos_minimum_bandwidth_rule_update(self):
policy = qos_policy.QoSPolicy.new(id=QOS_POLICY_ID)
self._verify2('openstack.network.v2._proxy.Proxy._update',
self.proxy.update_qos_minimum_bandwidth_rule,
method_args=['rule_id', policy],
method_kwargs={'foo': 'bar'},
expected_args=[
qos_minimum_bandwidth_rule.QoSMinimumBandwidthRule,
'rule_id'],
expected_kwargs={'qos_policy_id': QOS_POLICY_ID,
'foo': 'bar'})
self._verify(
'openstack.network.v2._proxy.Proxy._update',
self.proxy.update_qos_minimum_bandwidth_rule,
method_args=['rule_id', policy],
method_kwargs={'foo': 'bar'},
expected_args=[
qos_minimum_bandwidth_rule.QoSMinimumBandwidthRule,
'rule_id'],
expected_kwargs={
'qos_policy_id': QOS_POLICY_ID, 'foo': 'bar'})
def test_qos_policy_create_attrs(self):
self.verify_create(self.proxy.create_qos_policy, qos_policy.QoSPolicy)
@ -795,25 +801,27 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
def test_quota_get_details(self, mock_get):
fake_quota = mock.Mock(project_id='PROJECT')
mock_get.return_value = fake_quota
self._verify2("openstack.proxy.Proxy._get",
self.proxy.get_quota,
method_args=['QUOTA_ID'],
method_kwargs={'details': True},
expected_args=[quota.QuotaDetails],
expected_kwargs={'project': fake_quota.id,
'requires_id': False})
self._verify(
"openstack.proxy.Proxy._get",
self.proxy.get_quota,
method_args=['QUOTA_ID'],
method_kwargs={'details': True},
expected_args=[quota.QuotaDetails],
expected_kwargs={
'project': fake_quota.id, 'requires_id': False})
mock_get.assert_called_once_with(quota.Quota, 'QUOTA_ID')
@mock.patch.object(proxy_base.Proxy, "_get_resource")
def test_quota_default_get(self, mock_get):
fake_quota = mock.Mock(project_id='PROJECT')
mock_get.return_value = fake_quota
self._verify2("openstack.proxy.Proxy._get",
self.proxy.get_quota_default,
method_args=['QUOTA_ID'],
expected_args=[quota.QuotaDefault],
expected_kwargs={'project': fake_quota.id,
'requires_id': False})
self._verify(
"openstack.proxy.Proxy._get",
self.proxy.get_quota_default,
method_args=['QUOTA_ID'],
expected_args=[quota.QuotaDefault],
expected_kwargs={
'project': fake_quota.id, 'requires_id': False})
mock_get.assert_called_once_with(quota.Quota, 'QUOTA_ID')
def test_quotas(self):
@ -891,7 +899,7 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
x_router = router.Router.new(id="ROUTER_ID")
mock_get.return_value = x_router
self._verify2(
self._verify(
"openstack.network.v2.router.Router.add_interface",
self.proxy.add_interface_to_router,
method_args=["FAKE_ROUTER"],
@ -907,7 +915,7 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
x_router = router.Router.new(id="ROUTER_ID")
mock_get.return_value = x_router
self._verify2(
self._verify(
"openstack.network.v2.router.Router.add_interface",
self.proxy.add_interface_to_router,
method_args=["FAKE_ROUTER"],
@ -923,7 +931,7 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
x_router = router.Router.new(id="ROUTER_ID")
mock_get.return_value = x_router
self._verify2(
self._verify(
"openstack.network.v2.router.Router.remove_interface",
self.proxy.remove_interface_from_router,
method_args=["FAKE_ROUTER"],
@ -939,7 +947,7 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
x_router = router.Router.new(id="ROUTER_ID")
mock_get.return_value = x_router
self._verify2(
self._verify(
"openstack.network.v2.router.Router.remove_interface",
self.proxy.remove_interface_from_router,
method_args=["FAKE_ROUTER"],
@ -955,7 +963,7 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
x_router = router.Router.new(id="ROUTER_ID")
mock_get.return_value = x_router
self._verify2(
self._verify(
"openstack.network.v2.router.Router.add_extra_routes",
self.proxy.add_extra_routes_to_router,
method_args=["FAKE_ROUTER"],
@ -971,7 +979,7 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
x_router = router.Router.new(id="ROUTER_ID")
mock_get.return_value = x_router
self._verify2(
self._verify(
"openstack.network.v2.router.Router.remove_extra_routes",
self.proxy.remove_extra_routes_from_router,
method_args=["FAKE_ROUTER"],
@ -986,7 +994,7 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
x_router = router.Router.new(id="ROUTER_ID")
mock_get.return_value = x_router
self._verify2(
self._verify(
"openstack.network.v2.router.Router.add_gateway",
self.proxy.add_gateway_to_router,
method_args=["FAKE_ROUTER"],
@ -1001,7 +1009,7 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
x_router = router.Router.new(id="ROUTER_ID")
mock_get.return_value = x_router
self._verify2(
self._verify(
"openstack.network.v2.router.Router.remove_gateway",
self.proxy.remove_gateway_from_router,
method_args=["FAKE_ROUTER"],
@ -1357,7 +1365,7 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
def test_set_tags(self):
x_network = network.Network.new(id='NETWORK_ID')
self._verify2(
self._verify(
'openstack.network.v2.network.Network.set_tags',
self.proxy.set_tags,
method_args=[x_network, ['TAG1', 'TAG2']],
@ -1398,24 +1406,26 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
def test_find_floating_ip_port_forwarding(self):
fip = floating_ip.FloatingIP.new(id=FIP_ID)
self._verify2('openstack.proxy.Proxy._find',
self.proxy.find_floating_ip_port_forwarding,
method_args=[fip, 'port_forwarding_id'],
expected_args=[
port_forwarding.PortForwarding,
'port_forwarding_id'],
expected_kwargs={'ignore_missing': True,
'floatingip_id': FIP_ID})
self._verify(
'openstack.proxy.Proxy._find',
self.proxy.find_floating_ip_port_forwarding,
method_args=[fip, 'port_forwarding_id'],
expected_args=[
port_forwarding.PortForwarding,
'port_forwarding_id'],
expected_kwargs={
'ignore_missing': True, 'floatingip_id': FIP_ID})
def test_get_floating_ip_port_forwarding(self):
fip = floating_ip.FloatingIP.new(id=FIP_ID)
self._verify2('openstack.proxy.Proxy._get',
self.proxy.get_floating_ip_port_forwarding,
method_args=[fip, 'port_forwarding_id'],
expected_args=[
port_forwarding.PortForwarding,
'port_forwarding_id'],
expected_kwargs={'floatingip_id': FIP_ID})
self._verify(
'openstack.proxy.Proxy._get',
self.proxy.get_floating_ip_port_forwarding,
method_args=[fip, 'port_forwarding_id'],
expected_args=[
port_forwarding.PortForwarding,
'port_forwarding_id'],
expected_kwargs={'floatingip_id': FIP_ID})
def test_floating_ip_port_forwardings(self):
self.verify_list(self.proxy.floating_ip_port_forwardings,
@ -1425,15 +1435,15 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
def test_update_floating_ip_port_forwarding(self):
fip = floating_ip.FloatingIP.new(id=FIP_ID)
self._verify2('openstack.network.v2._proxy.Proxy._update',
self.proxy.update_floating_ip_port_forwarding,
method_args=[fip, 'port_forwarding_id'],
method_kwargs={'foo': 'bar'},
expected_args=[
port_forwarding.PortForwarding,
'port_forwarding_id'],
expected_kwargs={'floatingip_id': FIP_ID,
'foo': 'bar'})
self._verify(
'openstack.network.v2._proxy.Proxy._update',
self.proxy.update_floating_ip_port_forwarding,
method_args=[fip, 'port_forwarding_id'],
method_kwargs={'foo': 'bar'},
expected_args=[
port_forwarding.PortForwarding,
'port_forwarding_id'],
expected_kwargs={'floatingip_id': FIP_ID, 'foo': 'bar'})
def test_create_l3_conntrack_helper(self):
self.verify_create(self.proxy.create_conntrack_helper,
@ -1463,13 +1473,14 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
def test_get_l3_conntrack_helper(self):
r = router.Router.new(id=ROUTER_ID)
self._verify2('openstack.proxy.Proxy._get',
self.proxy.get_conntrack_helper,
method_args=['conntrack_helper_id', r],
expected_args=[
l3_conntrack_helper.ConntrackHelper,
'conntrack_helper_id'],
expected_kwargs={'router_id': ROUTER_ID})
self._verify(
'openstack.proxy.Proxy._get',
self.proxy.get_conntrack_helper,
method_args=['conntrack_helper_id', r],
expected_args=[
l3_conntrack_helper.ConntrackHelper,
'conntrack_helper_id'],
expected_kwargs={'router_id': ROUTER_ID})
def test_l3_conntrack_helpers(self):
self.verify_list(self.proxy.conntrack_helpers,
@ -1480,12 +1491,12 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
def test_update_l3_conntrack_helper(self):
r = router.Router.new(id=ROUTER_ID)
self._verify2('openstack.network.v2._proxy.Proxy._update',
self.proxy.update_conntrack_helper,
method_args=['conntrack_helper_id', r],
method_kwargs={'foo': 'bar'},
expected_args=[
l3_conntrack_helper.ConntrackHelper,
'conntrack_helper_id'],
expected_kwargs={'router_id': ROUTER_ID,
'foo': 'bar'})
self._verify(
'openstack.network.v2._proxy.Proxy._update',
self.proxy.update_conntrack_helper,
method_args=['conntrack_helper_id', r],
method_kwargs={'foo': 'bar'},
expected_args=[
l3_conntrack_helper.ConntrackHelper,
'conntrack_helper_id'],
expected_kwargs={'router_id': ROUTER_ID, 'foo': 'bar'})

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

@ -64,12 +64,13 @@ class TestObjectStoreProxy(test_proxy_base.TestProxyBase):
expected_kwargs={'name': 'container_name', "x": 1, "y": 2, "z": 3})
def test_object_metadata_get(self):
self._verify2("openstack.proxy.Proxy._head",
self.proxy.get_object_metadata,
method_args=['object'],
method_kwargs={'container': 'container'},
expected_args=[obj.Object, 'object'],
expected_kwargs={'container': 'container'})
self._verify(
"openstack.proxy.Proxy._head",
self.proxy.get_object_metadata,
method_args=['object'],
method_kwargs={'container': 'container'},
expected_args=[obj.Object, 'object'],
expected_kwargs={'container': 'container'})
def _test_object_delete(self, ignore):
expected_kwargs = {
@ -77,12 +78,13 @@ class TestObjectStoreProxy(test_proxy_base.TestProxyBase):
"container": "name",
}
self._verify2("openstack.proxy.Proxy._delete",
self.proxy.delete_object,
method_args=["resource"],
method_kwargs=expected_kwargs,
expected_args=[obj.Object, "resource"],
expected_kwargs=expected_kwargs)
self._verify(
"openstack.proxy.Proxy._delete",
self.proxy.delete_object,
method_args=["resource"],
method_kwargs=expected_kwargs,
expected_args=[obj.Object, "resource"],
expected_kwargs=expected_kwargs)
def test_object_delete(self):
self._test_object_delete(False)
@ -93,11 +95,12 @@ class TestObjectStoreProxy(test_proxy_base.TestProxyBase):
def test_object_create_attrs(self):
kwargs = {"name": "test", "data": "data", "container": "name"}
self._verify2("openstack.proxy.Proxy._create",
self.proxy.upload_object,
method_kwargs=kwargs,
expected_args=[obj.Object],
expected_kwargs=kwargs)
self._verify(
"openstack.proxy.Proxy._create",
self.proxy.upload_object,
method_kwargs=kwargs,
expected_args=[obj.Object],
expected_kwargs=kwargs)
def test_object_create_no_container(self):
self.assertRaises(TypeError, self.proxy.upload_object)

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

@ -51,23 +51,25 @@ class TestOrchestrationProxy(test_proxy_base.TestProxyBase):
# 'ignore_missing': False
# }
# method_args=["name_or_id"]
# self._verify2(mock_method, test_method,
# method_args=method_args,
# method_kwargs=method_kwargs,
# expected_args=[stack.Stack, "name_or_id"],
# expected_kwargs=method_kwargs,
# expected_result="result")
# self._verify(
# mock_method, test_method,
# method_args=method_args,
# method_kwargs=method_kwargs,
# expected_args=[stack.Stack, "name_or_id"],
# expected_kwargs=method_kwargs,
# expected_result="result")
#
# method_kwargs = {
# 'resolve_outputs': True,
# 'ignore_missing': True
# }
# self._verify2(mock_method, test_method,
# method_args=method_args,
# method_kwargs=method_kwargs,
# expected_args=[stack.Stack, "name_or_id"],
# expected_kwargs=method_kwargs,
# expected_result="result")
# self._verify(
# mock_method, test_method,
# method_args=method_args,
# method_kwargs=method_kwargs,
# expected_args=[stack.Stack, "name_or_id"],
# expected_kwargs=method_kwargs,
# expected_result="result")
def test_stacks(self):
self.verify_list(self.proxy.stacks, stack.Stack)
@ -81,27 +83,30 @@ class TestOrchestrationProxy(test_proxy_base.TestProxyBase):
'openstack.orchestration.v1.stack.Stack')
def test_update_stack(self):
self._verify2('openstack.orchestration.v1.stack.Stack.update',
self.proxy.update_stack,
expected_result='result',
method_args=['stack'],
method_kwargs={'preview': False},
expected_args=[self.proxy, False])
self._verify(
'openstack.orchestration.v1.stack.Stack.update',
self.proxy.update_stack,
expected_result='result',
method_args=['stack'],
method_kwargs={'preview': False},
expected_args=[self.proxy, False])
def test_update_stack_preview(self):
self._verify2('openstack.orchestration.v1.stack.Stack.update',
self.proxy.update_stack,
expected_result='result',
method_args=['stack'],
method_kwargs={'preview': True},
expected_args=[self.proxy, True])
self._verify(
'openstack.orchestration.v1.stack.Stack.update',
self.proxy.update_stack,
expected_result='result',
method_args=['stack'],
method_kwargs={'preview': True},
expected_args=[self.proxy, True])
def test_abandon_stack(self):
self._verify2('openstack.orchestration.v1.stack.Stack.abandon',
self.proxy.abandon_stack,
expected_result='result',
method_args=['stack'],
expected_args=[self.proxy])
self._verify(
'openstack.orchestration.v1.stack.Stack.abandon',
self.proxy.abandon_stack,
expected_result='result',
method_args=['stack'],
expected_args=[self.proxy])
def test_delete_stack(self):
self.verify_delete(self.proxy.delete_stack, stack.Stack, False)
@ -136,13 +141,15 @@ class TestOrchestrationProxy(test_proxy_base.TestProxyBase):
stk = stack.Stack(id=stack_id, name=stack_name)
mock_find.return_value = stk
self._verify2('openstack.proxy.Proxy._get',
self.proxy.get_stack_environment,
method_args=['IDENTITY'],
expected_args=[stack_environment.StackEnvironment],
expected_kwargs={'requires_id': False,
'stack_name': stack_name,
'stack_id': stack_id})
self._verify(
'openstack.proxy.Proxy._get',
self.proxy.get_stack_environment,
method_args=['IDENTITY'],
expected_args=[stack_environment.StackEnvironment],
expected_kwargs={
'requires_id': False,
'stack_name': stack_name,
'stack_id': stack_id})
mock_find.assert_called_once_with(mock.ANY, 'IDENTITY',
ignore_missing=False)
@ -151,13 +158,15 @@ class TestOrchestrationProxy(test_proxy_base.TestProxyBase):
stack_name = 'test_stack'
stk = stack.Stack(id=stack_id, name=stack_name)
self._verify2('openstack.proxy.Proxy._get',
self.proxy.get_stack_environment,
method_args=[stk],
expected_args=[stack_environment.StackEnvironment],
expected_kwargs={'requires_id': False,
'stack_name': stack_name,
'stack_id': stack_id})
self._verify(
'openstack.proxy.Proxy._get',
self.proxy.get_stack_environment,
method_args=[stk],
expected_args=[stack_environment.StackEnvironment],
expected_kwargs={
'requires_id': False,
'stack_name': stack_name,
'stack_id': stack_id})
@mock.patch.object(stack_files.StackFiles, 'fetch')
@mock.patch.object(stack.Stack, 'find')
@ -194,13 +203,15 @@ class TestOrchestrationProxy(test_proxy_base.TestProxyBase):
stk = stack.Stack(id=stack_id, name=stack_name)
mock_find.return_value = stk
self._verify2('openstack.proxy.Proxy._get',
self.proxy.get_stack_template,
method_args=['IDENTITY'],
expected_args=[stack_template.StackTemplate],
expected_kwargs={'requires_id': False,
'stack_name': stack_name,
'stack_id': stack_id})
self._verify(
'openstack.proxy.Proxy._get',
self.proxy.get_stack_template,
method_args=['IDENTITY'],
expected_args=[stack_template.StackTemplate],
expected_kwargs={
'requires_id': False,
'stack_name': stack_name,
'stack_id': stack_id})
mock_find.assert_called_once_with(mock.ANY, 'IDENTITY',
ignore_missing=False)
@ -209,13 +220,15 @@ class TestOrchestrationProxy(test_proxy_base.TestProxyBase):
stack_name = 'test_stack'
stk = stack.Stack(id=stack_id, name=stack_name)
self._verify2('openstack.proxy.Proxy._get',
self.proxy.get_stack_template,
method_args=[stk],
expected_args=[stack_template.StackTemplate],
expected_kwargs={'requires_id': False,
'stack_name': stack_name,
'stack_id': stack_id})
self._verify(
'openstack.proxy.Proxy._get',
self.proxy.get_stack_template,
method_args=[stk],
expected_args=[stack_template.StackTemplate],
expected_kwargs={
'requires_id': False,
'stack_name': stack_name,
'stack_id': stack_id})
@mock.patch.object(stack.Stack, 'find')
def test_resources_with_stack_object(self, mock_find):

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

@ -20,39 +20,11 @@ class TestProxyBase(base.TestCase):
super(TestProxyBase, self).setUp()
self.session = mock.Mock()
def _verify(self, mock_method, test_method,
method_args=None, method_kwargs=None,
expected_args=None, expected_kwargs=None,
expected_result=None):
with mock.patch(mock_method) as mocked:
mocked.return_value = expected_result
if any([method_args, method_kwargs,
expected_args, expected_kwargs]):
method_args = method_args or ()
method_kwargs = method_kwargs or {}
expected_args = expected_args or ()
expected_kwargs = expected_kwargs or {}
self.assertEqual(expected_result, test_method(*method_args,
**method_kwargs))
mocked.assert_called_with(test_method.__self__,
*expected_args, **expected_kwargs)
else:
self.assertEqual(expected_result, test_method())
mocked.assert_called_with(test_method.__self__)
# NOTE(briancurtin): This is a duplicate version of _verify that is
# temporarily here while we shift APIs. The difference is that
# calls from the Proxy classes aren't going to be going directly into
# the Resource layer anymore, so they don't pass in the session which
# was tested in assert_called_with.
# This is being done in lieu of adding logic and complicating
# the _verify method. It will be removed once there is one API to
# be verifying.
def _verify2(self, mock_method, test_method,
method_args=None, method_kwargs=None, method_result=None,
expected_args=None, expected_kwargs=None,
expected_result=None):
def _verify(
self, mock_method, test_method, *,
method_args=None, method_kwargs=None, method_result=None,
expected_args=None, expected_kwargs=None, expected_result=None,
):
with mock.patch(mock_method) as mocked:
mocked.return_value = expected_result
if any([
@ -120,7 +92,7 @@ class TestProxyBase(base.TestCase):
expected_kwargs = method_kwargs.copy()
expected_kwargs["base_path"] = base_path
self._verify2(
self._verify(
mock_method,
test_method,
method_args=method_args,
@ -142,7 +114,7 @@ class TestProxyBase(base.TestCase):
expected_args = expected_args or method_args.copy()
expected_kwargs = expected_kwargs or method_kwargs.copy()
self._verify2(
self._verify(
mock_method,
test_method,
method_args=method_args,
@ -166,7 +138,7 @@ class TestProxyBase(base.TestCase):
if expected_kwargs is None:
expected_kwargs = method_kwargs.copy()
self._verify2(
self._verify(
mock_method,
test_method,
method_args=method_args,
@ -197,7 +169,7 @@ class TestProxyBase(base.TestCase):
expected_args = expected_args or method_args.copy()
expected_kwargs = expected_kwargs or method_kwargs.copy()
self._verify2(
self._verify(
mock_method,
test_method,
method_args=method_args,
@ -219,7 +191,7 @@ class TestProxyBase(base.TestCase):
expected_args = expected_args or method_args.copy()
expected_kwargs = expected_kwargs or method_kwargs.copy()
self._verify2(
self._verify(
mock_method,
test_method,
method_args=method_args,
@ -247,7 +219,7 @@ class TestProxyBase(base.TestCase):
if base_path is not None:
expected_kwargs["base_path"] = base_path
self._verify2(
self._verify(
mock_method,
test_method,
method_args=method_args,
@ -272,7 +244,7 @@ class TestProxyBase(base.TestCase):
if expected_kwargs is None:
expected_kwargs = method_kwargs.copy()
self._verify2(
self._verify(
mock_method,
test_method,
method_args=method_args,
@ -286,4 +258,4 @@ class TestProxyBase(base.TestCase):
mock_method="openstack.resource.wait_for_status",
**kwargs,
):
self._verify2(mock_method, test_method, **kwargs)
self._verify(mock_method, test_method, **kwargs)