Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/google/auth/_helpers.py: 47%
55 statements
« prev ^ index » next coverage.py v7.2.2, created at 2023-03-26 07:30 +0000
« prev ^ index » next coverage.py v7.2.2, created at 2023-03-26 07:30 +0000
1# Copyright 2015 Google Inc.
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
15"""Helper functions for commonly used utilities."""
17import base64
18import calendar
19import datetime
20import sys
22import six
23from six.moves import urllib
25from google.auth import exceptions
27# Token server doesn't provide a new a token when doing refresh unless the
28# token is expiring within 30 seconds, so refresh threshold should not be
29# more than 30 seconds. Otherwise auth lib will send tons of refresh requests
30# until 30 seconds before the expiration, and cause a spike of CPU usage.
31REFRESH_THRESHOLD = datetime.timedelta(seconds=20)
34def copy_docstring(source_class):
35 """Decorator that copies a method's docstring from another class.
37 Args:
38 source_class (type): The class that has the documented method.
40 Returns:
41 Callable: A decorator that will copy the docstring of the same
42 named method in the source class to the decorated method.
43 """
45 def decorator(method):
46 """Decorator implementation.
48 Args:
49 method (Callable): The method to copy the docstring to.
51 Returns:
52 Callable: the same method passed in with an updated docstring.
54 Raises:
55 google.auth.exceptions.InvalidOperation: if the method already has a docstring.
56 """
57 if method.__doc__:
58 raise exceptions.InvalidOperation("Method already has a docstring.")
60 source_method = getattr(source_class, method.__name__)
61 method.__doc__ = source_method.__doc__
63 return method
65 return decorator
68def utcnow():
69 """Returns the current UTC datetime.
71 Returns:
72 datetime: The current time in UTC.
73 """
74 return datetime.datetime.utcnow()
77def datetime_to_secs(value):
78 """Convert a datetime object to the number of seconds since the UNIX epoch.
80 Args:
81 value (datetime): The datetime to convert.
83 Returns:
84 int: The number of seconds since the UNIX epoch.
85 """
86 return calendar.timegm(value.utctimetuple())
89def to_bytes(value, encoding="utf-8"):
90 """Converts a string value to bytes, if necessary.
92 Unfortunately, ``six.b`` is insufficient for this task since in
93 Python 2 because it does not modify ``unicode`` objects.
95 Args:
96 value (Union[str, bytes]): The value to be converted.
97 encoding (str): The encoding to use to convert unicode to bytes.
98 Defaults to "utf-8".
100 Returns:
101 bytes: The original value converted to bytes (if unicode) or as
102 passed in if it started out as bytes.
104 Raises:
105 google.auth.exceptions.InvalidValue: If the value could not be converted to bytes.
106 """
107 result = value.encode(encoding) if isinstance(value, six.text_type) else value
108 if isinstance(result, six.binary_type):
109 return result
110 else:
111 raise exceptions.InvalidValue(
112 "{0!r} could not be converted to bytes".format(value)
113 )
116def from_bytes(value):
117 """Converts bytes to a string value, if necessary.
119 Args:
120 value (Union[str, bytes]): The value to be converted.
122 Returns:
123 str: The original value converted to unicode (if bytes) or as passed in
124 if it started out as unicode.
126 Raises:
127 google.auth.exceptions.InvalidValue: If the value could not be converted to unicode.
128 """
129 result = value.decode("utf-8") if isinstance(value, six.binary_type) else value
130 if isinstance(result, six.text_type):
131 return result
132 else:
133 raise exceptions.InvalidValue(
134 "{0!r} could not be converted to unicode".format(value)
135 )
138def update_query(url, params, remove=None):
139 """Updates a URL's query parameters.
141 Replaces any current values if they are already present in the URL.
143 Args:
144 url (str): The URL to update.
145 params (Mapping[str, str]): A mapping of query parameter
146 keys to values.
147 remove (Sequence[str]): Parameters to remove from the query string.
149 Returns:
150 str: The URL with updated query parameters.
152 Examples:
154 >>> url = 'http://example.com?a=1'
155 >>> update_query(url, {'a': '2'})
156 http://example.com?a=2
157 >>> update_query(url, {'b': '3'})
158 http://example.com?a=1&b=3
159 >> update_query(url, {'b': '3'}, remove=['a'])
160 http://example.com?b=3
162 """
163 if remove is None:
164 remove = []
166 # Split the URL into parts.
167 parts = urllib.parse.urlparse(url)
168 # Parse the query string.
169 query_params = urllib.parse.parse_qs(parts.query)
170 # Update the query parameters with the new parameters.
171 query_params.update(params)
172 # Remove any values specified in remove.
173 query_params = {
174 key: value for key, value in six.iteritems(query_params) if key not in remove
175 }
176 # Re-encoded the query string.
177 new_query = urllib.parse.urlencode(query_params, doseq=True)
178 # Unsplit the url.
179 new_parts = parts._replace(query=new_query)
180 return urllib.parse.urlunparse(new_parts)
183def scopes_to_string(scopes):
184 """Converts scope value to a string suitable for sending to OAuth 2.0
185 authorization servers.
187 Args:
188 scopes (Sequence[str]): The sequence of scopes to convert.
190 Returns:
191 str: The scopes formatted as a single string.
192 """
193 return " ".join(scopes)
196def string_to_scopes(scopes):
197 """Converts stringifed scopes value to a list.
199 Args:
200 scopes (Union[Sequence, str]): The string of space-separated scopes
201 to convert.
202 Returns:
203 Sequence(str): The separated scopes.
204 """
205 if not scopes:
206 return []
208 return scopes.split(" ")
211def padded_urlsafe_b64decode(value):
212 """Decodes base64 strings lacking padding characters.
214 Google infrastructure tends to omit the base64 padding characters.
216 Args:
217 value (Union[str, bytes]): The encoded value.
219 Returns:
220 bytes: The decoded value
221 """
222 b64string = to_bytes(value)
223 padded = b64string + b"=" * (-len(b64string) % 4)
224 return base64.urlsafe_b64decode(padded)
227def unpadded_urlsafe_b64encode(value):
228 """Encodes base64 strings removing any padding characters.
230 `rfc 7515`_ defines Base64url to NOT include any padding
231 characters, but the stdlib doesn't do that by default.
233 _rfc7515: https://tools.ietf.org/html/rfc7515#page-6
235 Args:
236 value (Union[str|bytes]): The bytes-like value to encode
238 Returns:
239 Union[str|bytes]: The encoded value
240 """
241 return base64.urlsafe_b64encode(value).rstrip(b"=")
244def is_python_3():
245 """Check if the Python interpreter is Python 2 or 3.
247 Returns:
248 bool: True if the Python interpreter is Python 3 and False otherwise.
249 """
250 return sys.version_info > (3, 0)