Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified
📁 ..	-	2025-06-25 08:26
📄 __init__.py #!/usr/bin/env python # # 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. """Top-level imports for apitools base files.""" # pylint:disable=wildcard-import # pylint:disable=redefined-builtin from apitools.base.py.base_api import * from apitools.base.py.batch import * from apitools.base.py.credentials_lib import * from apitools.base.py.encoding import * from apitools.base.py.exceptions import * from apitools.base.py.extra_types import * from apitools.base.py.http_wrapper import * from apitools.base.py.list_pager import * from apitools.base.py.transfer import * from apitools.base.py.util import * try: # pylint:disable=no-name-in-module from apitools.base.py.internal import * except ImportError: pass	1.22 KB	2025-03-11 18:02
📄 base_api.py #!/usr/bin/env python # # 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. """Base class for api services.""" import base64 import contextlib import datetime import logging import pprint import six from six.moves import http_client from six.moves import urllib from apitools.base.protorpclite import message_types from apitools.base.protorpclite import messages from apitools.base.py import encoding from apitools.base.py import exceptions from apitools.base.py import http_wrapper from apitools.base.py import util __all__ = [ 'ApiMethodInfo', 'ApiUploadInfo', 'BaseApiClient', 'BaseApiService', 'NormalizeApiEndpoint', ] # TODO(craigcitro): Remove this once we quiet the spurious logging in # oauth2client (or drop oauth2client). logging.getLogger('oauth2client.util').setLevel(logging.ERROR) _MAX_URL_LENGTH = 2048 class ApiUploadInfo(messages.Message): """Media upload information for a method. Fields: accept: (repeated) MIME Media Ranges for acceptable media uploads to this method. max_size: (integer) Maximum size of a media upload, such as 3MB or 1TB (converted to an integer). resumable_path: Path to use for resumable uploads. resumable_multipart: (boolean) Whether or not the resumable endpoint supports multipart uploads. simple_path: Path to use for simple uploads. simple_multipart: (boolean) Whether or not the simple endpoint supports multipart uploads. """ accept = messages.StringField(1, repeated=True) max_size = messages.IntegerField(2) resumable_path = messages.StringField(3) resumable_multipart = messages.BooleanField(4) simple_path = messages.StringField(5) simple_multipart = messages.BooleanField(6) class ApiMethodInfo(messages.Message): """Configuration info for an API method. All fields are strings unless noted otherwise. Fields: relative_path: Relative path for this method. flat_path: Expanded version (if any) of relative_path. method_id: ID for this method. http_method: HTTP verb to use for this method. path_params: (repeated) path parameters for this method. query_params: (repeated) query parameters for this method. ordered_params: (repeated) ordered list of parameters for this method. description: description of this method. request_type_name: name of the request type. response_type_name: name of the response type. request_field: if not null, the field to pass as the body of this POST request. may also be the REQUEST_IS_BODY value below to indicate the whole message is the body. upload_config: (ApiUploadInfo) Information about the upload configuration supported by this method. supports_download: (boolean) If True, this method supports downloading the request via the `alt=media` query parameter. api_version_param: API version system parameter for this method. """ relative_path = messages.StringField(1) flat_path = messages.StringField(2) method_id = messages.StringField(3) http_method = messages.StringField(4) path_params = messages.StringField(5, repeated=True) query_params = messages.StringField(6, repeated=True) ordered_params = messages.StringField(7, repeated=True) description = messages.StringField(8) request_type_name = messages.StringField(9) response_type_name = messages.StringField(10) request_field = messages.StringField(11, default='') upload_config = messages.MessageField(ApiUploadInfo, 12) supports_download = messages.BooleanField(13, default=False) api_version_param = messages.StringField(14) REQUEST_IS_BODY = '<request>' def _LoadClass(name, messages_module): if name.startswith('message_types.'): _, _, classname = name.partition('.') return getattr(message_types, classname) elif '.' not in name: return getattr(messages_module, name) else: raise exceptions.GeneratedClientError('Unknown class %s' % name) def _RequireClassAttrs(obj, attrs): for attr in attrs: attr_name = attr.upper() if not hasattr(obj, '%s' % attr_name) or not getattr(obj, attr_name): msg = 'No %s specified for object of class %s.' % ( attr_name, type(obj).__name__) raise exceptions.GeneratedClientError(msg) def NormalizeApiEndpoint(api_endpoint): if not api_endpoint.endswith('/'): api_endpoint += '/' return api_endpoint def _urljoin(base, url): # pylint: disable=invalid-name """Custom urljoin replacement supporting : before / in url.""" # In general, it's unsafe to simply join base and url. However, for # the case of discovery documents, we know: # * base will never contain params, query, or fragment # * url will never contain a scheme or net_loc. # In general, this means we can safely join on /; we just need to # ensure we end up with precisely one / joining base and url. The # exception here is the case of media uploads, where url will be an # absolute url. if url.startswith('http://') or url.startswith('https://'): return urllib.parse.urljoin(base, url) new_base = base if base.endswith('/') else base + '/' new_url = url[1:] if url.startswith('/') else url return new_base + new_url class _UrlBuilder(object): """Convenient container for url data.""" def __init__(self, base_url, relative_path=None, query_params=None): components = urllib.parse.urlsplit(_urljoin( base_url, relative_path or '')) if components.fragment: raise exceptions.ConfigurationValueError( 'Unexpected url fragment: %s' % components.fragment) self.query_params = urllib.parse.parse_qs(components.query or '') if query_params is not None: self.query_params.update(query_params) self.__scheme = components.scheme self.__netloc = components.netloc self.relative_path = components.path or '' @classmethod def FromUrl(cls, url): urlparts = urllib.parse.urlsplit(url) query_params = urllib.parse.parse_qs(urlparts.query) base_url = urllib.parse.urlunsplit(( urlparts.scheme, urlparts.netloc, '', None, None)) relative_path = urlparts.path or '' return cls( base_url, relative_path=relative_path, query_params=query_params) @property def base_url(self): return urllib.parse.urlunsplit( (self.__scheme, self.__netloc, '', '', '')) @base_url.setter def base_url(self, value): components = urllib.parse.urlsplit(value) if components.path or components.query or components.fragment: raise exceptions.ConfigurationValueError( 'Invalid base url: %s' % value) self.__scheme = components.scheme self.__netloc = components.netloc @property def query(self): # TODO(craigcitro): In the case that some of the query params are # non-ASCII, we may silently fail to encode correctly. We should # figure out who is responsible for owning the object -> str # conversion. return urllib.parse.urlencode(self.query_params, True) @property def url(self): if '{' in self.relative_path or '}' in self.relative_path: raise exceptions.ConfigurationValueError( 'Cannot create url with relative path %s' % self.relative_path) return urllib.parse.urlunsplit(( self.__scheme, self.__netloc, self.relative_path, self.query, '')) def _SkipGetCredentials(): """Hook for skipping credentials. For internal use.""" return False class BaseApiClient(object): """Base class for client libraries.""" MESSAGES_MODULE = None _API_KEY = '' _CLIENT_ID = '' _CLIENT_SECRET = '' _PACKAGE = '' _SCOPES = [] _USER_AGENT = '' def __init__(self, url, credentials=None, get_credentials=True, http=None, model=None, log_request=False, log_response=False, num_retries=5, max_retry_wait=60, credentials_args=None, default_global_params=None, additional_http_headers=None, check_response_func=None, retry_func=None, response_encoding=None): _RequireClassAttrs(self, ('_package', '_scopes', 'messages_module')) if default_global_params is not None: util.Typecheck(default_global_params, self.params_type) self.__default_global_params = default_global_params self.log_request = log_request self.log_response = log_response self.__num_retries = 5 self.__max_retry_wait = 60 # We let the @property machinery below do our validation. self.num_retries = num_retries self.max_retry_wait = max_retry_wait self._credentials = credentials get_credentials = get_credentials and not _SkipGetCredentials() if get_credentials and not credentials: credentials_args = credentials_args or {} self._SetCredentials(credentials_args) self._url = NormalizeApiEndpoint(url) self._http = http or http_wrapper.GetHttp() # Note that "no credentials" is totally possible. if self._credentials is not None: self._http = self._credentials.authorize(self._http) # TODO(craigcitro): Remove this field when we switch to proto2. self.__include_fields = None self.additional_http_headers = additional_http_headers or {} self.check_response_func = check_response_func self.retry_func = retry_func self.response_encoding = response_encoding # Since we can't change the init arguments without regenerating clients, # offer this hook to affect FinalizeTransferUrl behavior. self.overwrite_transfer_urls_with_client_base = False # TODO(craigcitro): Finish deprecating these fields. _ = model self.__response_type_model = 'proto' def _SetCredentials(self, kwds): """Fetch credentials, and set them for this client. Note that we can't simply return credentials, since creating them may involve side-effecting self. Args: kwds: Additional keyword arguments are passed on to GetCredentials. Returns: None. Sets self._credentials. """ args = { 'api_key': self._API_KEY, 'client': self, 'client_id': self._CLIENT_ID, 'client_secret': self._CLIENT_SECRET, 'package_name': self._PACKAGE, 'scopes': self._SCOPES, 'user_agent': self._USER_AGENT, } args.update(kwds) # credentials_lib can be expensive to import so do it only if needed. from apitools.base.py import credentials_lib # TODO(craigcitro): It's a bit dangerous to pass this # still-half-initialized self into this method, but we might need # to set attributes on it associated with our credentials. # Consider another way around this (maybe a callback?) and whether # or not it's worth it. self._credentials = credentials_lib.GetCredentials(args) @classmethod def ClientInfo(cls): return { 'client_id': cls._CLIENT_ID, 'client_secret': cls._CLIENT_SECRET, 'scope': ' '.join(sorted(util.NormalizeScopes(cls._SCOPES))), 'user_agent': cls._USER_AGENT, } @property def base_model_class(self): return None @property def http(self): return self._http @property def url(self): return self._url @classmethod def GetScopes(cls): return cls._SCOPES @property def params_type(self): return _LoadClass('StandardQueryParameters', self.MESSAGES_MODULE) @property def user_agent(self): return self._USER_AGENT @property def _default_global_params(self): if self.__default_global_params is None: # pylint: disable=not-callable self.__default_global_params = self.params_type() return self.__default_global_params def AddGlobalParam(self, name, value): params = self._default_global_params setattr(params, name, value) @property def global_params(self): return encoding.CopyProtoMessage(self._default_global_params) @contextlib.contextmanager def IncludeFields(self, include_fields): self.__include_fields = include_fields yield self.__include_fields = None @property def response_type_model(self): return self.__response_type_model @contextlib.contextmanager def JsonResponseModel(self): """In this context, return raw JSON instead of proto.""" old_model = self.response_type_model self.__response_type_model = 'json' yield self.__response_type_model = old_model @property def num_retries(self): return self.__num_retries @num_retries.setter def num_retries(self, value): util.Typecheck(value, six.integer_types) if value < 0: raise exceptions.InvalidDataError( 'Cannot have negative value for num_retries') self.__num_retries = value @property def max_retry_wait(self): return self.__max_retry_wait @max_retry_wait.setter def max_retry_wait(self, value): util.Typecheck(value, six.integer_types) if value <= 0: raise exceptions.InvalidDataError( 'max_retry_wait must be a postiive integer') self.__max_retry_wait = value @contextlib.contextmanager def WithRetries(self, num_retries): old_num_retries = self.num_retries self.num_retries = num_retries yield self.num_retries = old_num_retries def ProcessRequest(self, method_config, request): """Hook for pre-processing of requests.""" if self.log_request: logging.info( 'Calling method %s with %s: %s', method_config.method_id, method_config.request_type_name, request) return request def ProcessHttpRequest(self, http_request): """Hook for pre-processing of http requests.""" http_request.headers.update(self.additional_http_headers) if self.log_request: logging.info('Making http %s to %s', http_request.http_method, http_request.url) logging.info('Headers: %s', pprint.pformat(http_request.headers)) if http_request.body: # TODO(craigcitro): Make this safe to print in the case of # non-printable body characters. logging.info('Body:\n%s', http_request.loggable_body or http_request.body) else: logging.info('Body: (none)') return http_request def ProcessResponse(self, method_config, response): if self.log_response: logging.info('Response of type %s: %s', method_config.response_type_name, response) return response # TODO(craigcitro): Decide where these two functions should live. def SerializeMessage(self, message): return encoding.MessageToJson( message, include_fields=self.__include_fields) def DeserializeMessage(self, response_type, data): """Deserialize the given data as method_config.response_type.""" try: message = encoding.JsonToMessage(response_type, data) except (exceptions.InvalidDataFromServerError, messages.ValidationError, ValueError) as e: raise exceptions.InvalidDataFromServerError( 'Error decoding response "%s" as type %s: %s' % ( data, response_type.__name__, e)) return message def FinalizeTransferUrl(self, url): """Modify the url for a given transfer, based on auth and version.""" url_builder = _UrlBuilder.FromUrl(url) if getattr(self.global_params, 'key', None): url_builder.query_params['key'] = self.global_params.key if self.overwrite_transfer_urls_with_client_base: client_url_builder = _UrlBuilder.FromUrl(self._url) url_builder.base_url = client_url_builder.base_url return url_builder.url class BaseApiService(object): """Base class for generated API services.""" def __init__(self, client): self.__client = client self._method_configs = {} self._upload_configs = {} @property def _client(self): return self.__client @property def client(self): return self.__client def GetMethodConfig(self, method): """Returns service cached method config for given method.""" method_config = self._method_configs.get(method) if method_config: return method_config func = getattr(self, method, None) if func is None: raise KeyError(method) method_config = getattr(func, 'method_config', None) if method_config is None: raise KeyError(method) self._method_configs[method] = config = method_config() return config @classmethod def GetMethodsList(cls): return [f.__name__ for f in six.itervalues(cls.__dict__) if getattr(f, 'method_config', None)] def GetUploadConfig(self, method): return self._upload_configs.get(method) def GetRequestType(self, method): method_config = self.GetMethodConfig(method) return getattr(self.client.MESSAGES_MODULE, method_config.request_type_name) def GetResponseType(self, method): method_config = self.GetMethodConfig(method) return getattr(self.client.MESSAGES_MODULE, method_config.response_type_name) def __CombineGlobalParams(self, global_params, default_params): """Combine the given params with the defaults.""" util.Typecheck(global_params, (type(None), self.__client.params_type)) result = self.__client.params_type() global_params = global_params or self.__client.params_type() for field in result.all_fields(): value = global_params.get_assigned_value(field.name) if value is None: value = default_params.get_assigned_value(field.name) if value not in (None, [], ()): setattr(result, field.name, value) return result def __EncodePrettyPrint(self, query_info): # The prettyPrint flag needs custom encoding: it should be encoded # as 0 if False, and ignored otherwise (True is the default). if not query_info.pop('prettyPrint', True): query_info['prettyPrint'] = 0 # The One Platform equivalent of prettyPrint is pp, which also needs # custom encoding. if not query_info.pop('pp', True): query_info['pp'] = 0 return query_info def __FinalUrlValue(self, value, field): """Encode value for the URL, using field to skip encoding for bytes.""" if isinstance(field, messages.BytesField) and value is not None: return base64.urlsafe_b64encode(value) elif isinstance(value, six.text_type): return value.encode('utf8') elif isinstance(value, six.binary_type): return value.decode('utf8') elif isinstance(value, datetime.datetime): return value.isoformat() return value def __ConstructQueryParams(self, query_params, request, global_params): """Construct a dictionary of query parameters for this request.""" # First, handle the global params. global_params = self.__CombineGlobalParams( global_params, self.__client.global_params) global_param_names = util.MapParamNames( [x.name for x in self.__client.params_type.all_fields()], self.__client.params_type) global_params_type = type(global_params) query_info = dict( (param, self.__FinalUrlValue(getattr(global_params, param), getattr(global_params_type, param))) for param in global_param_names) # Next, add the query params. query_param_names = util.MapParamNames(query_params, type(request)) request_type = type(request) query_info.update( (param, self.__FinalUrlValue(getattr(request, param, None), getattr(request_type, param))) for param in query_param_names) query_info = dict((k, v) for k, v in query_info.items() if v is not None) query_info = self.__EncodePrettyPrint(query_info) query_info = util.MapRequestParams(query_info, type(request)) return query_info def __ConstructRelativePath(self, method_config, request, relative_path=None): """Determine the relative path for request.""" python_param_names = util.MapParamNames( method_config.path_params, type(request)) params = dict([(param, getattr(request, param, None)) for param in python_param_names]) params = util.MapRequestParams(params, type(request)) return util.ExpandRelativePath(method_config, params, relative_path=relative_path) def __FinalizeRequest(self, http_request, url_builder): """Make any final general adjustments to the request.""" if (http_request.http_method == 'GET' and len(http_request.url) > _MAX_URL_LENGTH): http_request.http_method = 'POST' http_request.headers['x-http-method-override'] = 'GET' http_request.headers[ 'content-type'] = 'application/x-www-form-urlencoded' http_request.body = url_builder.query url_builder.query_params = {} http_request.url = url_builder.url def __ProcessHttpResponse(self, method_config, http_response, request): """Process the given http response.""" if http_response.status_code not in (http_client.OK, http_client.CREATED, http_client.NO_CONTENT): raise exceptions.HttpError.FromResponse( http_response, method_config=method_config, request=request) if http_response.status_code == http_client.NO_CONTENT: # TODO(craigcitro): Find out why _replace doesn't seem to work # here. http_response = http_wrapper.Response( info=http_response.info, content='{}', request_url=http_response.request_url) content = http_response.content if self._client.response_encoding and isinstance(content, bytes): content = content.decode(self._client.response_encoding) if self.__client.response_type_model == 'json': return content response_type = _LoadClass(method_config.response_type_name, self.__client.MESSAGES_MODULE) return self.__client.DeserializeMessage(response_type, content) def __SetBaseHeaders(self, http_request, client): """Fill in the basic headers on http_request.""" # TODO(craigcitro): Make the default a little better here, and # include the apitools version. user_agent = client.user_agent or 'apitools-client/1.0' http_request.headers['user-agent'] = user_agent http_request.headers['accept'] = 'application/json' http_request.headers['accept-encoding'] = 'gzip, deflate' def __SetBaseSystemParams(self, http_request, method_config): """Fill in the system parameters to always set for the method.""" if method_config.api_version_param: http_request.headers['X-Goog-Api-Version'] = ( method_config.api_version_param) def __SetBody(self, http_request, method_config, request, upload): """Fill in the body on http_request.""" if not method_config.request_field: return request_type = _LoadClass( method_config.request_type_name, self.__client.MESSAGES_MODULE) if method_config.request_field == REQUEST_IS_BODY: body_value = request body_type = request_type else: body_value = getattr(request, method_config.request_field) body_field = request_type.field_by_name( method_config.request_field) util.Typecheck(body_field, messages.MessageField) body_type = body_field.type # If there was no body provided, we use an empty message of the # appropriate type. body_value = body_value or body_type() if upload and not body_value: # We're going to fill in the body later. return util.Typecheck(body_value, body_type) http_request.headers['content-type'] = 'application/json' http_request.body = self.__client.SerializeMessage(body_value) def PrepareHttpRequest(self, method_config, request, global_params=None, upload=None, upload_config=None, download=None): """Prepares an HTTP request to be sent.""" request_type = _LoadClass( method_config.request_type_name, self.__client.MESSAGES_MODULE) util.Typecheck(request, request_type) request = self.__client.ProcessRequest(method_config, request) http_request = http_wrapper.Request( http_method=method_config.http_method) self.__SetBaseHeaders(http_request, self.__client) self.__SetBaseSystemParams(http_request, method_config) self.__SetBody(http_request, method_config, request, upload) url_builder = _UrlBuilder( self.__client.url, relative_path=method_config.relative_path) url_builder.query_params = self.__ConstructQueryParams( method_config.query_params, request, global_params) # It's important that upload and download go before we fill in the # relative path, so that they can replace it. if upload is not None: upload.ConfigureRequest(upload_config, http_request, url_builder) if download is not None: download.ConfigureRequest(http_request, url_builder) url_builder.relative_path = self.__ConstructRelativePath( method_config, request, relative_path=url_builder.relative_path) self.__FinalizeRequest(http_request, url_builder) return self.__client.ProcessHttpRequest(http_request) def _RunMethod(self, method_config, request, global_params=None, upload=None, upload_config=None, download=None): """Call this method with request.""" if upload is not None and download is not None: # TODO(craigcitro): This just involves refactoring the logic # below into callbacks that we can pass around; in particular, # the order should be that the upload gets the initial request, # and then passes its reply to a download if one exists, and # then that goes to ProcessResponse and is returned. raise exceptions.NotYetImplementedError( 'Cannot yet use both upload and download at once') http_request = self.PrepareHttpRequest( method_config, request, global_params, upload, upload_config, download) # TODO(craigcitro): Make num_retries customizable on Transfer # objects, and pass in self.__client.num_retries when initializing # an upload or download. if download is not None: download.InitializeDownload(http_request, client=self.client) return http_response = None if upload is not None: http_response = upload.InitializeUpload( http_request, client=self.client) if http_response is None: http = self.__client.http if upload and upload.bytes_http: http = upload.bytes_http opts = { 'retries': self.__client.num_retries, 'max_retry_wait': self.__client.max_retry_wait, } if self.__client.check_response_func: opts['check_response_func'] = self.__client.check_response_func if self.__client.retry_func: opts['retry_func'] = self.__client.retry_func http_response = http_wrapper.MakeRequest( http, http_request, **opts) return self.ProcessHttpResponse(method_config, http_response, request) def ProcessHttpResponse(self, method_config, http_response, request=None): """Convert an HTTP response to the expected message type.""" return self.__client.ProcessResponse( method_config, self.__ProcessHttpResponse(method_config, http_response, request))	29.29 KB	2025-03-11 18:02
📄 base_api_test.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. import base64 import datetime import sys import contextlib import unittest import six from six.moves import http_client from six.moves import urllib_parse from apitools.base.protorpclite import message_types 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 from apitools.base.py import http_wrapper @contextlib.contextmanager def mock(module, fn_name, patch): unpatch = getattr(module, fn_name) setattr(module, fn_name, patch) try: yield finally: setattr(module, fn_name, unpatch) class SimpleMessage(messages.Message): field = messages.StringField(1) bytes_field = messages.BytesField(2) class MessageWithTime(messages.Message): timestamp = message_types.DateTimeField(1) class MessageWithRemappings(messages.Message): class AnEnum(messages.Enum): value_one = 1 value_two = 2 str_field = messages.StringField(1) enum_field = messages.EnumField('AnEnum', 2) encoding.AddCustomJsonFieldMapping( MessageWithRemappings, 'str_field', 'remapped_field') encoding.AddCustomJsonEnumMapping( MessageWithRemappings.AnEnum, 'value_one', 'ONE/TWO') class StandardQueryParameters(messages.Message): field = messages.StringField(1) prettyPrint = messages.BooleanField( 5, default=True) # pylint: disable=invalid-name pp = messages.BooleanField(6, default=True) nextPageToken = messages.BytesField(7) # pylint:disable=invalid-name class FakeCredentials(object): def authorize(self, _): # pylint: disable=invalid-name return None class FakeClient(base_api.BaseApiClient): MESSAGES_MODULE = sys.modules[__name__] _PACKAGE = 'package' _SCOPES = ['scope1'] _CLIENT_ID = 'client_id' _CLIENT_SECRET = 'client_secret' class FakeService(base_api.BaseApiService): def __init__(self, client=None): client = client or FakeClient( 'http://www.example.com/', credentials=FakeCredentials()) super(FakeService, self).__init__(client) class BaseApiTest(unittest.TestCase): def __GetFakeClient(self): return FakeClient('', credentials=FakeCredentials()) def testUrlNormalization(self): client = FakeClient('http://www.googleapis.com', get_credentials=False) self.assertTrue(client.url.endswith('/')) def testNoCredentials(self): client = FakeClient('', get_credentials=False) self.assertIsNotNone(client) self.assertIsNone(client._credentials) def testIncludeEmptyFieldsClient(self): msg = SimpleMessage() client = self.__GetFakeClient() self.assertEqual('{}', client.SerializeMessage(msg)) with client.IncludeFields(('field',)): self.assertEqual('{"field": null}', client.SerializeMessage(msg)) def testJsonResponse(self): method_config = base_api.ApiMethodInfo( response_type_name='SimpleMessage') service = FakeService() http_response = http_wrapper.Response( info={'status': '200'}, content='{"field": "abc"}', request_url='http://www.google.com') response_message = SimpleMessage(field='abc') self.assertEqual(response_message, service.ProcessHttpResponse( method_config, http_response)) with service.client.JsonResponseModel(): self.assertEqual( http_response.content, service.ProcessHttpResponse(method_config, http_response)) def testJsonResponseEncoding(self): # On Python 3, httplib2 always returns bytes, so we need to check that # we can correctly decode the message content using the given encoding. method_config = base_api.ApiMethodInfo( response_type_name='SimpleMessage') service = FakeService(FakeClient( 'http://www.example.com/', credentials=FakeCredentials(), response_encoding='utf8')) http_response = http_wrapper.Response( info={'status': '200'}, content=b'{"field": "abc"}', request_url='http://www.google.com') response_message = SimpleMessage(field=u'abc') self.assertEqual(response_message, service.ProcessHttpResponse( method_config, http_response)) with service.client.JsonResponseModel(): self.assertEqual( http_response.content.decode('utf8'), service.ProcessHttpResponse(method_config, http_response)) def testAdditionalHeaders(self): additional_headers = {'Request-Is-Awesome': '1'} client = self.__GetFakeClient() # No headers to start http_request = http_wrapper.Request('http://www.example.com') new_request = client.ProcessHttpRequest(http_request) self.assertFalse('Request-Is-Awesome' in new_request.headers) # Add a new header and ensure it's added to the request. client.additional_http_headers = additional_headers http_request = http_wrapper.Request('http://www.example.com') new_request = client.ProcessHttpRequest(http_request) self.assertTrue('Request-Is-Awesome' in new_request.headers) def testCustomCheckResponse(self): def check_response(): pass def fakeMakeRequest(_, kwargs): self.assertEqual(check_response, kwargs['check_response_func']) return http_wrapper.Response( info={'status': '200'}, content='{"field": "abc"}', request_url='http://www.google.com') method_config = base_api.ApiMethodInfo( request_type_name='SimpleMessage', response_type_name='SimpleMessage') client = self.__GetFakeClient() client.check_response_func = check_response service = FakeService(client=client) request = SimpleMessage() with mock(base_api.http_wrapper, 'MakeRequest', fakeMakeRequest): service._RunMethod(method_config, request) def testCustomRetryFunc(self): def retry_func(): pass def fakeMakeRequest(_, *kwargs): self.assertEqual(retry_func, kwargs['retry_func']) return http_wrapper.Response( info={'status': '200'}, content='{"field": "abc"}', request_url='http://www.google.com') method_config = base_api.ApiMethodInfo( request_type_name='SimpleMessage', response_type_name='SimpleMessage') client = self.__GetFakeClient() client.retry_func = retry_func service = FakeService(client=client) request = SimpleMessage() with mock(base_api.http_wrapper, 'MakeRequest', fakeMakeRequest): service._RunMethod(method_config, request) def testHttpError(self): def fakeMakeRequest(unused_args, **unused_kwargs): return http_wrapper.Response( info={'status': http_client.BAD_REQUEST}, content='{"field": "abc"}', request_url='http://www.google.com') method_config = base_api.ApiMethodInfo( request_type_name='SimpleMessage', response_type_name='SimpleMessage') client = self.__GetFakeClient() service = FakeService(client=client) request = SimpleMessage() with mock(base_api.http_wrapper, 'MakeRequest', fakeMakeRequest): with self.assertRaises(exceptions.HttpBadRequestError) as err: service._RunMethod(method_config, request) http_error = err.exception self.assertEqual('http://www.google.com', http_error.url) self.assertEqual('{"field": "abc"}', http_error.content) self.assertEqual(method_config, http_error.method_config) self.assertEqual(request, http_error.request) def testQueryEncoding(self): method_config = base_api.ApiMethodInfo( request_type_name='MessageWithTime', query_params=['timestamp']) service = FakeService() request = MessageWithTime( timestamp=datetime.datetime(2014, 10, 0o7, 12, 53, 13)) http_request = service.PrepareHttpRequest(method_config, request) url_timestamp = urllib_parse.quote(request.timestamp.isoformat()) self.assertTrue(http_request.url.endswith(url_timestamp)) def testPrettyPrintEncoding(self): method_config = base_api.ApiMethodInfo( request_type_name='MessageWithTime', query_params=['timestamp']) service = FakeService() request = MessageWithTime( timestamp=datetime.datetime(2014, 10, 0o7, 12, 53, 13)) global_params = StandardQueryParameters() http_request = service.PrepareHttpRequest(method_config, request, global_params=global_params) self.assertFalse('prettyPrint' in http_request.url) self.assertFalse('pp' in http_request.url) global_params.prettyPrint = False # pylint: disable=invalid-name global_params.pp = False http_request = service.PrepareHttpRequest(method_config, request, global_params=global_params) self.assertTrue('prettyPrint=0' in http_request.url) self.assertTrue('pp=0' in http_request.url) def testQueryBytesRequest(self): method_config = base_api.ApiMethodInfo( request_type_name='SimpleMessage', query_params=['bytes_field']) service = FakeService() non_unicode_message = b''.join((six.int2byte(100), six.int2byte(200))) request = SimpleMessage(bytes_field=non_unicode_message) global_params = StandardQueryParameters() http_request = service.PrepareHttpRequest(method_config, request, global_params=global_params) want = urllib_parse.urlencode({ 'bytes_field': base64.urlsafe_b64encode(non_unicode_message), }) self.assertIn(want, http_request.url) def testQueryBytesGlobalParams(self): method_config = base_api.ApiMethodInfo( request_type_name='SimpleMessage', query_params=['bytes_field']) service = FakeService() non_unicode_message = b''.join((six.int2byte(100), six.int2byte(200))) request = SimpleMessage() global_params = StandardQueryParameters( nextPageToken=non_unicode_message) http_request = service.PrepareHttpRequest(method_config, request, global_params=global_params) want = urllib_parse.urlencode({ 'nextPageToken': base64.urlsafe_b64encode(non_unicode_message), }) self.assertIn(want, http_request.url) def testQueryRemapping(self): method_config = base_api.ApiMethodInfo( request_type_name='MessageWithRemappings', query_params=['remapped_field', 'enum_field']) request = MessageWithRemappings( str_field='foo', enum_field=MessageWithRemappings.AnEnum.value_one) http_request = FakeService().PrepareHttpRequest(method_config, request) result_params = urllib_parse.parse_qs( urllib_parse.urlparse(http_request.url).query) expected_params = {'enum_field': 'ONE%2FTWO', 'remapped_field': 'foo'} self.assertTrue(expected_params, result_params) def testPathRemapping(self): method_config = base_api.ApiMethodInfo( relative_path='parameters/{remapped_field}/remap/{enum_field}', request_type_name='MessageWithRemappings', path_params=['remapped_field', 'enum_field']) request = MessageWithRemappings( str_field='gonna', enum_field=MessageWithRemappings.AnEnum.value_one) service = FakeService() expected_url = service.client.url + 'parameters/gonna/remap/ONE%2FTWO' http_request = service.PrepareHttpRequest(method_config, request) self.assertEqual(expected_url, http_request.url) method_config.relative_path = ( 'parameters/{+remapped_field}/remap/{+enum_field}') expected_url = service.client.url + 'parameters/gonna/remap/ONE/TWO' http_request = service.PrepareHttpRequest(method_config, request) self.assertEqual(expected_url, http_request.url) def testColonInRelativePath(self): method_config = base_api.ApiMethodInfo( relative_path='path:withJustColon', request_type_name='SimpleMessage') service = FakeService() request = SimpleMessage() http_request = service.PrepareHttpRequest(method_config, request) self.assertEqual('http://www.example.com/path:withJustColon', http_request.url) def testOverwritesTransferUrlBase(self): client = self.__GetFakeClient() client.overwrite_transfer_urls_with_client_base = True client._url = 'http://custom.p.googleapis.com/' observed = client.FinalizeTransferUrl( 'http://normal.googleapis.com/path') expected = 'http://custom.p.googleapis.com/path' self.assertEqual(observed, expected) def testApiVersionSystemParameter(self): method_config = base_api.ApiMethodInfo( request_type_name='SimpleMessage', api_version_param='2024-01-01') service = FakeService() request = SimpleMessage() http_request = service.PrepareHttpRequest(method_config, request) self.assertIn('X-Goog-Api-Version', http_request.headers) self.assertEqual( '2024-01-01', http_request.headers['X-Goog-Api-Version']) if __name__ == '__main__': unittest.main()	14.12 KB	2025-03-11 18:02
📄 batch.py #!/usr/bin/env python # # 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. """Library for handling batch HTTP requests for apitools.""" import collections import email.generator as generator import email.mime.multipart as mime_multipart import email.mime.nonmultipart as mime_nonmultipart import email.parser as email_parser import itertools import time import uuid import six from six.moves import http_client from six.moves import urllib_parse from six.moves import range # pylint: disable=redefined-builtin from apitools.base.py import exceptions from apitools.base.py import http_wrapper __all__ = [ 'BatchApiRequest', ] class RequestResponseAndHandler(collections.namedtuple( 'RequestResponseAndHandler', ['request', 'response', 'handler'])): """Container for data related to completing an HTTP request. This contains an HTTP request, its response, and a callback for handling the response from the server. Attributes: request: An http_wrapper.Request object representing the HTTP request. response: The http_wrapper.Response object returned from the server. handler: A callback function accepting two arguments, response and exception. Response is an http_wrapper.Response object, and exception is an apiclient.errors.HttpError object if an error occurred, or otherwise None. """ class BatchApiRequest(object): """Batches multiple api requests into a single request.""" class ApiCall(object): """Holds request and response information for each request. ApiCalls are ultimately exposed to the client once the HTTP batch request has been completed. Attributes: http_request: A client-supplied http_wrapper.Request to be submitted to the server. response: A http_wrapper.Response object given by the server as a response to the user request, or None if an error occurred. exception: An apiclient.errors.HttpError object if an error occurred, or None. """ def __init__(self, request, retryable_codes, service, method_config): """Initialize an individual API request. Args: request: An http_wrapper.Request object. retryable_codes: A list of integer HTTP codes that can be retried. service: A service inheriting from base_api.BaseApiService. method_config: Method config for the desired API request. """ self.__retryable_codes = list( set(retryable_codes + [http_client.UNAUTHORIZED])) self.__http_response = None self.__service = service self.__method_config = method_config self.http_request = request # TODO(user): Add some validation to these fields. self.__response = None self.__exception = None @property def is_error(self): return self.exception is not None @property def response(self): return self.__response @property def exception(self): return self.__exception @property def authorization_failed(self): return (self.__http_response and ( self.__http_response.status_code == http_client.UNAUTHORIZED)) @property def terminal_state(self): if self.__http_response is None: return False response_code = self.__http_response.status_code return response_code not in self.__retryable_codes def HandleResponse(self, http_response, exception): """Handles incoming http response to the request in http_request. This is intended to be used as a callback function for BatchHttpRequest.Add. Args: http_response: Deserialized http_wrapper.Response object. exception: apiclient.errors.HttpError object if an error occurred. """ self.__http_response = http_response self.__exception = exception if self.terminal_state and not self.__exception: self.__response = self.__service.ProcessHttpResponse( self.__method_config, self.__http_response) def __init__(self, batch_url=None, retryable_codes=None, response_encoding=None): """Initialize a batch API request object. Args: batch_url: Base URL for batch API calls. retryable_codes: A list of integer HTTP codes that can be retried. response_encoding: The encoding type of response content. """ self.api_requests = [] self.retryable_codes = retryable_codes or [] self.batch_url = batch_url or 'https://www.googleapis.com/batch' self.response_encoding = response_encoding def Add(self, service, method, request, global_params=None): """Add a request to the batch. Args: service: A class inheriting base_api.BaseApiService. method: A string indicated desired method from the service. See the example in the class docstring. request: An input message appropriate for the specified service.method. global_params: Optional additional parameters to pass into method.PrepareHttpRequest. Returns: None """ # Retrieve the configs for the desired method and service. method_config = service.GetMethodConfig(method) upload_config = service.GetUploadConfig(method) # Prepare the HTTP Request. http_request = service.PrepareHttpRequest( method_config, request, global_params=global_params, upload_config=upload_config) # Create the request and add it to our master list. api_request = self.ApiCall( http_request, self.retryable_codes, service, method_config) self.api_requests.append(api_request) def Execute(self, http, sleep_between_polls=5, max_retries=5, max_batch_size=None, batch_request_callback=None): """Execute all of the requests in the batch. Args: http: httplib2.Http object for use in the request. sleep_between_polls: Integer number of seconds to sleep between polls. max_retries: Max retries. Any requests that have not succeeded by this number of retries simply report the last response or exception, whatever it happened to be. max_batch_size: int, if specified requests will be split in batches of given size. batch_request_callback: function of (http_response, exception) passed to BatchHttpRequest which will be run on any given results. Returns: List of ApiCalls. """ requests = [request for request in self.api_requests if not request.terminal_state] batch_size = max_batch_size or len(requests) for attempt in range(max_retries): if attempt: time.sleep(sleep_between_polls) for i in range(0, len(requests), batch_size): # Create a batch_http_request object and populate it with # incomplete requests. batch_http_request = BatchHttpRequest( batch_url=self.batch_url, callback=batch_request_callback, response_encoding=self.response_encoding ) for request in itertools.islice(requests, i, i + batch_size): batch_http_request.Add( request.http_request, request.HandleResponse) batch_http_request.Execute(http) if hasattr(http.request, 'credentials'): if any(request.authorization_failed for request in itertools.islice(requests, i, i + batch_size)): http.request.credentials.refresh(http) # Collect retryable requests. requests = [request for request in self.api_requests if not request.terminal_state] if not requests: break return self.api_requests class BatchHttpRequest(object): """Batches multiple http_wrapper.Request objects into a single request.""" def __init__(self, batch_url, callback=None, response_encoding=None): """Constructor for a BatchHttpRequest. Args: batch_url: URL to send batch requests to. callback: A callback to be called for each response, of the form callback(response, exception). The first parameter is the deserialized Response object. The second is an apiclient.errors.HttpError exception object if an HTTP error occurred while processing the request, or None if no error occurred. response_encoding: The encoding type of response content. """ # Endpoint to which these requests are sent. self.__batch_url = batch_url # Global callback to be called for each individual response in the # batch. self.__callback = callback # Response content will be decoded if this is provided. self.__response_encoding = response_encoding # List of requests, responses and handlers. self.__request_response_handlers = {} # The last auto generated id. self.__last_auto_id = itertools.count() # Unique ID on which to base the Content-ID headers. self.__base_id = uuid.uuid4() def _ConvertIdToHeader(self, request_id): """Convert an id to a Content-ID header value. Args: request_id: String identifier for a individual request. Returns: A Content-ID header with the id_ encoded into it. A UUID is prepended to the value because Content-ID headers are supposed to be universally unique. """ return '<%s+%s>' % (self.__base_id, urllib_parse.quote(request_id)) @staticmethod def _ConvertHeaderToId(header): """Convert a Content-ID header value to an id. Presumes the Content-ID header conforms to the format that _ConvertIdToHeader() returns. Args: header: A string indicating the Content-ID header value. Returns: The extracted id value. Raises: BatchError if the header is not in the expected format. """ if not (header.startswith('<') or header.endswith('>')): raise exceptions.BatchError( 'Invalid value for Content-ID: %s' % header) if '+' not in header: raise exceptions.BatchError( 'Invalid value for Content-ID: %s' % header) _, request_id = header[1:-1].rsplit('+', 1) return urllib_parse.unquote(request_id) def _SerializeRequest(self, request): """Convert a http_wrapper.Request object into a string. Args: request: A http_wrapper.Request to serialize. Returns: The request as a string in application/http format. """ # Construct status line parsed = urllib_parse.urlsplit(request.url) request_line = urllib_parse.urlunsplit( ('', '', parsed.path, parsed.query, '')) if not isinstance(request_line, six.text_type): request_line = request_line.decode('utf-8') status_line = u' '.join(( request.http_method, request_line, u'HTTP/1.1\n' )) major, minor = request.headers.get( 'content-type', 'application/json').split('/') msg = mime_nonmultipart.MIMENonMultipart(major, minor) # MIMENonMultipart adds its own Content-Type header. # Keep all of the other headers in `request.headers`. for key, value in request.headers.items(): if key == 'content-type': continue msg[key] = value msg['Host'] = parsed.netloc msg.set_unixfrom(None) if request.body is not None: msg.set_payload(request.body) # Serialize the mime message. str_io = six.StringIO() # maxheaderlen=0 means don't line wrap headers. gen = generator.Generator(str_io, maxheaderlen=0) gen.flatten(msg, unixfrom=False) body = str_io.getvalue() return status_line + body def _DeserializeResponse(self, payload): """Convert string into Response and content. Args: payload: Header and body string to be deserialized. Returns: A Response object """ # Strip off the status line. status_line, payload = payload.split('\n', 1) _, status, _ = status_line.split(' ', 2) # Parse the rest of the response. parser = email_parser.Parser() msg = parser.parsestr(payload) # Get the headers. info = dict(msg) info['status'] = status # Create Response from the parsed headers. content = msg.get_payload() return http_wrapper.Response(info, content, self.__batch_url) def _NewId(self): """Create a new id. Auto incrementing number that avoids conflicts with ids already used. Returns: A new unique id string. """ return str(next(self.__last_auto_id)) def Add(self, request, callback=None): """Add a new request. Args: request: A http_wrapper.Request to add to the batch. callback: A callback to be called for this response, of the form callback(response, exception). The first parameter is the deserialized response object. The second is an apiclient.errors.HttpError exception object if an HTTP error occurred while processing the request, or None if no errors occurred. Returns: None """ handler = RequestResponseAndHandler(request, None, callback) self.__request_response_handlers[self._NewId()] = handler def _Execute(self, http): """Serialize batch request, send to server, process response. Args: http: A httplib2.Http object to be used to make the request with. Raises: httplib2.HttpLib2Error if a transport error has occured. apiclient.errors.BatchError if the response is the wrong format. """ message = mime_multipart.MIMEMultipart('mixed') # Message should not write out its own headers. setattr(message, '_write_headers', lambda self: None) # Add all the individual requests. for key in self.__request_response_handlers: msg = mime_nonmultipart.MIMENonMultipart('application', 'http') msg['Content-Transfer-Encoding'] = 'binary' msg['Content-ID'] = self._ConvertIdToHeader(key) body = self._SerializeRequest( self.__request_response_handlers[key].request) msg.set_payload(body) message.attach(msg) request = http_wrapper.Request(self.__batch_url, 'POST') request.body = message.as_string() request.headers['content-type'] = ( 'multipart/mixed; boundary="%s"') % message.get_boundary() response = http_wrapper.MakeRequest(http, request) if response.status_code >= 300: raise exceptions.HttpError.FromResponse(response) # Prepend with a content-type header so Parser can handle it. header = 'content-type: %s\r\n\r\n' % response.info['content-type'] content = response.content if isinstance(content, bytes) and self.__response_encoding: content = response.content.decode(self.__response_encoding) parser = email_parser.Parser() mime_response = parser.parsestr(header + content) if not mime_response.is_multipart(): raise exceptions.BatchError( 'Response not in multipart/mixed format.') for part in mime_response.get_payload(): request_id = self._ConvertHeaderToId(part['Content-ID']) response = self._DeserializeResponse(part.get_payload()) # Disable protected access because namedtuple._replace(...) # is not actually meant to be protected. # pylint: disable=protected-access self.__request_response_handlers[request_id] = ( self.__request_response_handlers[request_id]._replace( response=response)) def Execute(self, http): """Execute all the requests as a single batched HTTP request. Args: http: A httplib2.Http object to be used with the request. Returns: None Raises: BatchError if the response is the wrong format. """ self._Execute(http) for key in self.__request_response_handlers: response = self.__request_response_handlers[key].response callback = self.__request_response_handlers[key].handler exception = None if response.status_code >= 300: exception = exceptions.HttpError.FromResponse(response) if callback is not None: callback(response, exception) if self.__callback is not None: self.__callback(response, exception)	17.93 KB	2025-03-11 18:02
📄 batch_test.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. """Tests for apitools.base.py.batch.""" import textwrap import unittest import mock from six.moves import http_client from six.moves import range # pylint:disable=redefined-builtin from six.moves.urllib import parse from apitools.base.py import batch from apitools.base.py import exceptions from apitools.base.py import http_wrapper class FakeCredentials(object): def __init__(self): self.num_refreshes = 0 def refresh(self, _): self.num_refreshes += 1 class FakeHttp(object): class FakeRequest(object): def __init__(self, credentials=None): if credentials is not None: self.credentials = credentials def __init__(self, credentials=None): self.request = FakeHttp.FakeRequest(credentials=credentials) class FakeService(object): """A service for testing.""" def GetMethodConfig(self, _): return {} def GetUploadConfig(self, _): return {} # pylint: disable=unused-argument def PrepareHttpRequest( self, method_config, request, global_params, upload_config): return global_params['desired_request'] # pylint: enable=unused-argument def ProcessHttpResponse(self, _, http_response): return http_response class BatchTest(unittest.TestCase): def assertUrlEqual(self, expected_url, provided_url): def parse_components(url): parsed = parse.urlsplit(url) query = parse.parse_qs(parsed.query) return parsed._replace(query=''), query expected_parse, expected_query = parse_components(expected_url) provided_parse, provided_query = parse_components(provided_url) self.assertEqual(expected_parse, provided_parse) self.assertEqual(expected_query, provided_query) def __ConfigureMock(self, mock_request, expected_request, response): if isinstance(response, list): response = list(response) def CheckRequest(_, request, *unused_kwds): self.assertUrlEqual(expected_request.url, request.url) self.assertEqual(expected_request.http_method, request.http_method) if isinstance(response, list): return response.pop(0) return response mock_request.side_effect = CheckRequest def testRequestServiceUnavailable(self): mock_service = FakeService() desired_url = 'https://www.example.com' batch_api_request = batch.BatchApiRequest(batch_url=desired_url, retryable_codes=[]) # The request to be added. The actual request sent will be somewhat # larger, as this is added to a batch. desired_request = http_wrapper.Request(desired_url, 'POST', { 'content-type': 'multipart/mixed; boundary="None"', 'content-length': 80, }, 'x' 80) with mock.patch.object(http_wrapper, 'MakeRequest', autospec=True) as mock_request: self.__ConfigureMock( mock_request, http_wrapper.Request(desired_url, 'POST', { 'content-type': 'multipart/mixed; boundary="None"', 'content-length': 419, }, 'x' * 419), http_wrapper.Response({ 'status': '200', 'content-type': 'multipart/mixed; boundary="boundary"', }, textwrap.dedent("""\ --boundary content-type: text/plain content-id: <id+0> HTTP/1.1 503 SERVICE UNAVAILABLE nope --boundary--"""), None)) batch_api_request.Add( mock_service, 'unused', None, global_params={'desired_request': desired_request}) api_request_responses = batch_api_request.Execute( FakeHttp(), sleep_between_polls=0) self.assertEqual(1, len(api_request_responses)) # Make sure we didn't retry non-retryable code 503. self.assertEqual(1, mock_request.call_count) self.assertTrue(api_request_responses[0].is_error) self.assertIsNone(api_request_responses[0].response) self.assertIsInstance(api_request_responses[0].exception, exceptions.HttpError) def testSingleRequestInBatch(self): desired_url = 'https://www.example.com' callback_was_called = [] def _Callback(response, exception): self.assertEqual({'status': '200'}, response.info) self.assertEqual('content', response.content) self.assertEqual(desired_url, response.request_url) self.assertIsNone(exception) callback_was_called.append(1) mock_service = FakeService() batch_api_request = batch.BatchApiRequest(batch_url=desired_url) # The request to be added. The actual request sent will be somewhat # larger, as this is added to a batch. desired_request = http_wrapper.Request(desired_url, 'POST', { 'content-type': 'multipart/mixed; boundary="None"', 'content-length': 80, }, 'x' * 80) with mock.patch.object(http_wrapper, 'MakeRequest', autospec=True) as mock_request: self.__ConfigureMock( mock_request, http_wrapper.Request(desired_url, 'POST', { 'content-type': 'multipart/mixed; boundary="None"', 'content-length': 419, }, 'x' * 419), http_wrapper.Response({ 'status': '200', 'content-type': 'multipart/mixed; boundary="boundary"', }, textwrap.dedent("""\ --boundary content-type: text/plain content-id: <id+0> HTTP/1.1 200 OK content --boundary--"""), None)) batch_api_request.Add(mock_service, 'unused', None, { 'desired_request': desired_request, }) api_request_responses = batch_api_request.Execute( FakeHttp(), batch_request_callback=_Callback) self.assertEqual(1, len(api_request_responses)) self.assertEqual(1, mock_request.call_count) self.assertFalse(api_request_responses[0].is_error) response = api_request_responses[0].response self.assertEqual({'status': '200'}, response.info) self.assertEqual('content', response.content) self.assertEqual(desired_url, response.request_url) self.assertEqual(1, len(callback_was_called)) def _MakeResponse(self, number_of_parts): return http_wrapper.Response( info={ 'status': '200', 'content-type': 'multipart/mixed; boundary="boundary"', }, content='--boundary\n' + '--boundary\n'.join( textwrap.dedent("""\ content-type: text/plain content-id: <id+{0}> HTTP/1.1 200 OK response {0} content """) .format(i) for i in range(number_of_parts)) + '--boundary--', request_url=None, ) def _MakeSampleRequest(self, url, name): return http_wrapper.Request(url, 'POST', { 'content-type': 'multipart/mixed; boundary="None"', 'content-length': 80, }, '{0} {1}'.format(name, 'x' * (79 - len(name)))) def testMultipleRequestInBatchWithMax(self): mock_service = FakeService() desired_url = 'https://www.example.com' batch_api_request = batch.BatchApiRequest(batch_url=desired_url) number_of_requests = 10 max_batch_size = 3 for i in range(number_of_requests): batch_api_request.Add( mock_service, 'unused', None, {'desired_request': self._MakeSampleRequest( desired_url, 'Sample-{0}'.format(i))}) responses = [] for i in range(0, number_of_requests, max_batch_size): responses.append( self._MakeResponse( min(number_of_requests - i, max_batch_size))) with mock.patch.object(http_wrapper, 'MakeRequest', autospec=True) as mock_request: self.__ConfigureMock( mock_request, expected_request=http_wrapper.Request(desired_url, 'POST', { 'content-type': 'multipart/mixed; boundary="None"', 'content-length': 1142, }, 'x' * 1142), response=responses) api_request_responses = batch_api_request.Execute( FakeHttp(), max_batch_size=max_batch_size) self.assertEqual(number_of_requests, len(api_request_responses)) self.assertEqual( -(-number_of_requests // max_batch_size), mock_request.call_count) def testRefreshOnAuthFailure(self): mock_service = FakeService() desired_url = 'https://www.example.com' batch_api_request = batch.BatchApiRequest(batch_url=desired_url) # The request to be added. The actual request sent will be somewhat # larger, as this is added to a batch. desired_request = http_wrapper.Request(desired_url, 'POST', { 'content-type': 'multipart/mixed; boundary="None"', 'content-length': 80, }, 'x' * 80) with mock.patch.object(http_wrapper, 'MakeRequest', autospec=True) as mock_request: self.__ConfigureMock( mock_request, http_wrapper.Request(desired_url, 'POST', { 'content-type': 'multipart/mixed; boundary="None"', 'content-length': 419, }, 'x' * 419), [ http_wrapper.Response({ 'status': '200', 'content-type': 'multipart/mixed; boundary="boundary"', }, textwrap.dedent("""\ --boundary content-type: text/plain content-id: <id+0> HTTP/1.1 401 UNAUTHORIZED Invalid grant --boundary--"""), None), http_wrapper.Response({ 'status': '200', 'content-type': 'multipart/mixed; boundary="boundary"', }, textwrap.dedent("""\ --boundary content-type: text/plain content-id: <id+0> HTTP/1.1 200 OK content --boundary--"""), None) ]) batch_api_request.Add(mock_service, 'unused', None, { 'desired_request': desired_request, }) credentials = FakeCredentials() api_request_responses = batch_api_request.Execute( FakeHttp(credentials=credentials), sleep_between_polls=0) self.assertEqual(1, len(api_request_responses)) self.assertEqual(2, mock_request.call_count) self.assertEqual(1, credentials.num_refreshes) self.assertFalse(api_request_responses[0].is_error) response = api_request_responses[0].response self.assertEqual({'status': '200'}, response.info) self.assertEqual('content', response.content) self.assertEqual(desired_url, response.request_url) def testNoAttempts(self): desired_url = 'https://www.example.com' batch_api_request = batch.BatchApiRequest(batch_url=desired_url) batch_api_request.Add(FakeService(), 'unused', None, { 'desired_request': http_wrapper.Request(desired_url, 'POST', { 'content-type': 'multipart/mixed; boundary="None"', 'content-length': 80, }, 'x' * 80), }) api_request_responses = batch_api_request.Execute(None, max_retries=0) self.assertEqual(1, len(api_request_responses)) self.assertIsNone(api_request_responses[0].response) self.assertIsNone(api_request_responses[0].exception) def _DoTestConvertIdToHeader(self, test_id, expected_result): batch_request = batch.BatchHttpRequest('https://www.example.com') self.assertEqual( expected_result % batch_request._BatchHttpRequest__base_id, batch_request._ConvertIdToHeader(test_id)) def testConvertIdSimple(self): self._DoTestConvertIdToHeader('blah', '<%s+blah>') def testConvertIdThatNeedsEscaping(self): self._DoTestConvertIdToHeader(' space1', '<%s+%%20space1>') def _DoTestConvertHeaderToId(self, header, expected_id): batch_request = batch.BatchHttpRequest('https://www.example.com') self.assertEqual(expected_id, batch_request._ConvertHeaderToId(header)) def testConvertHeaderToIdSimple(self): self._DoTestConvertHeaderToId('<hello+blah>', 'blah') def testConvertHeaderToIdWithLotsOfPlus(self): self._DoTestConvertHeaderToId('<a+++++plus>', 'plus') def _DoTestConvertInvalidHeaderToId(self, invalid_header): batch_request = batch.BatchHttpRequest('https://www.example.com') self.assertRaises(exceptions.BatchError, batch_request._ConvertHeaderToId, invalid_header) def testHeaderWithoutAngleBrackets(self): self._DoTestConvertInvalidHeaderToId('1+1') def testHeaderWithoutPlus(self): self._DoTestConvertInvalidHeaderToId('<HEADER>') def testSerializeRequest(self): request = http_wrapper.Request(body='Hello World', headers={ 'content-type': 'protocol/version', }) expected_serialized_request = '\n'.join([ 'GET HTTP/1.1', 'Content-Type: protocol/version', 'MIME-Version: 1.0', 'content-length: 11', 'Host: ', '', 'Hello World', ]) batch_request = batch.BatchHttpRequest('https://www.example.com') self.assertEqual(expected_serialized_request, batch_request._SerializeRequest(request)) def testSerializeRequestPreservesHeaders(self): # Now confirm that if an additional, arbitrary header is added # that it is successfully serialized to the request. Merely # check that it is included, because the order of the headers # in the request is arbitrary. request = http_wrapper.Request(body='Hello World', headers={ 'content-type': 'protocol/version', 'key': 'value', }) batch_request = batch.BatchHttpRequest('https://www.example.com') self.assertTrue( 'key: value\n' in batch_request._SerializeRequest(request)) def testSerializeRequestNoBody(self): request = http_wrapper.Request(body=None, headers={ 'content-type': 'protocol/version', }) expected_serialized_request = '\n'.join([ 'GET HTTP/1.1', 'Content-Type: protocol/version', 'MIME-Version: 1.0', 'Host: ', '', '', ]) batch_request = batch.BatchHttpRequest('https://www.example.com') self.assertEqual(expected_serialized_request, batch_request._SerializeRequest(request)) def testSerializeRequestWithPathAndQueryParams(self): request = http_wrapper.Request( url='my/path?query=param', body='Hello World', headers={'content-type': 'protocol/version'}) expected_serialized_request = '\n'.join([ 'GET my/path?query=param HTTP/1.1', 'Content-Type: protocol/version', 'MIME-Version: 1.0', 'content-length: 11', 'Host: ', '', 'Hello World', ]) batch_request = batch.BatchHttpRequest('https://www.example.com') self.assertEqual(expected_serialized_request, batch_request._SerializeRequest(request)) def testDeserializeRequest(self): serialized_payload = '\n'.join([ 'GET HTTP/1.1', 'Content-Type: protocol/version', 'MIME-Version: 1.0', 'content-length: 11', 'key: value', 'Host: ', '', 'Hello World', ]) example_url = 'https://www.example.com' expected_response = http_wrapper.Response({ 'content-length': str(len('Hello World')), 'Content-Type': 'protocol/version', 'key': 'value', 'MIME-Version': '1.0', 'status': '', 'Host': '' }, 'Hello World', example_url) batch_request = batch.BatchHttpRequest(example_url) self.assertEqual( expected_response, batch_request._DeserializeResponse(serialized_payload)) def testNewId(self): batch_request = batch.BatchHttpRequest('https://www.example.com') for i in range(100): self.assertEqual(str(i), batch_request._NewId()) def testAdd(self): batch_request = batch.BatchHttpRequest('https://www.example.com') for x in range(100): batch_request.Add(http_wrapper.Request(body=str(x))) for key in batch_request._BatchHttpRequest__request_response_handlers: value = batch_request._BatchHttpRequest__request_response_handlers[ key] self.assertEqual(key, value.request.body) self.assertFalse(value.request.url) self.assertEqual('GET', value.request.http_method) self.assertIsNone(value.response) self.assertIsNone(value.handler) def testInternalExecuteWithFailedRequest(self): with mock.patch.object(http_wrapper, 'MakeRequest', autospec=True) as mock_request: self.__ConfigureMock( mock_request, http_wrapper.Request('https://www.example.com', 'POST', { 'content-type': 'multipart/mixed; boundary="None"', 'content-length': 80, }, 'x' * 80), http_wrapper.Response({'status': '300'}, None, None)) batch_request = batch.BatchHttpRequest('https://www.example.com') self.assertRaises( exceptions.HttpError, batch_request._Execute, None) def testInternalExecuteWithNonMultipartResponse(self): with mock.patch.object(http_wrapper, 'MakeRequest', autospec=True) as mock_request: self.__ConfigureMock( mock_request, http_wrapper.Request('https://www.example.com', 'POST', { 'content-type': 'multipart/mixed; boundary="None"', 'content-length': 80, }, 'x' * 80), http_wrapper.Response({ 'status': '200', 'content-type': 'blah/blah' }, '', None)) batch_request = batch.BatchHttpRequest('https://www.example.com') self.assertRaises( exceptions.BatchError, batch_request._Execute, None) def testInternalExecute(self): with mock.patch.object(http_wrapper, 'MakeRequest', autospec=True) as mock_request: self.__ConfigureMock( mock_request, http_wrapper.Request('https://www.example.com', 'POST', { 'content-type': 'multipart/mixed; boundary="None"', 'content-length': 583, }, 'x' * 583), http_wrapper.Response({ 'status': '200', 'content-type': 'multipart/mixed; boundary="boundary"', }, textwrap.dedent("""\ --boundary content-type: text/plain content-id: <id+2> HTTP/1.1 200 OK Second response --boundary content-type: text/plain content-id: <id+1> HTTP/1.1 401 UNAUTHORIZED First response --boundary--"""), None)) test_requests = { '1': batch.RequestResponseAndHandler( http_wrapper.Request(body='first'), None, None), '2': batch.RequestResponseAndHandler( http_wrapper.Request(body='second'), None, None), } batch_request = batch.BatchHttpRequest('https://www.example.com') batch_request._BatchHttpRequest__request_response_handlers = ( test_requests) batch_request._Execute(FakeHttp()) test_responses = ( batch_request._BatchHttpRequest__request_response_handlers) self.assertEqual(http_client.UNAUTHORIZED, test_responses['1'].response.status_code) self.assertEqual(http_client.OK, test_responses['2'].response.status_code) self.assertIn( 'First response', test_responses['1'].response.content) self.assertIn( 'Second response', test_responses['2'].response.content) def testInternalExecuteWithEncodedResponse(self): with mock.patch.object(http_wrapper, 'MakeRequest', autospec=True) as mock_request: self.__ConfigureMock( mock_request, http_wrapper.Request('https://www.example.com', 'POST', { 'content-type': 'multipart/mixed; boundary="None"', 'content-length': 274, }, 'x' * 274), http_wrapper.Response({ 'status': '200', 'content-type': 'multipart/mixed; boundary="boundary"', }, textwrap.dedent("""\ --boundary content-type: text/plain content-id: <id+1> HTTP/1.1 200 OK response --boundary--""").encode('utf-8'), None)) test_request = { '1': batch.RequestResponseAndHandler( http_wrapper.Request(body='first'), None, None), } batch_request = batch.BatchHttpRequest('https://www.example.com', response_encoding='utf-8') batch_request._BatchHttpRequest__request_response_handlers = ( test_request) batch_request._Execute(FakeHttp()) test_responses = ( batch_request._BatchHttpRequest__request_response_handlers) self.assertEqual(http_client.OK, test_responses['1'].response.status_code) self.assertIn( 'response', test_responses['1'].response.content) def testPublicExecute(self): def LocalCallback(response, exception): self.assertEqual({'status': '418'}, response.info) self.assertEqual('Teapot', response.content) self.assertIsNone(response.request_url) self.assertIsInstance(exception, exceptions.HttpError) global_callback = mock.Mock() batch_request = batch.BatchHttpRequest( 'https://www.example.com', global_callback) with mock.patch.object(batch.BatchHttpRequest, '_Execute', autospec=True) as mock_execute: mock_execute.return_value = None test_requests = { '0': batch.RequestResponseAndHandler( None, http_wrapper.Response({'status': '200'}, 'Hello!', None), None), '1': batch.RequestResponseAndHandler( None, http_wrapper.Response({'status': '418'}, 'Teapot', None), LocalCallback), } batch_request._BatchHttpRequest__request_response_handlers = ( test_requests) batch_request.Execute(None) # Global callback was called once per handler. self.assertEqual(len(test_requests), global_callback.call_count)	24.73 KB	2025-03-11 18:02
📄 buffered_stream.py #!/usr/bin/env python # # 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. """Small helper class to provide a small slice of a stream. This class reads ahead to detect if we are at the end of the stream. """ from apitools.base.py import exceptions # TODO(user): Consider replacing this with a StringIO. class BufferedStream(object): """Buffers a stream, reading ahead to determine if we're at the end.""" def __init__(self, stream, start, size): self.__stream = stream self.__start_pos = start self.__buffer_pos = 0 self.__buffered_data = self.__stream.read(size) self.__stream_at_end = len(self.__buffered_data) < size self.__end_pos = self.__start_pos + len(self.__buffered_data) def __str__(self): return ('Buffered stream %s from position %s-%s with %s ' 'bytes remaining' % (self.__stream, self.__start_pos, self.__end_pos, self._bytes_remaining)) def __len__(self): return len(self.__buffered_data) @property def stream_exhausted(self): return self.__stream_at_end @property def stream_end_position(self): return self.__end_pos @property def _bytes_remaining(self): return len(self.__buffered_data) - self.__buffer_pos def read(self, size=None): # pylint: disable=invalid-name """Reads from the buffer.""" if size is None or size < 0: raise exceptions.NotYetImplementedError( 'Illegal read of size %s requested on BufferedStream. ' 'Wrapped stream %s is at position %s-%s, ' '%s bytes remaining.' % (size, self.__stream, self.__start_pos, self.__end_pos, self._bytes_remaining)) data = '' if self._bytes_remaining: size = min(size, self._bytes_remaining) data = self.__buffered_data[ self.__buffer_pos:self.__buffer_pos + size] self.__buffer_pos += size return data	2.52 KB	2025-03-11 18:02
📄 buffered_stream_test.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. """Tests for buffered_stream.""" import string import unittest import six from apitools.base.py import buffered_stream from apitools.base.py import exceptions class BufferedStreamTest(unittest.TestCase): def setUp(self): self.stream = six.StringIO(string.ascii_letters) self.value = self.stream.getvalue() self.stream.seek(0) def testEmptyBuffer(self): bs = buffered_stream.BufferedStream(self.stream, 0, 0) self.assertEqual('', bs.read(0)) self.assertEqual(0, bs.stream_end_position) def testOffsetStream(self): bs = buffered_stream.BufferedStream(self.stream, 50, 100) self.assertEqual(len(self.value), len(bs)) self.assertEqual(self.value, bs.read(len(self.value))) self.assertEqual(50 + len(self.value), bs.stream_end_position) def testUnexhaustedStream(self): bs = buffered_stream.BufferedStream(self.stream, 0, 50) self.assertEqual(50, bs.stream_end_position) self.assertEqual(False, bs.stream_exhausted) self.assertEqual(self.value[0:50], bs.read(50)) self.assertEqual(False, bs.stream_exhausted) self.assertEqual('', bs.read(0)) self.assertEqual('', bs.read(100)) def testExhaustedStream(self): bs = buffered_stream.BufferedStream(self.stream, 0, 100) self.assertEqual(len(self.value), bs.stream_end_position) self.assertEqual(True, bs.stream_exhausted) self.assertEqual(self.value, bs.read(100)) self.assertEqual('', bs.read(0)) self.assertEqual('', bs.read(100)) def testArbitraryLengthRead(self): bs = buffered_stream.BufferedStream(self.stream, 0, 20) with self.assertRaises(exceptions.NotYetImplementedError): bs.read() with self.assertRaises(exceptions.NotYetImplementedError): bs.read(size=-1)	2.4 KB	2025-03-11 18:02
📄 compression.py #!/usr/bin/env python # # Copyright 2017 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. """Compression support for apitools.""" from collections import deque from apitools.base.py import gzip __all__ = [ 'CompressStream', ] # pylint: disable=invalid-name # Note: Apitools only uses the default chunksize when compressing. def CompressStream(in_stream, length=None, compresslevel=2, chunksize=16777216): """Compresses an input stream into a file-like buffer. This reads from the input stream until either we've stored at least length compressed bytes, or the input stream has been exhausted. This supports streams of unknown size. Args: in_stream: The input stream to read from. length: The target number of compressed bytes to buffer in the output stream. If length is none, the input stream will be compressed until it's exhausted. The actual length of the output buffer can vary from the target. If the input stream is exhaused, the output buffer may be smaller than expected. If the data is incompressible, the maximum length can be exceeded by can be calculated to be: chunksize + 5 * (floor((chunksize - 1) / 16383) + 1) + 17 This accounts for additional header data gzip adds. For the default 16MiB chunksize, this results in the max size of the output buffer being: length + 16Mib + 5142 bytes compresslevel: Optional, defaults to 2. The desired compression level. chunksize: Optional, defaults to 16MiB. The chunk size used when reading data from the input stream to write into the output buffer. Returns: A file-like output buffer of compressed bytes, the number of bytes read from the input stream, and a flag denoting if the input stream was exhausted. """ in_read = 0 in_exhausted = False out_stream = StreamingBuffer() with gzip.GzipFile(mode='wb', fileobj=out_stream, compresslevel=compresslevel) as compress_stream: # Read until we've written at least length bytes to the output stream. while not length or out_stream.length < length: data = in_stream.read(chunksize) data_length = len(data) compress_stream.write(data) in_read += data_length # If we read less than requested, the stream is exhausted. if data_length < chunksize: in_exhausted = True break return out_stream, in_read, in_exhausted class StreamingBuffer(object): """Provides a file-like object that writes to a temporary buffer. When data is read from the buffer, it is permanently removed. This is useful when there are memory constraints preventing the entire buffer from being stored in memory. """ def __init__(self): # The buffer of byte arrays. self.__buf = deque() # The number of bytes in __buf. self.__size = 0 def __len__(self): return self.__size def __nonzero__(self): # For 32-bit python2.x, len() cannot exceed a 32-bit number; avoid # accidental len() calls from httplib in the form of "if this_object:". return bool(self.__size) @property def length(self): # For 32-bit python2.x, len() cannot exceed a 32-bit number. return self.__size def write(self, data): # Gzip can write many 0 byte chunks for highly compressible data. # Prevent them from being added internally. if data is not None and data: self.__buf.append(data) self.__size += len(data) def read(self, size=None): """Read at most size bytes from this buffer. Bytes read from this buffer are consumed and are permanently removed. Args: size: If provided, read no more than size bytes from the buffer. Otherwise, this reads the entire buffer. Returns: The bytes read from this buffer. """ if size is None: size = self.__size ret_list = [] while size > 0 and self.__buf: data = self.__buf.popleft() size -= len(data) ret_list.append(data) if size < 0: ret_list[-1], remainder = ret_list[-1][:size], ret_list[-1][size:] self.__buf.appendleft(remainder) ret = b''.join(ret_list) self.__size -= len(ret) return ret	5.02 KB	2025-03-11 18:02
📄 compression_test.py #!/usr/bin/env python # # Copyright 2017 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. """Tests for compression.""" import unittest from apitools.base.py import compression from apitools.base.py import gzip import six class CompressionTest(unittest.TestCase): def setUp(self): # Sample highly compressible data (~50MB). self.sample_data = b'abc' * 16777216 # Stream of the sample data. self.stream = six.BytesIO() self.stream.write(self.sample_data) self.length = self.stream.tell() self.stream.seek(0) def testCompressionExhausted(self): """Test full compression. Test that highly compressible data is actually compressed in entirety. """ output, read, exhausted = compression.CompressStream( self.stream, self.length, 9) # Ensure the compressed buffer is smaller than the input buffer. self.assertLess(output.length, self.length) # Ensure we read the entire input stream. self.assertEqual(read, self.length) # Ensure the input stream was exhausted. self.assertTrue(exhausted) def testCompressionUnbounded(self): """Test unbounded compression. Test that the input stream is exhausted when length is none. """ output, read, exhausted = compression.CompressStream( self.stream, None, 9) # Ensure the compressed buffer is smaller than the input buffer. self.assertLess(output.length, self.length) # Ensure we read the entire input stream. self.assertEqual(read, self.length) # Ensure the input stream was exhausted. self.assertTrue(exhausted) def testCompressionPartial(self): """Test partial compression. Test that the length parameter works correctly. The amount of data that's compressed can be greater than or equal to the requested length. """ output_length = 40 output, _, exhausted = compression.CompressStream( self.stream, output_length, 9) # Ensure the requested read size is <= the compressed buffer size. self.assertLessEqual(output_length, output.length) # Ensure the input stream was not exhausted. self.assertFalse(exhausted) def testCompressionIntegrity(self): """Test that compressed data can be decompressed.""" output, read, exhausted = compression.CompressStream( self.stream, self.length, 9) # Ensure uncompressed data matches the sample data. with gzip.GzipFile(fileobj=output) as f: original = f.read() self.assertEqual(original, self.sample_data) # Ensure we read the entire input stream. self.assertEqual(read, self.length) # Ensure the input stream was exhausted. self.assertTrue(exhausted) class StreamingBufferTest(unittest.TestCase): def setUp(self): self.stream = compression.StreamingBuffer() def testSimpleStream(self): """Test simple stream operations. Test that the stream can be written to and read from. Also test that reading from the stream consumes the bytes. """ # Ensure the stream is empty. self.assertEqual(self.stream.length, 0) # Ensure data is correctly written. self.stream.write(b'Sample data') self.assertEqual(self.stream.length, 11) # Ensure data can be read and the read data is purged from the stream. data = self.stream.read(11) self.assertEqual(data, b'Sample data') self.assertEqual(self.stream.length, 0) def testPartialReads(self): """Test partial stream reads. Test that the stream can be read in chunks while perserving the consumption mechanics. """ self.stream.write(b'Sample data') # Ensure data can be read and the read data is purged from the stream. data = self.stream.read(6) self.assertEqual(data, b'Sample') self.assertEqual(self.stream.length, 5) # Ensure the remaining data can be read. data = self.stream.read(5) self.assertEqual(data, b' data') self.assertEqual(self.stream.length, 0) def testTooShort(self): """Test excessive stream reads. Test that more data can be requested from the stream than available without raising an exception. """ self.stream.write(b'Sample') # Ensure requesting more data than available does not raise an # exception. data = self.stream.read(100) self.assertEqual(data, b'Sample') self.assertEqual(self.stream.length, 0)	5.19 KB	2025-03-11 18:02
📄 credentials_lib.py #!/usr/bin/env python # # 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. """Common credentials classes and constructors.""" from __future__ import print_function import argparse import contextlib import datetime import json import os import threading import warnings import httplib2 import oauth2client import oauth2client.client from oauth2client import service_account from oauth2client import tools # for gflags declarations import six from six.moves import http_client from six.moves import urllib from apitools.base.py import exceptions from apitools.base.py import util # App Engine does not support ctypes which are required for the # monotonic time used in fasteners. Conversely, App Engine does # not support colocated concurrent processes, so process locks # are not needed. try: import fasteners _FASTENERS_AVAILABLE = True except ImportError as import_error: server_env = os.environ.get('SERVER_SOFTWARE', '') if not (server_env.startswith('Development') or server_env.startswith('Google App Engine')): raise import_error _FASTENERS_AVAILABLE = False # Note: we try the oauth2client imports two ways, to accomodate layout # changes in oauth2client 2.0+. We can remove these once we no longer # support oauth2client < 2.0. # # pylint: disable=wrong-import-order,ungrouped-imports try: from oauth2client.contrib import gce except ImportError: from oauth2client import gce try: from oauth2client.contrib import multiprocess_file_storage _NEW_FILESTORE = True except ImportError: _NEW_FILESTORE = False try: from oauth2client.contrib import multistore_file except ImportError: from oauth2client import multistore_file try: import gflags FLAGS = gflags.FLAGS except ImportError: FLAGS = None __all__ = [ 'CredentialsFromFile', 'GaeAssertionCredentials', 'GceAssertionCredentials', 'GetCredentials', 'GetUserinfo', 'ServiceAccountCredentialsFromFile', ] # Lock when accessing the cache file to avoid resource contention. cache_file_lock = threading.Lock() def SetCredentialsCacheFileLock(lock): global cache_file_lock # pylint: disable=global-statement cache_file_lock = lock # List of additional methods we use when attempting to construct # credentials. Users can register their own methods here, which we try # before the defaults. _CREDENTIALS_METHODS = [] def _RegisterCredentialsMethod(method, position=None): """Register a new method for fetching credentials. This new method should be a function with signature: client_info, *kwds -> Credentials or None This method can be used as a decorator, unless position needs to be supplied. Note that method must always* accept arbitrary keyword arguments. Args: method: New credential-fetching method. position: (default: None) Where in the list of methods to add this; if None, we append. In all but rare cases, this should be either 0 or None. Returns: method, for use as a decorator. """ if position is None: position = len(_CREDENTIALS_METHODS) else: position = min(position, len(_CREDENTIALS_METHODS)) _CREDENTIALS_METHODS.insert(position, method) return method def GetCredentials(package_name, scopes, client_id, client_secret, user_agent, credentials_filename=None, api_key=None, # pylint: disable=unused-argument client=None, # pylint: disable=unused-argument oauth2client_args=None, kwds): """Attempt to get credentials, using an oauth dance as the last resort.""" scopes = util.NormalizeScopes(scopes) client_info = { 'client_id': client_id, 'client_secret': client_secret, 'scope': ' '.join(sorted(scopes)), 'user_agent': user_agent or '%s-generated/0.1' % package_name, } for method in _CREDENTIALS_METHODS: credentials = method(client_info, kwds) if credentials is not None: return credentials credentials_filename = credentials_filename or os.path.expanduser( '~/.apitools.token') credentials = CredentialsFromFile(credentials_filename, client_info, oauth2client_args=oauth2client_args) if credentials is not None: return credentials raise exceptions.CredentialsError('Could not create valid credentials') def ServiceAccountCredentialsFromFile(filename, scopes, user_agent=None): """Use the credentials in filename to create a token for scopes.""" filename = os.path.expanduser(filename) # We have two options, based on our version of oauth2client. if oauth2client.__version__ > '1.5.2': # oauth2client >= 2.0.0 credentials = ( service_account.ServiceAccountCredentials.from_json_keyfile_name( filename, scopes=scopes)) if credentials is not None: if user_agent is not None: credentials.user_agent = user_agent return credentials else: # oauth2client < 2.0.0 with open(filename) as keyfile: service_account_info = json.load(keyfile) account_type = service_account_info.get('type') if account_type != oauth2client.client.SERVICE_ACCOUNT: raise exceptions.CredentialsError( 'Invalid service account credentials: %s' % (filename,)) # pylint: disable=protected-access credentials = service_account._ServiceAccountCredentials( service_account_id=service_account_info['client_id'], service_account_email=service_account_info['client_email'], private_key_id=service_account_info['private_key_id'], private_key_pkcs8_text=service_account_info['private_key'], scopes=scopes, user_agent=user_agent) # pylint: enable=protected-access return credentials def ServiceAccountCredentialsFromP12File( service_account_name, private_key_filename, scopes, user_agent): """Create a new credential from the named .p12 keyfile.""" private_key_filename = os.path.expanduser(private_key_filename) scopes = util.NormalizeScopes(scopes) if oauth2client.__version__ > '1.5.2': # oauth2client >= 2.0.0 credentials = ( service_account.ServiceAccountCredentials.from_p12_keyfile( service_account_name, private_key_filename, scopes=scopes)) if credentials is not None: credentials.user_agent = user_agent return credentials else: # oauth2client < 2.0.0 with open(private_key_filename, 'rb') as key_file: return oauth2client.client.SignedJwtAssertionCredentials( service_account_name, key_file.read(), scopes, user_agent=user_agent) def _GceMetadataRequest(relative_url, use_metadata_ip=False): """Request the given url from the GCE metadata service.""" if use_metadata_ip: base_url = os.environ.get('GCE_METADATA_IP', '169.254.169.254') else: base_url = os.environ.get( 'GCE_METADATA_ROOT', 'metadata.google.internal') url = 'http://' + base_url + '/computeMetadata/v1/' + relative_url # Extra header requirement can be found here: # https://developers.google.com/compute/docs/metadata headers = {'Metadata-Flavor': 'Google'} request = urllib.request.Request(url, headers=headers) opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) try: response = opener.open(request) except urllib.error.URLError as e: raise exceptions.CommunicationError( 'Could not reach metadata service: %s' % e.reason) return response class GceAssertionCredentials(gce.AppAssertionCredentials): """Assertion credentials for GCE instances.""" def __init__(self, scopes=None, service_account_name='default', kwds): """Initializes the credentials instance. Args: scopes: The scopes to get. If None, whatever scopes that are available to the instance are used. service_account_name: The service account to retrieve the scopes from. kwds: Additional keyword args. """ # If there is a connectivity issue with the metadata server, # detection calls may fail even if we've already successfully # identified these scopes in the same execution. However, the # available scopes don't change once an instance is created, # so there is no reason to perform more than one query. self.__service_account_name = six.ensure_text( service_account_name, encoding='utf-8',) cached_scopes = None cache_filename = kwds.get('cache_filename') if cache_filename: cached_scopes = self._CheckCacheFileForMatch( cache_filename, scopes) scopes = cached_scopes or self._ScopesFromMetadataServer(scopes) if cache_filename and not cached_scopes: self._WriteCacheFile(cache_filename, scopes) # We check the scopes above, but don't need them again after # this point. Newer versions of oauth2client let us drop them # here, but since we support older versions as well, we just # catch and squelch the warning. with warnings.catch_warnings(): warnings.simplefilter('ignore') super(GceAssertionCredentials, self).__init__(scope=scopes, *kwds) @classmethod def Get(cls, args, *kwds): try: return cls(args, kwds) except exceptions.Error: return None def _CheckCacheFileForMatch(self, cache_filename, scopes): """Checks the cache file to see if it matches the given credentials. Args: cache_filename: Cache filename to check. scopes: Scopes for the desired credentials. Returns: List of scopes (if cache matches) or None. """ creds = { # Credentials metadata dict. 'scopes': sorted(list(scopes)) if scopes else None, 'svc_acct_name': self.__service_account_name, } cache_file = _MultiProcessCacheFile(cache_filename) try: cached_creds_str = cache_file.LockedRead() if not cached_creds_str: return None cached_creds = json.loads(cached_creds_str) if creds['svc_acct_name'] == cached_creds['svc_acct_name']: if creds['scopes'] in (None, cached_creds['scopes']): return cached_creds['scopes'] except KeyboardInterrupt: raise except: # pylint: disable=bare-except # Treat exceptions as a cache miss. pass def _WriteCacheFile(self, cache_filename, scopes): """Writes the credential metadata to the cache file. This does not save the credentials themselves (CredentialStore class optionally handles that after this class is initialized). Args: cache_filename: Cache filename to check. scopes: Scopes for the desired credentials. """ # Credentials metadata dict. scopes = sorted([six.ensure_text(scope) for scope in scopes]) creds = {'scopes': scopes, 'svc_acct_name': self.__service_account_name} creds_str = json.dumps(creds) cache_file = _MultiProcessCacheFile(cache_filename) try: cache_file.LockedWrite(creds_str) except KeyboardInterrupt: raise except: # pylint: disable=bare-except # Treat exceptions as a cache miss. pass def _ScopesFromMetadataServer(self, scopes): """Returns instance scopes based on GCE metadata server.""" if not util.DetectGce(): raise exceptions.ResourceUnavailableError( 'GCE credentials requested outside a GCE instance') if not self.GetServiceAccount(self.__service_account_name): raise exceptions.ResourceUnavailableError( 'GCE credentials requested but service account ' '%s does not exist.' % self.__service_account_name) if scopes: scope_ls = util.NormalizeScopes(scopes) instance_scopes = self.GetInstanceScopes() if scope_ls > instance_scopes: raise exceptions.CredentialsError( 'Instance did not have access to scopes %s' % ( sorted(list(scope_ls - instance_scopes)),)) else: scopes = self.GetInstanceScopes() return scopes def GetServiceAccount(self, account): relative_url = 'instance/service-accounts' response = _GceMetadataRequest(relative_url) response_lines = [six.ensure_str(line).rstrip(u'/\n\r') for line in response.readlines()] return account in response_lines def GetInstanceScopes(self): relative_url = 'instance/service-accounts/{0}/scopes'.format( self.__service_account_name) response = _GceMetadataRequest(relative_url) return util.NormalizeScopes(six.ensure_str(scope).strip() for scope in response.readlines()) # pylint: disable=arguments-differ def _refresh(self, do_request): """Refresh self.access_token. This function replaces AppAssertionCredentials._refresh, which does not use the credential store and is therefore poorly suited for multi-threaded scenarios. Args: do_request: A function matching httplib2.Http.request's signature. """ # pylint: disable=protected-access oauth2client.client.OAuth2Credentials._refresh(self, do_request) # pylint: enable=protected-access def _do_refresh_request(self, unused_http_request): """Refresh self.access_token by querying the metadata server. If self.store is initialized, store acquired credentials there. """ relative_url = 'instance/service-accounts/{0}/token'.format( self.__service_account_name) try: response = _GceMetadataRequest(relative_url) except exceptions.CommunicationError: self.invalid = True if self.store: self.store.locked_put(self) raise content = six.ensure_str(response.read()) try: credential_info = json.loads(content) except ValueError: raise exceptions.CredentialsError( 'Could not parse response as JSON: %s' % content) self.access_token = credential_info['access_token'] if 'expires_in' in credential_info: expires_in = int(credential_info['expires_in']) self.token_expiry = ( datetime.timedelta(seconds=expires_in) + datetime.datetime.now(tz=datetime.timezone.utc).replace(tzinfo=None)) else: self.token_expiry = None self.invalid = False if self.store: self.store.locked_put(self) def to_json(self): # OAuth2Client made gce.AppAssertionCredentials unserializable as of # v3.0, but we need those credentials to be serializable for use with # this library, so we use AppAssertionCredentials' parent's to_json # method. # pylint: disable=bad-super-call return super(gce.AppAssertionCredentials, self).to_json() @classmethod def from_json(cls, json_data): data = json.loads(json_data) kwargs = {} if 'cache_filename' in data.get('kwargs', []): kwargs['cache_filename'] = data['kwargs']['cache_filename'] # Newer versions of GceAssertionCredentials don't have a "scope" # attribute. scope_list = None if 'scope' in data: scope_list = [data['scope']] credentials = GceAssertionCredentials(scopes=scope_list, kwargs) if 'access_token' in data: credentials.access_token = data['access_token'] if 'token_expiry' in data: credentials.token_expiry = datetime.datetime.strptime( data['token_expiry'], oauth2client.client.EXPIRY_FORMAT) if 'invalid' in data: credentials.invalid = data['invalid'] return credentials @property def serialization_data(self): raise NotImplementedError( 'Cannot serialize credentials for GCE service accounts.') # TODO(craigcitro): Currently, we can't even load # `oauth2client.appengine` without being on appengine, because of how # it handles imports. Fix that by splitting that module into # GAE-specific and GAE-independent bits, and guarding imports. class GaeAssertionCredentials(oauth2client.client.AssertionCredentials): """Assertion credentials for Google App Engine apps.""" def __init__(self, scopes, kwds): if not util.DetectGae(): raise exceptions.ResourceUnavailableError( 'GCE credentials requested outside a GCE instance') self._scopes = list(util.NormalizeScopes(scopes)) super(GaeAssertionCredentials, self).__init__(None, kwds) @classmethod def Get(cls, args, kwds): try: return cls(args, kwds) except exceptions.Error: return None @classmethod def from_json(cls, json_data): data = json.loads(json_data) return GaeAssertionCredentials(data['_scopes']) def _refresh(self, _): """Refresh self.access_token. Args: _: (ignored) A function matching httplib2.Http.request's signature. """ # pylint: disable=import-error from google.appengine.api import app_identity try: token, _ = app_identity.get_access_token(self._scopes) except app_identity.Error as e: raise exceptions.CredentialsError(str(e)) self.access_token = token def sign_blob(self, blob): """Cryptographically sign a blob (of bytes). This method is provided to support a common interface, but the actual key used for a Google Compute Engine service account is not available, so it can't be used to sign content. Args: blob: bytes, Message to be signed. Raises: NotImplementedError, always. """ raise NotImplementedError( 'Compute Engine service accounts cannot sign blobs') def _GetRunFlowFlags(args=None): """Retrieves command line flags based on gflags module.""" # There's one rare situation where gsutil will not have argparse # available, but doesn't need anything depending on argparse anyway, # since they're bringing their own credentials. So we just allow this # to fail with an ImportError in those cases. # parser = argparse.ArgumentParser(parents=[tools.argparser]) # Get command line argparse flags. flags, _ = parser.parse_known_args(args=args) # Allow `gflags` and `argparse` to be used side-by-side. if hasattr(FLAGS, 'auth_host_name'): flags.auth_host_name = FLAGS.auth_host_name if hasattr(FLAGS, 'auth_host_port'): flags.auth_host_port = FLAGS.auth_host_port if hasattr(FLAGS, 'auth_local_webserver'): flags.noauth_local_webserver = (not FLAGS.auth_local_webserver) return flags # TODO(craigcitro): Switch this from taking a path to taking a stream. def CredentialsFromFile(path, client_info, oauth2client_args=None): """Read credentials from a file.""" user_agent = client_info['user_agent'] scope_key = client_info['scope'] if not isinstance(scope_key, six.string_types): scope_key = ':'.join(scope_key) storage_key = client_info['client_id'] + user_agent + scope_key if _NEW_FILESTORE: credential_store = multiprocess_file_storage.MultiprocessFileStorage( path, storage_key) else: credential_store = multistore_file.get_credential_storage_custom_string_key( # noqa path, storage_key) if hasattr(FLAGS, 'auth_local_webserver'): FLAGS.auth_local_webserver = False credentials = credential_store.get() if credentials is None or credentials.invalid: print('Generating new OAuth credentials ...') for _ in range(20): # If authorization fails, we want to retry, rather than let this # cascade up and get caught elsewhere. If users want out of the # retry loop, they can ^C. try: flow = oauth2client.client.OAuth2WebServerFlow(client_info) flags = _GetRunFlowFlags(args=oauth2client_args) credentials = tools.run_flow(flow, credential_store, flags) break except (oauth2client.client.FlowExchangeError, SystemExit) as e: # Here SystemExit is "no credential at all", and the # FlowExchangeError is "invalid" -- usually because # you reused a token. print('Invalid authorization: %s' % (e,)) except httplib2.HttpLib2Error as e: print('Communication error: %s' % (e,)) raise exceptions.CredentialsError( 'Communication error creating credentials: %s' % e) return credentials class _MultiProcessCacheFile(object): """Simple multithreading and multiprocessing safe cache file. Notes on behavior: * the fasteners.InterProcessLock object cannot reliably prevent threads from double-acquiring a lock. A threading lock is used in addition to the InterProcessLock. The threading lock is always acquired first and released last. * The interprocess lock will not deadlock. If a process can not acquire the interprocess lock within `_lock_timeout` the call will return as a cache miss or unsuccessful cache write. * App Engine environments cannot be process locked because (1) the runtime does not provide monotonic time and (2) different processes may or may not share the same machine. Because of this, process locks are disabled and locking is only guaranteed to protect against multithreaded access. """ _lock_timeout = 1 _encoding = 'utf-8' _thread_lock = threading.Lock() def __init__(self, filename): self._file = None self._filename = filename if _FASTENERS_AVAILABLE: self._process_lock_getter = self._ProcessLockAcquired self._process_lock = fasteners.InterProcessLock( '{0}.lock'.format(filename)) else: self._process_lock_getter = self._DummyLockAcquired self._process_lock = None @contextlib.contextmanager def _ProcessLockAcquired(self): """Context manager for process locks with timeout.""" try: is_locked = self._process_lock.acquire(timeout=self._lock_timeout) yield is_locked finally: if is_locked: self._process_lock.release() @contextlib.contextmanager def _DummyLockAcquired(self): """Lock context manager for environments without process locks.""" yield True def LockedRead(self): """Acquire an interprocess lock and dump cache contents. This method safely acquires the locks then reads a string from the cache file. If the file does not exist and cannot be created, it will return None. If the locks cannot be acquired, this will also return None. Returns: cache data - string if present, None on failure. """ file_contents = None with self._thread_lock: if not self._EnsureFileExists(): return None with self._process_lock_getter() as acquired_plock: if not acquired_plock: return None with open(self._filename, 'rb') as f: file_contents = f.read().decode(encoding=self._encoding) return file_contents def LockedWrite(self, cache_data): """Acquire an interprocess lock and write a string. This method safely acquires the locks then writes a string to the cache file. If the string is written successfully the function will return True, if the write fails for any reason it will return False. Args: cache_data: string or bytes to write. Returns: bool: success """ if isinstance(cache_data, six.text_type): cache_data = cache_data.encode(encoding=self._encoding) with self._thread_lock: if not self._EnsureFileExists(): return False with self._process_lock_getter() as acquired_plock: if not acquired_plock: return False with open(self._filename, 'wb') as f: f.write(cache_data) return True def _EnsureFileExists(self): """Touches a file; returns False on error, True on success.""" if not os.path.exists(self._filename): old_umask = os.umask(0o177) try: open(self._filename, 'a+b').close() except OSError: return False finally: os.umask(old_umask) return True # TODO(craigcitro): Push this into oauth2client. def GetUserinfo(credentials, http=None): # pylint: disable=invalid-name """Get the userinfo associated with the given credentials. This is dependent on the token having either the userinfo.email or userinfo.profile scope for the given token. Args: credentials: (oauth2client.client.Credentials) incoming credentials http: (httplib2.Http, optional) http instance to use Returns: The email address for this token, or None if the required scopes aren't available. """ http = http or httplib2.Http() url = _GetUserinfoUrl(credentials) # We ignore communication woes here (i.e. SSL errors, socket # timeout), as handling these should be done in a common location. response, content = http.request(url) if response.status == http_client.BAD_REQUEST: credentials.refresh(http) url = _GetUserinfoUrl(credentials) response, content = http.request(url) return json.loads(content or '{}') # Save ourselves from an empty reply. def _GetUserinfoUrl(credentials): url_root = 'https://oauth2.googleapis.com/tokeninfo' query_args = {'access_token': credentials.access_token} return '?'.join((url_root, urllib.parse.urlencode(query_args))) @_RegisterCredentialsMethod def _GetServiceAccountCredentials( client_info, service_account_name=None, service_account_keyfile=None, service_account_json_keyfile=None, unused_kwds): """Returns ServiceAccountCredentials from give file.""" scopes = client_info['scope'].split() user_agent = client_info['user_agent'] # Use the .json credentials, if provided. if service_account_json_keyfile: return ServiceAccountCredentialsFromFile( service_account_json_keyfile, scopes, user_agent=user_agent) # Fall back to .p12 if there's no .json credentials. if ((service_account_name and not service_account_keyfile) or (service_account_keyfile and not service_account_name)): raise exceptions.CredentialsError( 'Service account name or keyfile provided without the other') if service_account_name is not None: return ServiceAccountCredentialsFromP12File( service_account_name, service_account_keyfile, scopes, user_agent) @_RegisterCredentialsMethod def _GetGaeServiceAccount(client_info, unused_kwds): scopes = client_info['scope'].split(' ') return GaeAssertionCredentials.Get(scopes=scopes) @_RegisterCredentialsMethod def _GetGceServiceAccount(client_info, unused_kwds): scopes = client_info['scope'].split(' ') return GceAssertionCredentials.Get(scopes=scopes) @_RegisterCredentialsMethod def _GetApplicationDefaultCredentials( client_info, skip_application_default_credentials=False, unused_kwds): """Returns ADC with right scopes.""" scopes = client_info['scope'].split() if skip_application_default_credentials: return None gc = oauth2client.client.GoogleCredentials with cache_file_lock: try: # pylint: disable=protected-access # We've already done our own check for GAE/GCE # credentials, we don't want to pay for checking again. credentials = gc._implicit_credentials_from_files() except oauth2client.client.ApplicationDefaultCredentialsError: return None # If we got back a non-service account credential, we need to use # a heuristic to decide whether or not the application default # credential will work for us. We assume that if we're requesting # cloud-platform, our scopes are a subset of cloud scopes, and the # ADC will work. cp = 'https://www.googleapis.com/auth/cloud-platform' if credentials is None: return None if not isinstance(credentials, gc) or cp in scopes: return credentials.create_scoped(scopes) return None	29.42 KB	2025-03-11 18:02
📄 credentials_lib_test.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. import json import os.path import shutil import tempfile import unittest import mock import six from apitools.base.py import credentials_lib from apitools.base.py import util class MetadataMock(object): def __init__(self, scopes=None, service_account_name=None): self._scopes = scopes or ['scope1'] self._sa = service_account_name or 'default' def __call__(self, request_url): if request_url.endswith('scopes'): return six.StringIO(''.join(self._scopes)) elif request_url.endswith('service-accounts'): return six.StringIO(self._sa) elif request_url.endswith( '/service-accounts/%s/token' % self._sa): return six.StringIO('{"access_token": "token"}') self.fail('Unexpected HTTP request to %s' % request_url) class CredentialsLibTest(unittest.TestCase): def _RunGceAssertionCredentials( self, service_account_name=None, scopes=None, cache_filename=None): kwargs = {} if service_account_name is not None: kwargs['service_account_name'] = service_account_name if cache_filename is not None: kwargs['cache_filename'] = cache_filename service_account_name = service_account_name or 'default' credentials = credentials_lib.GceAssertionCredentials( scopes, **kwargs) self.assertIsNone(credentials._refresh(None)) return credentials def _GetServiceCreds(self, service_account_name=None, scopes=None): metadatamock = MetadataMock(scopes, service_account_name) with mock.patch.object(util, 'DetectGce', autospec=True) as gce_detect: gce_detect.return_value = True with mock.patch.object(credentials_lib, '_GceMetadataRequest', side_effect=metadatamock, autospec=True) as opener_mock: credentials = self._RunGceAssertionCredentials( service_account_name=service_account_name, scopes=scopes) self.assertEqual(3, opener_mock.call_count) return credentials def testGceServiceAccounts(self): scopes = ['scope1'] self._GetServiceCreds(service_account_name=None, scopes=None) self._GetServiceCreds(service_account_name=None, scopes=scopes) self._GetServiceCreds( service_account_name='my_service_account', scopes=scopes) def testGceAssertionCredentialsToJson(self): scopes = ['scope1'] service_account_name = 'my_service_account' # Ensure that we can obtain a JSON representation of # GceAssertionCredentials to put in a credential Storage object, and # that the JSON representation is valid. original_creds = self._GetServiceCreds( service_account_name=service_account_name, scopes=scopes) original_creds_json_str = original_creds.to_json() json.loads(original_creds_json_str) @mock.patch.object(util, 'DetectGce', autospec=True) def testGceServiceAccountsCached(self, mock_detect): mock_detect.return_value = True tempd = tempfile.mkdtemp() tempname = os.path.join(tempd, 'creds') scopes = ['scope1'] service_account_name = 'some_service_account_name' metadatamock = MetadataMock(scopes, service_account_name) with mock.patch.object(credentials_lib, '_GceMetadataRequest', side_effect=metadatamock, autospec=True) as opener_mock: try: creds1 = self._RunGceAssertionCredentials( service_account_name=service_account_name, cache_filename=tempname, scopes=scopes) pre_cache_call_count = opener_mock.call_count creds2 = self._RunGceAssertionCredentials( service_account_name=service_account_name, cache_filename=tempname, scopes=None) finally: shutil.rmtree(tempd) self.assertEqual(creds1.client_id, creds2.client_id) self.assertEqual(pre_cache_call_count, 3) # Caching obviates the need for extra metadata server requests. # Only one metadata request is made if the cache is hit. self.assertEqual(opener_mock.call_count, 4) def testGetServiceAccount(self): # We'd also like to test the metadata calls, which requires # having some knowledge about how HTTP calls are made (so that # we can mock them). It's unfortunate, but there's no way # around it. creds = self._GetServiceCreds() opener = mock.MagicMock() opener.open = mock.MagicMock() opener.open.return_value = six.StringIO('default/\nanother') with mock.patch.object(six.moves.urllib.request, 'build_opener', return_value=opener, autospec=True) as build_opener: creds.GetServiceAccount('default') self.assertEqual(1, build_opener.call_count) self.assertEqual(1, opener.open.call_count) req = opener.open.call_args[0][0] self.assertTrue(req.get_full_url().startswith( 'http://metadata.google.internal/')) # The urllib module does weird things with header case. self.assertEqual('Google', req.get_header('Metadata-flavor')) def testGetAdcNone(self): # Tests that we correctly return None when ADC aren't present in # the well-known file. creds = credentials_lib._GetApplicationDefaultCredentials( client_info={'scope': ''}) self.assertIsNone(creds) class TestGetRunFlowFlags(unittest.TestCase): def setUp(self): self._flags_actual = credentials_lib.FLAGS def tearDown(self): credentials_lib.FLAGS = self._flags_actual def test_with_gflags(self): HOST = 'myhostname' PORT = '144169' class MockFlags(object): auth_host_name = HOST auth_host_port = PORT auth_local_webserver = False credentials_lib.FLAGS = MockFlags flags = credentials_lib._GetRunFlowFlags([ '--auth_host_name=%s' % HOST, '--auth_host_port=%s' % PORT, '--noauth_local_webserver', ]) self.assertEqual(flags.auth_host_name, HOST) self.assertEqual(flags.auth_host_port, PORT) self.assertEqual(flags.logging_level, 'ERROR') self.assertEqual(flags.noauth_local_webserver, True) def test_without_gflags(self): credentials_lib.FLAGS = None flags = credentials_lib._GetRunFlowFlags([]) self.assertEqual(flags.auth_host_name, 'localhost') self.assertEqual(flags.auth_host_port, [8080, 8090]) self.assertEqual(flags.logging_level, 'ERROR') self.assertEqual(flags.noauth_local_webserver, False)	7.58 KB	2025-03-11 18:02
📄 encoding.py #!/usr/bin/env python # # 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. """Common code for converting proto to other formats, such as JSON.""" # pylint:disable=wildcard-import from apitools.base.py.encoding_helper import * import apitools.base.py.extra_types # pylint:disable=unused-import # pylint:disable=undefined-all-variable __all__ = [ 'CopyProtoMessage', 'JsonToMessage', 'MessageToJson', 'DictToMessage', 'MessageToDict', 'PyValueToMessage', 'MessageToPyValue', 'MessageToRepr', 'GetCustomJsonFieldMapping', 'AddCustomJsonFieldMapping', 'GetCustomJsonEnumMapping', 'AddCustomJsonEnumMapping', ]	1.16 KB	2025-03-11 18:02
📄 encoding_helper.py #!/usr/bin/env python # # Copyright 2018 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. """Common code for converting proto to other formats, such as JSON.""" import base64 import collections import datetime import json import six from apitools.base.protorpclite import message_types from apitools.base.protorpclite import messages from apitools.base.protorpclite import protojson from apitools.base.py import exceptions _Codec = collections.namedtuple('_Codec', ['encoder', 'decoder']) CodecResult = collections.namedtuple('CodecResult', ['value', 'complete']) class EdgeType(object): """The type of transition made by an edge.""" SCALAR = 1 REPEATED = 2 MAP = 3 class ProtoEdge(collections.namedtuple('ProtoEdge', ['type_', 'field', 'index'])): """A description of a one-level transition from a message to a value. Protobuf messages can be arbitrarily nested as fields can be defined with any "message" type. This nesting property means that there are often many levels of proto messages within a single message instance. This class can unambiguously describe a single step from a message to some nested value. Properties: type_: EdgeType, The type of transition represented by this edge. field: str, The name of the message-typed field. index: Any, Additional data needed to make the transition. The semantics of the "index" property change based on the value of "type_": SCALAR: ignored. REPEATED: a numeric index into "field"'s list. MAP: a key into "field"'s mapping. """ __slots__ = () def __str__(self): if self.type_ == EdgeType.SCALAR: return self.field else: return '{}[{}]'.format(self.field, self.index) # TODO(craigcitro): Make these non-global. _UNRECOGNIZED_FIELD_MAPPINGS = {} _CUSTOM_MESSAGE_CODECS = {} _CUSTOM_FIELD_CODECS = {} _FIELD_TYPE_CODECS = {} def MapUnrecognizedFields(field_name): """Register field_name as a container for unrecognized fields.""" def Register(cls): _UNRECOGNIZED_FIELD_MAPPINGS[cls] = field_name return cls return Register def RegisterCustomMessageCodec(encoder, decoder): """Register a custom encoder/decoder for this message class.""" def Register(cls): _CUSTOM_MESSAGE_CODECS[cls] = _Codec(encoder=encoder, decoder=decoder) return cls return Register def RegisterCustomFieldCodec(encoder, decoder): """Register a custom encoder/decoder for this field.""" def Register(field): _CUSTOM_FIELD_CODECS[field] = _Codec(encoder=encoder, decoder=decoder) return field return Register def RegisterFieldTypeCodec(encoder, decoder): """Register a custom encoder/decoder for all fields of this type.""" def Register(field_type): _FIELD_TYPE_CODECS[field_type] = _Codec( encoder=encoder, decoder=decoder) return field_type return Register def CopyProtoMessage(message): """Make a deep copy of a message.""" return JsonToMessage(type(message), MessageToJson(message)) def MessageToJson(message, include_fields=None): """Convert the given message to JSON.""" result = _ProtoJsonApiTools.Get().encode_message(message) return _IncludeFields(result, message, include_fields) def JsonToMessage(message_type, message): """Convert the given JSON to a message of type message_type.""" return _ProtoJsonApiTools.Get().decode_message(message_type, message) # TODO(craigcitro): Do this directly, instead of via JSON. def DictToMessage(d, message_type): """Convert the given dictionary to a message of type message_type.""" return JsonToMessage(message_type, json.dumps(d)) def MessageToDict(message): """Convert the given message to a dictionary.""" return json.loads(MessageToJson(message)) def DictToAdditionalPropertyMessage(properties, additional_property_type, sort_items=False): """Convert the given dictionary to an AdditionalProperty message.""" items = properties.items() if sort_items: items = sorted(items) map_ = [] for key, value in items: map_.append(additional_property_type.AdditionalProperty( key=key, value=value)) return additional_property_type(additionalProperties=map_) def PyValueToMessage(message_type, value): """Convert the given python value to a message of type message_type.""" return JsonToMessage(message_type, json.dumps(value)) def MessageToPyValue(message): """Convert the given message to a python value.""" return json.loads(MessageToJson(message)) def MessageToRepr(msg, multiline=False, kwargs): """Return a repr-style string for a protorpc message. protorpc.Message.__repr__ does not return anything that could be considered python code. Adding this function lets us print a protorpc message in such a way that it could be pasted into code later, and used to compare against other things. Args: msg: protorpc.Message, the message to be repr'd. multiline: bool, True if the returned string should have each field assignment on its own line. kwargs: {str:str}, Additional flags for how to format the string. Known *kwargs: shortstrings: bool, True if all string values should be truncated at 100 characters, since when mocking the contents typically don't matter except for IDs, and IDs are usually less than 100 characters. no_modules: bool, True if the long module name should not be printed with each type. Returns: str, A string of valid python (assuming the right imports have been made) that recreates the message passed into this function. """ # TODO(jasmuth): craigcitro suggests a pretty-printer from apitools/gen. indent = kwargs.get('indent', 0) def IndentKwargs(kwargs): kwargs = dict(kwargs) kwargs['indent'] = kwargs.get('indent', 0) + 4 return kwargs if isinstance(msg, list): s = '[' for item in msg: if multiline: s += '\n' + ' ' (indent + 4) s += MessageToRepr( item, multiline=multiline, *IndentKwargs(kwargs)) + ',' if multiline: s += '\n' + ' ' indent s += ']' return s if isinstance(msg, messages.Message): s = type(msg).__name__ + '(' if not kwargs.get('no_modules'): s = msg.__module__ + '.' + s names = sorted([field.name for field in msg.all_fields()]) for name in names: field = msg.field_by_name(name) if multiline: s += '\n' + ' ' * (indent + 4) value = getattr(msg, field.name) s += field.name + '=' + MessageToRepr( value, multiline=multiline, *IndentKwargs(kwargs)) + ',' if multiline: s += '\n' + ' ' indent s += ')' return s if isinstance(msg, six.string_types): if kwargs.get('shortstrings') and len(msg) > 100: msg = msg[:100] if isinstance(msg, datetime.datetime): class SpecialTZInfo(datetime.tzinfo): def __init__(self, offset): super(SpecialTZInfo, self).__init__() self.offset = offset def __repr__(self): s = 'TimeZoneOffset(' + repr(self.offset) + ')' if not kwargs.get('no_modules'): s = 'apitools.base.protorpclite.util.' + s return s msg = datetime.datetime( msg.year, msg.month, msg.day, msg.hour, msg.minute, msg.second, msg.microsecond, SpecialTZInfo(msg.tzinfo.utcoffset(0))) return repr(msg) def _GetField(message, field_path): for field in field_path: if field not in dir(message): raise KeyError('no field "%s"' % field) message = getattr(message, field) return message def _SetField(dictblob, field_path, value): for field in field_path[:-1]: dictblob = dictblob.setdefault(field, {}) dictblob[field_path[-1]] = value def _IncludeFields(encoded_message, message, include_fields): """Add the requested fields to the encoded message.""" if include_fields is None: return encoded_message result = json.loads(encoded_message) for field_name in include_fields: try: value = _GetField(message, field_name.split('.')) nullvalue = None if isinstance(value, list): nullvalue = [] except KeyError: raise exceptions.InvalidDataError( 'No field named %s in message of type %s' % ( field_name, type(message))) _SetField(result, field_name.split('.'), nullvalue) return json.dumps(result) def _GetFieldCodecs(field, attr): result = [ getattr(_CUSTOM_FIELD_CODECS.get(field), attr, None), getattr(_FIELD_TYPE_CODECS.get(type(field)), attr, None), ] return [x for x in result if x is not None] class _ProtoJsonApiTools(protojson.ProtoJson): """JSON encoder used by apitools clients.""" _INSTANCE = None @classmethod def Get(cls): if cls._INSTANCE is None: cls._INSTANCE = cls() return cls._INSTANCE def decode_message(self, message_type, encoded_message): if message_type in _CUSTOM_MESSAGE_CODECS: return _CUSTOM_MESSAGE_CODECS[ message_type].decoder(encoded_message) result = _DecodeCustomFieldNames(message_type, encoded_message) result = super(_ProtoJsonApiTools, self).decode_message( message_type, result) result = _ProcessUnknownEnums(result, encoded_message) result = _ProcessUnknownMessages(result, encoded_message) return _DecodeUnknownFields(result, encoded_message) def decode_field(self, field, value): """Decode the given JSON value. Args: field: a messages.Field for the field we're decoding. value: a python value we'd like to decode. Returns: A value suitable for assignment to field. """ for decoder in _GetFieldCodecs(field, 'decoder'): result = decoder(field, value) value = result.value if result.complete: return value if isinstance(field, messages.MessageField): field_value = self.decode_message( field.message_type, json.dumps(value)) elif isinstance(field, messages.EnumField): value = GetCustomJsonEnumMapping( field.type, json_name=value) or value try: field_value = super( _ProtoJsonApiTools, self).decode_field(field, value) except messages.DecodeError: if not isinstance(value, six.string_types): raise field_value = None else: field_value = super( _ProtoJsonApiTools, self).decode_field(field, value) return field_value def encode_message(self, message): if isinstance(message, messages.FieldList): return '[%s]' % (', '.join(self.encode_message(x) for x in message)) # pylint: disable=unidiomatic-typecheck if type(message) in _CUSTOM_MESSAGE_CODECS: return _CUSTOM_MESSAGE_CODECS[type(message)].encoder(message) message = _EncodeUnknownFields(message) result = super(_ProtoJsonApiTools, self).encode_message(message) result = _EncodeCustomFieldNames(message, result) return json.dumps(json.loads(result), sort_keys=True) def encode_field(self, field, value): """Encode the given value as JSON. Args: field: a messages.Field for the field we're encoding. value: a value for field. Returns: A python value suitable for json.dumps. """ for encoder in _GetFieldCodecs(field, 'encoder'): result = encoder(field, value) value = result.value if result.complete: return value if isinstance(field, messages.EnumField): if field.repeated: remapped_value = [GetCustomJsonEnumMapping( field.type, python_name=e.name) or e.name for e in value] else: remapped_value = GetCustomJsonEnumMapping( field.type, python_name=value.name) if remapped_value: return remapped_value if (isinstance(field, messages.MessageField) and not isinstance(field, message_types.DateTimeField)): value = json.loads(self.encode_message(value)) return super(_ProtoJsonApiTools, self).encode_field(field, value) # TODO(craigcitro): Fold this and _IncludeFields in as codecs. def _DecodeUnknownFields(message, encoded_message): """Rewrite unknown fields in message into message.destination.""" destination = _UNRECOGNIZED_FIELD_MAPPINGS.get(type(message)) if destination is None: return message pair_field = message.field_by_name(destination) if not isinstance(pair_field, messages.MessageField): raise exceptions.InvalidDataFromServerError( 'Unrecognized fields must be mapped to a compound ' 'message type.') pair_type = pair_field.message_type # TODO(craigcitro): Add more error checking around the pair # type being exactly what we suspect (field names, etc). if isinstance(pair_type.value, messages.MessageField): new_values = _DecodeUnknownMessages( message, json.loads(encoded_message), pair_type) else: new_values = _DecodeUnrecognizedFields(message, pair_type) setattr(message, destination, new_values) # We could probably get away with not setting this, but # why not clear it? setattr(message, '_Message__unrecognized_fields', {}) return message def _DecodeUnknownMessages(message, encoded_message, pair_type): """Process unknown fields in encoded_message of a message type.""" field_type = pair_type.value.type new_values = [] all_field_names = [x.name for x in message.all_fields()] for name, value_dict in six.iteritems(encoded_message): if name in all_field_names: continue value = PyValueToMessage(field_type, value_dict) if pair_type.value.repeated: value = _AsMessageList(value) new_pair = pair_type(key=name, value=value) new_values.append(new_pair) return new_values def _DecodeUnrecognizedFields(message, pair_type): """Process unrecognized fields in message.""" new_values = [] codec = _ProtoJsonApiTools.Get() for unknown_field in message.all_unrecognized_fields(): # TODO(craigcitro): Consider validating the variant if # the assignment below doesn't take care of it. It may # also be necessary to check it in the case that the # type has multiple encodings. value, _ = message.get_unrecognized_field_info(unknown_field) value_type = pair_type.field_by_name('value') if isinstance(value_type, messages.MessageField): decoded_value = DictToMessage(value, pair_type.value.message_type) else: decoded_value = codec.decode_field( pair_type.value, value) try: new_pair_key = str(unknown_field) except UnicodeEncodeError: new_pair_key = protojson.ProtoJson().decode_field( pair_type.key, unknown_field) new_pair = pair_type(key=new_pair_key, value=decoded_value) new_values.append(new_pair) return new_values def _CopyProtoMessageVanillaProtoJson(message): codec = protojson.ProtoJson() return codec.decode_message(type(message), codec.encode_message(message)) def _EncodeUnknownFields(message): """Remap unknown fields in message out of message.source.""" source = _UNRECOGNIZED_FIELD_MAPPINGS.get(type(message)) if source is None: return message # CopyProtoMessage uses _ProtoJsonApiTools, which uses this message. Use # the vanilla protojson-based copy function to avoid infinite recursion. result = _CopyProtoMessageVanillaProtoJson(message) pairs_field = message.field_by_name(source) if not isinstance(pairs_field, messages.MessageField): raise exceptions.InvalidUserInputError( 'Invalid pairs field %s' % pairs_field) pairs_type = pairs_field.message_type value_field = pairs_type.field_by_name('value') value_variant = value_field.variant pairs = getattr(message, source) codec = _ProtoJsonApiTools.Get() for pair in pairs: encoded_value = codec.encode_field(value_field, pair.value) result.set_unrecognized_field(pair.key, encoded_value, value_variant) setattr(result, source, []) return result def _SafeEncodeBytes(field, value): """Encode the bytes in value as urlsafe base64.""" try: if field.repeated: result = [base64.urlsafe_b64encode(byte) for byte in value] else: result = base64.urlsafe_b64encode(value) complete = True except TypeError: result = value complete = False return CodecResult(value=result, complete=complete) def _SafeDecodeBytes(unused_field, value): """Decode the urlsafe base64 value into bytes.""" try: result = base64.urlsafe_b64decode(str(value)) complete = True except TypeError: result = value complete = False return CodecResult(value=result, complete=complete) def _ProcessUnknownEnums(message, encoded_message): """Add unknown enum values from encoded_message as unknown fields. ProtoRPC diverges from the usual protocol buffer behavior here and doesn't allow unknown fields. Throwing on unknown fields makes it impossible to let servers add new enum values and stay compatible with older clients, which isn't reasonable for us. We simply store unrecognized enum values as unknown fields, and all is well. Args: message: Proto message we've decoded thus far. encoded_message: JSON string we're decoding. Returns: message, with any unknown enums stored as unrecognized fields. """ if not encoded_message: return message decoded_message = json.loads(six.ensure_str(encoded_message)) for field in message.all_fields(): if (isinstance(field, messages.EnumField) and field.name in decoded_message): value = message.get_assigned_value(field.name) if ((field.repeated and len(value) != len(decoded_message[field.name])) or value is None): message.set_unrecognized_field( field.name, decoded_message[field.name], messages.Variant.ENUM) return message def _ProcessUnknownMessages(message, encoded_message): """Store any remaining unknown fields as strings. ProtoRPC currently ignores unknown values for which no type can be determined (and logs a "No variant found" message). For the purposes of reserializing, this is quite harmful (since it throws away information). Here we simply add those as unknown fields of type string (so that they can easily be reserialized). Args: message: Proto message we've decoded thus far. encoded_message: JSON string we're decoding. Returns: message, with any remaining unrecognized fields saved. """ if not encoded_message: return message decoded_message = json.loads(six.ensure_str(encoded_message)) message_fields = [x.name for x in message.all_fields()] + list( message.all_unrecognized_fields()) missing_fields = [x for x in decoded_message.keys() if x not in message_fields] for field_name in missing_fields: message.set_unrecognized_field(field_name, decoded_message[field_name], messages.Variant.STRING) return message RegisterFieldTypeCodec(_SafeEncodeBytes, _SafeDecodeBytes)(messages.BytesField) # Note that these could share a dictionary, since they're keyed by # distinct types, but it's not really worth it. _JSON_ENUM_MAPPINGS = {} _JSON_FIELD_MAPPINGS = {} def AddCustomJsonEnumMapping(enum_type, python_name, json_name, package=None): # pylint: disable=unused-argument """Add a custom wire encoding for a given enum value. This is primarily used in generated code, to handle enum values which happen to be Python keywords. Args: enum_type: (messages.Enum) An enum type python_name: (basestring) Python name for this value. json_name: (basestring) JSON name to be used on the wire. package: (NoneType, optional) No effect, exists for legacy compatibility. """ if not issubclass(enum_type, messages.Enum): raise exceptions.TypecheckError( 'Cannot set JSON enum mapping for non-enum "%s"' % enum_type) if python_name not in enum_type.names(): raise exceptions.InvalidDataError( 'Enum value %s not a value for type %s' % (python_name, enum_type)) field_mappings = _JSON_ENUM_MAPPINGS.setdefault(enum_type, {}) _CheckForExistingMappings('enum', enum_type, python_name, json_name) field_mappings[python_name] = json_name def AddCustomJsonFieldMapping(message_type, python_name, json_name, package=None): # pylint: disable=unused-argument """Add a custom wire encoding for a given message field. This is primarily used in generated code, to handle enum values which happen to be Python keywords. Args: message_type: (messages.Message) A message type python_name: (basestring) Python name for this value. json_name: (basestring) JSON name to be used on the wire. package: (NoneType, optional) No effect, exists for legacy compatibility. """ if not issubclass(message_type, messages.Message): raise exceptions.TypecheckError( 'Cannot set JSON field mapping for ' 'non-message "%s"' % message_type) try: _ = message_type.field_by_name(python_name) except KeyError: raise exceptions.InvalidDataError( 'Field %s not recognized for type %s' % ( python_name, message_type)) field_mappings = _JSON_FIELD_MAPPINGS.setdefault(message_type, {}) _CheckForExistingMappings('field', message_type, python_name, json_name) field_mappings[python_name] = json_name def GetCustomJsonEnumMapping(enum_type, python_name=None, json_name=None): """Return the appropriate remapping for the given enum, or None.""" return _FetchRemapping(enum_type, 'enum', python_name=python_name, json_name=json_name, mappings=_JSON_ENUM_MAPPINGS) def GetCustomJsonFieldMapping(message_type, python_name=None, json_name=None): """Return the appropriate remapping for the given field, or None.""" return _FetchRemapping(message_type, 'field', python_name=python_name, json_name=json_name, mappings=_JSON_FIELD_MAPPINGS) def _FetchRemapping(type_name, mapping_type, python_name=None, json_name=None, mappings=None): """Common code for fetching a key or value from a remapping dict.""" if python_name and json_name: raise exceptions.InvalidDataError( 'Cannot specify both python_name and json_name ' 'for %s remapping' % mapping_type) if not (python_name or json_name): raise exceptions.InvalidDataError( 'Must specify either python_name or json_name for %s remapping' % ( mapping_type,)) field_remappings = mappings.get(type_name, {}) if field_remappings: if python_name: return field_remappings.get(python_name) elif json_name: if json_name in list(field_remappings.values()): return [k for k in field_remappings if field_remappings[k] == json_name][0] return None def _CheckForExistingMappings(mapping_type, message_type, python_name, json_name): """Validate that no mappings exist for the given values.""" if mapping_type == 'field': getter = GetCustomJsonFieldMapping elif mapping_type == 'enum': getter = GetCustomJsonEnumMapping remapping = getter(message_type, python_name=python_name) if remapping is not None and remapping != json_name: raise exceptions.InvalidDataError( 'Cannot add mapping for %s "%s", already mapped to "%s"' % ( mapping_type, python_name, remapping)) remapping = getter(message_type, json_name=json_name) if remapping is not None and remapping != python_name: raise exceptions.InvalidDataError( 'Cannot add mapping for %s "%s", already mapped to "%s"' % ( mapping_type, json_name, remapping)) def _EncodeCustomFieldNames(message, encoded_value): field_remappings = list(_JSON_FIELD_MAPPINGS.get(type(message), {}) .items()) if field_remappings: decoded_value = json.loads(encoded_value) for python_name, json_name in field_remappings: if python_name in encoded_value: decoded_value[json_name] = decoded_value.pop(python_name) encoded_value = json.dumps(decoded_value) return encoded_value def _DecodeCustomFieldNames(message_type, encoded_message): field_remappings = _JSON_FIELD_MAPPINGS.get(message_type, {}) if field_remappings: decoded_message = json.loads(encoded_message) for python_name, json_name in list(field_remappings.items()): if json_name in decoded_message: decoded_message[python_name] = decoded_message.pop(json_name) encoded_message = json.dumps(decoded_message) return encoded_message def _AsMessageList(msg): """Convert the provided list-as-JsonValue to a list.""" # This really needs to live in extra_types, but extra_types needs # to import this file to be able to register codecs. # TODO(craigcitro): Split out a codecs module and fix this ugly # import. from apitools.base.py import extra_types def _IsRepeatedJsonValue(msg): """Return True if msg is a repeated value as a JsonValue.""" if isinstance(msg, extra_types.JsonArray): return True if isinstance(msg, extra_types.JsonValue) and msg.array_value: return True return False if not _IsRepeatedJsonValue(msg): raise ValueError('invalid argument to _AsMessageList') if isinstance(msg, extra_types.JsonValue): msg = msg.array_value if isinstance(msg, extra_types.JsonArray): msg = msg.entries return msg def _IsMap(message, field): """Returns whether the "field" is actually a map-type.""" value = message.get_assigned_value(field.name) if not isinstance(value, messages.Message): return False try: additional_properties = value.field_by_name('additionalProperties') except KeyError: return False else: return additional_properties.repeated def _MapItems(message, field): """Yields the (key, value) pair of the map values.""" assert _IsMap(message, field) map_message = message.get_assigned_value(field.name) additional_properties = map_message.get_assigned_value( 'additionalProperties') for kv_pair in additional_properties: yield kv_pair.key, kv_pair.value def UnrecognizedFieldIter(message, _edges=()): # pylint: disable=invalid-name """Yields the locations of unrecognized fields within "message". If a sub-message is found to have unrecognized fields, that sub-message will not be searched any further. We prune the search of the sub-message because we assume it is malformed and further checks will not yield productive errors. Args: message: The Message instance to search. _edges: Internal arg for passing state. Yields: (edges_to_message, field_names): edges_to_message: List[ProtoEdge], The edges (relative to "message") describing the path to the sub-message where the unrecognized fields were found. field_names: List[Str], The names of the field(s) that were unrecognized in the sub-message. """ if not isinstance(message, messages.Message): # This is a primitive leaf, no errors found down this path. return field_names = message.all_unrecognized_fields() if field_names: # This message is malformed. Stop recursing and report it. yield _edges, field_names return # Recurse through all fields in the current message. for field in message.all_fields(): value = message.get_assigned_value(field.name) if field.repeated: for i, item in enumerate(value): repeated_edge = ProtoEdge(EdgeType.REPEATED, field.name, i) iter_ = UnrecognizedFieldIter(item, _edges + (repeated_edge,)) for (e, y) in iter_: yield e, y elif _IsMap(message, field): for key, item in _MapItems(message, field): map_edge = ProtoEdge(EdgeType.MAP, field.name, key) iter_ = UnrecognizedFieldIter(item, _edges + (map_edge,)) for (e, y) in iter_: yield e, y else: scalar_edge = ProtoEdge(EdgeType.SCALAR, field.name, None) iter_ = UnrecognizedFieldIter(value, _edges + (scalar_edge,)) for (e, y) in iter_: yield e, y	30 KB	2025-03-11 18:02
📄 encoding_test.py # -- coding: utf-8 -- # 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. import base64 import datetime import json import sys import unittest from apitools.base.protorpclite import message_types from apitools.base.protorpclite import messages from apitools.base.protorpclite import util from apitools.base.py import encoding from apitools.base.py import exceptions from apitools.base.py import extra_types class SimpleMessage(messages.Message): field = messages.StringField(1) repfield = messages.StringField(2, repeated=True) class BytesMessage(messages.Message): field = messages.BytesField(1) repfield = messages.BytesField(2, repeated=True) class TimeMessage(messages.Message): timefield = message_types.DateTimeField(3) @encoding.MapUnrecognizedFields('additionalProperties') class AdditionalPropertiesMessage(messages.Message): class AdditionalProperty(messages.Message): key = messages.StringField(1) value = messages.StringField(2) additionalProperties = messages.MessageField( 'AdditionalProperty', 1, repeated=True) @encoding.MapUnrecognizedFields('additionalProperties') class AdditionalIntPropertiesMessage(messages.Message): class AdditionalProperty(messages.Message): key = messages.StringField(1) value = messages.IntegerField(2) additionalProperties = messages.MessageField( 'AdditionalProperty', 1, repeated=True) @encoding.MapUnrecognizedFields('additionalProperties') class UnrecognizedEnumMessage(messages.Message): class ThisEnum(messages.Enum): VALUE_ONE = 1 VALUE_TWO = 2 class AdditionalProperty(messages.Message): key = messages.StringField(1) value = messages.EnumField('UnrecognizedEnumMessage.ThisEnum', 2) additionalProperties = messages.MessageField( AdditionalProperty, 1, repeated=True) class CompoundPropertyType(messages.Message): index = messages.IntegerField(1) name = messages.StringField(2) class MessageWithEnum(messages.Message): class ThisEnum(messages.Enum): VALUE_ONE = 1 VALUE_TWO = 2 field_one = messages.EnumField(ThisEnum, 1) field_two = messages.EnumField(ThisEnum, 2, default=ThisEnum.VALUE_TWO) field_three = messages.EnumField(ThisEnum, 3, repeated=True) ignored_field = messages.EnumField(ThisEnum, 4) @encoding.MapUnrecognizedFields('additionalProperties') class AdditionalMessagePropertiesMessage(messages.Message): class AdditionalProperty(messages.Message): key = messages.StringField(1) value = messages.MessageField(CompoundPropertyType, 2) additionalProperties = messages.MessageField( 'AdditionalProperty', 1, repeated=True) @encoding.MapUnrecognizedFields('additionalProperties') class MapToMessageWithEnum(messages.Message): class AdditionalProperty(messages.Message): key = messages.StringField(1) value = messages.MessageField(MessageWithEnum, 2) additionalProperties = messages.MessageField( 'AdditionalProperty', 1, repeated=True) @encoding.MapUnrecognizedFields('additionalProperties') class NestedAdditionalPropertiesWithEnumMessage(messages.Message): class AdditionalProperty(messages.Message): key = messages.StringField(1) value = messages.MessageField( MapToMessageWithEnum, 2) additionalProperties = messages.MessageField( 'AdditionalProperty', 1, repeated=True) @encoding.MapUnrecognizedFields('additionalProperties') class AdditionalPropertiesWithEnumMessage(messages.Message): class AdditionalProperty(messages.Message): key = messages.StringField(1) value = messages.MessageField(MessageWithEnum, 2) additionalProperties = messages.MessageField( 'AdditionalProperty', 1, repeated=True) class NestedMapMessage(messages.Message): msg_field = messages.MessageField(AdditionalPropertiesWithEnumMessage, 1) class RepeatedNestedMapMessage(messages.Message): map_field = messages.MessageField(NestedMapMessage, 1, repeated=True) class NestedWithEnumMessage(messages.Message): class ThisEnum(messages.Enum): VALUE_ONE = 1 VALUE_TWO = 2 msg_field = messages.MessageField(MessageWithEnum, 1) enum_field = messages.EnumField(ThisEnum, 2) class RepeatedNestedMessage(messages.Message): msg_field = messages.MessageField(SimpleMessage, 1, repeated=True) @encoding.MapUnrecognizedFields('additionalProperties') class MapToBytesValue(messages.Message): class AdditionalProperty(messages.Message): key = messages.StringField(1) value = messages.BytesField(2) additionalProperties = messages.MessageField('AdditionalProperty', 1, repeated=True) @encoding.MapUnrecognizedFields('additionalProperties') class MapToDateTimeValue(messages.Message): class AdditionalProperty(messages.Message): key = messages.StringField(1) value = message_types.DateTimeField(2) additionalProperties = messages.MessageField('AdditionalProperty', 1, repeated=True) class HasNestedMessage(messages.Message): nested = messages.MessageField(AdditionalPropertiesMessage, 1) nested_list = messages.StringField(2, repeated=True) class ExtraNestedMessage(messages.Message): nested = messages.MessageField(HasNestedMessage, 1) class MessageWithRemappings(messages.Message): class SomeEnum(messages.Enum): enum_value = 1 second_value = 2 enum_field = messages.EnumField(SomeEnum, 1) double_encoding = messages.EnumField(SomeEnum, 2) another_field = messages.StringField(3) repeated_enum = messages.EnumField(SomeEnum, 4, repeated=True) repeated_field = messages.StringField(5, repeated=True) class MessageWithPackageAndRemappings(messages.Message): class SomeEnum(messages.Enum): enum_value = 1 second_value = 2 enum_field = messages.EnumField(SomeEnum, 1) another_field = messages.StringField(2) @encoding.MapUnrecognizedFields('additionalProperties') class RepeatedJsonValueMessage(messages.Message): class AdditionalProperty(messages.Message): key = messages.StringField(1) value = messages.MessageField(extra_types.JsonValue, 2, repeated=True) additionalProperties = messages.MessageField('AdditionalProperty', 1, repeated=True) encoding.AddCustomJsonEnumMapping(MessageWithRemappings.SomeEnum, 'enum_value', 'wire_name') encoding.AddCustomJsonFieldMapping(MessageWithRemappings, 'double_encoding', 'doubleEncoding') encoding.AddCustomJsonFieldMapping(MessageWithRemappings, 'another_field', 'anotherField') encoding.AddCustomJsonFieldMapping(MessageWithRemappings, 'repeated_field', 'repeatedField') class EncodingTest(unittest.TestCase): def testCopyProtoMessage(self): msg = SimpleMessage(field='abc') new_msg = encoding.CopyProtoMessage(msg) self.assertEqual(msg.field, new_msg.field) msg.field = 'def' self.assertNotEqual(msg.field, new_msg.field) def testCopyProtoMessageInvalidEnum(self): json_msg = '{"field_one": "BAD_VALUE"}' orig_msg = encoding.JsonToMessage(MessageWithEnum, json_msg) new_msg = encoding.CopyProtoMessage(orig_msg) for msg in (orig_msg, new_msg): self.assertEqual(msg.all_unrecognized_fields(), ['field_one']) self.assertEqual( msg.get_unrecognized_field_info('field_one', value_default=None), ('BAD_VALUE', messages.Variant.ENUM)) def testCopyProtoMessageInvalidRepeatedEnum(self): json_msg = '{"field_three": ["VALUE_ONE", "BAD_VALUE"]}' orig_msg = encoding.JsonToMessage(MessageWithEnum, json_msg) new_msg = encoding.CopyProtoMessage(orig_msg) for msg in (orig_msg, new_msg): self.assertEqual(msg.all_unrecognized_fields(), ['field_three']) self.assertEqual( msg.get_unrecognized_field_info('field_three', value_default=None), (['VALUE_ONE', 'BAD_VALUE'], messages.Variant.ENUM)) def testCopyProtoMessageAdditionalProperties(self): msg = AdditionalPropertiesMessage(additionalProperties=[ AdditionalPropertiesMessage.AdditionalProperty( key='key', value='value')]) new_msg = encoding.CopyProtoMessage(msg) self.assertEqual(len(new_msg.additionalProperties), 1) self.assertEqual(new_msg.additionalProperties[0].key, 'key') self.assertEqual(new_msg.additionalProperties[0].value, 'value') def testCopyProtoMessageMappingInvalidEnum(self): json_msg = '{"key_one": {"field_one": "BAD_VALUE"}}' orig_msg = encoding.JsonToMessage(MapToMessageWithEnum, json_msg) new_msg = encoding.CopyProtoMessage(orig_msg) for msg in (orig_msg, new_msg): self.assertEqual( msg.additionalProperties[0].value.all_unrecognized_fields(), ['field_one']) self.assertEqual( msg.additionalProperties[0].value.get_unrecognized_field_info( 'field_one', value_default=None), ('BAD_VALUE', messages.Variant.ENUM)) def testCopyProtoMessageMappingInvalidRepeatedEnum(self): json_msg = '{"key_one": {"field_three": ["VALUE_ONE", "BAD_VALUE"]}}' orig_msg = encoding.JsonToMessage(MapToMessageWithEnum, json_msg) new_msg = encoding.CopyProtoMessage(orig_msg) for msg in (orig_msg, new_msg): self.assertEqual( msg.additionalProperties[0].value.all_unrecognized_fields(), ['field_three']) self.assertEqual( msg.additionalProperties[0].value.get_unrecognized_field_info( 'field_three', value_default=None), (['VALUE_ONE', 'BAD_VALUE'], messages.Variant.ENUM)) def testBytesEncoding(self): b64_str = 'AAc+' b64_msg = '{"field": "%s"}' % b64_str urlsafe_b64_str = 'AAc-' urlsafe_b64_msg = '{"field": "%s"}' % urlsafe_b64_str data = base64.b64decode(b64_str) msg = BytesMessage(field=data) self.assertEqual( msg, encoding.JsonToMessage(BytesMessage, urlsafe_b64_msg)) self.assertEqual(msg, encoding.JsonToMessage(BytesMessage, b64_msg)) self.assertEqual(urlsafe_b64_msg, encoding.MessageToJson(msg)) enc_rep_msg = '{"repfield": ["%(b)s", "%(b)s"]}' % { 'b': urlsafe_b64_str} rep_msg = BytesMessage(repfield=[data, data]) self.assertEqual( rep_msg, encoding.JsonToMessage(BytesMessage, enc_rep_msg)) self.assertEqual(enc_rep_msg, encoding.MessageToJson(rep_msg)) def testBase64RoundtripForMapFields(self): raw_data = b'\xFF\x0F\x80' encoded_data = '/w+A' # Has url-unsafe base64 characters safe_encoded_data = base64.urlsafe_b64encode(raw_data).decode("utf-8") self.assertEqual(raw_data, base64.b64decode(encoded_data)) # Use unsafe encoding, make sure we can load it. json_data = '{"1st": "%s"}' % encoded_data msg = encoding.JsonToMessage(MapToBytesValue, json_data) self.assertEqual(raw_data, msg.additionalProperties[0].value) # Now back to json and again to message from_msg_json_data = encoding.MessageToJson(msg) # Make sure now it is safe url encoded self.assertEqual(safe_encoded_data, json.loads(from_msg_json_data)['1st']) # Make sure we can also load url safe encoded bytes. redone_msg = encoding.JsonToMessage(MapToBytesValue, from_msg_json_data) # Still matches self.assertEqual(raw_data, redone_msg.additionalProperties[0].value) def testBytesEncodingInAMap(self): # Leading bit is 1 should not be interpreted as unicode. data1 = b'\xF0\x11\x0F' data2 = b'\xFF\xFF\xFF' msg = MapToBytesValue( additionalProperties=[ MapToBytesValue.AdditionalProperty(key='1st', value=data1), MapToBytesValue.AdditionalProperty(key='2nd', value=data2) ]) self.assertEqual( '{"1st": "%s", "2nd": "%s"}' % ( base64.b64encode(data1, b'-_').decode("utf-8"), base64.b64encode(data2, b'-_').decode("utf-8")), encoding.MessageToJson(msg)) def testDateTimeEncodingInAMap(self): msg = MapToDateTimeValue( additionalProperties=[ MapToDateTimeValue.AdditionalProperty( key='1st', value=datetime.datetime( 2014, 7, 2, 23, 33, 25, 541000, tzinfo=util.TimeZoneOffset(datetime.timedelta(0)))), MapToDateTimeValue.AdditionalProperty( key='2nd', value=datetime.datetime( 2015, 7, 2, 23, 33, 25, 541000, tzinfo=util.TimeZoneOffset(datetime.timedelta(0)))) ]) self.assertEqual( '{"1st": "2014-07-02T23:33:25.541000+00:00",' ' "2nd": "2015-07-02T23:33:25.541000+00:00"}', encoding.MessageToJson(msg)) def testInvalidEnumEncodingInAMap(self): json_msg = '{"key_one": {"field_one": "BAD_VALUE"}}' msg = encoding.JsonToMessage(MapToMessageWithEnum, json_msg) new_msg = encoding.MessageToJson(msg) self.assertEqual('{"key_one": {"field_one": "BAD_VALUE"}}', new_msg) def testInvalidRepeatedEnumEncodingInAMap(self): json_msg = '{"key_one": {"field_three": ["VALUE_ONE", "BAD_VALUE"]}}' msg = encoding.JsonToMessage(MapToMessageWithEnum, json_msg) new_msg = encoding.MessageToJson(msg) self.assertEqual( '{"key_one": {"field_three": ["VALUE_ONE", "BAD_VALUE"]}}', new_msg) def testIncludeFields(self): msg = SimpleMessage() self.assertEqual('{}', encoding.MessageToJson(msg)) self.assertEqual( '{"field": null}', encoding.MessageToJson(msg, include_fields=['field'])) self.assertEqual( '{"repfield": []}', encoding.MessageToJson(msg, include_fields=['repfield'])) def testNestedIncludeFields(self): msg = HasNestedMessage( nested=AdditionalPropertiesMessage( additionalProperties=[])) self.assertEqual( '{"nested": null}', encoding.MessageToJson(msg, include_fields=['nested'])) self.assertEqual( '{"nested": {"additionalProperties": []}}', encoding.MessageToJson( msg, include_fields=['nested.additionalProperties'])) msg = ExtraNestedMessage(nested=msg) self.assertEqual( '{"nested": {"nested": null}}', encoding.MessageToJson(msg, include_fields=['nested.nested'])) # When clearing 'nested.nested_list', its sibling ('nested.nested') # should remain unaffected. self.assertIn( encoding.MessageToJson(msg, include_fields=['nested.nested_list']), ['{"nested": {"nested": {}, "nested_list": []}}', '{"nested": {"nested_list": [], "nested": {}}}']) self.assertEqual( '{"nested": {"nested": {"additionalProperties": []}}}', encoding.MessageToJson( msg, include_fields=['nested.nested.additionalProperties'])) def testAdditionalPropertyMapping(self): msg = AdditionalPropertiesMessage() msg.additionalProperties = [ AdditionalPropertiesMessage.AdditionalProperty( key='key_one', value='value_one'), AdditionalPropertiesMessage.AdditionalProperty( key=u'key_twð', value='value_two'), ] encoded_msg = encoding.MessageToJson(msg) self.assertEqual( {'key_one': 'value_one', u'key_twð': 'value_two'}, json.loads(encoded_msg)) new_msg = encoding.JsonToMessage(type(msg), encoded_msg) self.assertEqual( set(('key_one', u'key_twð')), set([x.key for x in new_msg.additionalProperties])) self.assertIsNot(msg, new_msg) new_msg.additionalProperties.pop() self.assertEqual(1, len(new_msg.additionalProperties)) self.assertEqual(2, len(msg.additionalProperties)) def testNumericPropertyName(self): json_msg = '{"nested": {"123": "def"}}' msg = encoding.JsonToMessage(HasNestedMessage, json_msg) self.assertEqual(1, len(msg.nested.additionalProperties)) def testNumericPropertyValue(self): json_msg = '{"key_one": "123"}' msg = encoding.JsonToMessage(AdditionalIntPropertiesMessage, json_msg) self.assertEqual( AdditionalIntPropertiesMessage( additionalProperties=[ AdditionalIntPropertiesMessage.AdditionalProperty( key='key_one', value=123)]), msg) def testAdditionalMessageProperties(self): json_msg = '{"input": {"index": 0, "name": "output"}}' result = encoding.JsonToMessage( AdditionalMessagePropertiesMessage, json_msg) self.assertEqual(1, len(result.additionalProperties)) self.assertEqual(0, result.additionalProperties[0].value.index) def testUnrecognizedEnum(self): json_msg = '{"input": "VALUE_ONE"}' result = encoding.JsonToMessage( UnrecognizedEnumMessage, json_msg) self.assertEqual(1, len(result.additionalProperties)) self.assertEqual(UnrecognizedEnumMessage.ThisEnum.VALUE_ONE, result.additionalProperties[0].value) def testNestedFieldMapping(self): nested_msg = AdditionalPropertiesMessage() nested_msg.additionalProperties = [ AdditionalPropertiesMessage.AdditionalProperty( key='key_one', value='value_one'), AdditionalPropertiesMessage.AdditionalProperty( key='key_two', value='value_two'), ] msg = HasNestedMessage(nested=nested_msg) encoded_msg = encoding.MessageToJson(msg) self.assertEqual( {'nested': {'key_one': 'value_one', 'key_two': 'value_two'}}, json.loads(encoded_msg)) new_msg = encoding.JsonToMessage(type(msg), encoded_msg) self.assertEqual( set(('key_one', 'key_two')), set([x.key for x in new_msg.nested.additionalProperties])) new_msg.nested.additionalProperties.pop() self.assertEqual(1, len(new_msg.nested.additionalProperties)) self.assertEqual(2, len(msg.nested.additionalProperties)) def testValidEnums(self): message_json = '{"field_one": "VALUE_ONE"}' message = encoding.JsonToMessage(MessageWithEnum, message_json) self.assertEqual(MessageWithEnum.ThisEnum.VALUE_ONE, message.field_one) self.assertEqual(MessageWithEnum.ThisEnum.VALUE_TWO, message.field_two) self.assertEqual(json.loads(message_json), json.loads(encoding.MessageToJson(message))) def testIgnoredEnums(self): json_with_typo = '{"field_one": "VALUE_OEN"}' message = encoding.JsonToMessage(MessageWithEnum, json_with_typo) self.assertEqual(None, message.field_one) self.assertEqual(('VALUE_OEN', messages.Variant.ENUM), message.get_unrecognized_field_info('field_one')) self.assertEqual(json.loads(json_with_typo), json.loads(encoding.MessageToJson(message))) empty_json = '' message = encoding.JsonToMessage(MessageWithEnum, empty_json) self.assertEqual(None, message.field_one) def testIgnoredEnumsWithDefaults(self): json_with_typo = '{"field_two": "VALUE_OEN"}' message = encoding.JsonToMessage(MessageWithEnum, json_with_typo) self.assertEqual(MessageWithEnum.ThisEnum.VALUE_TWO, message.field_two) self.assertEqual(json.loads(json_with_typo), json.loads(encoding.MessageToJson(message))) def testUnknownNestedRoundtrip(self): json_message = '{"field": "abc", "submessage": {"a": 1, "b": "foo"}}' message = encoding.JsonToMessage(SimpleMessage, json_message) self.assertEqual(json.loads(json_message), json.loads(encoding.MessageToJson(message))) def testUnknownEnumNestedRoundtrip(self): json_with_typo = ('{"outer_key": {"key_one": {"field_one": ' '"VALUE_OEN", "field_two": "VALUE_OEN", ' '"field_three": ["VALUE_ONE", "BAD_VALUE"]}}}') msg = encoding.JsonToMessage(NestedAdditionalPropertiesWithEnumMessage, json_with_typo) self.assertEqual(json.loads(json_with_typo), json.loads(encoding.MessageToJson(msg))) def testJsonDatetime(self): msg = TimeMessage(timefield=datetime.datetime( 2014, 7, 2, 23, 33, 25, 541000, tzinfo=util.TimeZoneOffset(datetime.timedelta(0)))) self.assertEqual( '{"timefield": "2014-07-02T23:33:25.541000+00:00"}', encoding.MessageToJson(msg)) def testEnumRemapping(self): msg = MessageWithRemappings( enum_field=MessageWithRemappings.SomeEnum.enum_value) json_message = encoding.MessageToJson(msg) self.assertEqual('{"enum_field": "wire_name"}', json_message) self.assertEqual( msg, encoding.JsonToMessage(MessageWithRemappings, json_message)) def testRepeatedEnumRemapping(self): msg = MessageWithRemappings( repeated_enum=[ MessageWithRemappings.SomeEnum.enum_value, MessageWithRemappings.SomeEnum.second_value, ]) json_message = encoding.MessageToJson(msg) self.assertEqual('{"repeated_enum": ["wire_name", "second_value"]}', json_message) self.assertEqual( msg, encoding.JsonToMessage(MessageWithRemappings, json_message)) def testFieldRemapping(self): msg = MessageWithRemappings(another_field='abc') json_message = encoding.MessageToJson(msg) self.assertEqual('{"anotherField": "abc"}', json_message) self.assertEqual( msg, encoding.JsonToMessage(MessageWithRemappings, json_message)) def testFieldRemappingWithPackage(self): this_module = sys.modules[__name__] package_name = 'my_package' try: setattr(this_module, 'package', package_name) encoding.AddCustomJsonFieldMapping( MessageWithPackageAndRemappings, 'another_field', 'wire_field_name', package=package_name) msg = MessageWithPackageAndRemappings(another_field='my value') json_message = encoding.MessageToJson(msg) self.assertEqual('{"wire_field_name": "my value"}', json_message) self.assertEqual( msg, encoding.JsonToMessage(MessageWithPackageAndRemappings, json_message)) finally: delattr(this_module, 'package') def testEnumRemappingWithPackage(self): this_module = sys.modules[__name__] package_name = 'my_package' try: setattr(this_module, 'package', package_name) encoding.AddCustomJsonEnumMapping( MessageWithPackageAndRemappings.SomeEnum, 'enum_value', 'other_wire_name', package=package_name) msg = MessageWithPackageAndRemappings( enum_field=MessageWithPackageAndRemappings.SomeEnum.enum_value) json_message = encoding.MessageToJson(msg) self.assertEqual('{"enum_field": "other_wire_name"}', json_message) self.assertEqual( msg, encoding.JsonToMessage(MessageWithPackageAndRemappings, json_message)) finally: delattr(this_module, 'package') def testRepeatedFieldRemapping(self): msg = MessageWithRemappings(repeated_field=['abc', 'def']) json_message = encoding.MessageToJson(msg) self.assertEqual('{"repeatedField": ["abc", "def"]}', json_message) self.assertEqual( msg, encoding.JsonToMessage(MessageWithRemappings, json_message)) def testMultipleRemapping(self): msg = MessageWithRemappings( double_encoding=MessageWithRemappings.SomeEnum.enum_value) json_message = encoding.MessageToJson(msg) self.assertEqual('{"doubleEncoding": "wire_name"}', json_message) self.assertEqual( msg, encoding.JsonToMessage(MessageWithRemappings, json_message)) def testRepeatedRemapping(self): # Should allow remapping if the mapping remains the same. encoding.AddCustomJsonEnumMapping(MessageWithRemappings.SomeEnum, 'enum_value', 'wire_name') encoding.AddCustomJsonFieldMapping(MessageWithRemappings, 'double_encoding', 'doubleEncoding') encoding.AddCustomJsonFieldMapping(MessageWithRemappings, 'another_field', 'anotherField') encoding.AddCustomJsonFieldMapping(MessageWithRemappings, 'repeated_field', 'repeatedField') # Should raise errors if the remapping changes the mapping. self.assertRaises( exceptions.InvalidDataError, encoding.AddCustomJsonFieldMapping, MessageWithRemappings, 'double_encoding', 'something_else') self.assertRaises( exceptions.InvalidDataError, encoding.AddCustomJsonFieldMapping, MessageWithRemappings, 'enum_field', 'anotherField') self.assertRaises( exceptions.InvalidDataError, encoding.AddCustomJsonEnumMapping, MessageWithRemappings.SomeEnum, 'enum_value', 'another_name') self.assertRaises( exceptions.InvalidDataError, encoding.AddCustomJsonEnumMapping, MessageWithRemappings.SomeEnum, 'second_value', 'wire_name') def testMessageToRepr(self): # Using the same string returned by MessageToRepr, with the # module names fixed. # pylint: disable=bad-whitespace msg = SimpleMessage(field='field', repfield=['field', 'field', ],) # pylint: enable=bad-whitespace self.assertEqual( encoding.MessageToRepr(msg), r"%s.SimpleMessage(field='field',repfield=['field','field',],)" % ( __name__,)) self.assertEqual( encoding.MessageToRepr(msg, no_modules=True), r"SimpleMessage(field='field',repfield=['field','field',],)") def testMessageToReprWithTime(self): msg = TimeMessage(timefield=datetime.datetime( 2014, 7, 2, 23, 33, 25, 541000, tzinfo=util.TimeZoneOffset(datetime.timedelta(0)))) self.assertEqual( encoding.MessageToRepr(msg, multiline=True), ('%s.TimeMessage(\n ' 'timefield=datetime.datetime(2014, 7, 2, 23, 33, 25, 541000, ' 'tzinfo=apitools.base.protorpclite.util.TimeZoneOffset(' 'datetime.timedelta(0))),\n)') % __name__) self.assertEqual( encoding.MessageToRepr(msg, multiline=True, no_modules=True), 'TimeMessage(\n ' 'timefield=datetime.datetime(2014, 7, 2, 23, 33, 25, 541000, ' 'tzinfo=TimeZoneOffset(datetime.timedelta(0))),\n)') def testRepeatedJsonValuesAsRepeatedProperty(self): encoded_msg = '{"a": [{"one": 1}]}' msg = encoding.JsonToMessage(RepeatedJsonValueMessage, encoded_msg) self.assertEqual(encoded_msg, encoding.MessageToJson(msg)) def testDictToAdditionalPropertyMessage(self): dict_ = {'key': 'value'} encoded_msg = encoding.DictToAdditionalPropertyMessage( dict_, AdditionalPropertiesMessage) expected_msg = AdditionalPropertiesMessage() expected_msg.additionalProperties = [ AdditionalPropertiesMessage.AdditionalProperty( key='key', value='value') ] self.assertEqual(encoded_msg, expected_msg) def testDictToAdditionalPropertyMessageSorted(self): tuples = [('key{0:02}'.format(i), 'value') for i in range(100)] dict_ = dict(tuples) encoded_msg = encoding.DictToAdditionalPropertyMessage( dict_, AdditionalPropertiesMessage, sort_items=True) expected_msg = AdditionalPropertiesMessage() expected_msg.additionalProperties = [ AdditionalPropertiesMessage.AdditionalProperty( key=key, value=value) for key, value in tuples ] self.assertEqual(encoded_msg, expected_msg) def testDictToAdditionalPropertyMessageNumeric(self): dict_ = {'key': 1} encoded_msg = encoding.DictToAdditionalPropertyMessage( dict_, AdditionalIntPropertiesMessage) expected_msg = AdditionalIntPropertiesMessage() expected_msg.additionalProperties = [ AdditionalIntPropertiesMessage.AdditionalProperty( key='key', value=1) ] self.assertEqual(encoded_msg, expected_msg) def testUnrecognizedFieldIter(self): m = encoding.DictToMessage({ 'nested': { 'nested': {'a': 'b'}, 'nested_list': ['foo'], 'extra_field': 'foo', } }, ExtraNestedMessage) results = list(encoding.UnrecognizedFieldIter(m)) self.assertEqual(1, len(results)) edges, fields = results[0] expected_edge = encoding.ProtoEdge( encoding.EdgeType.SCALAR, 'nested', None) self.assertEqual((expected_edge,), edges) self.assertEqual(['extra_field'], fields) def testUnrecognizedFieldIterRepeated(self): m = encoding.DictToMessage({ 'msg_field': [ {'field': 'foo'}, {'not_a_field': 'bar'} ] }, RepeatedNestedMessage) results = list(encoding.UnrecognizedFieldIter(m)) self.assertEqual(1, len(results)) edges, fields = results[0] expected_edge = encoding.ProtoEdge( encoding.EdgeType.REPEATED, 'msg_field', 1) self.assertEqual((expected_edge,), edges) self.assertEqual(['not_a_field'], fields) def testUnrecognizedFieldIterNestedMap(self): m = encoding.DictToMessage({ 'map_field': [{ 'msg_field': { 'foo': {'field_one': 1}, 'bar': {'not_a_field': 1}, } }] }, RepeatedNestedMapMessage) results = list(encoding.UnrecognizedFieldIter(m)) self.assertEqual(1, len(results)) edges, fields = results[0] expected_edges = ( encoding.ProtoEdge(encoding.EdgeType.REPEATED, 'map_field', 0), encoding.ProtoEdge(encoding.EdgeType.MAP, 'msg_field', 'bar'), ) self.assertEqual(expected_edges, edges) self.assertEqual(['not_a_field'], fields) def testUnrecognizedFieldIterAbortAfterFirstError(self): m = encoding.DictToMessage({ 'msg_field': {'field_one': 3}, 'enum_field': 3, }, NestedWithEnumMessage) self.assertEqual(1, len(list(encoding.UnrecognizedFieldIter(m))))	31.97 KB	2025-03-11 18:02
📄 exceptions.py #!/usr/bin/env python # # 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. """Exceptions for generated client libraries.""" class Error(Exception): """Base class for all exceptions.""" class TypecheckError(Error, TypeError): """An object of an incorrect type is provided.""" class NotFoundError(Error): """A specified resource could not be found.""" class UserError(Error): """Base class for errors related to user input.""" class InvalidDataError(Error): """Base class for any invalid data error.""" class CommunicationError(Error): """Any communication error talking to an API server.""" class HttpError(CommunicationError): """Error making a request. Soon to be HttpError.""" def __init__(self, response, content, url, method_config=None, request=None): error_message = HttpError._build_message(response, content, url) super(HttpError, self).__init__(error_message) self.response = response self.content = content self.url = url self.method_config = method_config self.request = request def __str__(self): return HttpError._build_message(self.response, self.content, self.url) @staticmethod def _build_message(response, content, url): if isinstance(content, bytes): content = content.decode('ascii', 'replace') return 'HttpError accessing <%s>: response: <%s>, content <%s>' % ( url, response, content) @property def status_code(self): # TODO(craigcitro): Turn this into something better than a # KeyError if there is no status. return int(self.response['status']) @classmethod def FromResponse(cls, http_response, kwargs): try: status_code = int(http_response.info.get('status')) error_cls = _HTTP_ERRORS.get(status_code, cls) except ValueError: error_cls = cls return error_cls(http_response.info, http_response.content, http_response.request_url, kwargs) class HttpBadRequestError(HttpError): """HTTP 400 Bad Request.""" class HttpUnauthorizedError(HttpError): """HTTP 401 Unauthorized.""" class HttpForbiddenError(HttpError): """HTTP 403 Forbidden.""" class HttpNotFoundError(HttpError): """HTTP 404 Not Found.""" class HttpConflictError(HttpError): """HTTP 409 Conflict.""" _HTTP_ERRORS = { 400: HttpBadRequestError, 401: HttpUnauthorizedError, 403: HttpForbiddenError, 404: HttpNotFoundError, 409: HttpConflictError, } class InvalidUserInputError(InvalidDataError): """User-provided input is invalid.""" class InvalidDataFromServerError(InvalidDataError, CommunicationError): """Data received from the server is malformed.""" class BatchError(Error): """Error generated while constructing a batch request.""" class ConfigurationError(Error): """Base class for configuration errors.""" class GeneratedClientError(Error): """The generated client configuration is invalid.""" class ConfigurationValueError(UserError): """Some part of the user-specified client configuration is invalid.""" class ResourceUnavailableError(Error): """User requested an unavailable resource.""" class CredentialsError(Error): """Errors related to invalid credentials.""" class TransferError(CommunicationError): """Errors related to transfers.""" class TransferRetryError(TransferError): """Retryable errors related to transfers.""" class TransferInvalidError(TransferError): """The given transfer is invalid.""" class RequestError(CommunicationError): """The request was not successful.""" class RetryAfterError(HttpError): """The response contained a retry-after header.""" def __init__(self, response, content, url, retry_after, kwargs): super(RetryAfterError, self).__init__(response, content, url, kwargs) self.retry_after = int(retry_after) @classmethod def FromResponse(cls, http_response, kwargs): return cls(http_response.info, http_response.content, http_response.request_url, http_response.retry_after, kwargs) class BadStatusCodeError(HttpError): """The request completed but returned a bad status code.""" class NotYetImplementedError(GeneratedClientError): """This functionality is not yet implemented.""" class StreamExhausted(Error): """Attempted to read more bytes from a stream than were available."""	4.98 KB	2025-03-11 18:02
📄 exceptions_test.py # Copyright 2017 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. import unittest from apitools.base.py import exceptions from apitools.base.py import http_wrapper def _MakeResponse(status_code): return http_wrapper.Response( info={'status': status_code}, content='{"field": "abc"}', request_url='http://www.google.com') class HttpErrorFromResponseTest(unittest.TestCase): """Tests for exceptions.HttpError.FromResponse.""" def testBadRequest(self): err = exceptions.HttpError.FromResponse(_MakeResponse(400)) self.assertIsInstance(err, exceptions.HttpError) self.assertIsInstance(err, exceptions.HttpBadRequestError) self.assertEqual(err.status_code, 400) def testUnauthorized(self): err = exceptions.HttpError.FromResponse(_MakeResponse(401)) self.assertIsInstance(err, exceptions.HttpError) self.assertIsInstance(err, exceptions.HttpUnauthorizedError) self.assertEqual(err.status_code, 401) def testForbidden(self): err = exceptions.HttpError.FromResponse(_MakeResponse(403)) self.assertIsInstance(err, exceptions.HttpError) self.assertIsInstance(err, exceptions.HttpForbiddenError) self.assertEqual(err.status_code, 403) def testExceptionMessageIncludesErrorDetails(self): err = exceptions.HttpError.FromResponse(_MakeResponse(403)) self.assertIn('403', repr(err)) self.assertIn('http://www.google.com', repr(err)) self.assertIn('{"field": "abc"}', repr(err)) def testNotFound(self): err = exceptions.HttpError.FromResponse(_MakeResponse(404)) self.assertIsInstance(err, exceptions.HttpError) self.assertIsInstance(err, exceptions.HttpNotFoundError) self.assertEqual(err.status_code, 404) def testConflict(self): err = exceptions.HttpError.FromResponse(_MakeResponse(409)) self.assertIsInstance(err, exceptions.HttpError) self.assertIsInstance(err, exceptions.HttpConflictError) self.assertEqual(err.status_code, 409) def testUnknownStatus(self): err = exceptions.HttpError.FromResponse(_MakeResponse(499)) self.assertIsInstance(err, exceptions.HttpError) self.assertEqual(err.status_code, 499) def testMalformedStatus(self): err = exceptions.HttpError.FromResponse(_MakeResponse('BAD')) self.assertIsInstance(err, exceptions.HttpError)	2.88 KB	2025-03-11 18:02
📄 extra_types.py #!/usr/bin/env python # # 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. """Extra types understood by apitools.""" import datetime import json import numbers import six from apitools.base.protorpclite import message_types from apitools.base.protorpclite import messages from apitools.base.protorpclite import protojson from apitools.base.py import encoding_helper as encoding from apitools.base.py import exceptions from apitools.base.py import util if six.PY3: from collections.abc import Iterable else: from collections import Iterable __all__ = [ 'DateField', 'DateTimeMessage', 'JsonArray', 'JsonObject', 'JsonValue', 'JsonProtoEncoder', 'JsonProtoDecoder', ] # pylint:disable=invalid-name DateTimeMessage = message_types.DateTimeMessage # pylint:enable=invalid-name # We insert our own metaclass here to avoid letting ProtoRPC # register this as the default field type for strings. # * since ProtoRPC does this via metaclasses, we don't have any # choice but to use one ourselves # * since a subclass's metaclass must inherit from its superclass's # metaclass, we're forced to have this hard-to-read inheritance. # # pylint: disable=protected-access class _FieldMeta(messages._FieldMeta): def __init__(cls, name, bases, dct): # pylint: disable=no-self-argument # pylint: disable=super-init-not-called,non-parent-init-called type.__init__(cls, name, bases, dct) # pylint: enable=protected-access class DateField(six.with_metaclass(_FieldMeta, messages.Field)): """Field definition for Date values.""" VARIANTS = frozenset([messages.Variant.STRING]) DEFAULT_VARIANT = messages.Variant.STRING type = datetime.date def _ValidateJsonValue(json_value): entries = [(f, json_value.get_assigned_value(f.name)) for f in json_value.all_fields()] assigned_entries = [(f, value) for f, value in entries if value is not None] if len(assigned_entries) != 1: raise exceptions.InvalidDataError( 'Malformed JsonValue: %s' % json_value) def _JsonValueToPythonValue(json_value): """Convert the given JsonValue to a json string.""" util.Typecheck(json_value, JsonValue) _ValidateJsonValue(json_value) if json_value.is_null: return None entries = [(f, json_value.get_assigned_value(f.name)) for f in json_value.all_fields()] assigned_entries = [(f, value) for f, value in entries if value is not None] field, value = assigned_entries[0] if not isinstance(field, messages.MessageField): return value elif field.message_type is JsonObject: return _JsonObjectToPythonValue(value) elif field.message_type is JsonArray: return _JsonArrayToPythonValue(value) def _JsonObjectToPythonValue(json_value): util.Typecheck(json_value, JsonObject) return dict([(prop.key, _JsonValueToPythonValue(prop.value)) for prop in json_value.properties]) def _JsonArrayToPythonValue(json_value): util.Typecheck(json_value, JsonArray) return [_JsonValueToPythonValue(e) for e in json_value.entries] _MAXINT64 = 2 << 63 - 1 _MININT64 = -(2 << 63) def _PythonValueToJsonValue(py_value): """Convert the given python value to a JsonValue.""" if py_value is None: return JsonValue(is_null=True) if isinstance(py_value, bool): return JsonValue(boolean_value=py_value) if isinstance(py_value, six.string_types): return JsonValue(string_value=py_value) if isinstance(py_value, numbers.Number): if isinstance(py_value, six.integer_types): if _MININT64 < py_value < _MAXINT64: return JsonValue(integer_value=py_value) return JsonValue(double_value=float(py_value)) if isinstance(py_value, dict): return JsonValue(object_value=_PythonValueToJsonObject(py_value)) if isinstance(py_value, Iterable): return JsonValue(array_value=_PythonValueToJsonArray(py_value)) raise exceptions.InvalidDataError( 'Cannot convert "%s" to JsonValue' % py_value) def _PythonValueToJsonObject(py_value): util.Typecheck(py_value, dict) return JsonObject( properties=[ JsonObject.Property(key=key, value=_PythonValueToJsonValue(value)) for key, value in py_value.items()]) def _PythonValueToJsonArray(py_value): return JsonArray(entries=list(map(_PythonValueToJsonValue, py_value))) class JsonValue(messages.Message): """Any valid JSON value.""" # Is this JSON object `null`? is_null = messages.BooleanField(1, default=False) # Exactly one of the following is provided if is_null is False; none # should be provided if is_null is True. boolean_value = messages.BooleanField(2) string_value = messages.StringField(3) # We keep two numeric fields to keep int64 round-trips exact. double_value = messages.FloatField(4, variant=messages.Variant.DOUBLE) integer_value = messages.IntegerField(5, variant=messages.Variant.INT64) # Compound types object_value = messages.MessageField('JsonObject', 6) array_value = messages.MessageField('JsonArray', 7) class JsonObject(messages.Message): """A JSON object value. Messages: Property: A property of a JsonObject. Fields: properties: A list of properties of a JsonObject. """ class Property(messages.Message): """A property of a JSON object. Fields: key: Name of the property. value: A JsonValue attribute. """ key = messages.StringField(1) value = messages.MessageField(JsonValue, 2) properties = messages.MessageField(Property, 1, repeated=True) class JsonArray(messages.Message): """A JSON array value.""" entries = messages.MessageField(JsonValue, 1, repeated=True) _JSON_PROTO_TO_PYTHON_MAP = { JsonArray: _JsonArrayToPythonValue, JsonObject: _JsonObjectToPythonValue, JsonValue: _JsonValueToPythonValue, } _JSON_PROTO_TYPES = tuple(_JSON_PROTO_TO_PYTHON_MAP.keys()) def _JsonProtoToPythonValue(json_proto): util.Typecheck(json_proto, _JSON_PROTO_TYPES) return _JSON_PROTO_TO_PYTHON_MAP[type(json_proto)](json_proto) def _PythonValueToJsonProto(py_value): if isinstance(py_value, dict): return _PythonValueToJsonObject(py_value) if (isinstance(py_value, Iterable) and not isinstance(py_value, six.string_types)): return _PythonValueToJsonArray(py_value) return _PythonValueToJsonValue(py_value) def _JsonProtoToJson(json_proto, unused_encoder=None): return json.dumps(_JsonProtoToPythonValue(json_proto)) def _JsonToJsonProto(json_data, unused_decoder=None): return _PythonValueToJsonProto(json.loads(json_data)) def _JsonToJsonValue(json_data, unused_decoder=None): result = _PythonValueToJsonProto(json.loads(json_data)) if isinstance(result, JsonValue): return result elif isinstance(result, JsonObject): return JsonValue(object_value=result) elif isinstance(result, JsonArray): return JsonValue(array_value=result) else: raise exceptions.InvalidDataError( 'Malformed JsonValue: %s' % json_data) # pylint:disable=invalid-name JsonProtoEncoder = _JsonProtoToJson JsonProtoDecoder = _JsonToJsonProto # pylint:enable=invalid-name encoding.RegisterCustomMessageCodec( encoder=JsonProtoEncoder, decoder=_JsonToJsonValue)(JsonValue) encoding.RegisterCustomMessageCodec( encoder=JsonProtoEncoder, decoder=JsonProtoDecoder)(JsonObject) encoding.RegisterCustomMessageCodec( encoder=JsonProtoEncoder, decoder=JsonProtoDecoder)(JsonArray) def _EncodeDateTimeField(field, value): result = protojson.ProtoJson().encode_field(field, value) return encoding.CodecResult(value=result, complete=True) def _DecodeDateTimeField(unused_field, value): result = protojson.ProtoJson().decode_field( message_types.DateTimeField(1), value) return encoding.CodecResult(value=result, complete=True) encoding.RegisterFieldTypeCodec(_EncodeDateTimeField, _DecodeDateTimeField)( message_types.DateTimeField) def _EncodeInt64Field(field, value): """Handle the special case of int64 as a string.""" capabilities = [ messages.Variant.INT64, messages.Variant.UINT64, ] if field.variant not in capabilities: return encoding.CodecResult(value=value, complete=False) if field.repeated: result = [str(x) for x in value] else: result = str(value) return encoding.CodecResult(value=result, complete=True) def _DecodeInt64Field(unused_field, value): # Don't need to do anything special, they're decoded just fine return encoding.CodecResult(value=value, complete=False) encoding.RegisterFieldTypeCodec(_EncodeInt64Field, _DecodeInt64Field)( messages.IntegerField) def _EncodeDateField(field, value): """Encoder for datetime.date objects.""" if field.repeated: result = [d.isoformat() for d in value] else: result = value.isoformat() return encoding.CodecResult(value=result, complete=True) def _DecodeDateField(unused_field, value): date = datetime.datetime.strptime(value, '%Y-%m-%d').date() return encoding.CodecResult(value=date, complete=True) encoding.RegisterFieldTypeCodec(_EncodeDateField, _DecodeDateField)(DateField)	9.74 KB	2025-03-11 18:02
📄 extra_types_test.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. import datetime import json import math import unittest from apitools.base.protorpclite import messages from apitools.base.py import encoding from apitools.base.py import exceptions from apitools.base.py import extra_types class ExtraTypesTest(unittest.TestCase): def assertRoundTrip(self, value): if isinstance(value, extra_types._JSON_PROTO_TYPES): self.assertEqual( value, extra_types._PythonValueToJsonProto( extra_types._JsonProtoToPythonValue(value))) else: self.assertEqual( value, extra_types._JsonProtoToPythonValue( extra_types._PythonValueToJsonProto(value))) def assertTranslations(self, py_value, json_proto): self.assertEqual( py_value, extra_types._JsonProtoToPythonValue(json_proto)) self.assertEqual( json_proto, extra_types._PythonValueToJsonProto(py_value)) def testInvalidProtos(self): with self.assertRaises(exceptions.InvalidDataError): extra_types._ValidateJsonValue(extra_types.JsonValue()) with self.assertRaises(exceptions.InvalidDataError): extra_types._ValidateJsonValue( extra_types.JsonValue(is_null=True, string_value='a')) with self.assertRaises(exceptions.InvalidDataError): extra_types._ValidateJsonValue( extra_types.JsonValue(integer_value=3, string_value='a')) def testNullEncoding(self): self.assertTranslations(None, extra_types.JsonValue(is_null=True)) def testJsonNumberEncoding(self): seventeen = extra_types.JsonValue(integer_value=17) self.assertRoundTrip(17) self.assertRoundTrip(seventeen) self.assertTranslations(17, seventeen) json_pi = extra_types.JsonValue(double_value=math.pi) self.assertRoundTrip(math.pi) self.assertRoundTrip(json_pi) self.assertTranslations(math.pi, json_pi) def testArrayEncoding(self): array = [3, 'four', False] json_array = extra_types.JsonArray(entries=[ extra_types.JsonValue(integer_value=3), extra_types.JsonValue(string_value='four'), extra_types.JsonValue(boolean_value=False), ]) self.assertRoundTrip(array) self.assertRoundTrip(json_array) self.assertTranslations(array, json_array) def testArrayAsValue(self): array_json = '[3, "four", false]' array = [3, 'four', False] value = encoding.JsonToMessage(extra_types.JsonValue, array_json) self.assertTrue(isinstance(value, extra_types.JsonValue)) self.assertEqual(array, encoding.MessageToPyValue(value)) def testObjectAsValue(self): obj_json = '{"works": true}' obj = {'works': True} value = encoding.JsonToMessage(extra_types.JsonValue, obj_json) self.assertTrue(isinstance(value, extra_types.JsonValue)) self.assertEqual(obj, encoding.MessageToPyValue(value)) def testDictEncoding(self): d = {'a': 6, 'b': 'eleventeen'} json_d = extra_types.JsonObject(properties=[ extra_types.JsonObject.Property( key='a', value=extra_types.JsonValue(integer_value=6)), extra_types.JsonObject.Property( key='b', value=extra_types.JsonValue(string_value='eleventeen')), ]) self.assertRoundTrip(d) # We don't know json_d will round-trip, because of randomness in # python dictionary iteration ordering. We also need to force # comparison as lists, since hashing protos isn't helpful. translated_properties = extra_types._PythonValueToJsonProto( d).properties for p in json_d.properties: self.assertIn(p, translated_properties) for p in translated_properties: self.assertIn(p, json_d.properties) def testJsonObjectPropertyTranslation(self): value = extra_types.JsonValue(string_value='abc') obj = extra_types.JsonObject(properties=[ extra_types.JsonObject.Property(key='attr_name', value=value)]) json_value = '"abc"' json_obj = '{"attr_name": "abc"}' self.assertRoundTrip(value) self.assertRoundTrip(obj) self.assertRoundTrip(json_value) self.assertRoundTrip(json_obj) self.assertEqual(json_value, encoding.MessageToJson(value)) self.assertEqual(json_obj, encoding.MessageToJson(obj)) def testJsonValueAsFieldTranslation(self): class HasJsonValueMsg(messages.Message): some_value = messages.MessageField(extra_types.JsonValue, 1) msg_json = '{"some_value": [1, 2, 3]}' msg = HasJsonValueMsg( some_value=encoding.PyValueToMessage( extra_types.JsonValue, [1, 2, 3])) self.assertEqual(msg, encoding.JsonToMessage(HasJsonValueMsg, msg_json)) self.assertEqual(msg_json, encoding.MessageToJson(msg)) def testDateField(self): class DateMsg(messages.Message): start_date = extra_types.DateField(1) all_dates = extra_types.DateField(2, repeated=True) msg = DateMsg( start_date=datetime.date(1752, 9, 9), all_dates=[ datetime.date(1979, 5, 6), datetime.date(1980, 10, 24), datetime.date(1981, 1, 19), ]) msg_dict = { 'start_date': '1752-09-09', 'all_dates': ['1979-05-06', '1980-10-24', '1981-01-19'], } self.assertEqual(msg_dict, json.loads(encoding.MessageToJson(msg))) self.assertEqual( msg, encoding.JsonToMessage(DateMsg, json.dumps(msg_dict))) def testInt64(self): # Testing roundtrip of type 'long' class DogeMsg(messages.Message): such_string = messages.StringField(1) wow = messages.IntegerField(2, variant=messages.Variant.INT64) very_unsigned = messages.IntegerField( 3, variant=messages.Variant.UINT64) much_repeated = messages.IntegerField( 4, variant=messages.Variant.INT64, repeated=True) def MtoJ(msg): return encoding.MessageToJson(msg) def JtoM(class_type, json_str): return encoding.JsonToMessage(class_type, json_str) def DoRoundtrip(class_type, json_msg=None, message=None, times=4): if json_msg: json_msg = MtoJ(JtoM(class_type, json_msg)) if message: message = JtoM(class_type, MtoJ(message)) if times == 0: result = json_msg if json_msg else message return result return DoRoundtrip(class_type=class_type, json_msg=json_msg, message=message, times=times - 1) # Single json_msg = ('{"such_string": "poot", "wow": "-1234", ' '"very_unsigned": "999", "much_repeated": ["123", "456"]}') out_json = MtoJ(JtoM(DogeMsg, json_msg)) self.assertEqual(json.loads(out_json)['wow'], '-1234') # Repeated test case msg = DogeMsg(such_string='wow', wow=-1234, very_unsigned=800, much_repeated=[123, 456]) self.assertEqual(msg, DoRoundtrip(DogeMsg, message=msg))	7.81 KB	2025-03-11 18:02
📄 gzip.py # Copyright (c) 2001, 2002, 2003, 2004, 2005, 2006, 2007, 2008, 2009, 2010, # 2011, 2012, 2013, 2014, 2015, 2016, 2017 Python Software Foundation; All # Rights Reserved # # This is a backport from python 3.4 into python 2.7. Text and exclusive mode # support are removed as they're unsupported in 2.7. This backport patches a # streaming bug that exists in python 2.7. """Functions that read and write gzipped files. The user of the file doesn't have to worry about the compression, but random access is not allowed.""" # based on Andrew Kuchling's minigzip.py distributed with the zlib module import six from six.moves import builtins from six.moves import range import struct import sys import time import os import zlib import io __all__ = ["GzipFile", "open", "compress", "decompress"] FTEXT, FHCRC, FEXTRA, FNAME, FCOMMENT = 1, 2, 4, 8, 16 READ, WRITE = 1, 2 def open(filename, mode="rb", compresslevel=9): """Shorthand for GzipFile(filename, mode, compresslevel). The filename argument is required; mode defaults to 'rb' and compresslevel defaults to 9. """ return GzipFile(filename, mode, compresslevel) def write32u(output, value): # The L format writes the bit pattern correctly whether signed # or unsigned. output.write(struct.pack("<L", value)) class _PaddedFile(object): """Minimal read-only file object that prepends a string to the contents of an actual file. Shouldn't be used outside of gzip.py, as it lacks essential functionality.""" def __init__(self, f, prepend=b''): self._buffer = prepend self._length = len(prepend) self.file = f self._read = 0 def read(self, size): if self._read is None: return self.file.read(size) if self._read + size <= self._length: read = self._read self._read += size return self._buffer[read:self._read] else: read = self._read self._read = None return self._buffer[read:] + \ self.file.read(size - self._length + read) def prepend(self, prepend=b'', readprevious=False): if self._read is None: self._buffer = prepend elif readprevious and len(prepend) <= self._read: self._read -= len(prepend) return else: self._buffer = self._buffer[self._read:] + prepend self._length = len(self._buffer) self._read = 0 def unused(self): if self._read is None: return b'' return self._buffer[self._read:] def seek(self, offset, whence=0): # This is only ever called with offset=whence=0 if whence == 1 and self._read is not None: if 0 <= offset + self._read <= self._length: self._read += offset return else: offset += self._length - self._read self._read = None self._buffer = None return self.file.seek(offset, whence) def __getattr__(self, name): return getattr(self.file, name) class GzipFile(io.BufferedIOBase): """The GzipFile class simulates most of the methods of a file object with the exception of the readinto() and truncate() methods. This class only supports opening files in binary mode. If you need to open a compressed file in text mode, use the gzip.open() function. """ myfileobj = None max_read_chunk = 10 * 1024 * 1024 # 10Mb def __init__(self, filename=None, mode=None, compresslevel=9, fileobj=None, mtime=None): """Constructor for the GzipFile class. At least one of fileobj and filename must be given a non-trivial value. The new class instance is based on fileobj, which can be a regular file, an io.BytesIO object, or any other object which simulates a file. It defaults to None, in which case filename is opened to provide a file object. When fileobj is not None, the filename argument is only used to be included in the gzip file header, which may includes the original filename of the uncompressed file. It defaults to the filename of fileobj, if discernible; otherwise, it defaults to the empty string, and in this case the original filename is not included in the header. The mode argument can be any of 'r', 'rb', 'a', 'ab', 'w', or 'wb', depending on whether the file will be read or written. The default is the mode of fileobj if discernible; otherwise, the default is 'rb'. A mode of 'r' is equivalent to one of 'rb', and similarly for 'w' and 'wb', and 'a' and 'ab'. The compresslevel argument is an integer from 0 to 9 controlling the level of compression; 1 is fastest and produces the least compression, and 9 is slowest and produces the most compression. 0 is no compression at all. The default is 9. The mtime argument is an optional numeric timestamp to be written to the stream when compressing. All gzip compressed streams are required to contain a timestamp. If omitted or None, the current time is used. This module ignores the timestamp when decompressing; however, some programs, such as gunzip, make use of it. The format of the timestamp is the same as that of the return value of time.time() and of the st_mtime member of the object returned by os.stat(). """ if mode and ('t' in mode or 'U' in mode): raise ValueError("Invalid mode: {!r}".format(mode)) if mode and 'b' not in mode: mode += 'b' if fileobj is None: fileobj = self.myfileobj = builtins.open(filename, mode or 'rb') if filename is None: filename = getattr(fileobj, 'name', '') if not isinstance(filename, six.string_types): filename = '' if mode is None: mode = getattr(fileobj, 'mode', 'rb') if mode.startswith('r'): self.mode = READ # Set flag indicating start of a new member self._new_member = True # Buffer data read from gzip file. extrastart is offset in # stream where buffer starts. extrasize is number of # bytes remaining in buffer from current stream position. self.extrabuf = b"" self.extrasize = 0 self.extrastart = 0 self.name = filename # Starts small, scales exponentially self.min_readsize = 100 fileobj = _PaddedFile(fileobj) elif mode.startswith(('w', 'a')): self.mode = WRITE self._init_write(filename) self.compress = zlib.compressobj(compresslevel, zlib.DEFLATED, -zlib.MAX_WBITS, zlib.DEF_MEM_LEVEL, 0) else: raise ValueError("Invalid mode: {!r}".format(mode)) self.fileobj = fileobj self.offset = 0 self.mtime = mtime if self.mode == WRITE: self._write_gzip_header() @property def filename(self): import warnings warnings.warn("use the name attribute", DeprecationWarning, 2) if self.mode == WRITE and self.name[-3:] != ".gz": return self.name + ".gz" return self.name def __repr__(self): fileobj = self.fileobj if isinstance(fileobj, _PaddedFile): fileobj = fileobj.file s = repr(fileobj) return '<gzip ' + s[1:-1] + ' ' + hex(id(self)) + '>' def _check_closed(self): """Raises a ValueError if the underlying file object has been closed. """ if self.closed: raise ValueError('I/O operation on closed file.') def _init_write(self, filename): self.name = filename self.crc = zlib.crc32(b"") & 0xffffffff self.size = 0 self.writebuf = [] self.bufsize = 0 def _write_gzip_header(self): self.fileobj.write(b'\037\213') # magic header self.fileobj.write(b'\010') # compression method try: # RFC 1952 requires the FNAME field to be Latin-1. Do not # include filenames that cannot be represented that way. fname = os.path.basename(self.name) if not isinstance(fname, six.binary_type): fname = fname.encode('latin-1') if fname.endswith(b'.gz'): fname = fname[:-3] except UnicodeEncodeError: fname = b'' flags = 0 if fname: flags = FNAME self.fileobj.write(six.unichr(flags).encode('latin-1')) mtime = self.mtime if mtime is None: mtime = time.time() write32u(self.fileobj, int(mtime)) self.fileobj.write(b'\002') self.fileobj.write(b'\377') if fname: self.fileobj.write(fname + b'\000') def _init_read(self): self.crc = zlib.crc32(b"") & 0xffffffff self.size = 0 def _read_exact(self, n): data = self.fileobj.read(n) while len(data) < n: b = self.fileobj.read(n - len(data)) if not b: raise EOFError("Compressed file ended before the " "end-of-stream marker was reached") data += b return data def _read_gzip_header(self): magic = self.fileobj.read(2) if magic == b'': return False if magic != b'\037\213': raise OSError('Not a gzipped file') method, flag, self.mtime = struct.unpack("<BBIxx", self._read_exact(8)) if method != 8: raise OSError('Unknown compression method') if flag & FEXTRA: # Read & discard the extra field, if present extra_len, = struct.unpack("<H", self._read_exact(2)) self._read_exact(extra_len) if flag & FNAME: # Read and discard a null-terminated string containing the filename while True: s = self.fileobj.read(1) if not s or s == b'\000': break if flag & FCOMMENT: # Read and discard a null-terminated string containing a comment while True: s = self.fileobj.read(1) if not s or s == b'\000': break if flag & FHCRC: self._read_exact(2) # Read & discard the 16-bit header CRC unused = self.fileobj.unused() if unused: uncompress = self.decompress.decompress(unused) self._add_read_data(uncompress) return True def write(self, data): self._check_closed() if self.mode != WRITE: import errno raise OSError(errno.EBADF, "write() on read-only GzipFile object") if self.fileobj is None: raise ValueError("write() on closed GzipFile object") # Convert data type if called by io.BufferedWriter. if isinstance(data, memoryview): data = data.tobytes() if len(data) > 0: self.fileobj.write(self.compress.compress(data)) self.size += len(data) self.crc = zlib.crc32(data, self.crc) & 0xffffffff self.offset += len(data) return len(data) def read(self, size=-1): self._check_closed() if self.mode != READ: import errno raise OSError(errno.EBADF, "read() on write-only GzipFile object") if self.extrasize <= 0 and self.fileobj is None: return b'' readsize = 1024 if size < 0: # get the whole thing while self._read(readsize): readsize = min(self.max_read_chunk, readsize * 2) size = self.extrasize else: # just get some more of it while size > self.extrasize: if not self._read(readsize): if size > self.extrasize: size = self.extrasize break readsize = min(self.max_read_chunk, readsize * 2) offset = self.offset - self.extrastart chunk = self.extrabuf[offset: offset + size] self.extrasize = self.extrasize - size self.offset += size return chunk def read1(self, size=-1): self._check_closed() if self.mode != READ: import errno raise OSError(errno.EBADF, "read1() on write-only GzipFile object") if self.extrasize <= 0 and self.fileobj is None: return b'' # For certain input data, a single call to _read() may not return # any data. In this case, retry until we get some data or reach EOF. while self.extrasize <= 0 and self._read(): pass if size < 0 or size > self.extrasize: size = self.extrasize offset = self.offset - self.extrastart chunk = self.extrabuf[offset: offset + size] self.extrasize -= size self.offset += size return chunk def peek(self, n): if self.mode != READ: import errno raise OSError(errno.EBADF, "peek() on write-only GzipFile object") # Do not return ridiculously small buffers, for one common idiom # is to call peek(1) and expect more bytes in return. if n < 100: n = 100 if self.extrasize == 0: if self.fileobj is None: return b'' # Ensure that we don't return b"" if we haven't reached EOF. # 1024 is the same buffering heuristic used in read() while self.extrasize == 0 and self._read(max(n, 1024)): pass offset = self.offset - self.extrastart remaining = self.extrasize assert remaining == len(self.extrabuf) - offset return self.extrabuf[offset:offset + n] def _unread(self, buf): self.extrasize = len(buf) + self.extrasize self.offset -= len(buf) def _read(self, size=1024): if self.fileobj is None: return False if self._new_member: # If the _new_member flag is set, we have to # jump to the next member, if there is one. self._init_read() if not self._read_gzip_header(): return False self.decompress = zlib.decompressobj(-zlib.MAX_WBITS) self._new_member = False # Read a chunk of data from the file buf = self.fileobj.read(size) # If the EOF has been reached, flush the decompression object # and mark this object as finished. if buf == b"": uncompress = self.decompress.flush() # Prepend the already read bytes to the fileobj to they can be # seen by _read_eof() self.fileobj.prepend(self.decompress.unused_data, True) self._read_eof() self._add_read_data(uncompress) return False uncompress = self.decompress.decompress(buf) self._add_read_data(uncompress) if self.decompress.unused_data != b"": # Ending case: we've come to the end of a member in the file, # so seek back to the start of the unused data, finish up # this member, and read a new gzip header. # Prepend the already read bytes to the fileobj to they can be # seen by _read_eof() and _read_gzip_header() self.fileobj.prepend(self.decompress.unused_data, True) # Check the CRC and file size, and set the flag so we read # a new member on the next call self._read_eof() self._new_member = True return True def _add_read_data(self, data): self.crc = zlib.crc32(data, self.crc) & 0xffffffff offset = self.offset - self.extrastart self.extrabuf = self.extrabuf[offset:] + data self.extrasize = self.extrasize + len(data) self.extrastart = self.offset self.size = self.size + len(data) def _read_eof(self): # We've read to the end of the file # We check the that the computed CRC and size of the # uncompressed data matches the stored values. Note that the size # stored is the true file size mod 2*32. crc32, isize = struct.unpack("<II", self._read_exact(8)) if crc32 != self.crc: raise OSError("CRC check failed %s != %s" % (hex(crc32), hex(self.crc))) elif isize != (self.size & 0xffffffff): raise OSError("Incorrect length of data produced") # Gzip files can be padded with zeroes and still have archives. # Consume all zero bytes and set the file position to the first # non-zero byte. See http://www.gzip.org/#faq8 c = b"\x00" while c == b"\x00": c = self.fileobj.read(1) if c: self.fileobj.prepend(c, True) @property def closed(self): return self.fileobj is None def close(self): fileobj = self.fileobj if fileobj is None: return self.fileobj = None try: if self.mode == WRITE: fileobj.write(self.compress.flush()) write32u(fileobj, self.crc) # self.size may exceed 2GB, or even 4GB write32u(fileobj, self.size & 0xffffffff) finally: myfileobj = self.myfileobj if myfileobj: self.myfileobj = None myfileobj.close() def flush(self, zlib_mode=zlib.Z_SYNC_FLUSH): self._check_closed() if self.mode == WRITE: # Ensure the compressor's buffer is flushed self.fileobj.write(self.compress.flush(zlib_mode)) self.fileobj.flush() def fileno(self): """Invoke the underlying file object's fileno() method. This will raise AttributeError if the underlying file object doesn't support fileno(). """ return self.fileobj.fileno() def rewind(self): '''Return the uncompressed stream file position indicator to the beginning of the file''' if self.mode != READ: raise OSError("Can't rewind in write mode") self.fileobj.seek(0) self._new_member = True self.extrabuf = b"" self.extrasize = 0 self.extrastart = 0 self.offset = 0 def readable(self): return self.mode == READ def writable(self): return self.mode == WRITE def seekable(self): return True def seek(self, offset, whence=0): if whence: if whence == 1: offset = self.offset + offset else: raise ValueError('Seek from end not supported') if self.mode == WRITE: if offset < self.offset: raise OSError('Negative seek in write mode') count = offset - self.offset chunk = bytes(1024) for i in range(count // 1024): self.write(chunk) self.write(bytes(count % 1024)) elif self.mode == READ: if offset < self.offset: # for negative seek, rewind and do positive seek self.rewind() count = offset - self.offset for i in range(count // 1024): self.read(1024) self.read(count % 1024) return self.offset def readline(self, size=-1): if size < 0: # Shortcut common case - newline found in buffer. offset = self.offset - self.extrastart i = self.extrabuf.find(b'\n', offset) + 1 if i > 0: self.extrasize -= i - offset self.offset += i - offset return self.extrabuf[offset: i] size = sys.maxsize readsize = self.min_readsize else: readsize = size bufs = [] while size != 0: c = self.read(readsize) i = c.find(b'\n') # We set i=size to break out of the loop under two # conditions: 1) there's no newline, and the chunk is # larger than size, or 2) there is a newline, but the # resulting line would be longer than 'size'. if (size <= i) or (i == -1 and len(c) > size): i = size - 1 if i >= 0 or c == b'': bufs.append(c[:i + 1]) # Add portion of last chunk self._unread(c[i + 1:]) # Push back rest of chunk break # Append chunk to list, decrease 'size', bufs.append(c) size = size - len(c) readsize = min(size, readsize 2) if readsize > self.min_readsize: self.min_readsize = min(readsize, self.min_readsize * 2, 512) return b''.join(bufs) # Return resulting line def compress(data, compresslevel=9): """Compress data in one shot and return the compressed string. Optional argument is the compression level, in range of 0-9. """ buf = io.BytesIO() with GzipFile(fileobj=buf, mode='wb', compresslevel=compresslevel) as f: f.write(data) return buf.getvalue() def decompress(data): """Decompress a gzip compressed string in one shot. Return the decompressed string. """ with GzipFile(fileobj=io.BytesIO(data)) as f: return f.read()	21.39 KB	2025-03-11 18:02
📄 gzip_test.py # Copyright (c) 2001, 2002, 2003, 2004, 2005, 2006, 2007, 2008, 2009, 2010, # 2011, 2012, 2013, 2014, 2015, 2016, 2017 Python Software Foundation; All # Rights Reserved # # This is a backport from python 3.4 into python 2.7. Text and exclusive mode # support are removed as they're unsupported in 2.7. This backport patches a # streaming bug that exists in python 2.7. """Test script for the gzip module. """ import six from six.moves import range import unittest import os import io import struct from apitools.base.py import gzip from io import open data1 = b""" int length=DEFAULTALLOC, err = Z_OK; PyObject RetVal; int flushmode = Z_FINISH; unsigned long start_total_out; """ data2 = b"""/ zlibmodule.c -- gzip-compatible data compression / / See http://www.gzip.org/zlib/ /* See http://www.winimage.com/zLibDll for Windows / """ def unlink(filename): try: os.unlink(filename) except: pass class UnseekableIO(io.BytesIO): def seekable(self): return False def tell(self): raise io.UnsupportedOperation def seek(self, args): raise io.UnsupportedOperation class BaseTest(unittest.TestCase): filename = "@test" def setUp(self): unlink(self.filename) def tearDown(self): unlink(self.filename) class TestGzip(BaseTest): def write_and_read_back(self, data, mode='b'): b_data = bytes(data) with gzip.GzipFile(self.filename, 'w' + mode) as f: l = f.write(data) self.assertEqual(l, len(b_data)) with gzip.GzipFile(self.filename, 'r' + mode) as f: self.assertEqual(f.read(), b_data) def test_write(self): with gzip.GzipFile(self.filename, 'wb') as f: f.write(data1 * 50) # Try flush and fileno. f.flush() f.fileno() if hasattr(os, 'fsync'): os.fsync(f.fileno()) f.close() # Test multiple close() calls. f.close() # The following test_write_xy methods test that write accepts # the corresponding bytes-like object type as input # and that the data written equals bytes(xy) in all cases. def test_write_memoryview(self): data = memoryview(data1 * 50) self.write_and_read_back(data.tobytes()) data = memoryview(bytes(range(256))) self.write_and_read_back(data.tobytes()) def test_write_incompatible_type(self): # Test that non-bytes-like types raise TypeError. # Issue #21560: attempts to write incompatible types # should not affect the state of the fileobject with gzip.GzipFile(self.filename, 'wb') as f: if six.PY2: with self.assertRaises(UnicodeEncodeError): f.write(u'\xff') elif six.PY3: with self.assertRaises(TypeError): f.write(u'\xff') with self.assertRaises(TypeError): f.write([1]) f.write(data1) with gzip.GzipFile(self.filename, 'rb') as f: self.assertEqual(f.read(), data1) def test_read(self): self.test_write() # Try reading. with gzip.GzipFile(self.filename, 'r') as f: d = f.read() self.assertEqual(d, data1 * 50) def test_read1(self): self.test_write() blocks = [] nread = 0 with gzip.GzipFile(self.filename, 'r') as f: while True: d = f.read1() if not d: break blocks.append(d) nread += len(d) # Check that position was updated correctly (see issue10791). self.assertEqual(f.tell(), nread) self.assertEqual(b''.join(blocks), data1 * 50) def test_io_on_closed_object(self): # Test that I/O operations on closed GzipFile objects raise a # ValueError, just like the corresponding functions on file objects. # Write to a file, open it for reading, then close it. self.test_write() f = gzip.GzipFile(self.filename, 'r') f.close() with self.assertRaises(ValueError): f.read(1) with self.assertRaises(ValueError): f.seek(0) with self.assertRaises(ValueError): f.tell() # Open the file for writing, then close it. f = gzip.GzipFile(self.filename, 'w') f.close() with self.assertRaises(ValueError): f.write(b'') with self.assertRaises(ValueError): f.flush() def test_append(self): self.test_write() # Append to the previous file with gzip.GzipFile(self.filename, 'ab') as f: f.write(data2 * 15) with gzip.GzipFile(self.filename, 'rb') as f: d = f.read() self.assertEqual(d, (data1 * 50) + (data2 * 15)) def test_many_append(self): # Bug #1074261 was triggered when reading a file that contained # many, many members. Create such a file and verify that reading it # works. with gzip.GzipFile(self.filename, 'wb', 9) as f: f.write(b'a') for i in range(0, 200): with gzip.GzipFile(self.filename, "ab", 9) as f: # append f.write(b'a') # Try reading the file with gzip.GzipFile(self.filename, "rb") as zgfile: contents = b"" while 1: ztxt = zgfile.read(8192) contents += ztxt if not ztxt: break self.assertEqual(contents, b'a' * 201) def test_buffered_reader(self): # Issue #7471: a GzipFile can be wrapped in a BufferedReader for # performance. self.test_write() with gzip.GzipFile(self.filename, 'rb') as f: with io.BufferedReader(f) as r: lines = [line for line in r] self.assertEqual(lines, 50 * data1.splitlines(True)) def test_readline(self): self.test_write() # Try .readline() with varying line lengths with gzip.GzipFile(self.filename, 'rb') as f: line_length = 0 while 1: L = f.readline(line_length) if not L and line_length != 0: break self.assertTrue(len(L) <= line_length) line_length = (line_length + 1) % 50 def test_readlines(self): self.test_write() # Try .readlines() with gzip.GzipFile(self.filename, 'rb') as f: L = f.readlines() with gzip.GzipFile(self.filename, 'rb') as f: while 1: L = f.readlines(150) if L == []: break def test_seek_read(self): self.test_write() # Try seek, read test with gzip.GzipFile(self.filename) as f: while 1: oldpos = f.tell() line1 = f.readline() if not line1: break newpos = f.tell() f.seek(oldpos) # negative seek if len(line1) > 10: amount = 10 else: amount = len(line1) line2 = f.read(amount) self.assertEqual(line1[:amount], line2) f.seek(newpos) # positive seek def test_seek_whence(self): self.test_write() # Try seek(whence=1), read test with gzip.GzipFile(self.filename) as f: f.read(10) f.seek(10, whence=1) y = f.read(10) self.assertEqual(y, data1[20:30]) def test_seek_write(self): # Try seek, write test with gzip.GzipFile(self.filename, 'w') as f: for pos in range(0, 256, 16): f.seek(pos) f.write(b'GZ\n') def test_mode(self): self.test_write() with gzip.GzipFile(self.filename, 'r') as f: self.assertEqual(f.myfileobj.mode, 'rb') def test_1647484(self): for mode in ('wb', 'rb'): with gzip.GzipFile(self.filename, mode) as f: self.assertTrue(hasattr(f, "name")) self.assertEqual(f.name, self.filename) def test_paddedfile_getattr(self): self.test_write() with gzip.GzipFile(self.filename, 'rb') as f: self.assertTrue(hasattr(f.fileobj, "name")) self.assertEqual(f.fileobj.name, self.filename) def test_mtime(self): mtime = 123456789 with gzip.GzipFile(self.filename, 'w', mtime=mtime) as fWrite: fWrite.write(data1) with gzip.GzipFile(self.filename) as fRead: dataRead = fRead.read() self.assertEqual(dataRead, data1) self.assertTrue(hasattr(fRead, 'mtime')) self.assertEqual(fRead.mtime, mtime) def test_metadata(self): mtime = 123456789 with gzip.GzipFile(self.filename, 'w', mtime=mtime) as fWrite: fWrite.write(data1) with open(self.filename, 'rb') as fRead: # see RFC 1952: http://www.faqs.org/rfcs/rfc1952.html idBytes = fRead.read(2) self.assertEqual(idBytes, b'\x1f\x8b') # gzip ID cmByte = fRead.read(1) self.assertEqual(cmByte, b'\x08') # deflate flagsByte = fRead.read(1) self.assertEqual(flagsByte, b'\x08') # only the FNAME flag is set mtimeBytes = fRead.read(4) self.assertEqual(mtimeBytes, struct.pack( '<i', mtime)) # little-endian xflByte = fRead.read(1) self.assertEqual(xflByte, b'\x02') # maximum compression osByte = fRead.read(1) self.assertEqual(osByte, b'\xff') # OS "unknown" (OS-independent) # Since the FNAME flag is set, the zero-terminated filename # follows. RFC 1952 specifies that this is the name of the input # file, if any. However, the gzip module defaults to storing the # name of the output file in this field. expected = self.filename.encode('Latin-1') + b'\x00' nameBytes = fRead.read(len(expected)) self.assertEqual(nameBytes, expected) # Since no other flags were set, the header ends here. Rather than # process the compressed data, let's seek to the trailer. fRead.seek(os.stat(self.filename).st_size - 8) crc32Bytes = fRead.read(4) # CRC32 of uncompressed data [data1] self.assertEqual(crc32Bytes, b'\xaf\xd7d\x83') isizeBytes = fRead.read(4) self.assertEqual(isizeBytes, struct.pack('<i', len(data1))) def test_with_open(self): # GzipFile supports the context management protocol with gzip.GzipFile(self.filename, "wb") as f: f.write(b"xxx") f = gzip.GzipFile(self.filename, "rb") f.close() try: with f: pass except ValueError: pass else: self.fail("__enter__ on a closed file didn't raise an exception") try: with gzip.GzipFile(self.filename, "wb") as f: 1 / 0 except ZeroDivisionError: pass else: self.fail("1/0 didn't raise an exception") def test_zero_padded_file(self): with gzip.GzipFile(self.filename, "wb") as f: f.write(data1 * 50) # Pad the file with zeroes with open(self.filename, "ab") as f: f.write(b"\x00" * 50) with gzip.GzipFile(self.filename, "rb") as f: d = f.read() self.assertEqual(d, data1 * 50, "Incorrect data in file") def test_non_seekable_file(self): uncompressed = data1 * 50 buf = UnseekableIO() with gzip.GzipFile(fileobj=buf, mode="wb") as f: f.write(uncompressed) compressed = buf.getvalue() buf = UnseekableIO(compressed) with gzip.GzipFile(fileobj=buf, mode="rb") as f: self.assertEqual(f.read(), uncompressed) def test_peek(self): uncompressed = data1 * 200 with gzip.GzipFile(self.filename, "wb") as f: f.write(uncompressed) def sizes(): while True: for n in range(5, 50, 10): yield n with gzip.GzipFile(self.filename, "rb") as f: f.max_read_chunk = 33 nread = 0 for n in sizes(): s = f.peek(n) if s == b'': break self.assertEqual(f.read(len(s)), s) nread += len(s) self.assertEqual(f.read(100), b'') self.assertEqual(nread, len(uncompressed)) def test_textio_readlines(self): # Issue #10791: TextIOWrapper.readlines() fails when wrapping GzipFile. lines = (data1 * 50).decode("ascii").splitlines(True) self.test_write() with gzip.GzipFile(self.filename, 'r') as f: with io.TextIOWrapper(f, encoding="ascii") as t: self.assertEqual(t.readlines(), lines) def test_fileobj_from_fdopen(self): # Issue #13781: Opening a GzipFile for writing fails when using a # fileobj created with os.fdopen(). fd = os.open(self.filename, os.O_WRONLY \| os.O_CREAT) with os.fdopen(fd, "wb") as f: with gzip.GzipFile(fileobj=f, mode="w") as g: pass def test_bytes_filename(self): str_filename = self.filename try: bytes_filename = str_filename.encode("ascii") except UnicodeEncodeError: self.skipTest("Temporary file name needs to be ASCII") with gzip.GzipFile(bytes_filename, "wb") as f: f.write(data1 * 50) with gzip.GzipFile(bytes_filename, "rb") as f: self.assertEqual(f.read(), data1 * 50) # Sanity check that we are actually operating on the right file. with gzip.GzipFile(str_filename, "rb") as f: self.assertEqual(f.read(), data1 * 50) # Testing compress/decompress shortcut functions def test_compress(self): for data in [data1, data2]: for args in [(), (1,), (6,), (9,)]: datac = gzip.compress(data, args) self.assertEqual(type(datac), bytes) with gzip.GzipFile(fileobj=io.BytesIO(datac), mode="rb") as f: self.assertEqual(f.read(), data) def test_decompress(self): for data in (data1, data2): buf = io.BytesIO() with gzip.GzipFile(fileobj=buf, mode="wb") as f: f.write(data) self.assertEqual(gzip.decompress(buf.getvalue()), data) # Roundtrip with compress datac = gzip.compress(data) self.assertEqual(gzip.decompress(datac), data) def test_read_truncated(self): data = data1 50 # Drop the CRC (4 bytes) and file size (4 bytes). truncated = gzip.compress(data)[:-8] with gzip.GzipFile(fileobj=io.BytesIO(truncated)) as f: self.assertRaises(EOFError, f.read) with gzip.GzipFile(fileobj=io.BytesIO(truncated)) as f: self.assertEqual(f.read(len(data)), data) self.assertRaises(EOFError, f.read, 1) # Incomplete 10-byte header. for i in range(2, 10): with gzip.GzipFile(fileobj=io.BytesIO(truncated[:i])) as f: self.assertRaises(EOFError, f.read, 1) def test_read_with_extra(self): # Gzip data with an extra field gzdata = (b'\x1f\x8b\x08\x04\xb2\x17cQ\x02\xff' b'\x05\x00Extra' b'\x0bI-.\x01\x002\xd1Mx\x04\x00\x00\x00') with gzip.GzipFile(fileobj=io.BytesIO(gzdata)) as f: self.assertEqual(f.read(), b'Test') def test_prepend_error(self): # See issue #20875 with gzip.open(self.filename, "wb") as f: f.write(data1) with gzip.open(self.filename, "rb") as f: f.fileobj.prepend() class TestOpen(BaseTest): def test_binary_modes(self): uncompressed = data1 * 50 with gzip.open(self.filename, "wb") as f: f.write(uncompressed) with open(self.filename, "rb") as f: file_data = gzip.decompress(f.read()) self.assertEqual(file_data, uncompressed) with gzip.open(self.filename, "rb") as f: self.assertEqual(f.read(), uncompressed) with gzip.open(self.filename, "ab") as f: f.write(uncompressed) with open(self.filename, "rb") as f: file_data = gzip.decompress(f.read()) self.assertEqual(file_data, uncompressed * 2) def test_implicit_binary_modes(self): # Test implicit binary modes (no "b" or "t" in mode string). uncompressed = data1 * 50 with gzip.open(self.filename, "w") as f: f.write(uncompressed) with open(self.filename, "rb") as f: file_data = gzip.decompress(f.read()) self.assertEqual(file_data, uncompressed) with gzip.open(self.filename, "r") as f: self.assertEqual(f.read(), uncompressed) with gzip.open(self.filename, "a") as f: f.write(uncompressed) with open(self.filename, "rb") as f: file_data = gzip.decompress(f.read()) self.assertEqual(file_data, uncompressed * 2)	17.17 KB	2025-03-11 18:02
📄 http_wrapper.py #!/usr/bin/env python # # 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. """HTTP wrapper for apitools. This library wraps the underlying http library we use, which is currently httplib2. """ import collections import contextlib import logging import socket import time import httplib2 import six from six.moves import http_client from six.moves.urllib import parse from apitools.base.py import exceptions from apitools.base.py import util # pylint: disable=ungrouped-imports try: from oauth2client.client import HttpAccessTokenRefreshError as TokenRefreshError # noqa except ImportError: from oauth2client.client import AccessTokenRefreshError as TokenRefreshError # noqa __all__ = [ 'CheckResponse', 'GetHttp', 'HandleExceptionsAndRebuildHttpConnections', 'MakeRequest', 'RebuildHttpConnections', 'Request', 'Response', 'RethrowExceptionHandler', ] # 308 and 429 don't have names in httplib. RESUME_INCOMPLETE = 308 TOO_MANY_REQUESTS = 429 _REDIRECT_STATUS_CODES = ( http_client.MOVED_PERMANENTLY, http_client.FOUND, http_client.SEE_OTHER, http_client.TEMPORARY_REDIRECT, RESUME_INCOMPLETE, ) # http: An httplib2.Http instance. # http_request: A http_wrapper.Request. # exc: Exception being raised. # num_retries: Number of retries consumed; used for exponential backoff. ExceptionRetryArgs = collections.namedtuple( 'ExceptionRetryArgs', ['http', 'http_request', 'exc', 'num_retries', 'max_retry_wait', 'total_wait_sec']) @contextlib.contextmanager def _Httplib2Debuglevel(http_request, level, http=None): """Temporarily change the value of httplib2.debuglevel, if necessary. If http_request has a `loggable_body` distinct from `body`, then we need to prevent httplib2 from logging the full body. This sets httplib2.debuglevel for the duration of the `with` block; however, that alone won't change the value of existing HTTP connections. If an httplib2.Http object is provided, we'll also change the level on any cached connections attached to it. Args: http_request: a Request we're logging. level: (int) the debuglevel for logging. http: (optional) an httplib2.Http whose connections we should set the debuglevel on. Yields: None. """ if http_request.loggable_body is None: yield return old_level = httplib2.debuglevel http_levels = {} httplib2.debuglevel = level if http is not None: for connection_key, connection in http.connections.items(): # httplib2 stores two kinds of values in this dict, connection # classes and instances. Since the connection types are all # old-style classes, we can't easily distinguish by connection # type -- so instead we use the key pattern. if ':' not in connection_key: continue http_levels[connection_key] = connection.debuglevel connection.set_debuglevel(level) yield httplib2.debuglevel = old_level if http is not None: for connection_key, old_level in http_levels.items(): if connection_key in http.connections: http.connections[connection_key].set_debuglevel(old_level) class Request(object): """Class encapsulating the data for an HTTP request.""" def __init__(self, url='', http_method='GET', headers=None, body=''): self.url = url self.http_method = http_method self.headers = headers or {} self.__body = None self.__loggable_body = None self.body = body @property def loggable_body(self): return self.__loggable_body @loggable_body.setter def loggable_body(self, value): if self.body is None: raise exceptions.RequestError( 'Cannot set loggable body on request with no body') self.__loggable_body = value @property def body(self): return self.__body @body.setter def body(self, value): """Sets the request body; handles logging and length measurement.""" self.__body = value if value is not None: # Avoid calling len() which cannot exceed 4GiB in 32-bit python. body_length = getattr( self.__body, 'length', None) or len(self.__body) self.headers['content-length'] = str(body_length) else: self.headers.pop('content-length', None) # This line ensures we don't try to print large requests. if not isinstance(value, (type(None), six.string_types)): self.loggable_body = '<media body>' # Note: currently the order of fields here is important, since we want # to be able to pass in the result from httplib2.request. class Response(collections.namedtuple( 'HttpResponse', ['info', 'content', 'request_url'])): """Class encapsulating data for an HTTP response.""" __slots__ = () def __len__(self): return self.length @property def length(self): """Return the length of this response. We expose this as an attribute since using len() directly can fail for responses larger than sys.maxint. Returns: Response length (as int or long) """ def ProcessContentRange(content_range): _, _, range_spec = content_range.partition(' ') byte_range, _, _ = range_spec.partition('/') start, _, end = byte_range.partition('-') return int(end) - int(start) + 1 if '-content-encoding' in self.info and 'content-range' in self.info: # httplib2 rewrites content-length in the case of a compressed # transfer; we can't trust the content-length header in that # case, but we can trust content-range, if it's present. return ProcessContentRange(self.info['content-range']) elif 'content-length' in self.info: return int(self.info.get('content-length')) elif 'content-range' in self.info: return ProcessContentRange(self.info['content-range']) return len(self.content) @property def status_code(self): return int(self.info['status']) @property def retry_after(self): if 'retry-after' in self.info: return int(self.info['retry-after']) @property def is_redirect(self): return (self.status_code in _REDIRECT_STATUS_CODES and 'location' in self.info) def CheckResponse(response): if response is None: # Caller shouldn't call us if the response is None, but handle anyway. raise exceptions.RequestError( 'Request to url %s did not return a response.' % response.request_url) elif (response.status_code >= 500 or response.status_code == TOO_MANY_REQUESTS): raise exceptions.BadStatusCodeError.FromResponse(response) elif response.retry_after: raise exceptions.RetryAfterError.FromResponse(response) def RebuildHttpConnections(http): """Rebuilds all http connections in the httplib2.Http instance. httplib2 overloads the map in http.connections to contain two different types of values: { scheme string: connection class } and { scheme + authority string : actual http connection } Here we remove all of the entries for actual connections so that on the next request httplib2 will rebuild them from the connection types. Args: http: An httplib2.Http instance. """ if getattr(http, 'connections', None): for conn_key in list(http.connections.keys()): if ':' in conn_key: del http.connections[conn_key] def RethrowExceptionHandler(unused_args): # pylint: disable=misplaced-bare-raise raise def HandleExceptionsAndRebuildHttpConnections(retry_args): """Exception handler for http failures. This catches known failures and rebuilds the underlying HTTP connections. Args: retry_args: An ExceptionRetryArgs tuple. """ # If the server indicates how long to wait, use that value. Otherwise, # calculate the wait time on our own. retry_after = None # Transport failures if isinstance(retry_args.exc, (http_client.BadStatusLine, http_client.IncompleteRead, http_client.ResponseNotReady)): logging.debug('Caught HTTP error %s, retrying: %s', type(retry_args.exc).__name__, retry_args.exc) elif isinstance(retry_args.exc, socket.error): logging.debug('Caught socket error, retrying: %s', retry_args.exc) elif isinstance(retry_args.exc, socket.gaierror): logging.debug( 'Caught socket address error, retrying: %s', retry_args.exc) elif isinstance(retry_args.exc, socket.timeout): logging.debug( 'Caught socket timeout error, retrying: %s', retry_args.exc) elif isinstance(retry_args.exc, httplib2.ServerNotFoundError): logging.debug( 'Caught server not found error, retrying: %s', retry_args.exc) elif isinstance(retry_args.exc, ValueError): # oauth2client tries to JSON-decode the response, which can result # in a ValueError if the response was invalid. Until that is fixed in # oauth2client, need to handle it here. logging.debug('Response content was invalid (%s), retrying', retry_args.exc) elif (isinstance(retry_args.exc, TokenRefreshError) and hasattr(retry_args.exc, 'status') and (retry_args.exc.status == TOO_MANY_REQUESTS or retry_args.exc.status >= 500)): logging.debug( 'Caught transient credential refresh error (%s), retrying', retry_args.exc) elif isinstance(retry_args.exc, exceptions.RequestError): logging.debug('Request returned no response, retrying') # API-level failures elif isinstance(retry_args.exc, exceptions.BadStatusCodeError): logging.debug('Response returned status %s, retrying', retry_args.exc.status_code) elif isinstance(retry_args.exc, exceptions.RetryAfterError): logging.debug('Response returned a retry-after header, retrying') retry_after = retry_args.exc.retry_after else: raise retry_args.exc RebuildHttpConnections(retry_args.http) logging.debug('Retrying request to url %s after exception %s', retry_args.http_request.url, retry_args.exc) time.sleep( retry_after or util.CalculateWaitForRetry( retry_args.num_retries, max_wait=retry_args.max_retry_wait)) def MakeRequest(http, http_request, retries=7, max_retry_wait=60, redirections=5, retry_func=HandleExceptionsAndRebuildHttpConnections, check_response_func=CheckResponse): """Send http_request via the given http, performing error/retry handling. Args: http: An httplib2.Http instance, or a http multiplexer that delegates to an underlying http, for example, HTTPMultiplexer. http_request: A Request to send. retries: (int, default 7) Number of retries to attempt on retryable replies (such as 429 or 5XX). max_retry_wait: (int, default 60) Maximum number of seconds to wait when retrying. redirections: (int, default 5) Number of redirects to follow. retry_func: Function to handle retries on exceptions. Argument is an ExceptionRetryArgs tuple. check_response_func: Function to validate the HTTP response. Arguments are (Response, response content, url). Raises: InvalidDataFromServerError: if there is no response after retries. Returns: A Response object. """ retry = 0 first_req_time = time.time() # Provide compatibility for breaking change in httplib2 0.16.0+: # https://github.com/googleapis/google-api-python-client/issues/803 if hasattr(http, 'redirect_codes'): http.redirect_codes = set(http.redirect_codes) - {308} while True: try: return _MakeRequestNoRetry( http, http_request, redirections=redirections, check_response_func=check_response_func) # retry_func will consume the exception types it handles and raise. # pylint: disable=broad-except except Exception as e: retry += 1 if retry >= retries: raise else: total_wait_sec = time.time() - first_req_time retry_func(ExceptionRetryArgs(http, http_request, e, retry, max_retry_wait, total_wait_sec)) def _MakeRequestNoRetry(http, http_request, redirections=5, check_response_func=CheckResponse): """Send http_request via the given http. This wrapper exists to handle translation between the plain httplib2 request/response types and the Request and Response types above. Args: http: An httplib2.Http instance, or a http multiplexer that delegates to an underlying http, for example, HTTPMultiplexer. http_request: A Request to send. redirections: (int, default 5) Number of redirects to follow. check_response_func: Function to validate the HTTP response. Arguments are (Response, response content, url). Returns: A Response object. Raises: RequestError if no response could be parsed. """ connection_type = None # Handle overrides for connection types. This is used if the caller # wants control over the underlying connection for managing callbacks # or hash digestion. if getattr(http, 'connections', None): url_scheme = parse.urlsplit(http_request.url).scheme if url_scheme and url_scheme in http.connections: connection_type = http.connections[url_scheme] # Custom printing only at debuglevel 4 new_debuglevel = 4 if httplib2.debuglevel == 4 else 0 with _Httplib2Debuglevel(http_request, new_debuglevel, http=http): info, content = http.request( str(http_request.url), method=str(http_request.http_method), body=http_request.body, headers=http_request.headers, redirections=redirections, connection_type=connection_type) if info is None: raise exceptions.RequestError() response = Response(info, content, http_request.url) check_response_func(response) return response _HTTP_FACTORIES = [] def _RegisterHttpFactory(factory): _HTTP_FACTORIES.append(factory) def GetHttp(kwds): for factory in _HTTP_FACTORIES: http = factory(kwds) if http is not None: return http return httplib2.Http(*kwds)	15.09 KB	2025-03-11 18:02
📄 http_wrapper_test.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. """Tests for http_wrapper.""" import socket import unittest import httplib2 from six.moves import http_client from mock import patch from apitools.base.py import exceptions from apitools.base.py import http_wrapper # pylint: disable=ungrouped-imports try: from oauth2client.client import HttpAccessTokenRefreshError from oauth2client.client import AccessTokenRefreshError _TOKEN_REFRESH_STATUS_AVAILABLE = True except ImportError: from oauth2client.client import AccessTokenRefreshError _TOKEN_REFRESH_STATUS_AVAILABLE = False class _MockHttpRequest(object): url = None class _MockHttpResponse(object): def __init__(self, status_code): self.response = {'status': status_code} class RaisesExceptionOnLen(object): """Supports length property but raises if __len__ is used.""" def __len__(self): raise Exception('len() called unnecessarily') def length(self): return 1 class HttpWrapperTest(unittest.TestCase): def testRequestBodyUsesLengthProperty(self): http_wrapper.Request(body=RaisesExceptionOnLen()) def testRequestBodyWithLen(self): http_wrapper.Request(body='burrito') @unittest.skipIf(not _TOKEN_REFRESH_STATUS_AVAILABLE, 'oauth2client<1.5 lacks HttpAccessTokenRefreshError.') def testExceptionHandlerHttpAccessTokenError(self): exception_arg = HttpAccessTokenRefreshError(status=503) retry_args = http_wrapper.ExceptionRetryArgs( http={'connections': {}}, http_request=_MockHttpRequest(), exc=exception_arg, num_retries=0, max_retry_wait=0, total_wait_sec=0) # Disable time.sleep for this handler as it is called with # a minimum value of 1 second. with patch('time.sleep', return_value=None): http_wrapper.HandleExceptionsAndRebuildHttpConnections( retry_args) @unittest.skipIf(not _TOKEN_REFRESH_STATUS_AVAILABLE, 'oauth2client<1.5 lacks HttpAccessTokenRefreshError.') def testExceptionHandlerHttpAccessTokenErrorRaises(self): exception_arg = HttpAccessTokenRefreshError(status=200) retry_args = http_wrapper.ExceptionRetryArgs( http={'connections': {}}, http_request=_MockHttpRequest(), exc=exception_arg, num_retries=0, max_retry_wait=0, total_wait_sec=0) # Disable time.sleep for this handler as it is called with # a minimum value of 1 second. with self.assertRaises(HttpAccessTokenRefreshError): with patch('time.sleep', return_value=None): http_wrapper.HandleExceptionsAndRebuildHttpConnections( retry_args) def testExceptionHandlerAccessTokenErrorRaises(self): exception_arg = AccessTokenRefreshError() retry_args = http_wrapper.ExceptionRetryArgs( http={'connections': {}}, http_request=_MockHttpRequest(), exc=exception_arg, num_retries=0, max_retry_wait=0, total_wait_sec=0) # Disable time.sleep for this handler as it is called with # a minimum value of 1 second. with self.assertRaises(AccessTokenRefreshError): with patch('time.sleep', return_value=None): http_wrapper.HandleExceptionsAndRebuildHttpConnections( retry_args) def testDefaultExceptionHandler(self): """Ensures exception handles swallows (retries)""" mock_http_content = 'content'.encode('utf8') for exception_arg in ( http_client.BadStatusLine('line'), http_client.IncompleteRead('partial'), http_client.ResponseNotReady(), socket.error(), socket.gaierror(), httplib2.ServerNotFoundError(), ValueError(), exceptions.RequestError(), exceptions.BadStatusCodeError( {'status': 503}, mock_http_content, 'url'), exceptions.RetryAfterError( {'status': 429}, mock_http_content, 'url', 0)): retry_args = http_wrapper.ExceptionRetryArgs( http={'connections': {}}, http_request=_MockHttpRequest(), exc=exception_arg, num_retries=0, max_retry_wait=0, total_wait_sec=0) # Disable time.sleep for this handler as it is called with # a minimum value of 1 second. with patch('time.sleep', return_value=None): http_wrapper.HandleExceptionsAndRebuildHttpConnections( retry_args)	5.09 KB	2025-03-11 18:02
📄 list_pager.py #!/usr/bin/env python # # 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. """A helper function that executes a series of List queries for many APIs.""" from apitools.base.py import encoding import six __all__ = [ 'YieldFromList', ] def _GetattrNested(message, attribute): """Gets a possibly nested attribute. Same as getattr() if attribute is a string; if attribute is a tuple, returns the nested attribute referred to by the fields in the tuple as if they were a dotted accessor path. (ex _GetattrNested(msg, ('foo', 'bar', 'baz')) gets msg.foo.bar.baz """ if isinstance(attribute, six.string_types): return getattr(message, attribute) elif len(attribute) == 0: return message else: return _GetattrNested(getattr(message, attribute[0]), attribute[1:]) def _SetattrNested(message, attribute, value): """Sets a possibly nested attribute. Same as setattr() if attribute is a string; if attribute is a tuple, sets the nested attribute referred to by the fields in the tuple as if they were a dotted accessor path. (ex _SetattrNested(msg, ('foo', 'bar', 'baz'), 'v') sets msg.foo.bar.baz to 'v' """ if isinstance(attribute, six.string_types): return setattr(message, attribute, value) elif len(attribute) < 1: raise ValueError("Need an attribute to set") elif len(attribute) == 1: return setattr(message, attribute[0], value) else: return setattr(_GetattrNested(message, attribute[:-1]), attribute[-1], value) def YieldFromList( service, request, global_params=None, limit=None, batch_size=100, method='List', field='items', predicate=None, current_token_attribute='pageToken', next_token_attribute='nextPageToken', batch_size_attribute='maxResults', get_field_func=_GetattrNested): """Make a series of List requests, keeping track of page tokens. Args: service: apitools_base.BaseApiService, A service with a .List() method. request: protorpc.messages.Message, The request message corresponding to the service's .List() method, with all the attributes populated except the .maxResults and .pageToken attributes. global_params: protorpc.messages.Message, The global query parameters to provide when calling the given method. limit: int, The maximum number of records to yield. None if all available records should be yielded. batch_size: int, The number of items to retrieve per request. method: str, The name of the method used to fetch resources. field: str, The field in the response that will be a list of items. predicate: lambda, A function that returns true for items to be yielded. current_token_attribute: str or tuple, The name of the attribute in a request message holding the page token for the page being requested. If a tuple, path to attribute. next_token_attribute: str or tuple, The name of the attribute in a response message holding the page token for the next page. If a tuple, path to the attribute. batch_size_attribute: str or tuple, The name of the attribute in a response message holding the maximum number of results to be returned. None if caller-specified batch size is unsupported. If a tuple, path to the attribute. get_field_func: Function that returns the items to be yielded. Argument is response message, and field. Yields: protorpc.message.Message, The resources listed by the service. """ request = encoding.CopyProtoMessage(request) _SetattrNested(request, current_token_attribute, None) while limit is None or limit: if batch_size_attribute: # On Py3, None is not comparable so min() below will fail. # On Py2, None is always less than any number so if batch_size # is None, the request_batch_size will always be None regardless # of the value of limit. This doesn't generally strike me as the # correct behavior, but this change preserves the existing Py2 # behavior on Py3. if batch_size is None: request_batch_size = None else: request_batch_size = min(batch_size, limit or batch_size) _SetattrNested(request, batch_size_attribute, request_batch_size) response = getattr(service, method)(request, global_params=global_params) items = get_field_func(response, field) if predicate: items = list(filter(predicate, items)) for item in items: yield item if limit is None: continue limit -= 1 if not limit: return token = _GetattrNested(response, next_token_attribute) if not token: return _SetattrNested(request, current_token_attribute, token)	5.49 KB	2025-03-11 18:02
📄 list_pager_test.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. """Tests for list_pager.""" import unittest from apitools.base.py import list_pager from apitools.base.py.testing import mock from samples.fusiontables_sample.fusiontables_v1 \ import fusiontables_v1_client as fusiontables from samples.fusiontables_sample.fusiontables_v1 \ import fusiontables_v1_messages as messages from samples.iam_sample.iam_v1 import iam_v1_client as iam_client from samples.iam_sample.iam_v1 import iam_v1_messages as iam_messages class Example(object): def __init__(self): self.a = 'aaa' self.b = 'bbb' self.c = 'ccc' class GetterSetterTest(unittest.TestCase): def testGetattrNested(self): o = Example() self.assertEqual(list_pager._GetattrNested(o, 'a'), 'aaa') self.assertEqual(list_pager._GetattrNested(o, ('a',)), 'aaa') o.b = Example() self.assertEqual(list_pager._GetattrNested(o, ('b', 'c')), 'ccc') def testSetattrNested(self): o = Example() list_pager._SetattrNested(o, 'b', Example()) self.assertEqual(o.b.a, 'aaa') list_pager._SetattrNested(o, ('b', 'a'), 'AAA') self.assertEqual(o.b.a, 'AAA') list_pager._SetattrNested(o, ('c',), 'CCC') self.assertEqual(o.c, 'CCC') class ListPagerTest(unittest.TestCase): def _AssertInstanceSequence(self, results, n): counter = 0 for instance in results: self.assertEqual(instance.name, 'c' + str(counter)) counter += 1 self.assertEqual(counter, n) def setUp(self): self.mocked_client = mock.Client(fusiontables.FusiontablesV1) self.mocked_client.Mock() self.addCleanup(self.mocked_client.Unmock) def testYieldFromList(self): self.mocked_client.column.List.Expect( messages.FusiontablesColumnListRequest( maxResults=100, pageToken=None, tableId='mytable', ), messages.ColumnList( items=[ messages.Column(name='c0'), messages.Column(name='c1'), messages.Column(name='c2'), messages.Column(name='c3'), ], nextPageToken='x', )) self.mocked_client.column.List.Expect( messages.FusiontablesColumnListRequest( maxResults=100, pageToken='x', tableId='mytable', ), messages.ColumnList( items=[ messages.Column(name='c4'), messages.Column(name='c5'), messages.Column(name='c6'), messages.Column(name='c7'), ], )) client = fusiontables.FusiontablesV1(get_credentials=False) request = messages.FusiontablesColumnListRequest(tableId='mytable') results = list_pager.YieldFromList(client.column, request) self._AssertInstanceSequence(results, 8) def testYieldNoRecords(self): client = fusiontables.FusiontablesV1(get_credentials=False) request = messages.FusiontablesColumnListRequest(tableId='mytable') results = list_pager.YieldFromList(client.column, request, limit=False) self.assertEqual(0, len(list(results))) def testYieldFromListPartial(self): self.mocked_client.column.List.Expect( messages.FusiontablesColumnListRequest( maxResults=6, pageToken=None, tableId='mytable', ), messages.ColumnList( items=[ messages.Column(name='c0'), messages.Column(name='c1'), messages.Column(name='c2'), messages.Column(name='c3'), ], nextPageToken='x', )) self.mocked_client.column.List.Expect( messages.FusiontablesColumnListRequest( maxResults=2, pageToken='x', tableId='mytable', ), messages.ColumnList( items=[ messages.Column(name='c4'), messages.Column(name='c5'), messages.Column(name='c6'), messages.Column(name='c7'), ], )) client = fusiontables.FusiontablesV1(get_credentials=False) request = messages.FusiontablesColumnListRequest(tableId='mytable') results = list_pager.YieldFromList(client.column, request, limit=6) self._AssertInstanceSequence(results, 6) def testYieldFromListPaging(self): self.mocked_client.column.List.Expect( messages.FusiontablesColumnListRequest( maxResults=5, pageToken=None, tableId='mytable', ), messages.ColumnList( items=[ messages.Column(name='c0'), messages.Column(name='c1'), messages.Column(name='c2'), messages.Column(name='c3'), messages.Column(name='c4'), ], nextPageToken='x', )) self.mocked_client.column.List.Expect( messages.FusiontablesColumnListRequest( maxResults=4, pageToken='x', tableId='mytable', ), messages.ColumnList( items=[ messages.Column(name='c5'), messages.Column(name='c6'), messages.Column(name='c7'), messages.Column(name='c8'), ], )) client = fusiontables.FusiontablesV1(get_credentials=False) request = messages.FusiontablesColumnListRequest(tableId='mytable') results = list_pager.YieldFromList(client.column, request, limit=9, batch_size=5) self._AssertInstanceSequence(results, 9) def testYieldFromListBatchSizeNone(self): self.mocked_client.column.List.Expect( messages.FusiontablesColumnListRequest( maxResults=None, pageToken=None, tableId='mytable', ), messages.ColumnList( items=[ messages.Column(name='c0'), messages.Column(name='c1'), messages.Column(name='c2'), messages.Column(name='c3'), messages.Column(name='c4'), messages.Column(name='c5'), messages.Column(name='c6'), ], nextPageToken='x', )) client = fusiontables.FusiontablesV1(get_credentials=False) request = messages.FusiontablesColumnListRequest(tableId='mytable') results = list_pager.YieldFromList(client.column, request, limit=5, batch_size=None) self._AssertInstanceSequence(results, 5) def testYieldFromListEmpty(self): self.mocked_client.column.List.Expect( messages.FusiontablesColumnListRequest( maxResults=6, pageToken=None, tableId='mytable', ), messages.ColumnList()) client = fusiontables.FusiontablesV1(get_credentials=False) request = messages.FusiontablesColumnListRequest(tableId='mytable') results = list_pager.YieldFromList(client.column, request, limit=6) self._AssertInstanceSequence(results, 0) def testYieldFromListWithPredicate(self): self.mocked_client.column.List.Expect( messages.FusiontablesColumnListRequest( maxResults=100, pageToken=None, tableId='mytable', ), messages.ColumnList( items=[ messages.Column(name='c0'), messages.Column(name='bad0'), messages.Column(name='c1'), messages.Column(name='bad1'), ], nextPageToken='x', )) self.mocked_client.column.List.Expect( messages.FusiontablesColumnListRequest( maxResults=100, pageToken='x', tableId='mytable', ), messages.ColumnList( items=[ messages.Column(name='c2'), ], )) client = fusiontables.FusiontablesV1(get_credentials=False) request = messages.FusiontablesColumnListRequest(tableId='mytable') results = list_pager.YieldFromList( client.column, request, predicate=lambda x: 'c' in x.name) self._AssertInstanceSequence(results, 3) def testYieldFromListWithCustomGetFieldFunction(self): self.mocked_client.column.List.Expect( messages.FusiontablesColumnListRequest( maxResults=100, pageToken=None, tableId='mytable', ), messages.ColumnList( items=[ messages.Column(name='c0') ] )) custom_getter_called = [] def Custom_Getter(message, attribute): custom_getter_called.append(True) return getattr(message, attribute) client = fusiontables.FusiontablesV1(get_credentials=False) request = messages.FusiontablesColumnListRequest(tableId='mytable') results = list_pager.YieldFromList( client.column, request, get_field_func=Custom_Getter) self._AssertInstanceSequence(results, 1) self.assertEqual(1, len(custom_getter_called)) class ListPagerAttributeTest(unittest.TestCase): def setUp(self): self.mocked_client = mock.Client(iam_client.IamV1) self.mocked_client.Mock() self.addCleanup(self.mocked_client.Unmock) def testYieldFromListWithAttributes(self): self.mocked_client.iamPolicies.GetPolicyDetails.Expect( iam_messages.GetPolicyDetailsRequest( pageSize=100, pageToken=None, fullResourcePath='myresource', ), iam_messages.GetPolicyDetailsResponse( policies=[ iam_messages.PolicyDetail(fullResourcePath='c0'), iam_messages.PolicyDetail(fullResourcePath='c1'), ], nextPageToken='x', )) self.mocked_client.iamPolicies.GetPolicyDetails.Expect( iam_messages.GetPolicyDetailsRequest( pageSize=100, pageToken='x', fullResourcePath='myresource', ), iam_messages.GetPolicyDetailsResponse( policies=[ iam_messages.PolicyDetail(fullResourcePath='c2'), ], )) client = iam_client.IamV1(get_credentials=False) request = iam_messages.GetPolicyDetailsRequest( fullResourcePath='myresource') results = list_pager.YieldFromList( client.iamPolicies, request, batch_size_attribute='pageSize', method='GetPolicyDetails', field='policies') i = 0 for i, instance in enumerate(results): self.assertEqual('c{0}'.format(i), instance.fullResourcePath) self.assertEqual(2, i) def testYieldFromListWithNoBatchSizeAttribute(self): self.mocked_client.iamPolicies.GetPolicyDetails.Expect( iam_messages.GetPolicyDetailsRequest( pageToken=None, fullResourcePath='myresource', ), iam_messages.GetPolicyDetailsResponse( policies=[ iam_messages.PolicyDetail(fullResourcePath='c0'), iam_messages.PolicyDetail(fullResourcePath='c1'), ], )) client = iam_client.IamV1(get_credentials=False) request = iam_messages.GetPolicyDetailsRequest( fullResourcePath='myresource') results = list_pager.YieldFromList( client.iamPolicies, request, batch_size_attribute=None, method='GetPolicyDetails', field='policies') i = 0 for i, instance in enumerate(results): self.assertEqual('c{0}'.format(i), instance.fullResourcePath) self.assertEqual(1, i)	13.08 KB	2025-03-11 18:02
📄 stream_slice.py #!/usr/bin/env python # # 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. """Small helper class to provide a small slice of a stream.""" from apitools.base.py import exceptions class StreamSlice(object): """Provides a slice-like object for streams.""" def __init__(self, stream, max_bytes): self.__stream = stream self.__remaining_bytes = max_bytes self.__max_bytes = max_bytes def __str__(self): return 'Slice of stream %s with %s/%s bytes not yet read' % ( self.__stream, self.__remaining_bytes, self.__max_bytes) def __len__(self): return self.__max_bytes def __nonzero__(self): # For 32-bit python2.x, len() cannot exceed a 32-bit number; avoid # accidental len() calls from httplib in the form of "if this_object:". return bool(self.__max_bytes) @property def length(self): # For 32-bit python2.x, len() cannot exceed a 32-bit number. return self.__max_bytes def read(self, size=None): # pylint: disable=missing-docstring """Read at most size bytes from this slice. Compared to other streams, there is one case where we may unexpectedly raise an exception on read: if the underlying stream is exhausted (i.e. returns no bytes on read), and the size of this slice indicates we should still be able to read more bytes, we raise exceptions.StreamExhausted. Args: size: If provided, read no more than size bytes from the stream. Returns: The bytes read from this slice. Raises: exceptions.StreamExhausted """ if size is not None: read_size = min(size, self.__remaining_bytes) else: read_size = self.__remaining_bytes data = self.__stream.read(read_size) if read_size > 0 and not data: raise exceptions.StreamExhausted( 'Not enough bytes in stream; expected %d, exhausted ' 'after %d' % ( self.__max_bytes, self.__max_bytes - self.__remaining_bytes)) self.__remaining_bytes -= len(data) return data	2.67 KB	2025-03-11 18:02
📄 stream_slice_test.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. """Tests for stream_slice.""" import string import unittest import six from apitools.base.py import exceptions from apitools.base.py import stream_slice class StreamSliceTest(unittest.TestCase): def setUp(self): self.stream = six.StringIO(string.ascii_letters) self.value = self.stream.getvalue() self.stream.seek(0) def testSimpleSlice(self): ss = stream_slice.StreamSlice(self.stream, 10) self.assertEqual('', ss.read(0)) self.assertEqual(self.value[0:3], ss.read(3)) self.assertIn('7/10', str(ss)) self.assertEqual(self.value[3:10], ss.read()) self.assertEqual('', ss.read()) self.assertEqual('', ss.read(10)) self.assertEqual(10, self.stream.tell()) def testEmptySlice(self): ss = stream_slice.StreamSlice(self.stream, 0) self.assertEqual('', ss.read(5)) self.assertEqual('', ss.read()) self.assertEqual(0, self.stream.tell()) def testOffsetStream(self): self.stream.seek(26) ss = stream_slice.StreamSlice(self.stream, 26) self.assertEqual(self.value[26:36], ss.read(10)) self.assertEqual(self.value[36:], ss.read()) self.assertEqual('', ss.read()) def testTooShortStream(self): ss = stream_slice.StreamSlice(self.stream, 1000) self.assertEqual(self.value, ss.read()) self.assertEqual('', ss.read(0)) with self.assertRaises(exceptions.StreamExhausted) as e: ss.read() with self.assertRaises(exceptions.StreamExhausted) as e: ss.read(10) self.assertIn('exhausted after %d' % len(self.value), str(e.exception))	2.2 KB	2025-03-11 18:02
📁 testing	-	2025-03-11 18:02
📄 transfer.py #!/usr/bin/env python # # 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. """Upload and download support for apitools.""" from __future__ import print_function import email.generator as email_generator import email.mime.multipart as mime_multipart import email.mime.nonmultipart as mime_nonmultipart import io import json import mimetypes import os import threading import six from six.moves import http_client from apitools.base.py import buffered_stream from apitools.base.py import compression from apitools.base.py import exceptions from apitools.base.py import http_wrapper from apitools.base.py import stream_slice from apitools.base.py import util __all__ = [ 'Download', 'Upload', 'RESUMABLE_UPLOAD', 'SIMPLE_UPLOAD', 'DownloadProgressPrinter', 'DownloadCompletePrinter', 'UploadProgressPrinter', 'UploadCompletePrinter', ] _RESUMABLE_UPLOAD_THRESHOLD = 5 << 20 SIMPLE_UPLOAD = 'simple' RESUMABLE_UPLOAD = 'resumable' def DownloadProgressPrinter(response, unused_download): """Print download progress based on response.""" if 'content-range' in response.info: print('Received %s' % response.info['content-range']) else: print('Received %d bytes' % response.length) def DownloadCompletePrinter(unused_response, unused_download): """Print information about a completed download.""" print('Download complete') def UploadProgressPrinter(response, unused_upload): """Print upload progress based on response.""" print('Sent %s' % response.info['range']) def UploadCompletePrinter(unused_response, unused_upload): """Print information about a completed upload.""" print('Upload complete') class _Transfer(object): """Generic bits common to Uploads and Downloads.""" def __init__(self, stream, close_stream=False, chunksize=None, auto_transfer=True, http=None, num_retries=5): self.__bytes_http = None self.__close_stream = close_stream self.__http = http self.__stream = stream self.__url = None self.__num_retries = 5 # Let the @property do validation self.num_retries = num_retries self.retry_func = ( http_wrapper.HandleExceptionsAndRebuildHttpConnections) self.auto_transfer = auto_transfer self.chunksize = chunksize or 1048576 def __repr__(self): return str(self) @property def close_stream(self): return self.__close_stream @property def http(self): return self.__http @property def bytes_http(self): return self.__bytes_http or self.http @bytes_http.setter def bytes_http(self, value): self.__bytes_http = value @property def num_retries(self): return self.__num_retries @num_retries.setter def num_retries(self, value): util.Typecheck(value, six.integer_types) if value < 0: raise exceptions.InvalidDataError( 'Cannot have negative value for num_retries') self.__num_retries = value @property def stream(self): return self.__stream @property def url(self): return self.__url def _Initialize(self, http, url): """Initialize this download by setting self.http and self.url. We want the user to be able to override self.http by having set the value in the constructor; in that case, we ignore the provided http. Args: http: An httplib2.Http instance or None. url: The url for this transfer. Returns: None. Initializes self. """ self.EnsureUninitialized() if self.http is None: self.__http = http or http_wrapper.GetHttp() self.__url = url @property def initialized(self): return self.url is not None and self.http is not None @property def _type_name(self): return type(self).__name__ def EnsureInitialized(self): if not self.initialized: raise exceptions.TransferInvalidError( 'Cannot use uninitialized %s' % self._type_name) def EnsureUninitialized(self): if self.initialized: raise exceptions.TransferInvalidError( 'Cannot re-initialize %s' % self._type_name) def __del__(self): if self.__close_stream: self.__stream.close() def _ExecuteCallback(self, callback, response): # TODO(craigcitro): Push these into a queue. if callback is not None: threading.Thread(target=callback, args=(response, self)).start() class Download(_Transfer): """Data for a single download. Public attributes: chunksize: default chunksize to use for transfers. """ _ACCEPTABLE_STATUSES = set(( http_client.OK, http_client.NO_CONTENT, http_client.PARTIAL_CONTENT, http_client.REQUESTED_RANGE_NOT_SATISFIABLE, )) _REQUIRED_SERIALIZATION_KEYS = set(( 'auto_transfer', 'progress', 'total_size', 'url')) def __init__(self, stream, progress_callback=None, finish_callback=None, kwds): total_size = kwds.pop('total_size', None) super(Download, self).__init__(stream, kwds) self.__initial_response = None self.__progress = 0 self.__total_size = total_size self.__encoding = None self.progress_callback = progress_callback self.finish_callback = finish_callback @property def progress(self): return self.__progress @property def encoding(self): return self.__encoding @classmethod def FromFile(cls, filename, overwrite=False, auto_transfer=True, kwds): """Create a new download object from a filename.""" path = os.path.expanduser(filename) if os.path.exists(path) and not overwrite: raise exceptions.InvalidUserInputError( 'File %s exists and overwrite not specified' % path) return cls(open(path, 'wb'), close_stream=True, auto_transfer=auto_transfer, kwds) @classmethod def FromStream(cls, stream, auto_transfer=True, total_size=None, kwds): """Create a new Download object from a stream.""" return cls(stream, auto_transfer=auto_transfer, total_size=total_size, kwds) @classmethod def FromData(cls, stream, json_data, http=None, auto_transfer=None, client=None, kwds): """Create a new Download object from a stream and serialized data.""" info = json.loads(json_data) missing_keys = cls._REQUIRED_SERIALIZATION_KEYS - set(info.keys()) if missing_keys: raise exceptions.InvalidDataError( 'Invalid serialization data, missing keys: %s' % ( ', '.join(missing_keys))) download = cls.FromStream(stream, kwds) if auto_transfer is not None: download.auto_transfer = auto_transfer else: download.auto_transfer = info['auto_transfer'] if client is not None: url = client.FinalizeTransferUrl(info['url']) else: url = info['url'] setattr(download, '_Download__progress', info['progress']) setattr(download, '_Download__total_size', info['total_size']) download._Initialize( # pylint: disable=protected-access http, url) return download @property def serialization_data(self): self.EnsureInitialized() return { 'auto_transfer': self.auto_transfer, 'progress': self.progress, 'total_size': self.total_size, 'url': self.url, } @property def total_size(self): return self.__total_size def __str__(self): if not self.initialized: return 'Download (uninitialized)' return 'Download with %d/%s bytes transferred from url %s' % ( self.progress, self.total_size, self.url) def ConfigureRequest(self, http_request, url_builder): url_builder.query_params['alt'] = 'media' # TODO(craigcitro): We need to send range requests because by # default httplib2 stores entire reponses in memory. Override # httplib2's download method (as gsutil does) so that this is not # necessary. http_request.headers['Range'] = 'bytes=0-%d' % (self.chunksize - 1,) def __SetTotal(self, info): """Sets the total size based off info if possible otherwise 0.""" if 'content-range' in info: _, _, total = info['content-range'].rpartition('/') if total != '': self.__total_size = int(total) # Note "total_size is None" means we don't know it; if no size # info was returned on our initial range request, that means we # have a 0-byte file. (That last statement has been verified # empirically, but is not clearly documented anywhere.) if self.total_size is None: self.__total_size = 0 def InitializeDownload(self, http_request, http=None, client=None): """Initialize this download by making a request. Args: http_request: The HttpRequest to use to initialize this download. http: The httplib2.Http instance for this request. client: If provided, let this client process the final URL before sending any additional requests. If client is provided and http is not, client.http will be used instead. """ self.EnsureUninitialized() if http is None and client is None: raise exceptions.UserError('Must provide client or http.') http = http or client.http if client is not None: http_request.url = client.FinalizeTransferUrl(http_request.url) url = http_request.url if self.auto_transfer: end_byte = self.__ComputeEndByte(0) self.__SetRangeHeader(http_request, 0, end_byte) response = http_wrapper.MakeRequest( self.bytes_http or http, http_request) if response.status_code not in self._ACCEPTABLE_STATUSES: raise exceptions.HttpError.FromResponse(response) self.__initial_response = response self.__SetTotal(response.info) url = response.info.get('content-location', response.request_url) if client is not None: url = client.FinalizeTransferUrl(url) self._Initialize(http, url) # Unless the user has requested otherwise, we want to just # go ahead and pump the bytes now. if self.auto_transfer: self.StreamInChunks() def __NormalizeStartEnd(self, start, end=None): """Normalizes start and end values based on total size.""" if end is not None: if start < 0: raise exceptions.TransferInvalidError( 'Cannot have end index with negative start index ' + '[start=%d, end=%d]' % (start, end)) elif start >= self.total_size: raise exceptions.TransferInvalidError( 'Cannot have start index greater than total size ' + '[start=%d, total_size=%d]' % (start, self.total_size)) end = min(end, self.total_size - 1) if end < start: raise exceptions.TransferInvalidError( 'Range requested with end[%s] < start[%s]' % (end, start)) return start, end else: if start < 0: start = max(0, start + self.total_size) return start, self.total_size - 1 def __SetRangeHeader(self, request, start, end=None): if start < 0: request.headers['range'] = 'bytes=%d' % start elif end is None or end < start: request.headers['range'] = 'bytes=%d-' % start else: request.headers['range'] = 'bytes=%d-%d' % (start, end) def __ComputeEndByte(self, start, end=None, use_chunks=True): """Compute the last byte to fetch for this request. This is all based on the HTTP spec for Range and Content-Range. Note that this is potentially confusing in several ways: the value for the last byte is 0-based, eg "fetch 10 bytes from the beginning" would return 9 here. * if we have no information about size, and don't want to use the chunksize, we'll return None. See the tests for more examples. Args: start: byte to start at. end: (int or None, default: None) Suggested last byte. use_chunks: (bool, default: True) If False, ignore self.chunksize. Returns: Last byte to use in a Range header, or None. """ end_byte = end if start < 0 and not self.total_size: return end_byte if use_chunks: alternate = start + self.chunksize - 1 if end_byte is not None: end_byte = min(end_byte, alternate) else: end_byte = alternate if self.total_size: alternate = self.total_size - 1 if end_byte is not None: end_byte = min(end_byte, alternate) else: end_byte = alternate return end_byte def __GetChunk(self, start, end, additional_headers=None): """Retrieve a chunk, and return the full response.""" self.EnsureInitialized() request = http_wrapper.Request(url=self.url) self.__SetRangeHeader(request, start, end=end) if additional_headers is not None: request.headers.update(additional_headers) return http_wrapper.MakeRequest( self.bytes_http, request, retry_func=self.retry_func, retries=self.num_retries) def __ProcessResponse(self, response): """Process response (by updating self and writing to self.stream).""" if response.status_code not in self._ACCEPTABLE_STATUSES: # We distinguish errors that mean we made a mistake in setting # up the transfer versus something we should attempt again. if response.status_code in (http_client.FORBIDDEN, http_client.NOT_FOUND): raise exceptions.HttpError.FromResponse(response) else: raise exceptions.TransferRetryError(response.content) if response.status_code in (http_client.OK, http_client.PARTIAL_CONTENT): try: self.stream.write(six.ensure_binary(response.content)) except TypeError: self.stream.write(six.ensure_text(response.content)) self.__progress += response.length if response.info and 'content-encoding' in response.info: # TODO(craigcitro): Handle the case where this changes over a # download. self.__encoding = response.info['content-encoding'] elif response.status_code == http_client.NO_CONTENT: # It's important to write something to the stream for the case # of a 0-byte download to a file, as otherwise python won't # create the file. self.stream.write('') return response def GetRange(self, start, end=None, additional_headers=None, use_chunks=True): """Retrieve a given byte range from this download, inclusive. Range must be of one of these three forms: * 0 <= start, end = None: Fetch from start to the end of the file. * 0 <= start <= end: Fetch the bytes from start to end. * start < 0, end = None: Fetch the last -start bytes of the file. (These variations correspond to those described in the HTTP 1.1 protocol for range headers in RFC 2616, sec. 14.35.1.) Args: start: (int) Where to start fetching bytes. (See above.) end: (int, optional) Where to stop fetching bytes. (See above.) additional_headers: (bool, optional) Any additional headers to pass with the request. use_chunks: (bool, default: True) If False, ignore self.chunksize and fetch this range in a single request. Returns: None. Streams bytes into self.stream. """ self.EnsureInitialized() progress_end_normalized = False if self.total_size is not None: progress, end_byte = self.__NormalizeStartEnd(start, end) progress_end_normalized = True else: progress = start end_byte = end while (not progress_end_normalized or end_byte is None or progress <= end_byte): end_byte = self.__ComputeEndByte(progress, end=end_byte, use_chunks=use_chunks) response = self.__GetChunk(progress, end_byte, additional_headers=additional_headers) if not progress_end_normalized: self.__SetTotal(response.info) progress, end_byte = self.__NormalizeStartEnd(start, end) progress_end_normalized = True response = self.__ProcessResponse(response) progress += response.length if response.length == 0: if response.status_code == http_client.OK: # There can legitimately be no Content-Length header sent # in some cases (e.g., when there's a Transfer-Encoding # header) and if this was a 200 response (as opposed to # 206 Partial Content) we know we're done now without # looping further on received length. return raise exceptions.TransferRetryError( 'Zero bytes unexpectedly returned in download response') def StreamInChunks(self, callback=None, finish_callback=None, additional_headers=None): """Stream the entire download in chunks.""" self.StreamMedia(callback=callback, finish_callback=finish_callback, additional_headers=additional_headers, use_chunks=True) def StreamMedia(self, callback=None, finish_callback=None, additional_headers=None, use_chunks=True): """Stream the entire download. Args: callback: (default: None) Callback to call as each chunk is completed. finish_callback: (default: None) Callback to call when the download is complete. additional_headers: (default: None) Additional headers to include in fetching bytes. use_chunks: (bool, default: True) If False, ignore self.chunksize and stream this download in a single request. Returns: None. Streams bytes into self.stream. """ callback = callback or self.progress_callback finish_callback = finish_callback or self.finish_callback self.EnsureInitialized() while True: if self.__initial_response is not None: response = self.__initial_response self.__initial_response = None else: end_byte = self.__ComputeEndByte(self.progress, use_chunks=use_chunks) response = self.__GetChunk( self.progress, end_byte, additional_headers=additional_headers) if self.total_size is None: self.__SetTotal(response.info) response = self.__ProcessResponse(response) self._ExecuteCallback(callback, response) if (response.status_code == http_client.OK or self.progress >= self.total_size): break self._ExecuteCallback(finish_callback, response) if six.PY3: class MultipartBytesGenerator(email_generator.BytesGenerator): """Generates a bytes Message object tree for multipart messages This is a BytesGenerator that has been modified to not attempt line termination character modification in the bytes payload. Known to work with the compat32 policy only. It may work on others, but not tested. The outfp object must accept bytes in its write method. """ def _handle_text(self, msg): # If the string has surrogates the original source was bytes, so # just write it back out. if msg._payload is None: return self.write(msg._payload) def _encode(self, s): return s.encode('ascii', 'surrogateescape') # Default body handler _writeBody = _handle_text class Upload(_Transfer): """Data for a single Upload. Fields: stream: The stream to upload. mime_type: MIME type of the upload. total_size: (optional) Total upload size for the stream. close_stream: (default: False) Whether or not we should close the stream when finished with the upload. auto_transfer: (default: True) If True, stream all bytes as soon as the upload is created. """ _REQUIRED_SERIALIZATION_KEYS = set(( 'auto_transfer', 'mime_type', 'total_size', 'url')) def __init__(self, stream, mime_type, total_size=None, http=None, close_stream=False, chunksize=None, auto_transfer=True, progress_callback=None, finish_callback=None, gzip_encoded=False, kwds): super(Upload, self).__init__( stream, close_stream=close_stream, chunksize=chunksize, auto_transfer=auto_transfer, http=http, kwds) self.__complete = False self.__final_response = None self.__mime_type = mime_type self.__progress = 0 self.__server_chunk_granularity = None self.__strategy = None self.__total_size = None self.__gzip_encoded = gzip_encoded self.progress_callback = progress_callback self.finish_callback = finish_callback self.total_size = total_size @property def progress(self): return self.__progress @classmethod def FromFile(cls, filename, mime_type=None, auto_transfer=True, gzip_encoded=False, kwds): """Create a new Upload object from a filename.""" path = os.path.expanduser(filename) if not os.path.exists(path): raise exceptions.NotFoundError('Could not find file %s' % path) if not mime_type: mime_type, _ = mimetypes.guess_type(path) if mime_type is None: raise exceptions.InvalidUserInputError( 'Could not guess mime type for %s' % path) size = os.stat(path).st_size return cls(open(path, 'rb'), mime_type, total_size=size, close_stream=True, auto_transfer=auto_transfer, gzip_encoded=gzip_encoded, kwds) @classmethod def FromStream(cls, stream, mime_type, total_size=None, auto_transfer=True, gzip_encoded=False, kwds): """Create a new Upload object from a stream.""" if mime_type is None: raise exceptions.InvalidUserInputError( 'No mime_type specified for stream') return cls(stream, mime_type, total_size=total_size, close_stream=False, auto_transfer=auto_transfer, gzip_encoded=gzip_encoded, kwds) @classmethod def FromData(cls, stream, json_data, http, auto_transfer=None, gzip_encoded=False, client=None, kwds): """Create a new Upload of stream from serialized json_data and http.""" info = json.loads(json_data) missing_keys = cls._REQUIRED_SERIALIZATION_KEYS - set(info.keys()) if missing_keys: raise exceptions.InvalidDataError( 'Invalid serialization data, missing keys: %s' % ( ', '.join(missing_keys))) if 'total_size' in kwds: raise exceptions.InvalidUserInputError( 'Cannot override total_size on serialized Upload') upload = cls.FromStream(stream, info['mime_type'], total_size=info.get('total_size'), gzip_encoded=gzip_encoded, kwds) if isinstance(stream, io.IOBase) and not stream.seekable(): raise exceptions.InvalidUserInputError( 'Cannot restart resumable upload on non-seekable stream') if auto_transfer is not None: upload.auto_transfer = auto_transfer else: upload.auto_transfer = info['auto_transfer'] if client is not None: url = client.FinalizeTransferUrl(info['url']) else: url = info['url'] upload.strategy = RESUMABLE_UPLOAD upload._Initialize( # pylint: disable=protected-access http, url) upload.RefreshResumableUploadState() upload.EnsureInitialized() if upload.auto_transfer: upload.StreamInChunks() return upload @property def serialization_data(self): self.EnsureInitialized() if self.strategy != RESUMABLE_UPLOAD: raise exceptions.InvalidDataError( 'Serialization only supported for resumable uploads') return { 'auto_transfer': self.auto_transfer, 'mime_type': self.mime_type, 'total_size': self.total_size, 'url': self.url, } @property def complete(self): return self.__complete @property def mime_type(self): return self.__mime_type def __str__(self): if not self.initialized: return 'Upload (uninitialized)' return 'Upload with %d/%s bytes transferred for url %s' % ( self.progress, self.total_size or '???', self.url) @property def strategy(self): return self.__strategy @strategy.setter def strategy(self, value): if value not in (SIMPLE_UPLOAD, RESUMABLE_UPLOAD): raise exceptions.UserError(( 'Invalid value "%s" for upload strategy, must be one of ' '"simple" or "resumable".') % value) self.__strategy = value @property def total_size(self): return self.__total_size @total_size.setter def total_size(self, value): self.EnsureUninitialized() self.__total_size = value def __SetDefaultUploadStrategy(self, upload_config, http_request): """Determine and set the default upload strategy for this upload. We generally prefer simple or multipart, unless we're forced to use resumable. This happens when any of (1) the upload is too large, (2) the simple endpoint doesn't support multipart requests and we have metadata, or (3) there is no simple upload endpoint. Args: upload_config: Configuration for the upload endpoint. http_request: The associated http request. Returns: None. """ if upload_config.resumable_path is None: self.strategy = SIMPLE_UPLOAD if self.strategy is not None: return strategy = SIMPLE_UPLOAD if (self.total_size is not None and self.total_size > _RESUMABLE_UPLOAD_THRESHOLD): strategy = RESUMABLE_UPLOAD if http_request.body and not upload_config.simple_multipart: strategy = RESUMABLE_UPLOAD if not upload_config.simple_path: strategy = RESUMABLE_UPLOAD self.strategy = strategy def ConfigureRequest(self, upload_config, http_request, url_builder): """Configure the request and url for this upload.""" # Validate total_size vs. max_size if (self.total_size and upload_config.max_size and self.total_size > upload_config.max_size): raise exceptions.InvalidUserInputError( 'Upload too big: %s larger than max size %s' % ( self.total_size, upload_config.max_size)) # Validate mime type if not util.AcceptableMimeType(upload_config.accept, self.mime_type): raise exceptions.InvalidUserInputError( 'MIME type %s does not match any accepted MIME ranges %s' % ( self.mime_type, upload_config.accept)) self.__SetDefaultUploadStrategy(upload_config, http_request) if self.strategy == SIMPLE_UPLOAD: url_builder.relative_path = upload_config.simple_path if http_request.body: url_builder.query_params['uploadType'] = 'multipart' self.__ConfigureMultipartRequest(http_request) else: url_builder.query_params['uploadType'] = 'media' self.__ConfigureMediaRequest(http_request) # Once the entire body is written, compress the body if configured # to. Both multipart and media request uploads will read the # entire stream into memory, which means full compression is also # safe to perform. Because the strategy is set to SIMPLE_UPLOAD, # StreamInChunks throws an exception, meaning double compression # cannot happen. if self.__gzip_encoded: http_request.headers['Content-Encoding'] = 'gzip' # Turn the body into a stream so that we can compress it, then # read the compressed bytes. In the event of a retry (e.g. if # our access token has expired), we need to be able to re-read # the body, which we can't do with a stream. So, we consume the # bytes from the stream now and store them in a re-readable # bytes container. http_request.body = ( compression.CompressStream( six.BytesIO(http_request.body))[0].read()) else: url_builder.relative_path = upload_config.resumable_path url_builder.query_params['uploadType'] = 'resumable' self.__ConfigureResumableRequest(http_request) def __ConfigureMediaRequest(self, http_request): """Configure http_request as a simple request for this upload.""" http_request.headers['content-type'] = self.mime_type http_request.body = self.stream.read() http_request.loggable_body = '<media body>' def __ConfigureMultipartRequest(self, http_request): """Configure http_request as a multipart request for this upload.""" # This is a multipart/related upload. msg_root = mime_multipart.MIMEMultipart('related') # msg_root should not write out its own headers setattr(msg_root, '_write_headers', lambda self: None) # attach the body as one part msg = mime_nonmultipart.MIMENonMultipart( http_request.headers['content-type'].split('/')) msg.set_payload(http_request.body) msg_root.attach(msg) # attach the media as the second part msg = mime_nonmultipart.MIMENonMultipart(self.mime_type.split('/')) msg['Content-Transfer-Encoding'] = 'binary' msg.set_payload(self.stream.read()) msg_root.attach(msg) # NOTE: We encode the body, but can't use # `email.message.Message.as_string` because it prepends # `> ` to `From ` lines. fp = six.BytesIO() if six.PY3: generator_class = MultipartBytesGenerator else: generator_class = email_generator.Generator g = generator_class(fp, mangle_from_=False) g.flatten(msg_root, unixfrom=False) http_request.body = fp.getvalue() multipart_boundary = msg_root.get_boundary() http_request.headers['content-type'] = ( 'multipart/related; boundary=%r' % multipart_boundary) if isinstance(multipart_boundary, six.text_type): multipart_boundary = multipart_boundary.encode('ascii') body_components = http_request.body.split(multipart_boundary) headers, _, _ = body_components[-2].partition(b'\n\n') body_components[-2] = b'\n\n'.join([headers, b'<media body>\n\n--']) http_request.loggable_body = multipart_boundary.join(body_components) def __ConfigureResumableRequest(self, http_request): http_request.headers['X-Upload-Content-Type'] = self.mime_type if self.total_size is not None: http_request.headers[ 'X-Upload-Content-Length'] = str(self.total_size) def RefreshResumableUploadState(self): """Talk to the server and refresh the state of this resumable upload. Returns: Response if the upload is complete. """ if self.strategy != RESUMABLE_UPLOAD: return self.EnsureInitialized() refresh_request = http_wrapper.Request( url=self.url, http_method='PUT', headers={'Content-Range': 'bytes /'}) refresh_response = http_wrapper.MakeRequest( self.http, refresh_request, redirections=0, retries=self.num_retries) range_header = self._GetRangeHeaderFromResponse(refresh_response) if refresh_response.status_code in (http_client.OK, http_client.CREATED): self.__complete = True self.__progress = self.total_size self.stream.seek(self.progress) # If we're finished, the refresh response will contain the metadata # originally requested. Cache it so it can be returned in # StreamInChunks. self.__final_response = refresh_response elif refresh_response.status_code == http_wrapper.RESUME_INCOMPLETE: if range_header is None: self.__progress = 0 else: self.__progress = self.__GetLastByte(range_header) + 1 self.stream.seek(self.progress) else: raise exceptions.HttpError.FromResponse(refresh_response) def _GetRangeHeaderFromResponse(self, response): return response.info.get('Range', response.info.get('range')) def InitializeUpload(self, http_request, http=None, client=None): """Initialize this upload from the given http_request.""" if self.strategy is None: raise exceptions.UserError( 'No upload strategy set; did you call ConfigureRequest?') if http is None and client is None: raise exceptions.UserError('Must provide client or http.') if self.strategy != RESUMABLE_UPLOAD: return http = http or client.http if client is not None: http_request.url = client.FinalizeTransferUrl(http_request.url) self.EnsureUninitialized() http_response = http_wrapper.MakeRequest(http, http_request, retries=self.num_retries) if http_response.status_code != http_client.OK: raise exceptions.HttpError.FromResponse(http_response) self.__server_chunk_granularity = http_response.info.get( 'X-Goog-Upload-Chunk-Granularity') url = http_response.info['location'] if client is not None: url = client.FinalizeTransferUrl(url) self._Initialize(http, url) # Unless the user has requested otherwise, we want to just # go ahead and pump the bytes now. if self.auto_transfer: return self.StreamInChunks() return http_response def __GetLastByte(self, range_header): _, _, end = range_header.partition('-') # TODO(craigcitro): Validate start == 0? return int(end) def __ValidateChunksize(self, chunksize=None): if self.__server_chunk_granularity is None: return chunksize = chunksize or self.chunksize if chunksize % self.__server_chunk_granularity: raise exceptions.ConfigurationValueError( 'Server requires chunksize to be a multiple of %d' % self.__server_chunk_granularity) def __IsRetryable(self, response): return (response.status_code >= 500 or response.status_code == http_wrapper.TOO_MANY_REQUESTS or response.retry_after) def __StreamMedia(self, callback=None, finish_callback=None, additional_headers=None, use_chunks=True): """Helper function for StreamMedia / StreamInChunks.""" if self.strategy != RESUMABLE_UPLOAD: raise exceptions.InvalidUserInputError( 'Cannot stream non-resumable upload') callback = callback or self.progress_callback finish_callback = finish_callback or self.finish_callback # final_response is set if we resumed an already-completed upload. response = self.__final_response def CallSendChunk(start): return self.__SendChunk( start, additional_headers=additional_headers) def CallSendMediaBody(start): return self.__SendMediaBody( start, additional_headers=additional_headers) send_func = CallSendChunk if use_chunks else CallSendMediaBody if not use_chunks and self.__gzip_encoded: raise exceptions.InvalidUserInputError( 'Cannot gzip encode non-chunked upload') if use_chunks: self.__ValidateChunksize(self.chunksize) self.EnsureInitialized() while not self.complete: response = send_func(self.stream.tell()) if response.status_code in (http_client.OK, http_client.CREATED): self.__complete = True break if response.status_code not in ( http_client.OK, http_client.CREATED, http_wrapper.RESUME_INCOMPLETE): # Only raise an exception if the error is something we can't # recover from. if (self.strategy != RESUMABLE_UPLOAD or not self.__IsRetryable(response)): raise exceptions.HttpError.FromResponse(response) # We want to reset our state to wherever the server left us # before this failed request, and then raise. self.RefreshResumableUploadState() self._ExecuteCallback(callback, response) continue self.__progress = self.__GetLastByte( self._GetRangeHeaderFromResponse(response)) if self.progress + 1 != self.stream.tell(): # TODO(craigcitro): Add a better way to recover here. raise exceptions.CommunicationError( 'Failed to transfer all bytes in chunk, upload paused at ' 'byte %d' % self.progress) self._ExecuteCallback(callback, response) if self.__complete and hasattr(self.stream, 'seek'): current_pos = self.stream.tell() self.stream.seek(0, os.SEEK_END) end_pos = self.stream.tell() self.stream.seek(current_pos) if current_pos != end_pos: raise exceptions.TransferInvalidError( 'Upload complete with %s additional bytes left in stream' % (int(end_pos) - int(current_pos))) self._ExecuteCallback(finish_callback, response) return response def StreamMedia(self, callback=None, finish_callback=None, additional_headers=None): """Send this resumable upload in a single request. Args: callback: Progress callback function with inputs (http_wrapper.Response, transfer.Upload) finish_callback: Final callback function with inputs (http_wrapper.Response, transfer.Upload) additional_headers: Dict of headers to include with the upload http_wrapper.Request. Returns: http_wrapper.Response of final response. """ return self.__StreamMedia( callback=callback, finish_callback=finish_callback, additional_headers=additional_headers, use_chunks=False) def StreamInChunks(self, callback=None, finish_callback=None, additional_headers=None): """Send this (resumable) upload in chunks.""" return self.__StreamMedia( callback=callback, finish_callback=finish_callback, additional_headers=additional_headers) def __SendMediaRequest(self, request, end): """Request helper function for SendMediaBody & SendChunk.""" def CheckResponse(response): if response is None: # Caller shouldn't call us if the response is None, # but handle anyway. raise exceptions.RequestError( 'Request to url %s did not return a response.' % response.request_url) response = http_wrapper.MakeRequest( self.bytes_http, request, retry_func=self.retry_func, retries=self.num_retries, check_response_func=CheckResponse) if response.status_code == http_wrapper.RESUME_INCOMPLETE: last_byte = self.__GetLastByte( self._GetRangeHeaderFromResponse(response)) if last_byte + 1 != end: self.stream.seek(last_byte + 1) return response def __SendMediaBody(self, start, additional_headers=None): """Send the entire media stream in a single request.""" self.EnsureInitialized() if self.total_size is None: raise exceptions.TransferInvalidError( 'Total size must be known for SendMediaBody') body_stream = stream_slice.StreamSlice( self.stream, self.total_size - start) request = http_wrapper.Request(url=self.url, http_method='PUT', body=body_stream) request.headers['Content-Type'] = self.mime_type if start == self.total_size: # End of an upload with 0 bytes left to send; just finalize. range_string = 'bytes /%s' % self.total_size else: range_string = 'bytes %s-%s/%s' % (start, self.total_size - 1, self.total_size) request.headers['Content-Range'] = range_string if additional_headers: request.headers.update(additional_headers) return self.__SendMediaRequest(request, self.total_size) def __SendChunk(self, start, additional_headers=None): """Send the specified chunk.""" self.EnsureInitialized() no_log_body = self.total_size is None request = http_wrapper.Request(url=self.url, http_method='PUT') if self.__gzip_encoded: request.headers['Content-Encoding'] = 'gzip' body_stream, read_length, exhausted = compression.CompressStream( self.stream, self.chunksize) end = start + read_length # If the stream length was previously unknown and the input stream # is exhausted, then we're at the end of the stream. if self.total_size is None and exhausted: self.__total_size = end elif self.total_size is None: # For the streaming resumable case, we need to detect when # we're at the end of the stream. body_stream = buffered_stream.BufferedStream( self.stream, start, self.chunksize) end = body_stream.stream_end_position if body_stream.stream_exhausted: self.__total_size = end # TODO: Here, change body_stream from a stream to a string object, # which means reading a chunk into memory. This works around # https://code.google.com/p/httplib2/issues/detail?id=176 which can # cause httplib2 to skip bytes on 401's for file objects. # Rework this solution to be more general. body_stream = body_stream.read(self.chunksize) else: end = min(start + self.chunksize, self.total_size) body_stream = stream_slice.StreamSlice(self.stream, end - start) # TODO(craigcitro): Think about clearer errors on "no data in # stream". request.body = body_stream request.headers['Content-Type'] = self.mime_type if no_log_body: # Disable logging of streaming body. # TODO: Remove no_log_body and rework as part of a larger logs # refactor. request.loggable_body = '<media body>' if self.total_size is None: # Streaming resumable upload case, unknown total size. range_string = 'bytes %s-%s/' % (start, end - 1) elif end == start: # End of an upload with 0 bytes left to send; just finalize. range_string = 'bytes */%s' % self.total_size else: # Normal resumable upload case with known sizes. range_string = 'bytes %s-%s/%s' % (start, end - 1, self.total_size) request.headers['Content-Range'] = range_string if additional_headers: request.headers.update(additional_headers) return self.__SendMediaRequest(request, end)	45.42 KB	2025-03-11 18:02
📄 transfer_test.py # -- coding: utf-8 -- # # 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. """Tests for transfer.py.""" import string import unittest import httplib2 import json import mock import six from six.moves import http_client from apitools.base.py import base_api from apitools.base.py import exceptions from apitools.base.py import gzip from apitools.base.py import http_wrapper from apitools.base.py import transfer class TransferTest(unittest.TestCase): def assertRangeAndContentRangeCompatible(self, request, response): request_prefix = 'bytes=' self.assertIn('range', request.headers) self.assertTrue(request.headers['range'].startswith(request_prefix)) request_range = request.headers['range'][len(request_prefix):] response_prefix = 'bytes ' self.assertIn('content-range', response.info) response_header = response.info['content-range'] self.assertTrue(response_header.startswith(response_prefix)) response_range = ( response_header[len(response_prefix):].partition('/')[0]) msg = ('Request range ({0}) not a prefix of ' 'response_range ({1})').format( request_range, response_range) self.assertTrue(response_range.startswith(request_range), msg=msg) def testComputeEndByte(self): total_size = 100 chunksize = 10 download = transfer.Download.FromStream( six.StringIO(), chunksize=chunksize, total_size=total_size) self.assertEqual(chunksize - 1, download._Download__ComputeEndByte(0, end=50)) def testComputeEndByteReturnNone(self): download = transfer.Download.FromStream(six.StringIO()) self.assertIsNone( download._Download__ComputeEndByte(0, use_chunks=False)) def testComputeEndByteNoChunks(self): total_size = 100 download = transfer.Download.FromStream( six.StringIO(), chunksize=10, total_size=total_size) for end in (None, 1000): self.assertEqual( total_size - 1, download._Download__ComputeEndByte(0, end=end, use_chunks=False), msg='Failed on end={0}'.format(end)) def testComputeEndByteNoTotal(self): download = transfer.Download.FromStream(six.StringIO()) default_chunksize = download.chunksize for chunksize in (100, default_chunksize): download.chunksize = chunksize for start in (0, 10): self.assertEqual( download.chunksize + start - 1, download._Download__ComputeEndByte(start), msg='Failed on start={0}, chunksize={1}'.format( start, chunksize)) def testComputeEndByteSmallTotal(self): total_size = 100 download = transfer.Download.FromStream(six.StringIO(), total_size=total_size) for start in (0, 10): self.assertEqual(total_size - 1, download._Download__ComputeEndByte(start), msg='Failed on start={0}'.format(start)) def testDownloadThenStream(self): bytes_http = object() http = object() download_stream = six.StringIO() download = transfer.Download.FromStream(download_stream, total_size=26) download.bytes_http = bytes_http base_url = 'https://part.one/' with mock.patch.object(http_wrapper, 'MakeRequest', autospec=True) as make_request: make_request.return_value = http_wrapper.Response( info={ 'content-range': 'bytes 0-25/26', 'status': http_client.OK, }, content=string.ascii_lowercase, request_url=base_url, ) request = http_wrapper.Request(url='https://part.one/') download.InitializeDownload(request, http=http) self.assertEqual(1, make_request.call_count) received_request = make_request.call_args[0][1] self.assertEqual(base_url, received_request.url) self.assertRangeAndContentRangeCompatible( received_request, make_request.return_value) with mock.patch.object(http_wrapper, 'MakeRequest', autospec=True) as make_request: make_request.return_value = http_wrapper.Response( info={ 'status': http_client.REQUESTED_RANGE_NOT_SATISFIABLE, }, content='error', request_url=base_url, ) download.StreamInChunks() self.assertEqual(1, make_request.call_count) received_request = make_request.call_args[0][1] self.assertEqual('bytes=26-', received_request.headers['range']) def testGetRange(self): for (start_byte, end_byte) in [(0, 25), (5, 15), (0, 0), (25, 25)]: bytes_http = object() http = object() download_stream = six.StringIO() download = transfer.Download.FromStream(download_stream, total_size=26, auto_transfer=False) download.bytes_http = bytes_http base_url = 'https://part.one/' with mock.patch.object(http_wrapper, 'MakeRequest', autospec=True) as make_request: make_request.return_value = http_wrapper.Response( info={ 'content-range': 'bytes %d-%d/26' % (start_byte, end_byte), 'status': http_client.OK, }, content=string.ascii_lowercase[start_byte:end_byte + 1], request_url=base_url, ) request = http_wrapper.Request(url='https://part.one/') download.InitializeDownload(request, http=http) download.GetRange(start_byte, end_byte) self.assertEqual(1, make_request.call_count) received_request = make_request.call_args[0][1] self.assertEqual(base_url, received_request.url) self.assertRangeAndContentRangeCompatible( received_request, make_request.return_value) def testNonChunkedDownload(self): bytes_http = object() http = object() download_stream = six.StringIO() download = transfer.Download.FromStream(download_stream, total_size=52) download.bytes_http = bytes_http base_url = 'https://part.one/' with mock.patch.object(http_wrapper, 'MakeRequest', autospec=True) as make_request: make_request.return_value = http_wrapper.Response( info={ 'content-range': 'bytes 0-51/52', 'status': http_client.OK, }, content=string.ascii_lowercase * 2, request_url=base_url, ) request = http_wrapper.Request(url='https://part.one/') download.InitializeDownload(request, http=http) self.assertEqual(1, make_request.call_count) received_request = make_request.call_args[0][1] self.assertEqual(base_url, received_request.url) self.assertRangeAndContentRangeCompatible( received_request, make_request.return_value) download_stream.seek(0) self.assertEqual(string.ascii_lowercase * 2, download_stream.getvalue()) def testChunkedDownload(self): bytes_http = object() http = object() download_stream = six.StringIO() download = transfer.Download.FromStream( download_stream, chunksize=26, total_size=52) download.bytes_http = bytes_http # Setting autospec on a mock with an iterable side_effect is # currently broken (http://bugs.python.org/issue17826), so # instead we write a little function. def _ReturnBytes(unused_http, http_request, unused_args, unused_kwds): url = http_request.url if url == 'https://part.one/': return http_wrapper.Response( info={ 'content-location': 'https://part.two/', 'content-range': 'bytes 0-25/52', 'status': http_client.PARTIAL_CONTENT, }, content=string.ascii_lowercase, request_url='https://part.one/', ) elif url == 'https://part.two/': return http_wrapper.Response( info={ 'content-range': 'bytes 26-51/52', 'status': http_client.OK, }, content=string.ascii_uppercase, request_url='https://part.two/', ) else: self.fail('Unknown URL requested: %s' % url) with mock.patch.object(http_wrapper, 'MakeRequest', autospec=True) as make_request: make_request.side_effect = _ReturnBytes request = http_wrapper.Request(url='https://part.one/') download.InitializeDownload(request, http=http) self.assertEqual(2, make_request.call_count) for call in make_request.call_args_list: self.assertRangeAndContentRangeCompatible( call[0][1], _ReturnBytes(call[0])) download_stream.seek(0) self.assertEqual(string.ascii_lowercase + string.ascii_uppercase, download_stream.getvalue()) # @mock.patch.object(transfer.Upload, 'RefreshResumableUploadState', # new=mock.Mock()) def testFinalizesTransferUrlIfClientPresent(self): """Tests download's enforcement of client custom endpoints.""" mock_client = mock.Mock() fake_json_data = json.dumps({ 'auto_transfer': False, 'progress': 0, 'total_size': 0, 'url': 'url', }) transfer.Download.FromData(six.BytesIO(), fake_json_data, client=mock_client) mock_client.FinalizeTransferUrl.assert_called_once_with('url') def testMultipartEncoding(self): # This is really a table test for various issues we've seen in # the past; see notes below for particular histories. test_cases = [ # Python's mime module by default encodes lines that start # with "From " as ">From ", which we need to make sure we # don't run afoul of when sending content that isn't # intended to be so encoded. This test calls out that we # get this right. We test for both the multipart and # non-multipart case. 'line one\nFrom \nline two', # We had originally used a `six.StringIO` to hold the http # request body in the case of a multipart upload; for # bytes being uploaded in Python3, however, this causes # issues like this: # https://github.com/GoogleCloudPlatform/gcloud-python/issues/1760 # We test below to ensure that we don't end up mangling # the body before sending. u'name,main_ingredient\nRäksmörgås,Räkor\nBaguette,Bröd', ] for upload_contents in test_cases: multipart_body = '{"body_field_one": 7}' upload_bytes = upload_contents.encode('ascii', 'backslashreplace') upload_config = base_api.ApiUploadInfo( accept=['/'], max_size=None, resumable_multipart=True, resumable_path=u'/resumable/upload', simple_multipart=True, simple_path=u'/upload', ) url_builder = base_api._UrlBuilder('http://www.uploads.com') # Test multipart: having a body argument in http_request forces # multipart here. upload = transfer.Upload.FromStream( six.BytesIO(upload_bytes), 'text/plain', total_size=len(upload_bytes)) http_request = http_wrapper.Request( 'http://www.uploads.com', headers={'content-type': 'text/plain'}, body=multipart_body) upload.ConfigureRequest(upload_config, http_request, url_builder) self.assertEqual( 'multipart', url_builder.query_params['uploadType']) rewritten_upload_contents = b'\n'.join( http_request.body.split(b'--')[2].splitlines()[1:]) self.assertTrue(rewritten_upload_contents.endswith(upload_bytes)) # Test non-multipart (aka media): no body argument means this is # sent as media. upload = transfer.Upload.FromStream( six.BytesIO(upload_bytes), 'text/plain', total_size=len(upload_bytes)) http_request = http_wrapper.Request( 'http://www.uploads.com', headers={'content-type': 'text/plain'}) upload.ConfigureRequest(upload_config, http_request, url_builder) self.assertEqual(url_builder.query_params['uploadType'], 'media') rewritten_upload_contents = http_request.body self.assertTrue(rewritten_upload_contents.endswith(upload_bytes)) class UploadTest(unittest.TestCase): def setUp(self): # Sample highly compressible data. self.sample_data = b'abc' * 200 # Stream of the sample data. self.sample_stream = six.BytesIO(self.sample_data) # Sample url_builder. self.url_builder = base_api._UrlBuilder('http://www.uploads.com') # Sample request. self.request = http_wrapper.Request( 'http://www.uploads.com', headers={'content-type': 'text/plain'}) # Sample successful response. self.response = http_wrapper.Response( info={'status': http_client.OK, 'location': 'http://www.uploads.com'}, content='', request_url='http://www.uploads.com',) # Sample failure response. self.fail_response = http_wrapper.Response( info={'status': http_client.SERVICE_UNAVAILABLE, 'location': 'http://www.uploads.com'}, content='', request_url='http://www.uploads.com',) def testStreamInChunksCompressed(self): """Test that StreamInChunks will handle compression correctly.""" # Create and configure the upload object. upload = transfer.Upload( stream=self.sample_stream, mime_type='text/plain', total_size=len(self.sample_data), close_stream=False, gzip_encoded=True) upload.strategy = transfer.RESUMABLE_UPLOAD # Set the chunk size so the entire stream is uploaded. upload.chunksize = len(self.sample_data) # Mock the upload to return the sample response. with mock.patch.object(transfer.Upload, '_Upload__SendMediaRequest') as mock_result, \ mock.patch.object(http_wrapper, 'MakeRequest') as make_request: mock_result.return_value = self.response make_request.return_value = self.response # Initialization. upload.InitializeUpload(self.request, 'http') upload.StreamInChunks() # Get the uploaded request and end position of the stream. (request, _), _ = mock_result.call_args_list[0] # Ensure the mock was called. self.assertTrue(mock_result.called) # Ensure the correct content encoding was set. self.assertEqual(request.headers['Content-Encoding'], 'gzip') # Ensure the stream was compresed. self.assertLess(len(request.body), len(self.sample_data)) def testStreamMediaCompressedFail(self): """Test that non-chunked uploads raise an exception. Ensure uploads with the compressed and resumable flags set called from StreamMedia raise an exception. Those uploads are unsupported. """ # Create the upload object. upload = transfer.Upload( stream=self.sample_stream, mime_type='text/plain', total_size=len(self.sample_data), close_stream=False, auto_transfer=True, gzip_encoded=True) upload.strategy = transfer.RESUMABLE_UPLOAD # Mock the upload to return the sample response. with mock.patch.object(http_wrapper, 'MakeRequest') as make_request: make_request.return_value = self.response # Initialization. upload.InitializeUpload(self.request, 'http') # Ensure stream media raises an exception when the upload is # compressed. Compression is not supported on non-chunked uploads. with self.assertRaises(exceptions.InvalidUserInputError): upload.StreamMedia() def testAutoTransferCompressed(self): """Test that automatic transfers are compressed. Ensure uploads with the compressed, resumable, and automatic transfer flags set call StreamInChunks. StreamInChunks is tested in an earlier test. """ # Create the upload object. upload = transfer.Upload( stream=self.sample_stream, mime_type='text/plain', total_size=len(self.sample_data), close_stream=False, gzip_encoded=True) upload.strategy = transfer.RESUMABLE_UPLOAD # Mock the upload to return the sample response. with mock.patch.object(transfer.Upload, 'StreamInChunks') as mock_result, \ mock.patch.object(http_wrapper, 'MakeRequest') as make_request: mock_result.return_value = self.response make_request.return_value = self.response # Initialization. upload.InitializeUpload(self.request, 'http') # Ensure the mock was called. self.assertTrue(mock_result.called) def testMultipartCompressed(self): """Test that multipart uploads are compressed.""" # Create the multipart configuration. upload_config = base_api.ApiUploadInfo( accept=['/'], max_size=None, simple_multipart=True, simple_path=u'/upload',) # Create the upload object. upload = transfer.Upload( stream=self.sample_stream, mime_type='text/plain', total_size=len(self.sample_data), close_stream=False, gzip_encoded=True) # Set a body to trigger multipart configuration. self.request.body = '{"body_field_one": 7}' # Configure the request. upload.ConfigureRequest(upload_config, self.request, self.url_builder) # Ensure the request is a multipart request now. self.assertEqual( self.url_builder.query_params['uploadType'], 'multipart') # Ensure the request is gzip encoded. self.assertEqual(self.request.headers['Content-Encoding'], 'gzip') # Ensure data is compressed self.assertLess(len(self.request.body), len(self.sample_data)) # Ensure uncompressed data includes the sample data. with gzip.GzipFile(fileobj=six.BytesIO(self.request.body)) as f: original = f.read() self.assertTrue(self.sample_data in original) def testMediaCompressed(self): """Test that media uploads are compressed.""" # Create the media configuration. upload_config = base_api.ApiUploadInfo( accept=['/'], max_size=None, simple_multipart=True, simple_path=u'/upload',) # Create the upload object. upload = transfer.Upload( stream=self.sample_stream, mime_type='text/plain', total_size=len(self.sample_data), close_stream=False, gzip_encoded=True) # Configure the request. upload.ConfigureRequest(upload_config, self.request, self.url_builder) # Ensure the request is a media request now. self.assertEqual(self.url_builder.query_params['uploadType'], 'media') # Ensure the request is gzip encoded. self.assertEqual(self.request.headers['Content-Encoding'], 'gzip') # Ensure data is compressed self.assertLess(len(self.request.body), len(self.sample_data)) # Ensure uncompressed data includes the sample data. with gzip.GzipFile(fileobj=six.BytesIO(self.request.body)) as f: original = f.read() self.assertTrue(self.sample_data in original) def HttpRequestSideEffect(self, responses=None): responses = [(response.info, response.content) for response in responses] def _side_effect(uri, **kwargs): # pylint: disable=unused-argument body = kwargs['body'] read_func = getattr(body, 'read', None) if read_func: # If the body is a stream, consume the stream. body = read_func() self.assertEqual(int(kwargs['headers']['content-length']), len(body)) return responses.pop(0) return _side_effect def testRetryRequestChunks(self): """Test that StreamInChunks will retry correctly.""" refresh_response = http_wrapper.Response( info={'status': http_wrapper.RESUME_INCOMPLETE, 'location': 'http://www.uploads.com'}, content='', request_url='http://www.uploads.com',) # Create and configure the upload object. bytes_http = httplib2.Http() upload = transfer.Upload( stream=self.sample_stream, mime_type='text/plain', total_size=len(self.sample_data), close_stream=False, http=bytes_http) upload.strategy = transfer.RESUMABLE_UPLOAD # Set the chunk size so the entire stream is uploaded. upload.chunksize = len(self.sample_data) # Mock the upload to return the sample response. with mock.patch.object(bytes_http, 'request') as make_request: # This side effect also checks the request body. responses = [ self.response, # Initial request in InitializeUpload(). self.fail_response, # 503 status code from server. refresh_response, # Refresh upload progress. self.response, # Successful request. ] make_request.side_effect = self.HttpRequestSideEffect(responses) # Initialization. upload.InitializeUpload(self.request, bytes_http) upload.StreamInChunks() # Ensure the mock was called the correct number of times. self.assertEqual(make_request.call_count, len(responses)) def testStreamInChunks(self): """Test StreamInChunks.""" resume_incomplete_responses = [http_wrapper.Response( info={'status': http_wrapper.RESUME_INCOMPLETE, 'location': 'http://www.uploads.com', 'range': '0-{}'.format(end)}, content='', request_url='http://www.uploads.com',) for end in [199, 399, 599]] responses = [ self.response # Initial request in InitializeUpload(). ] + resume_incomplete_responses + [ self.response, # Successful request. ] # Create and configure the upload object. bytes_http = httplib2.Http() upload = transfer.Upload( stream=self.sample_stream, mime_type='text/plain', total_size=len(self.sample_data), close_stream=False, http=bytes_http) upload.strategy = transfer.RESUMABLE_UPLOAD # Set the chunk size so the entire stream is uploaded. upload.chunksize = 200 # Mock the upload to return the sample response. with mock.patch.object(bytes_http, 'request') as make_request: # This side effect also checks the request body. make_request.side_effect = self.HttpRequestSideEffect(responses) # Initialization. upload.InitializeUpload(self.request, bytes_http) upload.StreamInChunks() # Ensure the mock was called the correct number of times. self.assertEqual(make_request.call_count, len(responses)) @mock.patch.object(transfer.Upload, 'RefreshResumableUploadState', new=mock.Mock()) def testFinalizesTransferUrlIfClientPresent(self): """Tests upload's enforcement of client custom endpoints.""" mock_client = mock.Mock() mock_http = mock.Mock() fake_json_data = json.dumps({ 'auto_transfer': False, 'mime_type': '', 'total_size': 0, 'url': 'url', }) transfer.Upload.FromData(self.sample_stream, fake_json_data, mock_http, client=mock_client) mock_client.FinalizeTransferUrl.assert_called_once_with('url')	26.06 KB	2025-03-11 18:02
📄 util.py #!/usr/bin/env python # # 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. """Assorted utilities shared between parts of apitools.""" import os import random import six from six.moves import http_client import six.moves.urllib.error as urllib_error import six.moves.urllib.parse as urllib_parse import six.moves.urllib.request as urllib_request from apitools.base.protorpclite import messages from apitools.base.py import encoding_helper as encoding from apitools.base.py import exceptions if six.PY3: from collections.abc import Iterable else: from collections import Iterable __all__ = [ 'DetectGae', 'DetectGce', ] _RESERVED_URI_CHARS = r":/?#[]@!$&'()+,;=" def DetectGae(): """Determine whether or not we're running on GAE. This is based on: https://developers.google.com/appengine/docs/python/#The_Environment Returns: True iff we're running on GAE. """ server_software = os.environ.get('SERVER_SOFTWARE', '') return (server_software.startswith('Development/') or server_software.startswith('Google App Engine/')) def DetectGce(): """Determine whether or not we're running on GCE. This is based on: https://cloud.google.com/compute/docs/metadata#runninggce Returns: True iff we're running on a GCE instance. """ metadata_url = 'http://{}'.format( os.environ.get('GCE_METADATA_ROOT', 'metadata.google.internal')) try: o = urllib_request.build_opener(urllib_request.ProxyHandler({})).open( urllib_request.Request( metadata_url, headers={'Metadata-Flavor': 'Google'})) except urllib_error.URLError: return False return (o.getcode() == http_client.OK and o.headers.get('metadata-flavor') == 'Google') def NormalizeScopes(scope_spec): """Normalize scope_spec to a set of strings.""" if isinstance(scope_spec, six.string_types): scope_spec = six.ensure_str(scope_spec) return set(scope_spec.split(' ')) elif isinstance(scope_spec, Iterable): scope_spec = [six.ensure_str(x) for x in scope_spec] return set(scope_spec) raise exceptions.TypecheckError( 'NormalizeScopes expected string or iterable, found %s' % ( type(scope_spec),)) def Typecheck(arg, arg_type, msg=None): if not isinstance(arg, arg_type): if msg is None: if isinstance(arg_type, tuple): msg = 'Type of arg is "%s", not one of %r' % ( type(arg), arg_type) else: msg = 'Type of arg is "%s", not "%s"' % (type(arg), arg_type) raise exceptions.TypecheckError(msg) return arg def ExpandRelativePath(method_config, params, relative_path=None): """Determine the relative path for request.""" path = relative_path or method_config.relative_path or '' for param in method_config.path_params: param_template = '{%s}' % param # For more details about "reserved word expansion", see: # http://tools.ietf.org/html/rfc6570#section-3.2.2 reserved_chars = '' reserved_template = '{+%s}' % param if reserved_template in path: reserved_chars = _RESERVED_URI_CHARS path = path.replace(reserved_template, param_template) if param_template not in path: raise exceptions.InvalidUserInputError( 'Missing path parameter %s' % param) try: # TODO(craigcitro): Do we want to support some sophisticated # mapping here? value = params[param] except KeyError: raise exceptions.InvalidUserInputError( 'Request missing required parameter %s' % param) if value is None: raise exceptions.InvalidUserInputError( 'Request missing required parameter %s' % param) try: if not isinstance(value, six.string_types): value = str(value) path = path.replace(param_template, urllib_parse.quote(value.encode('utf_8'), reserved_chars)) except TypeError as e: raise exceptions.InvalidUserInputError( 'Error setting required parameter %s to value %s: %s' % ( param, value, e)) return path def CalculateWaitForRetry(retry_attempt, max_wait=60): """Calculates amount of time to wait before a retry attempt. Wait time grows exponentially with the number of attempts. A random amount of jitter is added to spread out retry attempts from different clients. Args: retry_attempt: Retry attempt counter. max_wait: Upper bound for wait time [seconds]. Returns: Number of seconds to wait before retrying request. """ wait_time = 2 * retry_attempt max_jitter = wait_time / 4.0 wait_time += random.uniform(-max_jitter, max_jitter) return max(1, min(wait_time, max_wait)) def AcceptableMimeType(accept_patterns, mime_type): """Return True iff mime_type is acceptable for one of accept_patterns. Note that this function assumes that all patterns in accept_patterns will be simple types of the form "type/subtype", where one or both of these can be "". We do not support parameters (i.e. "; q=") in patterns. Args: accept_patterns: list of acceptable MIME types. mime_type: the mime type we would like to match. Returns: Whether or not mime_type matches (at least) one of these patterns. """ if '/' not in mime_type: raise exceptions.InvalidUserInputError( 'Invalid MIME type: "%s"' % mime_type) unsupported_patterns = [p for p in accept_patterns if ';' in p] if unsupported_patterns: raise exceptions.GeneratedClientError( 'MIME patterns with parameter unsupported: "%s"' % ', '.join( unsupported_patterns)) def MimeTypeMatches(pattern, mime_type): """Return True iff mime_type is acceptable for pattern.""" # Some systems use a single '' instead of '/'. if pattern == '': pattern = '/' return all(accept in ('', provided) for accept, provided in zip(pattern.split('/'), mime_type.split('/'))) return any(MimeTypeMatches(pattern, mime_type) for pattern in accept_patterns) def MapParamNames(params, request_type): """Reverse parameter remappings for URL construction.""" return [encoding.GetCustomJsonFieldMapping(request_type, json_name=p) or p for p in params] def MapRequestParams(params, request_type): """Perform any renames/remappings needed for URL construction. Currently, we have several ways to customize JSON encoding, in particular of field names and enums. This works fine for JSON bodies, but also needs to be applied for path and query parameters in the URL. This function takes a dictionary from param names to values, and performs any registered mappings. We also need the request type (to look up the mappings). Args: params: (dict) Map from param names to values request_type: (protorpc.messages.Message) request type for this API call Returns: A new dict of the same size, with all registered mappings applied. """ new_params = dict(params) for param_name, value in params.items(): field_remapping = encoding.GetCustomJsonFieldMapping( request_type, python_name=param_name) if field_remapping is not None: new_params[field_remapping] = new_params.pop(param_name) param_name = field_remapping if isinstance(value, messages.Enum): new_params[param_name] = encoding.GetCustomJsonEnumMapping( type(value), python_name=str(value)) or str(value) return new_params	8.28 KB	2025-03-11 18:02
📄 util_test.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. """Tests for util.py.""" import unittest from apitools.base.protorpclite import messages from apitools.base.py import encoding from apitools.base.py import exceptions from apitools.base.py import util class MockedMethodConfig(object): def __init__(self, relative_path, path_params): self.relative_path = relative_path self.path_params = path_params class MessageWithRemappings(messages.Message): class AnEnum(messages.Enum): value_one = 1 value_two = 2 str_field = messages.StringField(1) enum_field = messages.EnumField('AnEnum', 2) enum_field_remapping = messages.EnumField('AnEnum', 3) encoding.AddCustomJsonFieldMapping( MessageWithRemappings, 'str_field', 'path_field') encoding.AddCustomJsonFieldMapping( MessageWithRemappings, 'enum_field_remapping', 'enum_field_remapped') encoding.AddCustomJsonEnumMapping( MessageWithRemappings.AnEnum, 'value_one', 'ONE') class UtilTest(unittest.TestCase): def testExpand(self): method_config_xy = MockedMethodConfig(relative_path='{x}/y/{z}', path_params=['x', 'z']) self.assertEqual( util.ExpandRelativePath(method_config_xy, {'x': '1', 'z': '2'}), '1/y/2') self.assertEqual( util.ExpandRelativePath( method_config_xy, {'x': '1', 'z': '2'}, relative_path='{x}/y/{z}/q'), '1/y/2/q') def testReservedExpansion(self): method_config_reserved = MockedMethodConfig(relative_path='{+x}/baz', path_params=['x']) self.assertEqual('foo/:bar:/baz', util.ExpandRelativePath( method_config_reserved, {'x': 'foo/:bar:'})) method_config_no_reserved = MockedMethodConfig(relative_path='{x}/baz', path_params=['x']) self.assertEqual('foo%2F%3Abar%3A/baz', util.ExpandRelativePath( method_config_no_reserved, {'x': 'foo/:bar:'})) def testCalculateWaitForRetry(self): try0 = util.CalculateWaitForRetry(0) self.assertTrue(try0 >= 1.0) self.assertTrue(try0 <= 1.5) try1 = util.CalculateWaitForRetry(1) self.assertTrue(try1 >= 1.0) self.assertTrue(try1 <= 3.0) try2 = util.CalculateWaitForRetry(2) self.assertTrue(try2 >= 2.0) self.assertTrue(try2 <= 6.0) try3 = util.CalculateWaitForRetry(3) self.assertTrue(try3 >= 4.0) self.assertTrue(try3 <= 12.0) try4 = util.CalculateWaitForRetry(4) self.assertTrue(try4 >= 8.0) self.assertTrue(try4 <= 24.0) self.assertAlmostEqual(10, util.CalculateWaitForRetry(5, max_wait=10)) def testTypecheck(self): class Class1(object): pass class Class2(object): pass class Class3(object): pass instance_of_class1 = Class1() self.assertEqual( instance_of_class1, util.Typecheck(instance_of_class1, Class1)) self.assertEqual( instance_of_class1, util.Typecheck(instance_of_class1, ((Class1, Class2), Class3))) self.assertEqual( instance_of_class1, util.Typecheck(instance_of_class1, (Class1, (Class2, Class3)))) self.assertEqual( instance_of_class1, util.Typecheck(instance_of_class1, Class1, 'message')) self.assertEqual( instance_of_class1, util.Typecheck( instance_of_class1, ((Class1, Class2), Class3), 'message')) self.assertEqual( instance_of_class1, util.Typecheck( instance_of_class1, (Class1, (Class2, Class3)), 'message')) with self.assertRaises(exceptions.TypecheckError): util.Typecheck(instance_of_class1, Class2) with self.assertRaises(exceptions.TypecheckError): util.Typecheck(instance_of_class1, (Class2, Class3)) with self.assertRaises(exceptions.TypecheckError): util.Typecheck(instance_of_class1, Class2, 'message') with self.assertRaises(exceptions.TypecheckError): util.Typecheck(instance_of_class1, (Class2, Class3), 'message') def testAcceptableMimeType(self): valid_pairs = ( ('', 'text/plain'), ('/', 'text/plain'), ('text/', 'text/plain'), ('/plain', 'text/plain'), ('text/plain', 'text/plain'), ) for accept, mime_type in valid_pairs: self.assertTrue(util.AcceptableMimeType([accept], mime_type)) invalid_pairs = ( ('text/', 'application/json'), ('text/plain', 'application/json'), ) for accept, mime_type in invalid_pairs: self.assertFalse(util.AcceptableMimeType([accept], mime_type)) self.assertTrue(util.AcceptableMimeType(['application/json', '/'], 'text/plain')) self.assertFalse(util.AcceptableMimeType(['application/json', 'img/'], 'text/plain')) def testMalformedMimeType(self): self.assertRaises( exceptions.InvalidUserInputError, util.AcceptableMimeType, ['/*'], 'abcd') def testUnsupportedMimeType(self): self.assertRaises( exceptions.GeneratedClientError, util.AcceptableMimeType, ['text/html;q=0.9'], 'text/html') def testMapRequestParams(self): params = { 'str_field': 'foo', 'enum_field': MessageWithRemappings.AnEnum.value_one, 'enum_field_remapping': MessageWithRemappings.AnEnum.value_one, } remapped_params = { 'path_field': 'foo', 'enum_field': 'ONE', 'enum_field_remapped': 'ONE', } self.assertEqual(remapped_params, util.MapRequestParams(params, MessageWithRemappings)) params['enum_field'] = MessageWithRemappings.AnEnum.value_two remapped_params['enum_field'] = 'value_two' self.assertEqual(remapped_params, util.MapRequestParams(params, MessageWithRemappings)) def testMapParamNames(self): params = ['path_field', 'enum_field'] remapped_params = ['str_field', 'enum_field'] self.assertEqual(remapped_params, util.MapParamNames(params, MessageWithRemappings))	7.01 KB	2025-03-11 18:02

Edit File:

Terminal