X7ROOT File Manager
Current Path:
/opt/gsutil/third_party/apitools/apitools/base/py/testing
opt
/
gsutil
/
third_party
/
apitools
/
apitools
/
base
/
py
/
testing
/
📁
..
📄
__init__.py
(605 B)
📄
mock.py
(14.44 KB)
📄
mock_test.py
(12.87 KB)
Editing: mock.py
# # Copyright 2015 Google Inc. # # 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. """The mock module allows easy mocking of apitools clients. This module allows you to mock out the constructor of a particular apitools client, for a specific API and version. Then, when the client is created, it will be run against an expected session that you define. This way code that is not aware of the testing framework can construct new clients as normal, as long as it's all done within the context of a mock. """ import difflib import sys import six from apitools.base.protorpclite import messages from apitools.base.py import base_api from apitools.base.py import encoding from apitools.base.py import exceptions class Error(Exception): """Exceptions for this module.""" def _MessagesEqual(msg1, msg2): """Compare two protorpc messages for equality. Using python's == operator does not work in all cases, specifically when there is a list involved. Args: msg1: protorpc.messages.Message or [protorpc.messages.Message] or number or string, One of the messages to compare. msg2: protorpc.messages.Message or [protorpc.messages.Message] or number or string, One of the messages to compare. Returns: If the messages are isomorphic. """ if isinstance(msg1, list) and isinstance(msg2, list): if len(msg1) != len(msg2): return False return all(_MessagesEqual(x, y) for x, y in zip(msg1, msg2)) if (not isinstance(msg1, messages.Message) or not isinstance(msg2, messages.Message)): return msg1 == msg2 for field in msg1.all_fields(): field1 = getattr(msg1, field.name) field2 = getattr(msg2, field.name) if not _MessagesEqual(field1, field2): return False return True class UnexpectedRequestException(Error): def __init__(self, received_call, expected_call): expected_key, expected_request = expected_call received_key, received_request = received_call expected_repr = encoding.MessageToRepr( expected_request, multiline=True) received_repr = encoding.MessageToRepr( received_request, multiline=True) expected_lines = expected_repr.splitlines() received_lines = received_repr.splitlines() diff_lines = difflib.unified_diff(expected_lines, received_lines) diff = '\n'.join(diff_lines) if expected_key != received_key: msg = '\n'.join(( 'expected: {expected_key}({expected_request})', 'received: {received_key}({received_request})', '', )).format( expected_key=expected_key, expected_request=expected_repr, received_key=received_key, received_request=received_repr) super(UnexpectedRequestException, self).__init__(msg) else: msg = '\n'.join(( 'for request to {key},', 'expected: {expected_request}', 'received: {received_request}', 'diff: {diff}', '', )).format( key=expected_key, expected_request=expected_repr, received_request=received_repr, diff=diff) super(UnexpectedRequestException, self).__init__(msg) class ExpectedRequestsException(Error): def __init__(self, expected_calls): msg = 'expected:\n' for (key, request) in expected_calls: msg += '{key}({request})\n'.format( key=key, request=encoding.MessageToRepr(request, multiline=True)) super(ExpectedRequestsException, self).__init__(msg) class _ExpectedRequestResponse(object): """Encapsulation of an expected request and corresponding response.""" def __init__(self, key, request, response=None, exception=None): self.__key = key self.__request = request if response and exception: raise exceptions.ConfigurationValueError( 'Should specify at most one of response and exception') if response and isinstance(response, exceptions.Error): raise exceptions.ConfigurationValueError( 'Responses should not be an instance of Error') if exception and not isinstance(exception, exceptions.Error): raise exceptions.ConfigurationValueError( 'Exceptions must be instances of Error') self.__response = response self.__exception = exception @property def key(self): return self.__key @property def request(self): return self.__request def ValidateAndRespond(self, key, request): """Validate that key and request match expectations, and respond if so. Args: key: str, Actual key to compare against expectations. request: protorpc.messages.Message or [protorpc.messages.Message] or number or string, Actual request to compare againt expectations Raises: UnexpectedRequestException: If key or request dont match expectations. apitools_base.Error: If a non-None exception is specified to be thrown. Returns: The response that was specified to be returned. """ if key != self.__key or not (self.__request == request or _MessagesEqual(request, self.__request)): raise UnexpectedRequestException((key, request), (self.__key, self.__request)) if self.__exception: # Can only throw apitools_base.Error. raise self.__exception # pylint: disable=raising-bad-type return self.__response class _MockedMethod(object): """A mocked API service method.""" def __init__(self, key, mocked_client, real_method): self.__name__ = real_method.__name__ self.__key = key self.__mocked_client = mocked_client self.__real_method = real_method self.method_config = real_method.method_config config = self.method_config() self.__request_type = getattr(self.__mocked_client.MESSAGES_MODULE, config.request_type_name) self.__response_type = getattr(self.__mocked_client.MESSAGES_MODULE, config.response_type_name) def _TypeCheck(self, msg, is_request): """Ensure the given message is of the expected type of this method. Args: msg: The message instance to check. is_request: True to validate against the expected request type, False to validate against the expected response type. Raises: exceptions.ConfigurationValueError: If the type of the message was not correct. """ if is_request: mode = 'request' real_type = self.__request_type else: mode = 'response' real_type = self.__response_type if not isinstance(msg, real_type): raise exceptions.ConfigurationValueError( 'Expected {} is not of the correct type for method [{}].\n' ' Required: [{}]\n' ' Given: [{}]'.format( mode, self.__key, real_type, type(msg))) def Expect(self, request, response=None, exception=None, enable_type_checking=True, **unused_kwargs): """Add an expectation on the mocked method. Exactly one of response and exception should be specified. Args: request: The request that should be expected response: The response that should be returned or None if exception is provided. exception: An exception that should be thrown, or None. enable_type_checking: When true, the message type of the request and response (if provided) will be checked against the types required by this method. """ # TODO(jasmuth): the unused_kwargs provides a placeholder for # future things that can be passed to Expect(), like special # params to the method call. # Ensure that the registered request and response mocks actually # match what this method accepts and returns. if enable_type_checking: self._TypeCheck(request, is_request=True) if response: self._TypeCheck(response, is_request=False) # pylint: disable=protected-access # Class in same module. self.__mocked_client._request_responses.append( _ExpectedRequestResponse(self.__key, request, response=response, exception=exception)) # pylint: enable=protected-access def __call__(self, request, **unused_kwargs): # TODO(jasmuth): allow the testing code to expect certain # values in these currently unused_kwargs, especially the # upload parameter used by media-heavy services like bigquery # or bigstore. # pylint: disable=protected-access # Class in same module. if self.__mocked_client._request_responses: request_response = self.__mocked_client._request_responses.pop(0) else: raise UnexpectedRequestException( (self.__key, request), (None, None)) # pylint: enable=protected-access response = request_response.ValidateAndRespond(self.__key, request) if response is None and self.__real_method: response = self.__real_method(request) print(encoding.MessageToRepr( response, multiline=True, shortstrings=True)) return response return response def _MakeMockedService(api_name, collection_name, mock_client, service, real_service): class MockedService(base_api.BaseApiService): pass for method in service.GetMethodsList(): real_method = None if real_service: real_method = getattr(real_service, method) setattr(MockedService, method, _MockedMethod(api_name + '.' + collection_name + '.' + method, mock_client, real_method)) return MockedService class Client(object): """Mock an apitools client.""" def __init__(self, client_class, real_client=None): """Mock an apitools API, given its class. Args: client_class: The class for the API. eg, if you from apis.sqladmin import v1beta3 then you can pass v1beta3.SqladminV1beta3 to this class and anything within its context will use your mocked version. real_client: apitools Client, The client to make requests against when the expected response is None. """ if not real_client: real_client = client_class(get_credentials=False) self.__orig_class = self.__class__ self.__client_class = client_class self.__real_service_classes = {} self.__real_client = real_client self._request_responses = [] self.__real_include_fields = None def __enter__(self): return self.Mock() def Mock(self): """Stub out the client class with mocked services.""" client = self.__real_client or self.__client_class( get_credentials=False) class Patched(self.__class__, self.__client_class): pass self.__class__ = Patched for name in dir(self.__client_class): service_class = getattr(self.__client_class, name) if not isinstance(service_class, type): continue if not issubclass(service_class, base_api.BaseApiService): continue self.__real_service_classes[name] = service_class # pylint: disable=protected-access collection_name = service_class._NAME # pylint: enable=protected-access api_name = '%s_%s' % (self.__client_class._PACKAGE, self.__client_class._URL_VERSION) mocked_service_class = _MakeMockedService( api_name, collection_name, self, service_class, service_class(client) if self.__real_client else None) setattr(self.__client_class, name, mocked_service_class) setattr(self, collection_name, mocked_service_class(self)) self.__real_include_fields = self.__client_class.IncludeFields self.__client_class.IncludeFields = self.IncludeFields # pylint: disable=attribute-defined-outside-init self._url = client._url self._http = client._http return self def __exit__(self, exc_type, value, traceback): is_active_exception = value is not None self.Unmock(suppress=is_active_exception) if is_active_exception: six.reraise(exc_type, value, traceback) return True def Unmock(self, suppress=False): self.__class__ = self.__orig_class for name, service_class in self.__real_service_classes.items(): setattr(self.__client_class, name, service_class) delattr(self, service_class._NAME) self.__real_service_classes = {} del self._url del self._http self.__client_class.IncludeFields = self.__real_include_fields self.__real_include_fields = None requests = [(rq_rs.key, rq_rs.request) for rq_rs in self._request_responses] self._request_responses = [] if requests and not suppress and sys.exc_info()[1] is None: raise ExpectedRequestsException(requests) def IncludeFields(self, include_fields): if self.__real_client: return self.__real_include_fields(self.__real_client, include_fields)
Upload File
Create Folder