D7net
Home
Console
Upload
information
Create File
Create Folder
About
Tools
:
/
proc
/
3
/
cwd
/
opt
/
imh-python
/
lib
/
python3.9
/
site-packages
/
openstack
/
tests
/
unit
/
Filename :
test_proxy_base.py
back
Copy
# Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. from unittest import mock from openstack.tests.unit import base class TestProxyBase(base.TestCase): def setUp(self): super(TestProxyBase, self).setUp() self.session = mock.Mock() def _add_path_args_for_verify(self, path_args, method_args, expected_kwargs, value=None): if path_args is not None: if value is None: for key in path_args: method_args.append(path_args[key]) expected_kwargs['path_args'] = path_args def _verify(self, mock_method, test_method, method_args=None, method_kwargs=None, expected_args=None, expected_kwargs=None, 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): 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 {} if method_result: self.assertEqual(method_result, test_method(*method_args, **method_kwargs)) else: self.assertEqual(expected_result, test_method(*method_args, **method_kwargs)) # Check how the mock was called in detail (called_args, called_kwargs) = mocked.call_args self.assertEqual(list(called_args), expected_args) base_path = expected_kwargs.get('base_path', None) # NOTE(gtema): if base_path is not in expected_kwargs or empty # exclude it from the comparison, since some methods might # still invoke method with None value if not base_path: expected_kwargs.pop('base_path', None) called_kwargs.pop('base_path', None) self.assertDictEqual(called_kwargs, expected_kwargs) else: self.assertEqual(expected_result, test_method()) mocked.assert_called_with(test_method.__self__) def verify_create(self, test_method, resource_type, mock_method="openstack.proxy.Proxy._create", expected_result="result", **kwargs): the_kwargs = {"x": 1, "y": 2, "z": 3} method_kwargs = kwargs.pop("method_kwargs", the_kwargs) expected_args = kwargs.pop('expected_args', [resource_type]) # Default the_kwargs should be copied, since we might need to extend it expected_kwargs = kwargs.pop("expected_kwargs", the_kwargs.copy()) expected_kwargs["base_path"] = kwargs.pop("base_path", None) self._verify2(mock_method, test_method, expected_result=expected_result, method_kwargs=method_kwargs, expected_args=expected_args, expected_kwargs=expected_kwargs, **kwargs) def verify_delete(self, test_method, resource_type, ignore, input_path_args=None, expected_path_args=None, method_kwargs=None, expected_args=None, expected_kwargs=None, mock_method="openstack.proxy.Proxy._delete"): method_args = ["resource_or_id"] method_kwargs = method_kwargs or {} method_kwargs["ignore_missing"] = ignore if isinstance(input_path_args, dict): for key in input_path_args: method_kwargs[key] = input_path_args[key] elif isinstance(input_path_args, list): method_args = input_path_args expected_kwargs = expected_kwargs or {} expected_kwargs["ignore_missing"] = ignore if expected_path_args: expected_kwargs.update(expected_path_args) expected_args = expected_args or [resource_type, "resource_or_id"] self._verify2(mock_method, test_method, method_args=method_args, method_kwargs=method_kwargs, expected_args=expected_args, expected_kwargs=expected_kwargs) def verify_get(self, test_method, resource_type, value=None, args=None, mock_method="openstack.proxy.Proxy._get", ignore_value=False, **kwargs): the_value = value if value is None: the_value = [] if ignore_value else ["value"] expected_args = kwargs.pop("expected_args", []) expected_kwargs = kwargs.pop("expected_kwargs", {}) method_kwargs = kwargs.pop("method_kwargs", kwargs) if args: expected_kwargs["args"] = args if kwargs: expected_kwargs["path_args"] = kwargs if not expected_args: expected_args = [resource_type] + the_value self._verify2(mock_method, test_method, method_args=the_value, method_kwargs=method_kwargs or {}, expected_args=expected_args, expected_kwargs=expected_kwargs) def verify_get_overrided(self, proxy, resource_type, patch_target): with mock.patch(patch_target, autospec=True) as res: proxy._get_resource = mock.Mock(return_value=res) proxy._get(resource_type) res.fetch.assert_called_once_with( proxy, requires_id=True, base_path=None, error_message=mock.ANY) def verify_head(self, test_method, resource_type, mock_method="openstack.proxy.Proxy._head", value=None, **kwargs): the_value = [value] if value is not None else [] expected_kwargs = {"path_args": kwargs} if kwargs else {} self._verify2(mock_method, test_method, method_args=the_value, method_kwargs=kwargs, expected_args=[resource_type] + the_value, expected_kwargs=expected_kwargs) def verify_find(self, test_method, resource_type, value=None, mock_method="openstack.proxy.Proxy._find", path_args=None, **kwargs): method_args = value or ["name_or_id"] expected_kwargs = kwargs.pop('expected_kwargs', {}) self._add_path_args_for_verify(path_args, method_args, expected_kwargs, value=value) # TODO(briancurtin): if sub-tests worked in this mess of # test dependencies, the following would be a lot easier to work with. expected_kwargs["ignore_missing"] = False self._verify2(mock_method, test_method, method_args=method_args + [False], expected_args=[resource_type, "name_or_id"], expected_kwargs=expected_kwargs, expected_result="result", **kwargs) expected_kwargs["ignore_missing"] = True self._verify2(mock_method, test_method, method_args=method_args + [True], expected_args=[resource_type, "name_or_id"], expected_kwargs=expected_kwargs, expected_result="result", **kwargs) def verify_list(self, test_method, resource_type, mock_method="openstack.proxy.Proxy._list", **kwargs): expected_kwargs = kwargs.pop("expected_kwargs", {}) if 'paginated' in kwargs: expected_kwargs.update({"paginated": kwargs.pop('paginated')}) method_kwargs = kwargs.pop("method_kwargs", {}) expected_kwargs["base_path"] = kwargs.pop("base_path", None) self._verify2(mock_method, test_method, method_kwargs=method_kwargs, expected_args=[resource_type], expected_kwargs=expected_kwargs, expected_result=["result"], **kwargs) def verify_list_no_kwargs(self, test_method, resource_type, mock_method="openstack.proxy.Proxy._list"): self._verify2(mock_method, test_method, method_kwargs={}, expected_args=[resource_type], expected_kwargs={}, expected_result=["result"]) def verify_update(self, test_method, resource_type, value=None, mock_method="openstack.proxy.Proxy._update", expected_result="result", path_args=None, **kwargs): method_args = value or ["resource_or_id"] method_kwargs = kwargs.pop("method_kwargs", {}) method_kwargs.update({"x": 1, "y": 2, "z": 3}) expected_args = kwargs.pop("expected_args", ["resource_or_id"]) expected_kwargs = kwargs.pop("expected_kwargs", method_kwargs.copy()) expected_kwargs["base_path"] = kwargs.pop("base_path", None) self._add_path_args_for_verify(path_args, method_args, expected_kwargs, value=value) self._verify2(mock_method, test_method, expected_result=expected_result, method_args=method_args, method_kwargs=method_kwargs, expected_args=[resource_type] + expected_args, expected_kwargs=expected_kwargs, **kwargs) def verify_wait_for_status( self, test_method, mock_method="openstack.resource.wait_for_status", **kwargs): self._verify(mock_method, test_method, **kwargs)