openstack/deb-python-oauth2client
Code Issues Proposed changes

Merge pull request #579 from pferate/refactor_helpers

Browse Source 
Merge util.py and _helpers.py.
This commit is contained in:
Nathaniel Manista
2016-08-05 09:28:20 -07:00
committed by GitHub
parent 0dc30bc033 2f5c53b1fa
commit 248dc6c58d
18 changed files with 363 additions and 402 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
docs/source/oauth2client.rst
View File

@@ -20,7 +20,6 @@ Submodules
oauth2client.service_account
oauth2client.tools
oauth2client.transport
oauth2client.util
Module contents
---------------

7
docs/source/oauth2client.util.rst
View File

@@ -1,7 +0,0 @@
oauth2client.util module
========================
.. automodule:: oauth2client.util
:members:
:undoc-members:
:show-inheritance:

197
oauth2client/_helpers.py
View File

@@ -11,12 +11,209 @@
# 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.
"""Helper functions for commonly used utilities."""
import base64
import functools
import inspect
import json
import logging
import os
import warnings
import six
from six.moves import urllib
__author__ = (
'rafek@google.com (Rafe Kaplan)',
'guido@google.com (Guido van Rossum)',
)
logger = logging.getLogger(__name__)
POSITIONAL_WARNING = 'WARNING'
POSITIONAL_EXCEPTION = 'EXCEPTION'
POSITIONAL_IGNORE = 'IGNORE'
POSITIONAL_SET = frozenset([POSITIONAL_WARNING, POSITIONAL_EXCEPTION,
POSITIONAL_IGNORE])
positional_parameters_enforcement = POSITIONAL_WARNING
_SYM_LINK_MESSAGE = 'File: {0}: Is a symbolic link.'
_IS_DIR_MESSAGE = '{0}: Is a directory'
_MISSING_FILE_MESSAGE = 'Cannot access {0}: No such file or directory'
def positional(max_positional_args):
"""A decorator to declare that only the first N arguments my be positional.
This decorator makes it easy to support Python 3 style keyword-only
parameters. For example, in Python 3 it is possible to write::
def fn(pos1, *, kwonly1=None, kwonly1=None):
...
All named parameters after ``*`` must be a keyword::
fn(10, 'kw1', 'kw2') # Raises exception.
fn(10, kwonly1='kw1') # Ok.
Example
^^^^^^^
To define a function like above, do::
@positional(1)
def fn(pos1, kwonly1=None, kwonly2=None):
...
If no default value is provided to a keyword argument, it becomes a
required keyword argument::
@positional(0)
def fn(required_kw):
...
This must be called with the keyword parameter::
fn() # Raises exception.
fn(10) # Raises exception.
fn(required_kw=10) # Ok.
When defining instance or class methods always remember to account for
``self`` and ``cls``::
class MyClass(object):
@positional(2)
def my_method(self, pos1, kwonly1=None):
...
@classmethod
@positional(2)
def my_method(cls, pos1, kwonly1=None):
...
The positional decorator behavior is controlled by
``_helpers.positional_parameters_enforcement``, which may be set to
``POSITIONAL_EXCEPTION``, ``POSITIONAL_WARNING`` or
``POSITIONAL_IGNORE`` to raise an exception, log a warning, or do
nothing, respectively, if a declaration is violated.
Args:
max_positional_arguments: Maximum number of positional arguments. All
parameters after the this index must be
keyword only.
Returns:
A decorator that prevents using arguments after max_positional_args
from being used as positional parameters.
Raises:
TypeError: if a key-word only argument is provided as a positional
parameter, but only if
_helpers.positional_parameters_enforcement is set to
POSITIONAL_EXCEPTION.
"""
def positional_decorator(wrapped):
@functools.wraps(wrapped)
def positional_wrapper(*args, **kwargs):
if len(args) > max_positional_args:
plural_s = ''
if max_positional_args != 1:
plural_s = 's'
message = ('{function}() takes at most {args_max} positional '
'argument{plural} ({args_given} given)'.format(
function=wrapped.__name__,
args_max=max_positional_args,
args_given=len(args),
plural=plural_s))
if positional_parameters_enforcement == POSITIONAL_EXCEPTION:
raise TypeError(message)
elif positional_parameters_enforcement == POSITIONAL_WARNING:
logger.warning(message)
return wrapped(*args, **kwargs)
return positional_wrapper
if isinstance(max_positional_args, six.integer_types):
return positional_decorator
else:
args, _, _, defaults = inspect.getargspec(max_positional_args)
return positional(len(args) - len(defaults))(max_positional_args)
def scopes_to_string(scopes):
"""Converts scope value to a string.
If scopes is a string then it is simply passed through. If scopes is an
iterable then a string is returned that is all the individual scopes
concatenated with spaces.
Args:
scopes: string or iterable of strings, the scopes.
Returns:
The scopes formatted as a single string.
"""
if isinstance(scopes, six.string_types):
return scopes
else:
return ' '.join(scopes)
def string_to_scopes(scopes):
"""Converts stringifed scope value to a list.
If scopes is a list then it is simply passed through. If scopes is an
string then a list of each individual scope is returned.
Args:
scopes: a string or iterable of strings, the scopes.
Returns:
The scopes in a list.
"""
if not scopes:
return []
elif isinstance(scopes, six.string_types):
return scopes.split(' ')
else:
return scopes
def _add_query_parameter(url, name, value):
"""Adds a query parameter to a url.
Replaces the current value if it already exists in the URL.
Args:
url: string, url to add the query parameter to.
name: string, query parameter name.
value: string, query parameter value.
Returns:
Updated query parameter. Does not update the url if value is None.
"""
if value is None:
return url
else:
parsed = list(urllib.parse.urlparse(url))
query = dict(urllib.parse.parse_qsl(parsed[4]))
query[name] = value
parsed[4] = urllib.parse.urlencode(query)
return urllib.parse.urlunparse(parsed)
def validate_file(filename):
if os.path.islink(filename):
raise IOError(_SYM_LINK_MESSAGE.format(filename))
elif os.path.isdir(filename):
raise IOError(_IS_DIR_MESSAGE.format(filename))
elif not os.path.isfile(filename):
warnings.warn(_MISSING_FILE_MESSAGE.format(filename))
def _parse_pem_key(raw_key_input):

29
oauth2client/client.py
View File

@@ -36,7 +36,6 @@ import oauth2client
from oauth2client import _helpers
from oauth2client import clientsecrets
from oauth2client import transport
from oauth2client import util
__author__ = 'jcgregorio@google.com (Joe Gregorio)'
@@ -466,7 +465,7 @@ class OAuth2Credentials(Credentials):
OAuth2Credentials objects may be safely pickled and unpickled.
"""
@util.positional(8)
@_helpers.positional(8)
def __init__(self, access_token, client_id, client_secret, refresh_token,
token_expiry, token_uri, user_agent, revoke_uri=None,
id_token=None, token_response=None, scopes=None,
@@ -513,7 +512,7 @@ class OAuth2Credentials(Credentials):
self.revoke_uri = revoke_uri
self.id_token = id_token
self.token_response = token_response
self.scopes = set(util.string_to_scopes(scopes or []))
self.scopes = set(_helpers.string_to_scopes(scopes or []))
self.token_info_uri = token_info_uri
# True if the credentials have been revoked or expired and can't be
@@ -592,7 +591,7 @@ class OAuth2Credentials(Credentials):
not have scopes. In both cases, you can use refresh_scopes() to
obtain the canonical set of scopes.
"""
scopes = util.string_to_scopes(scopes)
scopes = _helpers.string_to_scopes(scopes)
return set(scopes).issubset(self.scopes)
def retrieve_scopes(self, http):
@@ -908,7 +907,7 @@ class OAuth2Credentials(Credentials):
content = _helpers._from_bytes(content)
if resp.status == http_client.OK:
d = json.loads(content)
self.scopes = set(util.string_to_scopes(d.get('scope', '')))
self.scopes = set(_helpers.string_to_scopes(d.get('scope', '')))
else:
error_msg = 'Invalid response {0}.'.format(resp.status)
try:
@@ -1469,7 +1468,7 @@ class AssertionCredentials(GoogleCredentials):
AssertionCredentials objects may be safely pickled and unpickled.
"""
@util.positional(2)
@_helpers.positional(2)
def __init__(self, assertion_type, user_agent=None,
token_uri=oauth2client.GOOGLE_TOKEN_URI,
revoke_uri=oauth2client.GOOGLE_REVOKE_URI,
@@ -1545,7 +1544,7 @@ def _require_crypto_or_die():
raise CryptoUnavailableError('No crypto library available')
@util.positional(2)
@_helpers.positional(2)
def verify_id_token(id_token, audience, http=None,
cert_uri=ID_TOKEN_VERIFICATION_CERTS):
"""Verifies a signed JWT id_token.
@@ -1633,7 +1632,7 @@ def _parse_exchange_token_response(content):
return resp
@util.positional(4)
@_helpers.positional(4)
def credentials_from_code(client_id, client_secret, scope, code,
redirect_uri='postmessage', http=None,
user_agent=None,
@@ -1684,7 +1683,7 @@ def credentials_from_code(client_id, client_secret, scope, code,
return credentials
@util.positional(3)
@_helpers.positional(3)
def credentials_from_clientsecrets_and_code(filename, scope, code,
message=None,
redirect_uri='postmessage',
@@ -1803,7 +1802,7 @@ class OAuth2WebServerFlow(Flow):
OAuth2WebServerFlow objects may be safely pickled and unpickled.
"""
@util.positional(4)
@_helpers.positional(4)
def __init__(self, client_id,
client_secret=None,
scope=None,
@@ -1862,7 +1861,7 @@ class OAuth2WebServerFlow(Flow):
raise TypeError("The value of scope must not be None")
self.client_id = client_id
self.client_secret = client_secret
self.scope = util.scopes_to_string(scope)
self.scope = _helpers.scopes_to_string(scope)
self.redirect_uri = redirect_uri
self.login_hint = login_hint
self.user_agent = user_agent
@@ -1874,7 +1873,7 @@ class OAuth2WebServerFlow(Flow):
self.authorization_header = authorization_header
self.params = _oauth2_web_server_flow_params(kwargs)
@util.positional(1)
@_helpers.positional(1)
def step1_get_authorize_url(self, redirect_uri=None, state=None):
"""Returns a URI to redirect to the provider.
@@ -1915,7 +1914,7 @@ class OAuth2WebServerFlow(Flow):
query_params.update(self.params)
return _update_query_params(self.auth_uri, query_params)
@util.positional(1)
@_helpers.positional(1)
def step1_get_device_and_user_codes(self, http=None):
"""Returns a user code and the verification URL where to enter it
@@ -1963,7 +1962,7 @@ class OAuth2WebServerFlow(Flow):
pass
raise OAuth2DeviceCodeError(error_msg)
@util.positional(2)
@_helpers.positional(2)
def step2_exchange(self, code=None, http=None, device_flow_info=None):
"""Exchanges a code for OAuth2Credentials.
@@ -2060,7 +2059,7 @@ class OAuth2WebServerFlow(Flow):
raise FlowExchangeError(error_msg)
@util.positional(2)
@_helpers.positional(2)
def flow_from_clientsecrets(filename, scope, redirect_uri=None,
message=None, cache=None, login_hint=None,
device_uri=None):

3
oauth2client/contrib/_metadata.py
View File

@@ -25,7 +25,6 @@ from six.moves.urllib import parse as urlparse
from oauth2client import _helpers
from oauth2client import client
from oauth2client import util
METADATA_ROOT = 'http://metadata.google.internal/computeMetadata/v1/'
@@ -54,7 +53,7 @@ def get(http_request, path, root=METADATA_ROOT, recursive=None):
retrieving metadata.
"""
url = urlparse.urljoin(root, path)
url = util._add_query_parameter(url, 'recursive', recursive)
url = _helpers._add_query_parameter(url, 'recursive', recursive)
response, content = http_request(
url,

18
oauth2client/contrib/appengine.py
View File

@@ -32,10 +32,10 @@ from google.appengine.ext.webapp.util import login_required
import webapp2 as webapp
import oauth2client
from oauth2client import _helpers
from oauth2client import client
from oauth2client import clientsecrets
from oauth2client import transport
from oauth2client import util
from oauth2client.contrib import xsrfutil
# This is a temporary fix for a Google internal issue.
@@ -131,7 +131,7 @@ class AppAssertionCredentials(client.AssertionCredentials):
information to generate and refresh its own access tokens.
"""
@util.positional(2)
@_helpers.positional(2)
def __init__(self, scope, **kwargs):
"""Constructor for AppAssertionCredentials
@@ -143,7 +143,7 @@ class AppAssertionCredentials(client.AssertionCredentials):
or unspecified, the default service account for
the app is used.
"""
self.scope = util.scopes_to_string(scope)
self.scope = _helpers.scopes_to_string(scope)
self._kwargs = kwargs
self.service_account_id = kwargs.get('service_account_id', None)
self._service_account_email = None
@@ -305,7 +305,7 @@ class StorageByKeyName(client.Storage):
and that entities are stored by key_name.
"""
@util.positional(4)
@_helpers.positional(4)
def __init__(self, model, key_name, property_name, cache=None, user=None):
"""Constructor for Storage.
@@ -523,7 +523,7 @@ class OAuth2Decorator(object):
flow = property(get_flow, set_flow)
@util.positional(4)
@_helpers.positional(4)
def __init__(self, client_id, client_secret, scope,
auth_uri=oauth2client.GOOGLE_AUTH_URI,
token_uri=oauth2client.GOOGLE_TOKEN_URI,
@@ -590,7 +590,7 @@ class OAuth2Decorator(object):
self.credentials = None
self._client_id = client_id
self._client_secret = client_secret
self._scope = util.scopes_to_string(scope)
self._scope = _helpers.scopes_to_string(scope)
self._auth_uri = auth_uri
self._token_uri = token_uri
self._revoke_uri = revoke_uri
@@ -805,7 +805,7 @@ class OAuth2Decorator(object):
if (decorator._token_response_param and
credentials.token_response):
resp_json = json.dumps(credentials.token_response)
redirect_uri = util._add_query_parameter(
redirect_uri = _helpers._add_query_parameter(
redirect_uri, decorator._token_response_param,
resp_json)
@@ -849,7 +849,7 @@ class OAuth2DecoratorFromClientSecrets(OAuth2Decorator):
"""
@util.positional(3)
@_helpers.positional(3)
def __init__(self, filename, scope, message=None, cache=None, **kwargs):
"""Constructor
@@ -892,7 +892,7 @@ class OAuth2DecoratorFromClientSecrets(OAuth2Decorator):
self._message = 'Please configure your application for OAuth 2.0.'
@util.positional(2)
@_helpers.positional(2)
def oauth2decorator_from_clientsecrets(filename, scope,
message=None, cache=None):
"""Creates an OAuth2Decorator populated from a clientsecrets file.

5
oauth2client/contrib/xsrfutil.py
View File

@@ -20,7 +20,6 @@ import hmac
import time
from oauth2client import _helpers
from oauth2client import util
__authors__ = [
'"Doug Coker" <dcoker@google.com>',
@@ -34,7 +33,7 @@ DELIMITER = b':'
DEFAULT_TIMEOUT_SECS = 60 * 60
@util.positional(2)
@_helpers.positional(2)
def generate_token(key, user_id, action_id='', when=None):
"""Generates a URL-safe token for the given user, action, time tuple.
@@ -62,7 +61,7 @@ def generate_token(key, user_id, action_id='', when=None):
return token
@util.positional(3)
@_helpers.positional(3)
def validate_token(key, token, user_id, action_id="", current_time=None):
"""Validates that the given token authorizes the user for the action.

6
oauth2client/file.py
View File

@@ -21,8 +21,8 @@ credentials.
import os
import threading
from oauth2client import _helpers
from oauth2client import client
from oauth2client import util
__author__ = 'jcgregorio@google.com (Joe Gregorio)'
@@ -45,7 +45,7 @@ class Storage(client.Storage):
IOError if the file is a symbolic link.
"""
credentials = None
util.validate_file(self._filename)
_helpers.validate_file(self._filename)
try:
f = open(self._filename, 'rb')
content = f.read()
@@ -84,7 +84,7 @@ class Storage(client.Storage):
IOError if the file is a symbolic link.
"""
self._create_file_if_needed()
util.validate_file(self._filename)
_helpers.validate_file(self._filename)
f = open(self._filename, 'w')
f.write(credentials.to_json())
f.close()

3
oauth2client/service_account.py
View File

@@ -25,7 +25,6 @@ from oauth2client import _helpers
from oauth2client import client
from oauth2client import crypt
from oauth2client import transport
from oauth2client import util
_PASSWORD_DEFAULT = 'notasecret'
@@ -110,7 +109,7 @@ class ServiceAccountCredentials(client.AssertionCredentials):
self._service_account_email = service_account_email
self._signer = signer
self._scopes = util.scopes_to_string(scopes)
self._scopes = _helpers.scopes_to_string(scopes)
self._private_key_id = private_key_id
self.client_id = client_id
self._user_agent = user_agent

4
oauth2client/tools.py
View File

@@ -30,8 +30,8 @@ from six.moves import http_client
from six.moves import input
from six.moves import urllib
from oauth2client import _helpers
from oauth2client import client
from oauth2client import util
__author__ = 'jcgregorio@google.com (Joe Gregorio)'
@@ -138,7 +138,7 @@ class ClientRedirectHandler(BaseHTTPServer.BaseHTTPRequestHandler):
"""Do not log messages to stdout while running as cmd. line program."""
@util.positional(3)
@_helpers.positional(3)
def run_flow(flow, storage, flags=None, http=None):
"""Core code for a command-line application.

4
oauth2client/transport.py
View File

@@ -18,7 +18,7 @@ import httplib2
import six
from six.moves import http_client
from oauth2client._helpers import _to_bytes
from oauth2client import _helpers
_LOGGER = logging.getLogger(__name__)
@@ -127,7 +127,7 @@ def clean_headers(headers):
k = str(k)
if not isinstance(v, six.binary_type):
v = str(v)
clean[_to_bytes(k)] = _to_bytes(v)
clean[_helpers._to_bytes(k)] = _helpers._to_bytes(v)
except UnicodeEncodeError:
from oauth2client.client import NonAsciiHeaderError
raise NonAsciiHeaderError(k, ': ', v)

221
oauth2client/util.py
View File

@@ -1,221 +0,0 @@
# Copyright 2014 Google Inc. All rights reserved.
#
# 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 utility library."""
import functools
import inspect
import logging
import os
import warnings
import six
from six.moves import urllib
__author__ = [
'rafek@google.com (Rafe Kaplan)',
'guido@google.com (Guido van Rossum)',
]
__all__ = [
'positional',
'POSITIONAL_WARNING',
'POSITIONAL_EXCEPTION',
'POSITIONAL_IGNORE',
]
logger = logging.getLogger(__name__)
POSITIONAL_WARNING = 'WARNING'
POSITIONAL_EXCEPTION = 'EXCEPTION'
POSITIONAL_IGNORE = 'IGNORE'
POSITIONAL_SET = frozenset([POSITIONAL_WARNING, POSITIONAL_EXCEPTION,
POSITIONAL_IGNORE])
positional_parameters_enforcement = POSITIONAL_WARNING
_SYM_LINK_MESSAGE = 'File: {0}: Is a symbolic link.'
_IS_DIR_MESSAGE = '{0}: Is a directory'
_MISSING_FILE_MESSAGE = 'Cannot access {0}: No such file or directory'
def positional(max_positional_args):
"""A decorator to declare that only the first N arguments my be positional.
This decorator makes it easy to support Python 3 style keyword-only
parameters. For example, in Python 3 it is possible to write::
def fn(pos1, *, kwonly1=None, kwonly1=None):
...
All named parameters after ``*`` must be a keyword::
fn(10, 'kw1', 'kw2') # Raises exception.
fn(10, kwonly1='kw1') # Ok.
Example
^^^^^^^
To define a function like above, do::
@positional(1)
def fn(pos1, kwonly1=None, kwonly2=None):
...
If no default value is provided to a keyword argument, it becomes a
required keyword argument::
@positional(0)
def fn(required_kw):
...
This must be called with the keyword parameter::
fn() # Raises exception.
fn(10) # Raises exception.
fn(required_kw=10) # Ok.
When defining instance or class methods always remember to account for
``self`` and ``cls``::
class MyClass(object):
@positional(2)
def my_method(self, pos1, kwonly1=None):
...
@classmethod
@positional(2)
def my_method(cls, pos1, kwonly1=None):
...
The positional decorator behavior is controlled by
``util.positional_parameters_enforcement``, which may be set to
``POSITIONAL_EXCEPTION``, ``POSITIONAL_WARNING`` or
``POSITIONAL_IGNORE`` to raise an exception, log a warning, or do
nothing, respectively, if a declaration is violated.
Args:
max_positional_arguments: Maximum number of positional arguments. All
parameters after the this index must be
keyword only.
Returns:
A decorator that prevents using arguments after max_positional_args
from being used as positional parameters.
Raises:
TypeError: if a key-word only argument is provided as a positional
parameter, but only if
util.positional_parameters_enforcement is set to
POSITIONAL_EXCEPTION.
"""
def positional_decorator(wrapped):
@functools.wraps(wrapped)
def positional_wrapper(*args, **kwargs):
if len(args) > max_positional_args:
plural_s = ''
if max_positional_args != 1:
plural_s = 's'
message = ('{function}() takes at most {args_max} positional '
'argument{plural} ({args_given} given)'.format(
function=wrapped.__name__,
args_max=max_positional_args,
args_given=len(args),
plural=plural_s))
if positional_parameters_enforcement == POSITIONAL_EXCEPTION:
raise TypeError(message)
elif positional_parameters_enforcement == POSITIONAL_WARNING:
logger.warning(message)
return wrapped(*args, **kwargs)
return positional_wrapper
if isinstance(max_positional_args, six.integer_types):
return positional_decorator
else:
args, _, _, defaults = inspect.getargspec(max_positional_args)
return positional(len(args) - len(defaults))(max_positional_args)
def scopes_to_string(scopes):
"""Converts scope value to a string.
If scopes is a string then it is simply passed through. If scopes is an
iterable then a string is returned that is all the individual scopes
concatenated with spaces.
Args:
scopes: string or iterable of strings, the scopes.
Returns:
The scopes formatted as a single string.
"""
if isinstance(scopes, six.string_types):
return scopes
else:
return ' '.join(scopes)
def string_to_scopes(scopes):
"""Converts stringifed scope value to a list.
If scopes is a list then it is simply passed through. If scopes is an
string then a list of each individual scope is returned.
Args:
scopes: a string or iterable of strings, the scopes.
Returns:
The scopes in a list.
"""
if not scopes:
return []
if isinstance(scopes, six.string_types):
return scopes.split(' ')
else:
return scopes
def _add_query_parameter(url, name, value):
"""Adds a query parameter to a url.
Replaces the current value if it already exists in the URL.
Args:
url: string, url to add the query parameter to.
name: string, query parameter name.
value: string, query parameter value.
Returns:
Updated query parameter. Does not update the url if value is None.
"""
if value is None:
return url
else:
parsed = list(urllib.parse.urlparse(url))
q = dict(urllib.parse.parse_qsl(parsed[4]))
q[name] = value
parsed[4] = urllib.parse.urlencode(q)
return urllib.parse.urlunparse(parsed)
def validate_file(filename):
if os.path.islink(filename):
raise IOError(_SYM_LINK_MESSAGE.format(filename))
elif os.path.isdir(filename):
raise IOError(_IS_DIR_MESSAGE.format(filename))
elif not os.path.isfile(filename):
warnings.warn(_MISSING_FILE_MESSAGE.format(filename))

4
tests/__init__.py
View File

@@ -12,11 +12,11 @@
"""Test package set-up."""
from oauth2client import util
from oauth2client import _helpers
__author__ = 'afshar@google.com (Ali Afshar)'
def setup_package():
"""Run on testing package."""
util.positional_parameters_enforcement = util.POSITIONAL_EXCEPTION
_helpers.positional_parameters_enforcement = _helpers.POSITIONAL_EXCEPTION

4
tests/contrib/django_util/test_django_models.py
View File

@@ -24,7 +24,7 @@ from tests.contrib.django_util.models import CredentialsModel
import unittest2
from oauth2client._helpers import _from_bytes
from oauth2client import _helpers
from oauth2client.client import Credentials
from oauth2client.contrib.django_util.models import CredentialsField
@@ -36,7 +36,7 @@ class TestCredentialsField(unittest2.TestCase):
self.fake_model_field = self.fake_model._meta.get_field('credentials')
self.field = CredentialsField(null=True)
self.credentials = Credentials()
self.pickle_str = _from_bytes(
self.pickle_str = _helpers._from_bytes(
base64.b64encode(pickle.dumps(self.credentials)))
def test_field_is_text(self):

120
tests/test__helpers.py
View File

@@ -11,13 +11,133 @@
# 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.
"""Unit tests for oauth2client._helpers."""
import mock
import unittest2
from oauth2client import _helpers
__author__ = 'jcgregorio@google.com (Joe Gregorio)'
class PositionalTests(unittest2.TestCase):
def test_usage(self):
_helpers.positional_parameters_enforcement = (
_helpers.POSITIONAL_EXCEPTION)
# 1 positional arg, 1 keyword-only arg.
@_helpers.positional(1)
def function(pos, kwonly=None):
return True
self.assertTrue(function(1))
self.assertTrue(function(1, kwonly=2))
with self.assertRaises(TypeError):
function(1, 2)
# No positional, but a required keyword arg.
@_helpers.positional(0)
def function2(required_kw):
return True
self.assertTrue(function2(required_kw=1))
with self.assertRaises(TypeError):
function2(1)
# Unspecified positional, should automatically figure out 1 positional
# 1 keyword-only (same as first case above).
@_helpers.positional
def function3(pos, kwonly=None):
return True
self.assertTrue(function3(1))
self.assertTrue(function3(1, kwonly=2))
with self.assertRaises(TypeError):
function3(1, 2)
@mock.patch('oauth2client._helpers.logger')
def test_enforcement_warning(self, mock_logger):
_helpers.positional_parameters_enforcement = (
_helpers.POSITIONAL_WARNING)
@_helpers.positional(1)
def function(pos, kwonly=None):
return True
self.assertTrue(function(1, 2))
self.assertTrue(mock_logger.warning.called)
@mock.patch('oauth2client._helpers.logger')
def test_enforcement_ignore(self, mock_logger):
_helpers.positional_parameters_enforcement = _helpers.POSITIONAL_IGNORE
@_helpers.positional(1)
def function(pos, kwonly=None):
return True
self.assertTrue(function(1, 2))
self.assertFalse(mock_logger.warning.called)
class ScopeToStringTests(unittest2.TestCase):
def test_iterables(self):
cases = [
('', ''),
('', ()),
('', []),
('', ('',)),
('', ['', ]),
('a', ('a',)),
('b', ['b', ]),
('a b', ['a', 'b']),
('a b', ('a', 'b')),
('a b', 'a b'),
('a b', (s for s in ['a', 'b'])),
]
for expected, case in cases:
self.assertEqual(expected, _helpers.scopes_to_string(case))
class StringToScopeTests(unittest2.TestCase):
def test_conversion(self):
cases = [
(['a', 'b'], ['a', 'b']),
('', []),
('a', ['a']),
('a b c d e f', ['a', 'b', 'c', 'd', 'e', 'f']),
]
for case, expected in cases:
self.assertEqual(expected, _helpers.string_to_scopes(case))
class AddQueryParameterTests(unittest2.TestCase):
def test__add_query_parameter(self):
self.assertEqual(
_helpers._add_query_parameter('/action', 'a', None),
'/action')
self.assertEqual(
_helpers._add_query_parameter('/action', 'a', 'b'),
'/action?a=b')
self.assertEqual(
_helpers._add_query_parameter('/action?a=b', 'a', 'c'),
'/action?a=c')
# Order is non-deterministic.
self.assertIn(
_helpers._add_query_parameter('/action?a=b', 'c', 'd'),
['/action?a=b&c=d', '/action?c=d&a=b'])
self.assertEqual(
_helpers._add_query_parameter('/action', 'a', ' ='),
'/action?a=+%3D')
class Test__parse_pem_key(unittest2.TestCase):
def test_valid_input(self):

13
tests/test_client.py
View File

@@ -38,7 +38,6 @@ from oauth2client import _helpers
from oauth2client import client
from oauth2client import clientsecrets
from oauth2client import service_account
from oauth2client import util
from . import http_mock
__author__ = 'jcgregorio@google.com (Joe Gregorio)'
@@ -882,14 +881,14 @@ class BasicCredentialsTests(unittest2.TestCase):
user_agent, revoke_uri=oauth2client.GOOGLE_REVOKE_URI,
scopes='foo', token_info_uri=oauth2client.GOOGLE_TOKEN_INFO_URI)
# Provoke a failure if @util.positional is not respected.
# Provoke a failure if @_helpers.positional is not respected.
self.old_positional_enforcement = (
util.positional_parameters_enforcement)
util.positional_parameters_enforcement = (
util.POSITIONAL_EXCEPTION)
_helpers.positional_parameters_enforcement)
_helpers.positional_parameters_enforcement = (
_helpers.POSITIONAL_EXCEPTION)
def tearDown(self):
util.positional_parameters_enforcement = (
_helpers.positional_parameters_enforcement = (
self.old_positional_enforcement)
def test_token_refresh_success(self):
@@ -911,7 +910,7 @@ class BasicCredentialsTests(unittest2.TestCase):
# Tests that OAuth2Credentials doesn't introduce new method
# constraints. Formerly, OAuth2Credentials.authorize monkeypatched the
# request method of the passed in HTTP object with a wrapper annotated
# with @util.positional(1). Since the original method has no such
# with @_helpers.positional(1). Since the original method has no such
# annotation, that meant that the wrapper was violating the contract of
# the original method by adding a new requirement to it. And in fact
# the wrapper itself doesn't even respect that requirement. So before

4
tests/test_file.py
View File

@@ -31,9 +31,9 @@ import six
from six.moves import http_client
import unittest2
from oauth2client import _helpers
from oauth2client import client
from oauth2client import file
from oauth2client import util
from .http_mock import HttpMockSequence
try:
@@ -83,7 +83,7 @@ class OAuth2ClientFileTests(unittest2.TestCase):
storage = file.Storage(FILENAME)
credentials = storage.get()
warn_mock.assert_called_with(
util._MISSING_FILE_MESSAGE.format(FILENAME))
_helpers._MISSING_FILE_MESSAGE.format(FILENAME))
self.assertIsNone(credentials)
def test_directory_file_storage(self):

122
tests/test_util.py
View File

@@ -1,122 +0,0 @@
"""Unit tests for oauth2client.util."""
import mock
import unittest2
from oauth2client import util
__author__ = 'jcgregorio@google.com (Joe Gregorio)'
class PositionalTests(unittest2.TestCase):
def test_usage(self):
util.positional_parameters_enforcement = util.POSITIONAL_EXCEPTION
# 1 positional arg, 1 keyword-only arg.
@util.positional(1)
def fn(pos, kwonly=None):
return True
self.assertTrue(fn(1))
self.assertTrue(fn(1, kwonly=2))
with self.assertRaises(TypeError):
fn(1, 2)
# No positional, but a required keyword arg.
@util.positional(0)
def fn2(required_kw):
return True
self.assertTrue(fn2(required_kw=1))
with self.assertRaises(TypeError):
fn2(1)
# Unspecified positional, should automatically figure out 1 positional
# 1 keyword-only (same as first case above).
@util.positional
def fn3(pos, kwonly=None):
return True
self.assertTrue(fn3(1))
self.assertTrue(fn3(1, kwonly=2))
with self.assertRaises(TypeError):
fn3(1, 2)
@mock.patch('oauth2client.util.logger')
def test_enforcement_warning(self, mock_logger):
util.positional_parameters_enforcement = util.POSITIONAL_WARNING
@util.positional(1)
def fn(pos, kwonly=None):
return True
self.assertTrue(fn(1, 2))
self.assertTrue(mock_logger.warning.called)
@mock.patch('oauth2client.util.logger')
def test_enforcement_ignore(self, mock_logger):
util.positional_parameters_enforcement = util.POSITIONAL_IGNORE
@util.positional(1)
def fn(pos, kwonly=None):
return True
self.assertTrue(fn(1, 2))
self.assertFalse(mock_logger.warning.called)
class ScopeToStringTests(unittest2.TestCase):
def test_iterables(self):
cases = [
('', ''),
('', ()),
('', []),
('', ('',)),
('', ['', ]),
('a', ('a',)),
('b', ['b', ]),
('a b', ['a', 'b']),
('a b', ('a', 'b')),
('a b', 'a b'),
('a b', (s for s in ['a', 'b'])),
]
for expected, case in cases:
self.assertEqual(expected, util.scopes_to_string(case))
class StringToScopeTests(unittest2.TestCase):
def test_conversion(self):
cases = [
(['a', 'b'], ['a', 'b']),
('', []),
('a', ['a']),
('a b c d e f', ['a', 'b', 'c', 'd', 'e', 'f']),
]
for case, expected in cases:
self.assertEqual(expected, util.string_to_scopes(case))
class AddQueryParameterTests(unittest2.TestCase):
def test__add_query_parameter(self):
self.assertEqual(
util._add_query_parameter('/action', 'a', None),
'/action')
self.assertEqual(
util._add_query_parameter('/action', 'a', 'b'),
'/action?a=b')
self.assertEqual(
util._add_query_parameter('/action?a=b', 'a', 'c'),
'/action?a=c')
# Order is non-deterministic.
self.assertIn(
util._add_query_parameter('/action?a=b', 'c', 'd'),
['/action?a=b&c=d', '/action?c=d&a=b'])
self.assertEqual(
util._add_query_parameter('/action', 'a', ' ='),
'/action?a=+%3D')