Initial tests for OAuthSignIn
This commit is contained in:
@@ -3,7 +3,7 @@ import attr
|
|||||||
import json
|
import json
|
||||||
|
|
||||||
from rauth import OAuth2Service
|
from rauth import OAuth2Service
|
||||||
from flask import current_app, url_for, request, redirect, session
|
from flask import current_app, url_for, request, redirect, session, Response
|
||||||
|
|
||||||
|
|
||||||
@attr.s
|
@attr.s
|
||||||
@@ -16,18 +16,29 @@ class OAuthUserResponse:
|
|||||||
email = attr.ib(validator=attr.validators.instance_of(str))
|
email = attr.ib(validator=attr.validators.instance_of(str))
|
||||||
|
|
||||||
|
|
||||||
|
class ProviderConfigurationMissing(ValueError):
|
||||||
|
"""Raised when an OAuth provider is used but not configured."""
|
||||||
|
|
||||||
|
|
||||||
|
class ProviderNotImplemented(ValueError):
|
||||||
|
"""Raised when a provider is requested that does not exist."""
|
||||||
|
|
||||||
|
|
||||||
class OAuthSignIn(metaclass=abc.ABCMeta):
|
class OAuthSignIn(metaclass=abc.ABCMeta):
|
||||||
providers = None
|
_providers = None # initialized in get_provider()
|
||||||
|
|
||||||
def __init__(self, provider_name):
|
def __init__(self, provider_name):
|
||||||
self.provider_name = provider_name
|
self.provider_name = provider_name
|
||||||
|
try:
|
||||||
credentials = current_app.config['OAUTH_CREDENTIALS'][provider_name]
|
credentials = current_app.config['OAUTH_CREDENTIALS'][provider_name]
|
||||||
|
except KeyError:
|
||||||
|
raise ProviderConfigurationMissing(f'Missing OAuth credentials for {provider_name}')
|
||||||
self.consumer_id = credentials['id']
|
self.consumer_id = credentials['id']
|
||||||
self.consumer_secret = credentials['secret']
|
self.consumer_secret = credentials['secret']
|
||||||
|
|
||||||
@abc.abstractmethod
|
@abc.abstractmethod
|
||||||
def authorize(self) -> redirect:
|
def authorize(self) -> Response:
|
||||||
"""Redirect to the corret authorization endpoint for the current provider
|
"""Redirect to the correct authorization endpoint for the current provider.
|
||||||
|
|
||||||
Depending on the provider, we sometimes have to specify a different
|
Depending on the provider, we sometimes have to specify a different
|
||||||
'scope'.
|
'scope'.
|
||||||
@@ -36,7 +47,7 @@ class OAuthSignIn(metaclass=abc.ABCMeta):
|
|||||||
|
|
||||||
@abc.abstractmethod
|
@abc.abstractmethod
|
||||||
def callback(self) -> OAuthUserResponse:
|
def callback(self) -> OAuthUserResponse:
|
||||||
"""Callback performed after authorizing the user
|
"""Callback performed after authorizing the user.
|
||||||
|
|
||||||
This is usually a request to a protected /me endpoint to query for
|
This is usually a request to a protected /me endpoint to query for
|
||||||
user information, such as user id and email address.
|
user information, such as user id and email address.
|
||||||
@@ -48,14 +59,17 @@ class OAuthSignIn(metaclass=abc.ABCMeta):
|
|||||||
_external=True)
|
_external=True)
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def get_provider(cls, provider_name):
|
def get_provider(cls, provider_name) -> 'OAuthSignIn':
|
||||||
if cls.providers is None:
|
if cls._providers is None:
|
||||||
cls.providers = {}
|
cls._providers = {}
|
||||||
# TODO convert to the new __init_subclass__
|
# TODO convert to the new __init_subclass__
|
||||||
for provider_class in cls.__subclasses__():
|
for provider_class in cls.__subclasses__():
|
||||||
provider = provider_class()
|
provider = provider_class()
|
||||||
cls.providers[provider.provider_name] = provider
|
cls._providers[provider.provider_name] = provider
|
||||||
return cls.providers[provider_name]
|
try:
|
||||||
|
return cls._providers[provider_name]
|
||||||
|
except KeyError:
|
||||||
|
raise ProviderNotImplemented(f'No such OAuth provider {provider_name}')
|
||||||
|
|
||||||
|
|
||||||
class BlenderIdSignIn(OAuthSignIn):
|
class BlenderIdSignIn(OAuthSignIn):
|
||||||
|
@@ -93,13 +93,15 @@ FULL_FILE_ACCESS_ROLES = {'admin', 'subscriber', 'demo'}
|
|||||||
BLENDER_ID_CLIENT_ID = 'SPECIAL-SNOWFLAKE-57'
|
BLENDER_ID_CLIENT_ID = 'SPECIAL-SNOWFLAKE-57'
|
||||||
BLENDER_ID_SUBCLIENT_ID = 'PILLAR'
|
BLENDER_ID_SUBCLIENT_ID = 'PILLAR'
|
||||||
|
|
||||||
# Collection of supported OAuth providers (Blender ID, Facebook and Google). Example entry:
|
# Collection of supported OAuth providers (Blender ID, Facebook and Google).
|
||||||
|
# Example entry:
|
||||||
|
# OAUTH_CREDENTIALS = {
|
||||||
# 'blender-id': {
|
# 'blender-id': {
|
||||||
# 'id': 'CLOUD-OF-SNOWFLAKES-43',
|
# 'id': 'CLOUD-OF-SNOWFLAKES-43',
|
||||||
# 'secret': 'thesecret',
|
# 'secret': 'thesecret',
|
||||||
# 'base_url': 'http://blender_id:8000/'
|
# 'base_url': 'http://blender_id:8000/'
|
||||||
# }
|
# }
|
||||||
|
# }
|
||||||
OAUTH_CREDENTIALS = {}
|
OAUTH_CREDENTIALS = {}
|
||||||
|
|
||||||
# See https://docs.python.org/2/library/logging.config.html#configuration-dictionary-schema
|
# See https://docs.python.org/2/library/logging.config.html#configuration-dictionary-schema
|
||||||
|
@@ -16,3 +16,19 @@ STORAGE_BACKEND = 'local'
|
|||||||
EXTERNAL_SUBSCRIPTIONS_MANAGEMENT_SERVER = "http://store.localhost/api"
|
EXTERNAL_SUBSCRIPTIONS_MANAGEMENT_SERVER = "http://store.localhost/api"
|
||||||
|
|
||||||
SECRET_KEY = '12345'
|
SECRET_KEY = '12345'
|
||||||
|
|
||||||
|
OAUTH_CREDENTIALS = {
|
||||||
|
'blender-id': {
|
||||||
|
'id': 'blender-id-app-id',
|
||||||
|
'secret': 'blender-id–secret',
|
||||||
|
'base_url': 'http://blender_id:8000/'
|
||||||
|
},
|
||||||
|
'facebook': {
|
||||||
|
'id': 'fb-app-id',
|
||||||
|
'secret': 'facebook-secret'
|
||||||
|
},
|
||||||
|
'google': {
|
||||||
|
'id': 'google-app-id',
|
||||||
|
'secret': 'google-secret'
|
||||||
|
}
|
||||||
|
}
|
||||||
|
31
tests/test_api/test_oauth.py
Normal file
31
tests/test_api/test_oauth.py
Normal file
@@ -0,0 +1,31 @@
|
|||||||
|
from pillar.tests import AbstractPillarTest
|
||||||
|
|
||||||
|
|
||||||
|
class OAuthTests(AbstractPillarTest):
|
||||||
|
def setUp(self, **kwargs):
|
||||||
|
super().setUp(**kwargs)
|
||||||
|
self.enter_app_context()
|
||||||
|
|
||||||
|
def test_providers_init(self):
|
||||||
|
from pillar.auth.oauth import OAuthSignIn, BlenderIdSignIn
|
||||||
|
|
||||||
|
blender_id_oauth_provider = OAuthSignIn.get_provider('blender-id')
|
||||||
|
self.assertIsInstance(blender_id_oauth_provider, BlenderIdSignIn)
|
||||||
|
self.assertEqual(blender_id_oauth_provider.service.base_url, 'http://blender_id:8000/api/')
|
||||||
|
|
||||||
|
def test_provider_not_implemented(self):
|
||||||
|
from pillar.auth.oauth import OAuthSignIn, ProviderNotImplemented
|
||||||
|
|
||||||
|
with self.assertRaises(ProviderNotImplemented):
|
||||||
|
OAuthSignIn.get_provider('jonny')
|
||||||
|
|
||||||
|
def test_provider_not_configured(self):
|
||||||
|
from pillar.auth.oauth import OAuthSignIn, ProviderConfigurationMissing
|
||||||
|
|
||||||
|
# Before we start this test, the providers dict
|
||||||
|
# may not be initialized yet.
|
||||||
|
self.assertIsNone(OAuthSignIn._providers)
|
||||||
|
|
||||||
|
del self.app.config['OAUTH_CREDENTIALS']['blender-id']
|
||||||
|
with self.assertRaises(ProviderConfigurationMissing):
|
||||||
|
OAuthSignIn.get_provider('blender-id')
|
Reference in New Issue
Block a user