openstack/openstacksdk
Code Issues Proposed changes

Merge "tests: Remove final use of 'TestProxyBase._verify'"

Browse Source
This commit is contained in:
Zuul
2021-05-19 21:31:24 +00:00
committed by Gerrit Code Review
parent 32d2a5522c 6f98539497
commit 4feb26d46f
8 changed files with 437 additions and 320 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -88,10 +88,11 @@ class TestVolumeProxy(test_proxy_base.TestProxyBase):
self.verify_delete(self.proxy.delete_volume, volume.Volume, True) self.verify_delete(self.proxy.delete_volume, volume.Volume, True)
def test_volume_extend(self): def test_volume_extend(self):
self._verify("openstack.block_storage.v2.volume.Volume.extend", self._verify2(
self.proxy.extend_volume, "openstack.block_storage.v2.volume.Volume.extend",
method_args=["value", "new-size"], self.proxy.extend_volume,
expected_args=["new-size"]) method_args=["value", "new-size"],
expected_args=[self.proxy, "new-size"])
def test_backend_pools(self): def test_backend_pools(self):
self.verify_list(self.proxy.backend_pools, stats.Pools) self.verify_list(self.proxy.backend_pools, stats.Pools)

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

@@ -156,22 +156,25 @@ class TestVolumeProxy(test_proxy_base.TestProxyBase):
self.verify_delete(self.proxy.delete_volume, volume.Volume, True) self.verify_delete(self.proxy.delete_volume, volume.Volume, True)
def test_volume_extend(self): def test_volume_extend(self):
self._verify("openstack.block_storage.v3.volume.Volume.extend", self._verify2(
self.proxy.extend_volume, "openstack.block_storage.v3.volume.Volume.extend",
method_args=["value", "new-size"], self.proxy.extend_volume,
expected_args=["new-size"]) method_args=["value", "new-size"],
expected_args=[self.proxy, "new-size"])
def test_volume_set_readonly_no_argument(self): def test_volume_set_readonly_no_argument(self):
self._verify("openstack.block_storage.v3.volume.Volume.set_readonly", self._verify2(
self.proxy.set_volume_readonly, "openstack.block_storage.v3.volume.Volume.set_readonly",
method_args=["value"], self.proxy.set_volume_readonly,
expected_args=[True]) method_args=["value"],
expected_args=[self.proxy, True])
def test_volume_set_readonly_false(self): def test_volume_set_readonly_false(self):
self._verify("openstack.block_storage.v3.volume.Volume.set_readonly", self._verify2(
self.proxy.set_volume_readonly, "openstack.block_storage.v3.volume.Volume.set_readonly",
method_args=["value", False], self.proxy.set_volume_readonly,
expected_args=[False]) method_args=["value", False],
expected_args=[self.proxy, False])
def test_volume_retype_without_migration_policy(self): def test_volume_retype_without_migration_policy(self):
self._verify("openstack.block_storage.v3.volume.Volume.retype", self._verify("openstack.block_storage.v3.volume.Volume.retype",

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

@@ -91,9 +91,11 @@ class TestClusterProxy(test_proxy_base.TestProxyBase):
self.verify_delete(self.proxy.delete_cluster, cluster.Cluster, True) self.verify_delete(self.proxy.delete_cluster, cluster.Cluster, True)
def test_cluster_force_delete(self): def test_cluster_force_delete(self):
self._verify("openstack.clustering.v1.cluster.Cluster.force_delete", self._verify2(
self.proxy.delete_cluster, "openstack.clustering.v1.cluster.Cluster.force_delete",
method_args=["value", False, True]) self.proxy.delete_cluster,
method_args=["value", False, True],
expected_args=[self.proxy])
def test_cluster_find(self): def test_cluster_find(self):
self.verify_find(self.proxy.find_cluster, cluster.Cluster) self.verify_find(self.proxy.find_cluster, cluster.Cluster)
@@ -117,21 +119,25 @@ class TestClusterProxy(test_proxy_base.TestProxyBase):
def test_resize_cluster(self, mock_find): def test_resize_cluster(self, mock_find):
mock_cluster = cluster.Cluster.new(id='FAKE_CLUSTER') mock_cluster = cluster.Cluster.new(id='FAKE_CLUSTER')
mock_find.return_value = mock_cluster mock_find.return_value = mock_cluster
self._verify("openstack.clustering.v1.cluster.Cluster.resize", self._verify2(
self.proxy.resize_cluster, "openstack.clustering.v1.cluster.Cluster.resize",
method_args=["FAKE_CLUSTER"], self.proxy.resize_cluster,
method_kwargs={'k1': 'v1', 'k2': 'v2'}, method_args=["FAKE_CLUSTER"],
expected_kwargs={'k1': 'v1', 'k2': 'v2'}) method_kwargs={'k1': 'v1', 'k2': 'v2'},
expected_args=[self.proxy],
expected_kwargs={'k1': 'v1', 'k2': 'v2'})
mock_find.assert_called_once_with(cluster.Cluster, "FAKE_CLUSTER", mock_find.assert_called_once_with(cluster.Cluster, "FAKE_CLUSTER",
ignore_missing=False) ignore_missing=False)
def test_resize_cluster_with_obj(self): def test_resize_cluster_with_obj(self):
mock_cluster = cluster.Cluster.new(id='FAKE_CLUSTER') mock_cluster = cluster.Cluster.new(id='FAKE_CLUSTER')
self._verify("openstack.clustering.v1.cluster.Cluster.resize", self._verify2(
self.proxy.resize_cluster, "openstack.clustering.v1.cluster.Cluster.resize",
method_args=[mock_cluster], self.proxy.resize_cluster,
method_kwargs={'k1': 'v1', 'k2': 'v2'}, method_args=[mock_cluster],
expected_kwargs={'k1': 'v1', 'k2': 'v2'}) method_kwargs={'k1': 'v1', 'k2': 'v2'},
expected_args=[self.proxy],
expected_kwargs={'k1': 'v1', 'k2': 'v2'})
def test_collect_cluster_attrs(self): def test_collect_cluster_attrs(self):
self.verify_list(self.proxy.collect_cluster_attrs, self.verify_list(self.proxy.collect_cluster_attrs,
@@ -145,18 +151,22 @@ class TestClusterProxy(test_proxy_base.TestProxyBase):
def test_cluster_check(self, mock_get): def test_cluster_check(self, mock_get):
mock_cluster = cluster.Cluster.new(id='FAKE_CLUSTER') mock_cluster = cluster.Cluster.new(id='FAKE_CLUSTER')
mock_get.return_value = mock_cluster mock_get.return_value = mock_cluster
self._verify("openstack.clustering.v1.cluster.Cluster.check", self._verify2(
self.proxy.check_cluster, "openstack.clustering.v1.cluster.Cluster.check",
method_args=["FAKE_CLUSTER"]) self.proxy.check_cluster,
method_args=["FAKE_CLUSTER"],
expected_args=[self.proxy])
mock_get.assert_called_once_with(cluster.Cluster, "FAKE_CLUSTER") mock_get.assert_called_once_with(cluster.Cluster, "FAKE_CLUSTER")
@mock.patch.object(proxy_base.Proxy, '_get_resource') @mock.patch.object(proxy_base.Proxy, '_get_resource')
def test_cluster_recover(self, mock_get): def test_cluster_recover(self, mock_get):
mock_cluster = cluster.Cluster.new(id='FAKE_CLUSTER') mock_cluster = cluster.Cluster.new(id='FAKE_CLUSTER')
mock_get.return_value = mock_cluster mock_get.return_value = mock_cluster
self._verify("openstack.clustering.v1.cluster.Cluster.recover", self._verify2(
self.proxy.recover_cluster, "openstack.clustering.v1.cluster.Cluster.recover",
method_args=["FAKE_CLUSTER"]) self.proxy.recover_cluster,
method_args=["FAKE_CLUSTER"],
expected_args=[self.proxy])
mock_get.assert_called_once_with(cluster.Cluster, "FAKE_CLUSTER") mock_get.assert_called_once_with(cluster.Cluster, "FAKE_CLUSTER")
def test_node_create(self): def test_node_create(self):
@@ -169,9 +179,11 @@ class TestClusterProxy(test_proxy_base.TestProxyBase):
self.verify_delete(self.proxy.delete_node, node.Node, True) self.verify_delete(self.proxy.delete_node, node.Node, True)
def test_node_force_delete(self): def test_node_force_delete(self):
self._verify("openstack.clustering.v1.node.Node.force_delete", self._verify2(
self.proxy.delete_node, "openstack.clustering.v1.node.Node.force_delete",
method_args=["value", False, True]) self.proxy.delete_node,
method_args=["value", False, True],
expected_args=[self.proxy])
def test_node_find(self): def test_node_find(self):
self.verify_find(self.proxy.find_node, node.Node) self.verify_find(self.proxy.find_node, node.Node)
@@ -200,28 +212,34 @@ class TestClusterProxy(test_proxy_base.TestProxyBase):
def test_node_check(self, mock_get): def test_node_check(self, mock_get):
mock_node = node.Node.new(id='FAKE_NODE') mock_node = node.Node.new(id='FAKE_NODE')
mock_get.return_value = mock_node mock_get.return_value = mock_node
self._verify("openstack.clustering.v1.node.Node.check", self._verify2(
self.proxy.check_node, "openstack.clustering.v1.node.Node.check",
method_args=["FAKE_NODE"]) self.proxy.check_node,
method_args=["FAKE_NODE"],
expected_args=[self.proxy])
mock_get.assert_called_once_with(node.Node, "FAKE_NODE") mock_get.assert_called_once_with(node.Node, "FAKE_NODE")
@mock.patch.object(proxy_base.Proxy, '_get_resource') @mock.patch.object(proxy_base.Proxy, '_get_resource')
def test_node_recover(self, mock_get): def test_node_recover(self, mock_get):
mock_node = node.Node.new(id='FAKE_NODE') mock_node = node.Node.new(id='FAKE_NODE')
mock_get.return_value = mock_node mock_get.return_value = mock_node
self._verify("openstack.clustering.v1.node.Node.recover", self._verify2(
self.proxy.recover_node, "openstack.clustering.v1.node.Node.recover",
method_args=["FAKE_NODE"]) self.proxy.recover_node,
method_args=["FAKE_NODE"],
expected_args=[self.proxy])
mock_get.assert_called_once_with(node.Node, "FAKE_NODE") mock_get.assert_called_once_with(node.Node, "FAKE_NODE")
@mock.patch.object(proxy_base.Proxy, '_get_resource') @mock.patch.object(proxy_base.Proxy, '_get_resource')
def test_node_adopt(self, mock_get): def test_node_adopt(self, mock_get):
mock_node = node.Node.new() mock_node = node.Node.new()
mock_get.return_value = mock_node mock_get.return_value = mock_node
self._verify("openstack.clustering.v1.node.Node.adopt", self._verify2(
self.proxy.adopt_node, "openstack.clustering.v1.node.Node.adopt",
method_kwargs={"preview": False, "foo": "bar"}, self.proxy.adopt_node,
expected_kwargs={"preview": False, "foo": "bar"}) method_kwargs={"preview": False, "foo": "bar"},
expected_args=[self.proxy],
expected_kwargs={"preview": False, "foo": "bar"})
mock_get.assert_called_once_with(node.Node, None) mock_get.assert_called_once_with(node.Node, None)
@@ -229,10 +247,12 @@ class TestClusterProxy(test_proxy_base.TestProxyBase):
def test_node_adopt_preview(self, mock_get): def test_node_adopt_preview(self, mock_get):
mock_node = node.Node.new() mock_node = node.Node.new()
mock_get.return_value = mock_node mock_get.return_value = mock_node
self._verify("openstack.clustering.v1.node.Node.adopt", self._verify2(
self.proxy.adopt_node, "openstack.clustering.v1.node.Node.adopt",
method_kwargs={"preview": True, "foo": "bar"}, self.proxy.adopt_node,
expected_kwargs={"preview": True, "foo": "bar"}) method_kwargs={"preview": True, "foo": "bar"},
expected_args=[self.proxy],
expected_kwargs={"preview": True, "foo": "bar"})
mock_get.assert_called_once_with(node.Node, None) mock_get.assert_called_once_with(node.Node, None)

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

@@ -193,58 +193,61 @@ class TestFlavor(TestComputeProxy):
) )
def test_flavor_get_access(self): def test_flavor_get_access(self):
self._verify("openstack.compute.v2.flavor.Flavor.get_access", self._verify2(
self.proxy.get_flavor_access, "openstack.compute.v2.flavor.Flavor.get_access",
method_args=["value"], self.proxy.get_flavor_access,
expected_args=[]) method_args=["value"],
expected_args=[self.proxy])
def test_flavor_add_tenant_access(self): def test_flavor_add_tenant_access(self):
self._verify("openstack.compute.v2.flavor.Flavor.add_tenant_access", self._verify2(
self.proxy.flavor_add_tenant_access, "openstack.compute.v2.flavor.Flavor.add_tenant_access",
method_args=["value", "fake-tenant"], self.proxy.flavor_add_tenant_access,
expected_args=["fake-tenant"]) method_args=["value", "fake-tenant"],
expected_args=[self.proxy, "fake-tenant"])
def test_flavor_remove_tenant_access(self): def test_flavor_remove_tenant_access(self):
self._verify("openstack.compute.v2.flavor.Flavor.remove_tenant_access", self._verify2(
self.proxy.flavor_remove_tenant_access, "openstack.compute.v2.flavor.Flavor.remove_tenant_access",
method_args=["value", "fake-tenant"], self.proxy.flavor_remove_tenant_access,
expected_args=["fake-tenant"]) method_args=["value", "fake-tenant"],
expected_args=[self.proxy, "fake-tenant"])
def test_flavor_fetch_extra_specs(self): def test_flavor_fetch_extra_specs(self):
self._verify("openstack.compute.v2.flavor.Flavor.fetch_extra_specs", self._verify2(
self.proxy.fetch_flavor_extra_specs, "openstack.compute.v2.flavor.Flavor.fetch_extra_specs",
method_args=["value"], self.proxy.fetch_flavor_extra_specs,
expected_args=[]) method_args=["value"],
expected_args=[self.proxy])
def test_create_flavor_extra_specs(self): def test_create_flavor_extra_specs(self):
specs = { self._verify2(
'a': 'b' "openstack.compute.v2.flavor.Flavor.create_extra_specs",
} self.proxy.create_flavor_extra_specs,
self._verify("openstack.compute.v2.flavor.Flavor.create_extra_specs", method_args=["value", {'a': 'b'}],
self.proxy.create_flavor_extra_specs, expected_args=[self.proxy],
method_args=["value", specs], expected_kwargs={"specs": {'a': 'b'}})
expected_kwargs={"specs": specs})
def test_get_flavor_extra_specs_prop(self): def test_get_flavor_extra_specs_prop(self):
self._verify( self._verify2(
"openstack.compute.v2.flavor.Flavor.get_extra_specs_property", "openstack.compute.v2.flavor.Flavor.get_extra_specs_property",
self.proxy.get_flavor_extra_specs_property, self.proxy.get_flavor_extra_specs_property,
method_args=["value", "prop"], method_args=["value", "prop"],
expected_args=["prop"]) expected_args=[self.proxy, "prop"])
def test_update_flavor_extra_specs_prop(self): def test_update_flavor_extra_specs_prop(self):
self._verify( self._verify2(
"openstack.compute.v2.flavor.Flavor.update_extra_specs_property", "openstack.compute.v2.flavor.Flavor.update_extra_specs_property",
self.proxy.update_flavor_extra_specs_property, self.proxy.update_flavor_extra_specs_property,
method_args=["value", "prop", "val"], method_args=["value", "prop", "val"],
expected_args=["prop", "val"]) expected_args=[self.proxy, "prop", "val"])
def test_delete_flavor_extra_specs_prop(self): def test_delete_flavor_extra_specs_prop(self):
self._verify( self._verify2(
"openstack.compute.v2.flavor.Flavor.delete_extra_specs_property", "openstack.compute.v2.flavor.Flavor.delete_extra_specs_property",
self.proxy.delete_flavor_extra_specs_property, self.proxy.delete_flavor_extra_specs_property,
method_args=["value", "prop"], method_args=["value", "prop"],
expected_args=["prop"]) expected_args=[self.proxy, "prop"])
class TestKeyPair(TestComputeProxy): class TestKeyPair(TestComputeProxy):
@@ -321,36 +324,39 @@ class TestAggregate(TestComputeProxy):
self.verify_update(self.proxy.update_aggregate, aggregate.Aggregate) self.verify_update(self.proxy.update_aggregate, aggregate.Aggregate)
def test_aggregate_add_host(self): def test_aggregate_add_host(self):
self._verify("openstack.compute.v2.aggregate.Aggregate.add_host", self._verify2(
self.proxy.add_host_to_aggregate, "openstack.compute.v2.aggregate.Aggregate.add_host",
method_args=["value", "host"], self.proxy.add_host_to_aggregate,
expected_args=["host"]) method_args=["value", "host"],
expected_args=[self.proxy, "host"])
def test_aggregate_remove_host(self): def test_aggregate_remove_host(self):
self._verify("openstack.compute.v2.aggregate.Aggregate.remove_host", self._verify2(
self.proxy.remove_host_from_aggregate, "openstack.compute.v2.aggregate.Aggregate.remove_host",
method_args=["value", "host"], self.proxy.remove_host_from_aggregate,
expected_args=["host"]) method_args=["value", "host"],
expected_args=[self.proxy, "host"])
def test_aggregate_set_metadata(self): def test_aggregate_set_metadata(self):
self._verify("openstack.compute.v2.aggregate.Aggregate.set_metadata", self._verify2(
self.proxy.set_aggregate_metadata, "openstack.compute.v2.aggregate.Aggregate.set_metadata",
method_args=["value", {'a': 'b'}], self.proxy.set_aggregate_metadata,
expected_args=[{'a': 'b'}]) method_args=["value", {'a': 'b'}],
expected_args=[self.proxy, {'a': 'b'}])
def test_aggregate_precache_image(self): def test_aggregate_precache_image(self):
self._verify( self._verify2(
"openstack.compute.v2.aggregate.Aggregate.precache_images", "openstack.compute.v2.aggregate.Aggregate.precache_images",
self.proxy.aggregate_precache_images, self.proxy.aggregate_precache_images,
method_args=["value", '1'], method_args=["value", '1'],
expected_args=[[{'id': '1'}]]) expected_args=[self.proxy, [{'id': '1'}]])
def test_aggregate_precache_images(self): def test_aggregate_precache_images(self):
self._verify( self._verify2(
"openstack.compute.v2.aggregate.Aggregate.precache_images", "openstack.compute.v2.aggregate.Aggregate.precache_images",
self.proxy.aggregate_precache_images, self.proxy.aggregate_precache_images,
method_args=["value", ['1', '2']], method_args=["value", ['1', '2']],
expected_args=[[{'id': '1'}, {'id': '2'}]]) expected_args=[self.proxy, [{'id': '1'}, {'id': '2'}]])
class TestService(TestComputeProxy): class TestService(TestComputeProxy):
@@ -510,11 +516,11 @@ class TestHypervisor(TestComputeProxy):
hypervisor.Hypervisor) hypervisor.Hypervisor)
def test_get_hypervisor_uptime(self): def test_get_hypervisor_uptime(self):
self._verify( self._verify2(
"openstack.compute.v2.hypervisor.Hypervisor.get_uptime", "openstack.compute.v2.hypervisor.Hypervisor.get_uptime",
self.proxy.get_hypervisor_uptime, self.proxy.get_hypervisor_uptime,
method_args=["value"], method_args=["value"],
expected_args=[]) expected_args=[self.proxy])
class TestCompute(TestComputeProxy): class TestCompute(TestComputeProxy):
@@ -649,9 +655,11 @@ class TestCompute(TestComputeProxy):
self.verify_delete(self.proxy.delete_server, server.Server, True) self.verify_delete(self.proxy.delete_server, server.Server, True)
def test_server_force_delete(self): def test_server_force_delete(self):
self._verify("openstack.compute.v2.server.Server.force_delete", self._verify2(
self.proxy.delete_server, "openstack.compute.v2.server.Server.force_delete",
method_args=["value", False, True]) self.proxy.delete_server,
method_args=["value", False, True],
expected_args=[self.proxy])
def test_server_find(self): def test_server_find(self):
self.verify_find(self.proxy.find_server, server.Server) self.verify_find(self.proxy.find_server, server.Server)
@@ -680,23 +688,28 @@ class TestCompute(TestComputeProxy):
self.verify_wait_for_status( self.verify_wait_for_status(
self.proxy.wait_for_server, self.proxy.wait_for_server,
method_args=[value], method_args=[value],
expected_args=[value, 'ACTIVE', ['ERROR'], 2, 120]) expected_args=[self.proxy, value, 'ACTIVE', ['ERROR'], 2, 120])
def test_server_resize(self): def test_server_resize(self):
self._verify("openstack.compute.v2.server.Server.resize", self._verify2(
self.proxy.resize_server, "openstack.compute.v2.server.Server.resize",
method_args=["value", "test-flavor"], self.proxy.resize_server,
expected_args=["test-flavor"]) method_args=["value", "test-flavor"],
expected_args=[self.proxy, "test-flavor"])
def test_server_confirm_resize(self): def test_server_confirm_resize(self):
self._verify("openstack.compute.v2.server.Server.confirm_resize", self._verify2(
self.proxy.confirm_server_resize, "openstack.compute.v2.server.Server.confirm_resize",
method_args=["value"]) self.proxy.confirm_server_resize,
method_args=["value"],
expected_args=[self.proxy])
def test_server_revert_resize(self): def test_server_revert_resize(self):
self._verify("openstack.compute.v2.server.Server.revert_resize", self._verify2(
self.proxy.revert_server_resize, "openstack.compute.v2.server.Server.revert_resize",
method_args=["value"]) self.proxy.revert_server_resize,
method_args=["value"],
expected_args=[self.proxy])
def test_server_rebuild(self): def test_server_rebuild(self):
id = 'test_image_id' id = 'test_image_id'
@@ -705,156 +718,191 @@ class TestCompute(TestComputeProxy):
# Case1: image object is provided # Case1: image object is provided
# NOTE: Inside of Server.rebuild is where image_obj gets converted # NOTE: Inside of Server.rebuild is where image_obj gets converted
# to an ID instead of object. # to an ID instead of object.
self._verify('openstack.compute.v2.server.Server.rebuild', self._verify2(
self.proxy.rebuild_server, 'openstack.compute.v2.server.Server.rebuild',
method_args=["value", "test_server", "test_pass"], self.proxy.rebuild_server,
method_kwargs={"metadata": {"k1": "v1"}, method_args=["value", "test_server", "test_pass"],
"image": image_obj}, method_kwargs={"metadata": {"k1": "v1"}, "image": image_obj},
expected_args=["test_server", "test_pass"], expected_args=[self.proxy, "test_server", "test_pass"],
expected_kwargs={"metadata": {"k1": "v1"}, expected_kwargs={"metadata": {"k1": "v1"}, "image": image_obj})
"image": image_obj})
# Case2: image name or id is provided # Case2: image name or id is provided
self._verify('openstack.compute.v2.server.Server.rebuild', self._verify2(
self.proxy.rebuild_server, 'openstack.compute.v2.server.Server.rebuild',
method_args=["value", "test_server", "test_pass"], self.proxy.rebuild_server,
method_kwargs={"metadata": {"k1": "v1"}, method_args=["value", "test_server", "test_pass"],
"image": id}, method_kwargs={"metadata": {"k1": "v1"}, "image": id},
expected_args=["test_server", "test_pass"], expected_args=[self.proxy, "test_server", "test_pass"],
expected_kwargs={"metadata": {"k1": "v1"}, expected_kwargs={"metadata": {"k1": "v1"}, "image": id})
"image": id})
def test_add_fixed_ip_to_server(self): def test_add_fixed_ip_to_server(self):
self._verify("openstack.compute.v2.server.Server.add_fixed_ip", self._verify2(
self.proxy.add_fixed_ip_to_server, "openstack.compute.v2.server.Server.add_fixed_ip",
method_args=["value", "network-id"], self.proxy.add_fixed_ip_to_server,
expected_args=["network-id"]) method_args=["value", "network-id"],
expected_args=[self.proxy, "network-id"])
def test_fixed_ip_from_server(self): def test_fixed_ip_from_server(self):
self._verify("openstack.compute.v2.server.Server.remove_fixed_ip", self._verify2(
self.proxy.remove_fixed_ip_from_server, "openstack.compute.v2.server.Server.remove_fixed_ip",
method_args=["value", "address"], self.proxy.remove_fixed_ip_from_server,
expected_args=["address"]) method_args=["value", "address"],
expected_args=[self.proxy, "address"])
def test_floating_ip_to_server(self): def test_floating_ip_to_server(self):
self._verify("openstack.compute.v2.server.Server.add_floating_ip", self._verify2(
self.proxy.add_floating_ip_to_server, "openstack.compute.v2.server.Server.add_floating_ip",
method_args=["value", "floating-ip"], self.proxy.add_floating_ip_to_server,
expected_args=["floating-ip"], method_args=["value", "floating-ip"],
expected_kwargs={'fixed_address': None}) expected_args=[self.proxy, "floating-ip"],
expected_kwargs={'fixed_address': None})
def test_add_floating_ip_to_server_with_fixed_addr(self): def test_add_floating_ip_to_server_with_fixed_addr(self):
self._verify("openstack.compute.v2.server.Server.add_floating_ip", self._verify2(
self.proxy.add_floating_ip_to_server, "openstack.compute.v2.server.Server.add_floating_ip",
method_args=["value", "floating-ip", 'fixed-addr'], self.proxy.add_floating_ip_to_server,
expected_args=["floating-ip"], method_args=["value", "floating-ip", 'fixed-addr'],
expected_kwargs={'fixed_address': 'fixed-addr'}) expected_args=[self.proxy, "floating-ip"],
expected_kwargs={'fixed_address': 'fixed-addr'})
def test_remove_floating_ip_from_server(self): def test_remove_floating_ip_from_server(self):
self._verify("openstack.compute.v2.server.Server.remove_floating_ip", self._verify2(
self.proxy.remove_floating_ip_from_server, "openstack.compute.v2.server.Server.remove_floating_ip",
method_args=["value", "address"], self.proxy.remove_floating_ip_from_server,
expected_args=["address"]) method_args=["value", "address"],
expected_args=[self.proxy, "address"])
def test_server_backup(self): def test_server_backup(self):
self._verify("openstack.compute.v2.server.Server.backup", self._verify2(
self.proxy.backup_server, "openstack.compute.v2.server.Server.backup",
method_args=["value", "name", "daily", 1], self.proxy.backup_server,
expected_args=["name", "daily", 1]) method_args=["value", "name", "daily", 1],
expected_args=[self.proxy, "name", "daily", 1])
def test_server_pause(self): def test_server_pause(self):
self._verify("openstack.compute.v2.server.Server.pause", self._verify2(
self.proxy.pause_server, "openstack.compute.v2.server.Server.pause",
method_args=["value"]) self.proxy.pause_server,
method_args=["value"],
expected_args=[self.proxy])
def test_server_unpause(self): def test_server_unpause(self):
self._verify("openstack.compute.v2.server.Server.unpause", self._verify2(
self.proxy.unpause_server, "openstack.compute.v2.server.Server.unpause",
method_args=["value"]) self.proxy.unpause_server,
method_args=["value"],
expected_args=[self.proxy])
def test_server_suspend(self): def test_server_suspend(self):
self._verify("openstack.compute.v2.server.Server.suspend", self._verify2(
self.proxy.suspend_server, "openstack.compute.v2.server.Server.suspend",
method_args=["value"]) self.proxy.suspend_server,
method_args=["value"],
expected_args=[self.proxy])
def test_server_resume(self): def test_server_resume(self):
self._verify("openstack.compute.v2.server.Server.resume", self._verify2(
self.proxy.resume_server, "openstack.compute.v2.server.Server.resume",
method_args=["value"]) self.proxy.resume_server,
method_args=["value"],
expected_args=[self.proxy])
def test_server_lock(self): def test_server_lock(self):
self._verify("openstack.compute.v2.server.Server.lock", self._verify2(
self.proxy.lock_server, "openstack.compute.v2.server.Server.lock",
method_args=["value"]) self.proxy.lock_server,
method_args=["value"],
expected_args=[self.proxy])
def test_server_unlock(self): def test_server_unlock(self):
self._verify("openstack.compute.v2.server.Server.unlock", self._verify2(
self.proxy.unlock_server, "openstack.compute.v2.server.Server.unlock",
method_args=["value"]) self.proxy.unlock_server,
method_args=["value"],
expected_args=[self.proxy])
def test_server_rescue(self): def test_server_rescue(self):
self._verify("openstack.compute.v2.server.Server.rescue", self._verify2(
self.proxy.rescue_server, "openstack.compute.v2.server.Server.rescue",
method_args=["value"], self.proxy.rescue_server,
expected_kwargs={"admin_pass": None, "image_ref": None}) method_args=["value"],
expected_args=[self.proxy],
expected_kwargs={"admin_pass": None, "image_ref": None})
def test_server_rescue_with_options(self): def test_server_rescue_with_options(self):
self._verify("openstack.compute.v2.server.Server.rescue", self._verify2(
self.proxy.rescue_server, "openstack.compute.v2.server.Server.rescue",
method_args=["value", 'PASS', 'IMG'], self.proxy.rescue_server,
expected_kwargs={"admin_pass": 'PASS', method_args=["value", 'PASS', 'IMG'],
"image_ref": 'IMG'}) expected_args=[self.proxy],
expected_kwargs={"admin_pass": 'PASS', "image_ref": 'IMG'})
def test_server_unrescue(self): def test_server_unrescue(self):
self._verify("openstack.compute.v2.server.Server.unrescue", self._verify2(
self.proxy.unrescue_server, "openstack.compute.v2.server.Server.unrescue",
method_args=["value"]) self.proxy.unrescue_server,
method_args=["value"],
expected_args=[self.proxy])
def test_server_evacuate(self): def test_server_evacuate(self):
self._verify("openstack.compute.v2.server.Server.evacuate", self._verify2(
self.proxy.evacuate_server, "openstack.compute.v2.server.Server.evacuate",
method_args=["value"], self.proxy.evacuate_server,
expected_kwargs={"host": None, "admin_pass": None, method_args=["value"],
"force": None}) expected_args=[self.proxy],
expected_kwargs={"host": None, "admin_pass": None, "force": None})
def test_server_evacuate_with_options(self): def test_server_evacuate_with_options(self):
self._verify("openstack.compute.v2.server.Server.evacuate", self._verify2(
self.proxy.evacuate_server, "openstack.compute.v2.server.Server.evacuate",
method_args=["value", 'HOST2', 'NEW_PASS', True], self.proxy.evacuate_server,
expected_kwargs={"host": "HOST2", method_args=["value", 'HOST2', 'NEW_PASS', True],
"admin_pass": 'NEW_PASS', expected_args=[self.proxy],
"force": True}) expected_kwargs={
"host": "HOST2", "admin_pass": 'NEW_PASS', "force": True})
def test_server_start(self): def test_server_start(self):
self._verify("openstack.compute.v2.server.Server.start", self._verify2(
self.proxy.start_server, "openstack.compute.v2.server.Server.start",
method_args=["value"]) self.proxy.start_server,
method_args=["value"],
expected_args=[self.proxy])
def test_server_stop(self): def test_server_stop(self):
self._verify("openstack.compute.v2.server.Server.stop", self._verify2(
self.proxy.stop_server, "openstack.compute.v2.server.Server.stop",
method_args=["value"]) self.proxy.stop_server,
method_args=["value"],
expected_args=[self.proxy])
def test_server_shelve(self): def test_server_shelve(self):
self._verify("openstack.compute.v2.server.Server.shelve", self._verify2(
self.proxy.shelve_server, "openstack.compute.v2.server.Server.shelve",
method_args=["value"]) self.proxy.shelve_server,
method_args=["value"],
expected_args=[self.proxy])
def test_server_unshelve(self): def test_server_unshelve(self):
self._verify("openstack.compute.v2.server.Server.unshelve", self._verify2(
self.proxy.unshelve_server, "openstack.compute.v2.server.Server.unshelve",
method_args=["value"]) self.proxy.unshelve_server,
method_args=["value"],
expected_args=[self.proxy])
def test_get_server_output(self): def test_get_server_output(self):
self._verify("openstack.compute.v2.server.Server.get_console_output", self._verify2(
self.proxy.get_server_console_output, "openstack.compute.v2.server.Server.get_console_output",
method_args=["value"], self.proxy.get_server_console_output,
expected_kwargs={"length": None}) method_args=["value"],
expected_args=[self.proxy],
expected_kwargs={"length": None})
self._verify("openstack.compute.v2.server.Server.get_console_output", self._verify2(
self.proxy.get_server_console_output, "openstack.compute.v2.server.Server.get_console_output",
method_args=["value", 1], self.proxy.get_server_console_output,
expected_kwargs={"length": 1}) method_args=["value", 1],
expected_args=[self.proxy],
expected_kwargs={"length": 1})
def test_availability_zones_not_detailed(self): def test_availability_zones_not_detailed(self):
self.verify_list(self.proxy.availability_zones, self.verify_list(self.proxy.availability_zones,
@@ -947,32 +995,33 @@ class TestCompute(TestComputeProxy):
self.verify_list(self.proxy.server_groups, server_group.ServerGroup) self.verify_list(self.proxy.server_groups, server_group.ServerGroup)
def test_live_migrate_server(self): def test_live_migrate_server(self):
self._verify('openstack.compute.v2.server.Server.live_migrate', self._verify2(
self.proxy.live_migrate_server, 'openstack.compute.v2.server.Server.live_migrate',
method_args=["value", "host1", False], self.proxy.live_migrate_server,
expected_args=["host1"], method_args=["value", "host1", False],
expected_kwargs={'force': False, 'block_migration': None}) expected_args=[self.proxy, "host1"],
expected_kwargs={'force': False, 'block_migration': None})
def test_fetch_security_groups(self): def test_fetch_security_groups(self):
self._verify( self._verify2(
'openstack.compute.v2.server.Server.fetch_security_groups', 'openstack.compute.v2.server.Server.fetch_security_groups',
self.proxy.fetch_server_security_groups, self.proxy.fetch_server_security_groups,
method_args=["value"], method_args=["value"],
expected_args=[]) expected_args=[self.proxy])
def test_add_security_groups(self): def test_add_security_groups(self):
self._verify( self._verify2(
'openstack.compute.v2.server.Server.add_security_group', 'openstack.compute.v2.server.Server.add_security_group',
self.proxy.add_security_group_to_server, self.proxy.add_security_group_to_server,
method_args=["value", {'id': 'id', 'name': 'sg'}], method_args=["value", {'id': 'id', 'name': 'sg'}],
expected_args=['sg']) expected_args=[self.proxy, 'sg'])
def test_remove_security_groups(self): def test_remove_security_groups(self):
self._verify( self._verify2(
'openstack.compute.v2.server.Server.remove_security_group', 'openstack.compute.v2.server.Server.remove_security_group',
self.proxy.remove_security_group_from_server, self.proxy.remove_security_group_from_server,
method_args=["value", {'id': 'id', 'name': 'sg'}], method_args=["value", {'id': 'id', 'name': 'sg'}],
expected_args=['sg']) expected_args=[self.proxy, 'sg'])
def test_create_server_remote_console(self): def test_create_server_remote_console(self):
self.verify_create( self.verify_create(
@@ -982,11 +1031,11 @@ class TestCompute(TestComputeProxy):
expected_kwargs={"server_id": "test_id", "type": "fake"}) expected_kwargs={"server_id": "test_id", "type": "fake"})
def test_get_console_url(self): def test_get_console_url(self):
self._verify( self._verify2(
'openstack.compute.v2.server.Server.get_console_url', 'openstack.compute.v2.server.Server.get_console_url',
self.proxy.get_server_console_url, self.proxy.get_server_console_url,
method_args=["value", "console_type"], method_args=["value", "console_type"],
expected_args=["console_type"]) expected_args=[self.proxy, "console_type"])
@mock.patch('openstack.utils.supports_microversion', autospec=True) @mock.patch('openstack.utils.supports_microversion', autospec=True)
@mock.patch('openstack.compute.v2._proxy.Proxy._create', autospec=True) @mock.patch('openstack.compute.v2._proxy.Proxy._create', autospec=True)

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

@@ -54,10 +54,11 @@ class TestImageProxy(test_proxy_base.TestProxyBase):
def test_image_import(self): def test_image_import(self):
original_image = image.Image(**EXAMPLE) original_image = image.Image(**EXAMPLE)
self._verify( self._verify2(
"openstack.image.v2.image.Image.import_image", "openstack.image.v2.image.Image.import_image",
self.proxy.import_image, self.proxy.import_image,
method_args=[original_image, "method", "uri"], method_args=[original_image, "method", "uri"],
expected_args=[self.proxy],
expected_kwargs={ expected_kwargs={
"method": "method", "method": "method",
"store": None, "store": None,
@@ -218,17 +219,21 @@ class TestImageProxy(test_proxy_base.TestProxyBase):
def test_image_download(self): def test_image_download(self):
original_image = image.Image(**EXAMPLE) original_image = image.Image(**EXAMPLE)
self._verify('openstack.image.v2.image.Image.download', self._verify2(
self.proxy.download_image, 'openstack.image.v2.image.Image.download',
method_args=[original_image], self.proxy.download_image,
method_kwargs={ method_args=[original_image],
'output': 'some_output', method_kwargs={
'chunk_size': 1, 'output': 'some_output',
'stream': True 'chunk_size': 1,
}, 'stream': True
expected_kwargs={'output': 'some_output', },
'chunk_size': 1, expected_args=[self.proxy],
'stream': True}) expected_kwargs={
'output': 'some_output',
'chunk_size': 1,
'stream': True,
})
@mock.patch("openstack.image.v2.image.Image.fetch") @mock.patch("openstack.image.v2.image.Image.fetch")
def test_image_stage(self, mock_fetch): def test_image_stage(self, mock_fetch):
@@ -289,26 +294,32 @@ class TestImageProxy(test_proxy_base.TestProxyBase):
self.verify_list(self.proxy.images, image.Image) self.verify_list(self.proxy.images, image.Image)
def test_add_tag(self): def test_add_tag(self):
self._verify("openstack.image.v2.image.Image.add_tag", self._verify2(
self.proxy.add_tag, "openstack.image.v2.image.Image.add_tag",
method_args=["image", "tag"], self.proxy.add_tag,
expected_args=["tag"]) method_args=["image", "tag"],
expected_args=[self.proxy, "tag"])
def test_remove_tag(self): def test_remove_tag(self):
self._verify("openstack.image.v2.image.Image.remove_tag", self._verify2(
self.proxy.remove_tag, "openstack.image.v2.image.Image.remove_tag",
method_args=["image", "tag"], self.proxy.remove_tag,
expected_args=["tag"]) method_args=["image", "tag"],
expected_args=[self.proxy, "tag"])
def test_deactivate_image(self): def test_deactivate_image(self):
self._verify("openstack.image.v2.image.Image.deactivate", self._verify2(
self.proxy.deactivate_image, "openstack.image.v2.image.Image.deactivate",
method_args=["image"]) self.proxy.deactivate_image,
method_args=["image"],
expected_args=[self.proxy])
def test_reactivate_image(self): def test_reactivate_image(self):
self._verify("openstack.image.v2.image.Image.reactivate", self._verify2(
self.proxy.reactivate_image, "openstack.image.v2.image.Image.reactivate",
method_args=["image"]) self.proxy.reactivate_image,
method_args=["image"],
expected_args=[self.proxy])
def test_member_create(self): def test_member_create(self):
self.verify_create(self.proxy.add_member, member.Member, self.verify_create(self.proxy.add_member, member.Member,

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

@@ -50,10 +50,11 @@ class TestMessageProxy(test_proxy_base.TestProxyBase):
def test_message_post(self, mock_get_resource): def test_message_post(self, mock_get_resource):
message_obj = message.Message(queue_name="test_queue") message_obj = message.Message(queue_name="test_queue")
mock_get_resource.return_value = message_obj mock_get_resource.return_value = message_obj
self._verify("openstack.message.v2.message.Message.post", self._verify2(
self.proxy.post_message, "openstack.message.v2.message.Message.post",
method_args=["test_queue", ["msg1", "msg2"]], self.proxy.post_message,
expected_args=[["msg1", "msg2"]]) method_args=["test_queue", ["msg1", "msg2"]],
expected_args=[self.proxy, ["msg1", "msg2"]])
mock_get_resource.assert_called_once_with(message.Message, None, mock_get_resource.assert_called_once_with(message.Message, None,
queue_name="test_queue") queue_name="test_queue")
@@ -128,10 +129,12 @@ class TestMessageProxy(test_proxy_base.TestProxyBase):
queue_name="test_queue") queue_name="test_queue")
def test_subscription_create(self): def test_subscription_create(self):
self._verify("openstack.message.v2.subscription.Subscription.create", self._verify2(
self.proxy.create_subscription, "openstack.message.v2.subscription.Subscription.create",
method_args=["test_queue"], self.proxy.create_subscription,
expected_kwargs={"base_path": None}) method_args=["test_queue"],
expected_args=[self.proxy],
expected_kwargs={"base_path": None})
@mock.patch.object(proxy_base.Proxy, '_get_resource') @mock.patch.object(proxy_base.Proxy, '_get_resource')
def test_subscription_get(self, mock_get_resource): def test_subscription_get(self, mock_get_resource):
@@ -178,10 +181,12 @@ class TestMessageProxy(test_proxy_base.TestProxyBase):
queue_name="test_queue") queue_name="test_queue")
def test_claim_create(self): def test_claim_create(self):
self._verify("openstack.message.v2.claim.Claim.create", self._verify2(
self.proxy.create_claim, "openstack.message.v2.claim.Claim.create",
method_args=["test_queue"], self.proxy.create_claim,
expected_kwargs={"base_path": None}) method_args=["test_queue"],
expected_args=[self.proxy],
expected_kwargs={"base_path": None})
def test_claim_get(self): def test_claim_get(self):
self._verify2("openstack.proxy.Proxy._get", self._verify2("openstack.proxy.Proxy._get",

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

@@ -891,11 +891,13 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
x_router = router.Router.new(id="ROUTER_ID") x_router = router.Router.new(id="ROUTER_ID")
mock_get.return_value = x_router mock_get.return_value = x_router
self._verify("openstack.network.v2.router.Router.add_interface", self._verify2(
self.proxy.add_interface_to_router, "openstack.network.v2.router.Router.add_interface",
method_args=["FAKE_ROUTER"], self.proxy.add_interface_to_router,
method_kwargs={"port_id": "PORT"}, method_args=["FAKE_ROUTER"],
expected_kwargs={"port_id": "PORT"}) method_kwargs={"port_id": "PORT"},
expected_args=[self.proxy],
expected_kwargs={"port_id": "PORT"})
mock_get.assert_called_once_with(router.Router, "FAKE_ROUTER") mock_get.assert_called_once_with(router.Router, "FAKE_ROUTER")
@mock.patch.object(proxy_base.Proxy, '_get_resource') @mock.patch.object(proxy_base.Proxy, '_get_resource')
@@ -905,11 +907,13 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
x_router = router.Router.new(id="ROUTER_ID") x_router = router.Router.new(id="ROUTER_ID")
mock_get.return_value = x_router mock_get.return_value = x_router
self._verify("openstack.network.v2.router.Router.add_interface", self._verify2(
self.proxy.add_interface_to_router, "openstack.network.v2.router.Router.add_interface",
method_args=["FAKE_ROUTER"], self.proxy.add_interface_to_router,
method_kwargs={"subnet_id": "SUBNET"}, method_args=["FAKE_ROUTER"],
expected_kwargs={"subnet_id": "SUBNET"}) method_kwargs={"subnet_id": "SUBNET"},
expected_args=[self.proxy],
expected_kwargs={"subnet_id": "SUBNET"})
mock_get.assert_called_once_with(router.Router, "FAKE_ROUTER") mock_get.assert_called_once_with(router.Router, "FAKE_ROUTER")
@mock.patch.object(proxy_base.Proxy, '_get_resource') @mock.patch.object(proxy_base.Proxy, '_get_resource')
@@ -919,11 +923,13 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
x_router = router.Router.new(id="ROUTER_ID") x_router = router.Router.new(id="ROUTER_ID")
mock_get.return_value = x_router mock_get.return_value = x_router
self._verify("openstack.network.v2.router.Router.remove_interface", self._verify2(
self.proxy.remove_interface_from_router, "openstack.network.v2.router.Router.remove_interface",
method_args=["FAKE_ROUTER"], self.proxy.remove_interface_from_router,
method_kwargs={"port_id": "PORT"}, method_args=["FAKE_ROUTER"],
expected_kwargs={"port_id": "PORT"}) method_kwargs={"port_id": "PORT"},
expected_args=[self.proxy],
expected_kwargs={"port_id": "PORT"})
mock_get.assert_called_once_with(router.Router, "FAKE_ROUTER") mock_get.assert_called_once_with(router.Router, "FAKE_ROUTER")
@mock.patch.object(proxy_base.Proxy, '_get_resource') @mock.patch.object(proxy_base.Proxy, '_get_resource')
@@ -933,11 +939,13 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
x_router = router.Router.new(id="ROUTER_ID") x_router = router.Router.new(id="ROUTER_ID")
mock_get.return_value = x_router mock_get.return_value = x_router
self._verify("openstack.network.v2.router.Router.remove_interface", self._verify2(
self.proxy.remove_interface_from_router, "openstack.network.v2.router.Router.remove_interface",
method_args=["FAKE_ROUTER"], self.proxy.remove_interface_from_router,
method_kwargs={"subnet_id": "SUBNET"}, method_args=["FAKE_ROUTER"],
expected_kwargs={"subnet_id": "SUBNET"}) method_kwargs={"subnet_id": "SUBNET"},
expected_args=[self.proxy],
expected_kwargs={"subnet_id": "SUBNET"})
mock_get.assert_called_once_with(router.Router, "FAKE_ROUTER") mock_get.assert_called_once_with(router.Router, "FAKE_ROUTER")
@mock.patch.object(proxy_base.Proxy, '_get_resource') @mock.patch.object(proxy_base.Proxy, '_get_resource')
@@ -947,11 +955,13 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
x_router = router.Router.new(id="ROUTER_ID") x_router = router.Router.new(id="ROUTER_ID")
mock_get.return_value = x_router mock_get.return_value = x_router
self._verify("openstack.network.v2.router.Router.add_extra_routes", self._verify2(
self.proxy.add_extra_routes_to_router, "openstack.network.v2.router.Router.add_extra_routes",
method_args=["FAKE_ROUTER"], self.proxy.add_extra_routes_to_router,
method_kwargs={"body": {"router": {"routes": []}}}, method_args=["FAKE_ROUTER"],
expected_kwargs={"body": {"router": {"routes": []}}}) method_kwargs={"body": {"router": {"routes": []}}},
expected_args=[self.proxy],
expected_kwargs={"body": {"router": {"routes": []}}})
mock_get.assert_called_once_with(router.Router, "FAKE_ROUTER") mock_get.assert_called_once_with(router.Router, "FAKE_ROUTER")
@mock.patch.object(proxy_base.Proxy, '_get_resource') @mock.patch.object(proxy_base.Proxy, '_get_resource')
@@ -961,11 +971,13 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
x_router = router.Router.new(id="ROUTER_ID") x_router = router.Router.new(id="ROUTER_ID")
mock_get.return_value = x_router mock_get.return_value = x_router
self._verify("openstack.network.v2.router.Router.remove_extra_routes", self._verify2(
self.proxy.remove_extra_routes_from_router, "openstack.network.v2.router.Router.remove_extra_routes",
method_args=["FAKE_ROUTER"], self.proxy.remove_extra_routes_from_router,
method_kwargs={"body": {"router": {"routes": []}}}, method_args=["FAKE_ROUTER"],
expected_kwargs={"body": {"router": {"routes": []}}}) method_kwargs={"body": {"router": {"routes": []}}},
expected_args=[self.proxy],
expected_kwargs={"body": {"router": {"routes": []}}})
mock_get.assert_called_once_with(router.Router, "FAKE_ROUTER") mock_get.assert_called_once_with(router.Router, "FAKE_ROUTER")
@mock.patch.object(proxy_base.Proxy, '_get_resource') @mock.patch.object(proxy_base.Proxy, '_get_resource')
@@ -974,11 +986,13 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
x_router = router.Router.new(id="ROUTER_ID") x_router = router.Router.new(id="ROUTER_ID")
mock_get.return_value = x_router mock_get.return_value = x_router
self._verify("openstack.network.v2.router.Router.add_gateway", self._verify2(
self.proxy.add_gateway_to_router, "openstack.network.v2.router.Router.add_gateway",
method_args=["FAKE_ROUTER"], self.proxy.add_gateway_to_router,
method_kwargs={"foo": "bar"}, method_args=["FAKE_ROUTER"],
expected_kwargs={"foo": "bar"}) method_kwargs={"foo": "bar"},
expected_args=[self.proxy],
expected_kwargs={"foo": "bar"})
mock_get.assert_called_once_with(router.Router, "FAKE_ROUTER") mock_get.assert_called_once_with(router.Router, "FAKE_ROUTER")
@mock.patch.object(proxy_base.Proxy, '_get_resource') @mock.patch.object(proxy_base.Proxy, '_get_resource')
@@ -987,11 +1001,13 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
x_router = router.Router.new(id="ROUTER_ID") x_router = router.Router.new(id="ROUTER_ID")
mock_get.return_value = x_router mock_get.return_value = x_router
self._verify("openstack.network.v2.router.Router.remove_gateway", self._verify2(
self.proxy.remove_gateway_from_router, "openstack.network.v2.router.Router.remove_gateway",
method_args=["FAKE_ROUTER"], self.proxy.remove_gateway_from_router,
method_kwargs={"foo": "bar"}, method_args=["FAKE_ROUTER"],
expected_kwargs={"foo": "bar"}) method_kwargs={"foo": "bar"},
expected_args=[self.proxy],
expected_kwargs={"foo": "bar"})
mock_get.assert_called_once_with(router.Router, "FAKE_ROUTER") mock_get.assert_called_once_with(router.Router, "FAKE_ROUTER")
def test_router_hosting_l3_agents_list(self): def test_router_hosting_l3_agents_list(self):
@@ -1341,11 +1357,12 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
def test_set_tags(self): def test_set_tags(self):
x_network = network.Network.new(id='NETWORK_ID') x_network = network.Network.new(id='NETWORK_ID')
self._verify('openstack.network.v2.network.Network.set_tags', self._verify2(
self.proxy.set_tags, 'openstack.network.v2.network.Network.set_tags',
method_args=[x_network, ['TAG1', 'TAG2']], self.proxy.set_tags,
expected_args=[['TAG1', 'TAG2']], method_args=[x_network, ['TAG1', 'TAG2']],
expected_result=mock.sentinel.result_set_tags) expected_args=[self.proxy, ['TAG1', 'TAG2']],
expected_result=mock.sentinel.result_set_tags)
@mock.patch('openstack.network.v2.network.Network.set_tags') @mock.patch('openstack.network.v2.network.Network.set_tags')
def test_set_tags_resource_without_tag_suport(self, mock_set_tags): def test_set_tags_resource_without_tag_suport(self, mock_set_tags):

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

@@ -55,22 +55,32 @@ class TestProxyBase(base.TestCase):
expected_result=None): expected_result=None):
with mock.patch(mock_method) as mocked: with mock.patch(mock_method) as mocked:
mocked.return_value = expected_result mocked.return_value = expected_result
if any([method_args, method_kwargs, if any([
expected_args, expected_kwargs]): method_args,
method_kwargs,
expected_args,
expected_kwargs,
]):
method_args = method_args or () method_args = method_args or ()
method_kwargs = method_kwargs or {} method_kwargs = method_kwargs or {}
expected_args = expected_args or () expected_args = expected_args or ()
expected_kwargs = expected_kwargs or {} expected_kwargs = expected_kwargs or {}
if method_result: if method_result:
self.assertEqual(method_result, test_method(*method_args, self.assertEqual(
**method_kwargs)) method_result,
test_method(*method_args, **method_kwargs),
)
else: else:
self.assertEqual(expected_result, test_method(*method_args, self.assertEqual(
**method_kwargs)) expected_result,
test_method(*method_args, **method_kwargs),
)
# Check how the mock was called in detail # Check how the mock was called in detail
(called_args, called_kwargs) = mocked.call_args called_args, called_kwargs = mocked.call_args
self.assertEqual(expected_args, list(called_args)) self.assertEqual(expected_args, list(called_args))
# NOTE(gtema): if base_path is not in expected_kwargs or empty # NOTE(gtema): if base_path is not in expected_kwargs or empty
# exclude it from the comparison, since some methods might # exclude it from the comparison, since some methods might
# still invoke method with None value # still invoke method with None value
@@ -88,6 +98,7 @@ class TestProxyBase(base.TestCase):
if ignore_missing is None: if ignore_missing is None:
expected_kwargs.pop('ignore_missing', None) expected_kwargs.pop('ignore_missing', None)
called_kwargs.pop('ignore_missing', None) called_kwargs.pop('ignore_missing', None)
self.assertDictEqual(expected_kwargs, called_kwargs) self.assertDictEqual(expected_kwargs, called_kwargs)
else: else:
self.assertEqual(expected_result, test_method()) self.assertEqual(expected_result, test_method())
@@ -275,4 +286,4 @@ class TestProxyBase(base.TestCase):
mock_method="openstack.resource.wait_for_status", mock_method="openstack.resource.wait_for_status",
**kwargs, **kwargs,
): ):
self._verify(mock_method, test_method, **kwargs) self._verify2(mock_method, test_method, **kwargs)