python/fastapi-users
1
0
Fork 0
mirror of https://github.com/fastapi-users/fastapi-users.git synced 2026-03-13 07:49:55 +08:00
Code Issues Packages Projects Releases Wiki Activity

Implement OAuth2 flow (#88)

Browse Source 
* Move users router in sub-module

* Factorize UserRouter into EventHandlersRouter

* Implement OAuth registration/login router

* Apply isort/black

* Remove temporary pytest marker

* Fix httpx-oauth version in lock file

* Ensure ON_AFTER_REGISTER event is triggered on OAuth registration

* Add API on FastAPIUsers to generate an OAuth router

* Improve test coverage of FastAPIUsers

* Small fixes

* Write the OAuth documentation

* Fix SQL unit-tests by avoiding collisions in SQLite db files
This commit is contained in:
François Voron
2020-01-17 11:43:17 +01:00
committed by GitHub
parent 54aefea59a
commit 88b133d41c
32 changed files with 1723 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@@ -48,7 +48,7 @@ coverage.xml
.pytest_cache/
junit/
junit.xml
test.db*
test*.db*
# Translations
*.mo

1
Pipfile
View File

@@ -24,6 +24,7 @@ markdown-include = "*"
pygments = "*"
pymdown-extensions = "*"
bumpversion = "*"
httpx-oauth = "*"
[packages]
fastapi = "==0.46.0"

81
Pipfile.lock generated
View File

@@ -1,7 +1,7 @@
{
"_meta": {
"hash": {
"sha256": "bbb79297ba3be63e11d0753c88bbf35c7ca6e7d5c0920083a39430322c5938c0"
"sha256": "3723c798fdb2c6d9980d81cfb6d3494c6c3d55cb8052168d93172a18b5231a59"
},
"pipfile-spec": 6,
"requires": {
@@ -238,9 +238,9 @@
},
"pypika": {
"hashes": [
"sha256:a4d80829a065047c39b7838aafb440761bd9a306977c0e6f78caf5eaa056c9e4"
"sha256:69552b11a400790bf4ebc76aff2b39ef1e1ac581ab53bca80144f8e9848b82b9"
],
"version": "==0.35.18"
"version": "==0.35.19"
},
"python-multipart": {
"hashes": [
@@ -251,10 +251,10 @@
},
"six": {
"hashes": [
"sha256:1f1b7d42e254082a9db6279deae68afb421ceba6158efa6131de7b3003ee93fd",
"sha256:30f610279e8b2578cab6db20741130331735c781b56053c59c4076da27f06b66"
"sha256:236bdbdce46e6e6a3d61a337c0f8b763ca1e8717c03b369e87a7ec7ce1319c0a",
"sha256:8f3cd2e254d8f793e7f3d6d9df77b92252b52637291d0f0da013c76ea2724b6c"
],
"version": "==1.13.0"
"version": "==1.14.0"
},
"sqlalchemy": {
"hashes": [
@@ -447,12 +447,61 @@
],
"version": "==2.2.0"
},
"h11": {
"hashes": [
"sha256:33d4bca7be0fa039f4e84d50ab00531047e53d6ee8ffbc83501ea602c169cae1",
"sha256:4bc6d6a1238b7615b266ada57e0618568066f57dd6fa967d1290ec9309b2f2f1"
],
"version": "==0.9.0"
},
"h2": {
"hashes": [
"sha256:ac377fcf586314ef3177bfd90c12c7826ab0840edeb03f0f24f511858326049e",
"sha256:b8a32bd282594424c0ac55845377eea13fa54fe4a8db012f3a198ed923dc3ab4"
],
"version": "==3.1.1"
},
"hpack": {
"hashes": [
"sha256:0edd79eda27a53ba5be2dfabf3b15780928a0dff6eb0c60a3d6767720e970c89",
"sha256:8eec9c1f4bfae3408a3f30500261f7e6a65912dc138526ea054f9ad98892e9d2"
],
"version": "==3.0.0"
},
"hstspreload": {
"hashes": [
"sha256:1cde56803877ad3aa0a4aa91d0c9c52203f4b7ceb93280c87ad399d3fd79679c"
],
"version": "==2020.1.17"
},
"htmlmin": {
"hashes": [
"sha256:50c1ef4630374a5d723900096a961cff426dff46b48f34d194a81bbe14eca178"
],
"version": "==0.1.12"
},
"httpx": {
"hashes": [
"sha256:2f2a7ce82cd4373af5a72d2646e20e4061da36b637ebcb88b93265083f132bff",
"sha256:abc46081611a86270d92c8fcaaf148f1eb33510c92096075fac27288bea3a9b2"
],
"version": "==0.11.0"
},
"httpx-oauth": {
"hashes": [
"sha256:940f82eed06fe2abe9c2ab414ad5c4f7feb03009dc24bd5ea0c01473334424b8",
"sha256:ed34cda0748b66eda7f874290ca5e7e522487cd78a15df25bf12e71bdb47ac27"
],
"index": "pypi",
"version": "==0.2.0"
},
"hyperframe": {
"hashes": [
"sha256:5187962cb16dcc078f23cb5a4b110098d546c3f41ff2d4038a9896893bbd0b40",
"sha256:a9f5c17f2cc3c719b917c4f33ed1c61bd1f8dfac4b1bd23b7c80b3400971b41f"
],
"version": "==5.2.0"
},
"idna": {
"hashes": [
"sha256:c357b3f628cf53ae2c4c05627ecc484553142ca23264e593d327bcde5e9c3407",
@@ -775,12 +824,26 @@
"index": "pypi",
"version": "==2.22.0"
},
"rfc3986": {
"hashes": [
"sha256:0344d0bd428126ce554e7ca2b61787b6a28d2bbd19fc70ed2dd85efe31176405",
"sha256:df4eba676077cefb86450c8f60121b9ae04b94f65f85b69f3f731af0516b7b18"
],
"version": "==1.3.2"
},
"six": {
"hashes": [
"sha256:1f1b7d42e254082a9db6279deae68afb421ceba6158efa6131de7b3003ee93fd",
"sha256:30f610279e8b2578cab6db20741130331735c781b56053c59c4076da27f06b66"
"sha256:236bdbdce46e6e6a3d61a337c0f8b763ca1e8717c03b369e87a7ec7ce1319c0a",
"sha256:8f3cd2e254d8f793e7f3d6d9df77b92252b52637291d0f0da013c76ea2724b6c"
],
"version": "==1.13.0"
"version": "==1.14.0"
},
"sniffio": {
"hashes": [
"sha256:20ed6d5b46f8ae136d00b9dcb807615d83ed82ceea6b2058cecb696765246da5",
"sha256:8e3810100f69fe0edd463d02ad407112542a11ffdc29f67db2bf3771afb87a21"
],
"version": "==1.1.0"
},
"snowballstemmer": {
"hashes": [

144
docs/configuration/oauth.md Normal file
View File

@@ -0,0 +1,144 @@
# OAuth2
FastAPI Users provides an optional OAuth2 authentication support. It relies on [HTTPX OAuth library](https://frankie567.github.io/httpx-oauth/), which is a pure-async implementation of OAuth2.
## Installation
You should install the library with the optional dependencies for OAuth:
```sh
pip install fastapi-users[sqlalchemy,oauth]
```
```sh
pip install fastapi-users[mongodb,oauth]
```
```sh
pip install fastapi-users[tortoise-orm,oauth]
```
## Configuration
### Instantiate an OAuth2 client
You first need to get an HTTPX OAuth client instance. [Read the documentation](https://frankie567.github.io/httpx-oauth/oauth2/) for more information.
```py
from httpx_oauth.clients.google import GoogleOAuth2
google_oauth_client = GoogleOAuth2("CLIENT_ID", "CLIENT_SECRET")
```
### Setup the models
The user models differ a bit from the standard one as we have to have a way to store the OAuth information (access tokens, account ids...).
```py
from fastapi_users import models
class User(models.BaseUser, models.BaseOAuthAccountMixin):
pass
class UserCreate(User, models.BaseUserCreate):
pass
class UserUpdate(User, models.BaseUserUpdate):
pass
class UserDB(User, models.BaseUserDB):
pass
```
Notice that we inherit from the `BaseOAuthAccountMixin`, which adds a `List` of `BaseOAuthAccount` objects. This object is structured like this:
* `id` (`str`) – Unique identifier of the user. Default to a **UUID4**.
* `oauth_name` (`str`) – Name of the OAuth service. It corresponds to the `name` property of the OAuth client.
* `access_token` (`str`) – Access token.
* `expires_at` (`int`) - Timestamp at which the access token is expired.
* `refresh_token` (`Optional[str]`) – On services that support it, a token to get a fresh access token.
* `account_id` (`str`) - Identifier of the OAuth account on the corresponding service.
* `account_email` (`str`) - Email address of the OAuth account on the corresponding service.
### Setup the database adapter
#### SQLAlchemy
You'll need to define the table for storing the OAuth account model. We provide a base one for this:
```py
from fastapi_users.db.sqlalchemy import SQLAlchemyBaseOAuthAccountTable
class OAuthAccount(SQLAlchemyBaseOAuthAccountTable, Base):
pass
```
Then, you should declare it on the database adapter:
```py
user_db = SQLAlchemyUserDatabase(UserDB, database, User.__table__, OAuthAccount.__table__)
```
#### MongoDB
Nothing to do, the [basic configuration](./databases/mongodb.md) is enough.
#### Tortoise ORM
You'll need to define the Tortoise model for storing the OAuth account model. We provide a base one for this:
```py
from fastapi_users.db.tortoise import TortoiseBaseOAuthAccountModel
class OAuthAccount(TortoiseBaseOAuthAccountModel):
user = fields.ForeignKeyField("models.User", related_name="oauth_accounts")
```
!!! warning
Note that you shouls define the foreign key yourself, so that you can point it the user model in your namespace.
Then, you should declare it on the database adapter:
```py
user_db = TortoiseUserDatabase(UserDB, User, OAuthAccount)
```
### Generate a router
Once you have a `FastAPIUsers` instance, you can make it generate a single OAuth router for the given client.
```py
from fastapi import FastAPI
from fastapi_users import FastAPIUsers
from httpx_oauth.clients.google import GoogleOAuth2
google_oauth_client = GoogleOAuth2("CLIENT_ID", "CLIENT_SECRET")
app = FastAPI()
fastapi_users = FastAPIUsers(
user_db, auth_backends, User, UserCreate, UserUpdate, UserDB, SECRET,
)
google_oauth_router = fastapi_users.get_oauth_router(google_oauth_client, SECRET)
app.include_router(google_oauth_router, prefix="/google-oauth", tags=["users"])
```
### Full example
``` py tab="SQLAlchemy"
{!./src/oauth_full_sqlalchemy.py!}
```
```py tab="MongoDB"
{!./src/oauth_full_mongodb.py!}
```
```py tab="Tortoise ORM"
{!./src/oauth_full_tortoise.py!}
```

57
docs/src/oauth_full_mongodb.py Normal file
View File

@@ -0,0 +1,57 @@
import motor.motor_asyncio
from fastapi import FastAPI
from fastapi_users import FastAPIUsers, models
from fastapi_users.authentication import JWTAuthentication
from fastapi_users.db import MongoDBUserDatabase
from httpx_oauth.clients.google import GoogleOAuth2
DATABASE_URL = "mongodb://localhost:27017"
SECRET = "SECRET"
google_oauth_client = GoogleOAuth2("CLIENT_ID", "CLIENT_SECRET")
class User(models.BaseUser, models.BaseOAuthAccountMixin):
pass
class UserCreate(User, models.BaseUserCreate):
pass
class UserUpdate(User, models.BaseUserUpdate):
pass
class UserDB(User, models.BaseUserDB):
pass
client = motor.motor_asyncio.AsyncIOMotorClient(DATABASE_URL)
db = client["database_name"]
collection = db["users"]
user_db = MongoDBUserDatabase(UserDB, collection)
auth_backends = [
JWTAuthentication(secret=SECRET, lifetime_seconds=3600),
]
app = FastAPI()
fastapi_users = FastAPIUsers(
user_db, auth_backends, User, UserCreate, UserUpdate, UserDB, SECRET,
)
app.include_router(fastapi_users.router, prefix="/users", tags=["users"])
google_oauth_router = fastapi_users.get_oauth_router(google_oauth_client, SECRET)
app.include_router(google_oauth_router, prefix="/google-oauth", tags=["users"])
@fastapi_users.on_after_register()
def on_after_register(user: User):
print(f"User {user.id} has registered.")
@fastapi_users.on_after_forgot_password()
def on_after_forgot_password(user: User, token: str):
print(f"User {user.id} has forgot their password. Reset token: {token}")

22
docs/src/full_sqlalchemy.py → docs/src/oauth_full_sqlalchemy.py
View File

@@ -3,14 +3,22 @@ import sqlalchemy
from fastapi import FastAPI
from fastapi_users import FastAPIUsers, models
from fastapi_users.authentication import JWTAuthentication
from fastapi_users.db import SQLAlchemyBaseUserTable, SQLAlchemyUserDatabase
from fastapi_users.db import (
SQLAlchemyBaseOAuthAccountTable,
SQLAlchemyBaseUserTable,
SQLAlchemyUserDatabase,
)
from httpx_oauth.clients.google import GoogleOAuth2
from sqlalchemy.ext.declarative import DeclarativeMeta, declarative_base
DATABASE_URL = "sqlite:///./test.db"
SECRET = "SECRET"
class User(models.BaseUser):
google_oauth_client = GoogleOAuth2("CLIENT_ID", "CLIENT_SECRET")
class User(models.BaseUser, models.BaseOAuthAccountMixin):
pass
@@ -34,13 +42,18 @@ class UserTable(Base, SQLAlchemyBaseUserTable):
pass
class OAuthAccount(SQLAlchemyBaseOAuthAccountTable, Base):
pass
engine = sqlalchemy.create_engine(
DATABASE_URL, connect_args={"check_same_thread": False}
)
Base.metadata.create_all(engine)
users = UserTable.__table__
user_db = SQLAlchemyUserDatabase(UserDB, database, users)
oauth_accounts = OAuthAccount.__table__
user_db = SQLAlchemyUserDatabase(UserDB, database, users, oauth_accounts)
auth_backends = [
@@ -53,6 +66,9 @@ fastapi_users = FastAPIUsers(
)
app.include_router(fastapi_users.router, prefix="/users", tags=["users"])
google_oauth_router = fastapi_users.get_oauth_router(google_oauth_client, SECRET)
app.include_router(google_oauth_router, prefix="/google-oauth", tags=["users"])
@fastapi_users.on_after_register()
def on_after_register(user: User):

22
docs/src/full_tortoise.py → docs/src/oauth_full_tortoise.py
View File

@@ -1,14 +1,23 @@
from fastapi import FastAPI
from fastapi_users import FastAPIUsers, models
from fastapi_users.authentication import JWTAuthentication
from fastapi_users.db import TortoiseBaseUserModel, TortoiseUserDatabase
from fastapi_users.db import (
TortoiseBaseOAuthAccountModel,
TortoiseBaseUserModel,
TortoiseUserDatabase,
)
from httpx_oauth.clients.google import GoogleOAuth2
from tortoise import fields
from tortoise.contrib.starlette import register_tortoise
DATABASE_URL = "sqlite://./test.db"
SECRET = "SECRET"
class User(models.BaseUser):
google_oauth_client = GoogleOAuth2("CLIENT_ID", "CLIENT_SECRET")
class User(models.BaseUser, models.BaseOAuthAccountMixin):
pass
@@ -28,7 +37,11 @@ class UserModel(TortoiseBaseUserModel):
pass
user_db = TortoiseUserDatabase(UserDB, UserModel)
class OAuthAccountModel(TortoiseBaseOAuthAccountModel):
user = fields.ForeignKeyField("models.UserModel", related_name="oauth_accounts")
user_db = TortoiseUserDatabase(UserDB, UserModel, OAuthAccountModel)
app = FastAPI()
register_tortoise(app, db_url=DATABASE_URL, modules={"models": ["test"]})
@@ -41,6 +54,9 @@ fastapi_users = FastAPIUsers(
)
app.include_router(fastapi_users.router, prefix="/users", tags=["users"])
google_oauth_router = fastapi_users.get_oauth_router(google_oauth_client, SECRET)
app.include_router(google_oauth_router, prefix="/google-oauth", tags=["users"])
@fastapi_users.on_after_register()
def on_after_register(user: User):

45
docs/usage/routes.md
View File

@@ -97,6 +97,51 @@ Reset a password. Requires the token generated by the `/forgot-password` route.
}
```
### OAuth routes
Each OAuth router you define will expose the two following routes.
#### `GET /authorize`
Return the authorization URL for the OAuth service where you should redirect your user.
!!! abstract "Query parameters"
* `authentication_backend`: Name of a defined [authentication method](../configuration/authentication/index.md) to use to authenticate the user on successful callback.
* `scopes`: Optional list of scopes to ask for. Expected format: `scopes=a&scopes=b`.
!!! success "`200 OK`"
```json
{
"authorization_url": "https://www.tintagel.bt/oauth/authorize?client_id=CLIENT_ID&scopes=a+b&redirect_uri=https://www.camelot.bt/oauth/callback"
}
```
!!! fail "`422 Validation Error`"
!!! fail "`400 Bad Request`"
Unknown authentication backend.
#### `GET /callback`
Handle the OAuth callback.
!!! abstract "Query parameters"
* `code`: OAuth callback code.
* `state`: State token.
* `error`: OAuth error.
Depending on the situation, several things can happen:
* The OAuth account exists in database and is linked to a user:
* OAuth account is updated in database with fresh access token.
* The user is authenticated following the chosen [authentication method](../configuration/authentication/index.md).
* The OAuth account doesn't exist in database but a user with the same email address exists:
* OAuth account is linked to the user.
* The user is authenticated following the chosen [authentication method](../configuration/authentication/index.md).
* The OAuth account doesn't exist in database and no user with the email address exists:
* A new user is created and linked to the OAuth account.
* The user is authenticated following the chosen [authentication method](../configuration/authentication/index.md).
## Authenticated
### `GET /me`

2
fastapi_users/db/__init__.py
View File

@@ -7,6 +7,7 @@ except ImportError: # pragma: no cover
try:
from fastapi_users.db.sqlalchemy import ( # noqa: F401
SQLAlchemyBaseOAuthAccountTable,
SQLAlchemyBaseUserTable,
SQLAlchemyUserDatabase,
)
@@ -15,6 +16,7 @@ except ImportError: # pragma: no cover
try:
from fastapi_users.db.tortoise import ( # noqa: F401
TortoiseBaseOAuthAccountModel,
TortoiseBaseUserModel,
TortoiseUserDatabase,
)

4
fastapi_users/db/base.py
View File

@@ -30,6 +30,10 @@ class BaseUserDatabase(Generic[UD]):
"""Get a single user by email."""
raise NotImplementedError()
async def get_by_oauth_account(self, oauth: str, account_id: str) -> Optional[UD]:
"""Get a single user by OAuth account id."""
raise NotImplementedError()
async def create(self, user: UD) -> UD:
"""Create a user."""
raise NotImplementedError()

9
fastapi_users/db/mongodb.py
View File

@@ -33,6 +33,15 @@ class MongoDBUserDatabase(BaseUserDatabase[UD]):
user = await self.collection.find_one({"email": email})
return self.user_db_model(**user) if user else None
async def get_by_oauth_account(self, oauth: str, account_id: str) -> Optional[UD]:
user = await self.collection.find_one(
{
"oauth_accounts.oauth_name": oauth,
"oauth_accounts.account_id": account_id,
}
)
return self.user_db_model(**user) if user else None
async def create(self, user: UD) -> UD:
await self.collection.insert_one(user.dict())
return user

119
fastapi_users/db/sqlalchemy.py
View File

@@ -1,7 +1,8 @@
from typing import List, Optional, Type
from typing import List, Mapping, Optional, Type
from databases import Database
from sqlalchemy import Boolean, Column, String, Table
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String, Table, select
from sqlalchemy.ext.declarative import declared_attr
from fastapi_users.db.base import BaseUserDatabase
from fastapi_users.models import UD
@@ -19,6 +20,35 @@ class SQLAlchemyBaseUserTable:
is_superuser = Column(Boolean, default=False, nullable=False)
class SQLAlchemyBaseOAuthAccountTable:
"""Base SQLAlchemy OAuth account table definition."""
__tablename__ = "oauth_account"
id = Column(String, primary_key=True)
oauth_name = Column(String, index=True, nullable=False)
access_token = Column(String, nullable=False)
expires_at = Column(Integer, nullable=False)
refresh_token = Column(String, nullable=True)
account_id = Column(String, index=True, nullable=False)
account_email = Column(String, nullable=False)
@declared_attr
def user_id(cls):
return Column(String, ForeignKey("user.id", ondelete="cascade"), nullable=False)
class NotSetOAuthAccountTableError(Exception):
"""
OAuth table was not set in DB adapter but was needed.
Raised when trying to create/update a user with OAuth accounts set
but no table were specified in the DB adapter.
"""
pass
class SQLAlchemyUserDatabase(BaseUserDatabase[UD]):
"""
Database adapter for SQLAlchemy.
@@ -26,43 +56,110 @@ class SQLAlchemyUserDatabase(BaseUserDatabase[UD]):
:param user_db_model: Pydantic model of a DB representation of a user.
:param database: `Database` instance from `encode/databases`.
:param users: SQLAlchemy users table instance.
:param oauth_accounts: Optional SQLAlchemy OAuth accounts table instance.
"""
database: Database
users: Table
oauth_accounts: Optional[Table]
def __init__(self, user_db_model: Type[UD], database: Database, users: Table):
def __init__(
self,
user_db_model: Type[UD],
database: Database,
users: Table,
oauth_accounts: Optional[Table] = None,
):
super().__init__(user_db_model)
self.database = database
self.users = users
self.oauth_accounts = oauth_accounts
async def list(self) -> List[UD]:
query = self.users.select()
users = await self.database.fetch_all(query)
return [self.user_db_model(**user) for user in users]
return [await self._make_user(user) for user in users]
async def get(self, id: str) -> Optional[UD]:
query = self.users.select().where(self.users.c.id == id)
user = await self.database.fetch_one(query)
return self.user_db_model(**user) if user else None
return await self._make_user(user) if user else None
async def get_by_email(self, email: str) -> Optional[UD]:
query = self.users.select().where(self.users.c.email == email)
user = await self.database.fetch_one(query)
return self.user_db_model(**user) if user else None
return await self._make_user(user) if user else None
async def get_by_oauth_account(self, oauth: str, account_id: str) -> Optional[UD]:
if self.oauth_accounts is not None:
query = (
select([self.users])
.select_from(self.users.join(self.oauth_accounts))
.where(self.oauth_accounts.c.oauth_name == oauth)
.where(self.oauth_accounts.c.account_id == account_id)
)
user = await self.database.fetch_one(query)
return await self._make_user(user) if user else None
raise NotSetOAuthAccountTableError()
async def create(self, user: UD) -> UD:
query = self.users.insert().values(**user.dict())
await self.database.execute(query)
user_dict = user.dict()
oauth_accounts_values = None
if "oauth_accounts" in user_dict:
oauth_accounts_values = []
oauth_accounts = user_dict.pop("oauth_accounts")
for oauth_account in oauth_accounts:
oauth_accounts_values.append({"user_id": user.id, **oauth_account})
query = self.users.insert()
await self.database.execute(query, user_dict)
if oauth_accounts_values is not None:
if self.oauth_accounts is None:
raise NotSetOAuthAccountTableError()
query = self.oauth_accounts.insert()
await self.database.execute_many(query, oauth_accounts_values)
return user
async def update(self, user: UD) -> UD:
query = (
self.users.update().where(self.users.c.id == user.id).values(**user.dict())
)
user_dict = user.dict()
if "oauth_accounts" in user_dict:
if self.oauth_accounts is None:
raise NotSetOAuthAccountTableError()
query = self.oauth_accounts.delete().where(
self.oauth_accounts.c.user_id == user.id
)
await self.database.execute(query)
oauth_accounts_values = []
oauth_accounts = user_dict.pop("oauth_accounts")
for oauth_account in oauth_accounts:
oauth_accounts_values.append({"user_id": user.id, **oauth_account})
query = self.oauth_accounts.insert()
await self.database.execute_many(query, oauth_accounts_values)
query = self.users.update().where(self.users.c.id == user.id).values(user_dict)
await self.database.execute(query)
return user
async def delete(self, user: UD) -> None:
query = self.users.delete().where(self.users.c.id == user.id)
await self.database.execute(query)
async def _make_user(self, user: Mapping) -> UD:
user_dict = {**user}
if self.oauth_accounts is not None:
query = self.oauth_accounts.select().where(
self.oauth_accounts.c.user_id == user["id"]
)
oauth_accounts = await self.database.fetch_all(query)
user_dict["oauth_accounts"] = oauth_accounts
return self.user_db_model(**user_dict)

107
fastapi_users/db/tortoise.py
View File

@@ -14,6 +14,27 @@ class TortoiseBaseUserModel(Model):
is_active = fields.BooleanField(default=True, null=False)
is_superuser = fields.BooleanField(default=False, null=False)
async def to_dict(self):
d = {}
for field in self._meta.db_fields:
d[field] = getattr(self, field)
for field in self._meta.backward_fk_fields:
d[field] = await getattr(self, field).all().values()
return d
class Meta:
abstract = True
class TortoiseBaseOAuthAccountModel(Model):
id = fields.CharField(pk=True, generated=False, max_length=255)
oauth_name = fields.CharField(null=False, max_length=255)
access_token = fields.CharField(null=False, max_length=255)
expires_at = fields.IntField(null=False)
refresh_token = fields.CharField(null=True, max_length=255)
account_id = fields.CharField(index=True, null=False, max_length=255)
account_email = fields.CharField(null=False, max_length=255)
class Meta:
abstract = True
@@ -24,41 +45,109 @@ class TortoiseUserDatabase(BaseUserDatabase[UD]):
:param user_db_model: Pydantic model of a DB representation of a user.
:param model: Tortoise ORM model.
:param oauth_account_model: Optional Tortoise ORM model of a OAuth account.
"""
model: Type[TortoiseBaseUserModel]
oauth_account_model: Optional[Type[TortoiseBaseOAuthAccountModel]]
def __init__(self, user_db_model: Type[UD], model: Type[TortoiseBaseUserModel]):
def __init__(
self,
user_db_model: Type[UD],
model: Type[TortoiseBaseUserModel],
oauth_account_model: Optional[Type[TortoiseBaseOAuthAccountModel]] = None,
):
super().__init__(user_db_model)
self.model = model
self.oauth_account_model = oauth_account_model
async def list(self) -> List[UD]:
users = await self.model.all()
return [self.user_db_model.from_orm(user) for user in users]
query = self.model.all()
if self.oauth_account_model is not None:
query = query.prefetch_related("oauth_accounts")
users = await query
return [self.user_db_model(**await user.to_dict()) for user in users]
async def get(self, id: str) -> Optional[UD]:
try:
user = await self.model.get(id=id)
return self.user_db_model.from_orm(user)
query = self.model.get(id=id)
if self.oauth_account_model is not None:
query = query.prefetch_related("oauth_accounts")
user = await query
user_dict = await user.to_dict()
return self.user_db_model(**user_dict)
except DoesNotExist:
return None
async def get_by_email(self, email: str) -> Optional[UD]:
try:
user = await self.model.get(email=email)
return self.user_db_model.from_orm(user)
query = self.model.get(email=email)
if self.oauth_account_model is not None:
query = query.prefetch_related("oauth_accounts")
user = await query
user_dict = await user.to_dict()
return self.user_db_model(**user_dict)
except DoesNotExist:
return None
async def get_by_oauth_account(self, oauth: str, account_id: str) -> Optional[UD]:
try:
query = self.model.get(
oauth_accounts__oauth_name=oauth, oauth_accounts__account_id=account_id
).prefetch_related("oauth_accounts")
user = await query
user_dict = await user.to_dict()
return self.user_db_model(**user_dict)
except DoesNotExist:
return None
async def create(self, user: UD) -> UD:
model = self.model(**user.dict())
user_dict = user.dict()
oauth_accounts = user_dict.pop("oauth_accounts", None)
model = self.model(**user_dict)
await model.save()
if oauth_accounts and self.oauth_account_model:
oauth_account_objects = []
for oauth_account in oauth_accounts:
oauth_account_objects.append(
self.oauth_account_model(user=model, **oauth_account)
)
await self.oauth_account_model.bulk_create(oauth_account_objects)
return user
async def update(self, user: UD) -> UD:
user_dict = user.dict()
user_dict.pop("id") # Tortoise complains if we pass the PK again
await self.model.filter(id=user.id).update(**user_dict)
oauth_accounts = user_dict.pop("oauth_accounts", None)
model = await self.model.get(id=user.id)
for field in user_dict:
setattr(model, field, user_dict[field])
await model.save()
if oauth_accounts and self.oauth_account_model:
await model.oauth_accounts.all().delete()
oauth_account_objects = []
for oauth_account in oauth_accounts:
oauth_account_objects.append(
self.oauth_account_model(user=model, **oauth_account)
)
await self.oauth_account_model.bulk_create(oauth_account_objects)
return user
async def delete(self, user: UD) -> None:

52
fastapi_users/fastapi_users.py
View File

@@ -1,9 +1,17 @@
from typing import Callable, Sequence, Type
from collections import defaultdict
from typing import Callable, DefaultDict, List, Sequence, Type
from httpx_oauth.oauth2 import BaseOAuth2
from fastapi_users import models
from fastapi_users.authentication import Authenticator, BaseAuthentication
from fastapi_users.db import BaseUserDatabase
from fastapi_users.router import Event, UserRouter, get_user_router
from fastapi_users.router import (
Event,
EventHandlersRouter,
get_oauth_router,
get_user_router,
)
class FastAPIUsers:
@@ -20,12 +28,16 @@ class FastAPIUsers:
:param reset_password_token_lifetime_seconds: Lifetime of reset password token.
:attribute router: Router exposing authentication routes.
:attribute oauth_routers: List of OAuth routers created through `get_oauth_router`.
:attribute get_current_user: Dependency callable to inject authenticated user.
"""
db: BaseUserDatabase
authenticator: Authenticator
router: UserRouter
router: EventHandlersRouter
oauth_routers: List[EventHandlersRouter]
_user_db_model: Type[models.BaseUserDB]
_event_handlers: DefaultDict[Event, List[Callable]]
def __init__(
self,
@@ -50,6 +62,9 @@ class FastAPIUsers:
reset_password_token_secret,
reset_password_token_lifetime_seconds,
)
self.oauth_routers = []
self._user_db_model = user_db_model
self._event_handlers = defaultdict(list)
self.get_current_user = self.authenticator.get_current_user
self.get_current_active_user = self.authenticator.get_current_active_user
@@ -63,9 +78,40 @@ class FastAPIUsers:
"""Add an event handler on successful forgot password request."""
return self._on_event(Event.ON_AFTER_FORGOT_PASSWORD)
def get_oauth_router(
self, oauth_client: BaseOAuth2, state_secret: str, redirect_url: str = None
) -> EventHandlersRouter:
"""
Return an OAuth router for a given OAuth client.
:param oauth_client: The HTTPX OAuth client instance.
:param state_secret: Secret used to encode the state JWT.
:param redirect_url: Optional arbitrary redirect URL for the OAuth2 flow.
If not given, the URL to the callback endpoint will be generated.
"""
oauth_router = get_oauth_router(
oauth_client,
self.db,
self._user_db_model,
self.authenticator,
state_secret,
redirect_url,
)
for event_type in self._event_handlers:
for handler in self._event_handlers[event_type]:
oauth_router.add_event_handler(event_type, handler)
self.oauth_routers.append(oauth_router)
return oauth_router
def _on_event(self, event_type: Event) -> Callable:
def decorator(func: Callable) -> Callable:
self._event_handlers[event_type].append(func)
self.router.add_event_handler(event_type, func)
for oauth_router in self.oauth_routers:
oauth_router.add_event_handler(event_type, func)
return func
return decorator

30
fastapi_users/models.py
View File

@@ -1,5 +1,5 @@
import uuid
from typing import Optional, TypeVar
from typing import List, Optional, TypeVar
import pydantic
from pydantic import BaseModel, EmailStr
@@ -19,7 +19,8 @@ class BaseUser(BaseModel):
def create_update_dict(self):
return self.dict(
exclude_unset=True, exclude={"id", "is_superuser", "is_active"}
exclude_unset=True,
exclude={"id", "is_superuser", "is_active", "oauth_accounts"},
)
def create_update_dict_superuser(self):
@@ -44,3 +45,28 @@ class BaseUserDB(BaseUser):
UD = TypeVar("UD", bound=BaseUserDB)
class BaseOAuthAccount(BaseModel):
"""Base OAuth account model."""
id: Optional[str] = None
oauth_name: str
access_token: str
expires_at: int
refresh_token: Optional[str] = None
account_id: str
account_email: str
@pydantic.validator("id", pre=True, always=True)
def default_id(cls, v):
return v or str(uuid.uuid4())
class Config:
orm_mode = True
class BaseOAuthAccountMixin(BaseModel):
"""Adds OAuth accounts list to a User model."""
oauth_accounts: List[BaseOAuthAccount] = []

5
fastapi_users/password.py
View File

@@ -1,5 +1,6 @@
from typing import Tuple
from passlib import pwd
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
@@ -13,3 +14,7 @@ def verify_and_update_password(
def get_password_hash(password: str) -> str:
return pwd_context.hash(password)
def generate_password() -> str:
return pwd.genword()

7
fastapi_users/router/__init__.py Normal file
View File

@@ -0,0 +1,7 @@
from fastapi_users.router.common import ( # noqa: F401
ErrorCode,
Event,
EventHandlersRouter,
)
from fastapi_users.router.oauth import get_oauth_router # noqa: F401
from fastapi_users.router.users import get_user_router # noqa: F401

35
fastapi_users/router/common.py Normal file
View File

@@ -0,0 +1,35 @@
import asyncio
from collections import defaultdict
from enum import Enum, auto
from typing import Callable, DefaultDict, List
from fastapi import APIRouter
class ErrorCode:
REGISTER_USER_ALREADY_EXISTS = "REGISTER_USER_ALREADY_EXISTS"
LOGIN_BAD_CREDENTIALS = "LOGIN_BAD_CREDENTIALS"
RESET_PASSWORD_BAD_TOKEN = "RESET_PASSWORD_BAD_TOKEN"
class Event(Enum):
ON_AFTER_REGISTER = auto()
ON_AFTER_FORGOT_PASSWORD = auto()
class EventHandlersRouter(APIRouter):
event_handlers: DefaultDict[Event, List[Callable]]
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.event_handlers = defaultdict(list)
def add_event_handler(self, event_type: Event, func: Callable) -> None:
self.event_handlers[event_type].append(func)
async def run_handlers(self, event_type: Event, *args, **kwargs) -> None:
for handler in self.event_handlers[event_type]:
if asyncio.iscoroutinefunction(handler):
await handler(*args, **kwargs)
else:
handler(*args, **kwargs)

144
fastapi_users/router/oauth.py Normal file
View File

@@ -0,0 +1,144 @@
from typing import Dict, List, Type, cast
import jwt
from fastapi import Depends, HTTPException, Query
from httpx_oauth.integrations.fastapi import OAuth2AuthorizeCallback
from httpx_oauth.oauth2 import BaseOAuth2
from starlette import status
from starlette.requests import Request
from starlette.responses import Response
from fastapi_users import models
from fastapi_users.authentication import Authenticator
from fastapi_users.db import BaseUserDatabase
from fastapi_users.password import generate_password, get_password_hash
from fastapi_users.router.common import ErrorCode, Event, EventHandlersRouter
from fastapi_users.utils import JWT_ALGORITHM, generate_jwt
STATE_TOKEN_AUDIENCE = "fastapi-users:oauth-state"
def generate_state_token(
data: Dict[str, str], secret: str, lifetime_seconds: int = 3600
) -> str:
data["aud"] = STATE_TOKEN_AUDIENCE
return generate_jwt(data, lifetime_seconds, secret, JWT_ALGORITHM)
def decode_state_token(token: str, secret: str) -> Dict[str, str]:
return jwt.decode(
token, secret, audience=STATE_TOKEN_AUDIENCE, algorithms=[JWT_ALGORITHM],
)
def get_oauth_router(
oauth_client: BaseOAuth2,
user_db: BaseUserDatabase[models.BaseUserDB],
user_db_model: Type[models.BaseUserDB],
authenticator: Authenticator,
state_secret: str,
redirect_url: str = None,
) -> EventHandlersRouter:
"""Generate a router with the OAuth routes."""
router = EventHandlersRouter()
callback_route_name = f"{oauth_client.name}-callback"
if redirect_url is not None:
oauth2_authorize_callback = OAuth2AuthorizeCallback(
oauth_client, redirect_url=redirect_url,
)
else:
oauth2_authorize_callback = OAuth2AuthorizeCallback(
oauth_client, route_name=callback_route_name,
)
@router.get("/authorize")
async def authorize(
request: Request, authentication_backend: str, scopes: List[str] = Query(None),
):
# Check that authentication_backend exists
backend_exists = False
for backend in authenticator.backends:
if backend.name == authentication_backend:
backend_exists = True
break
if not backend_exists:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST)
redirect_url = request.url_for(callback_route_name)
state_data = {
"authentication_backend": authentication_backend,
}
state = generate_state_token(state_data, state_secret)
authorization_url = await oauth_client.get_authorization_url(
redirect_url, state, scopes,
)
return {"authorization_url": authorization_url}
@router.get("/callback", name=f"{oauth_client.name}-callback")
async def callback(
response: Response, access_token_state=Depends(oauth2_authorize_callback)
):
token, state = access_token_state
account_id, account_email = await oauth_client.get_id_email(
token["access_token"]
)
try:
state_data = decode_state_token(state, state_secret)
except jwt.DecodeError:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST)
user = await user_db.get_by_oauth_account(oauth_client.name, account_id)
new_oauth_account = models.BaseOAuthAccount(
oauth_name=oauth_client.name,
access_token=token["access_token"],
expires_at=token["expires_at"],
refresh_token=token.get("refresh_token"),
account_id=account_id,
account_email=account_email,
)
if not user:
user = await user_db.get_by_email(account_email)
if user:
# Link account
user.oauth_accounts.append(new_oauth_account) # type: ignore
await user_db.update(user)
else:
# Create account
password = generate_password()
user = user_db_model(
email=account_email,
hashed_password=get_password_hash(password),
oauth_accounts=[new_oauth_account],
)
await user_db.create(user)
await router.run_handlers(Event.ON_AFTER_REGISTER, user)
else:
# Update oauth
updated_oauth_accounts = []
for oauth_account in user.oauth_accounts: # type: ignore
if oauth_account.account_id == account_id:
updated_oauth_accounts.append(new_oauth_account)
else:
updated_oauth_accounts.append(oauth_account)
user.oauth_accounts = updated_oauth_accounts # type: ignore
await user_db.update(user)
if not user.is_active:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=ErrorCode.LOGIN_BAD_CREDENTIALS,
)
# Authenticate
for backend in authenticator.backends:
if backend.name == state_data["authentication_backend"]:
return await backend.get_login_response(
cast(models.BaseUserDB, user), response
)
return router

45
fastapi_users/router.py → fastapi_users/router/users.py
View File

@@ -1,10 +1,7 @@
import asyncio
from collections import defaultdict
from enum import Enum, auto
from typing import Any, Callable, DefaultDict, Dict, List, Type, cast
from typing import Any, Dict, List, Type, cast
import jwt
from fastapi import APIRouter, Body, Depends, HTTPException
from fastapi import Body, Depends, HTTPException
from fastapi.security import OAuth2PasswordRequestForm
from pydantic import EmailStr
from starlette import status
@@ -14,40 +11,14 @@ from fastapi_users import models
from fastapi_users.authentication import Authenticator, BaseAuthentication
from fastapi_users.db import BaseUserDatabase
from fastapi_users.password import get_password_hash
from fastapi_users.router.common import ErrorCode, Event, EventHandlersRouter
from fastapi_users.utils import JWT_ALGORITHM, generate_jwt
class ErrorCode:
REGISTER_USER_ALREADY_EXISTS = "REGISTER_USER_ALREADY_EXISTS"
LOGIN_BAD_CREDENTIALS = "LOGIN_BAD_CREDENTIALS"
RESET_PASSWORD_BAD_TOKEN = "RESET_PASSWORD_BAD_TOKEN"
class Event(Enum):
ON_AFTER_REGISTER = auto()
ON_AFTER_FORGOT_PASSWORD = auto()
class UserRouter(APIRouter):
event_handlers: DefaultDict[Event, List[Callable]]
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.event_handlers = defaultdict(list)
def add_event_handler(self, event_type: Event, func: Callable) -> None:
self.event_handlers[event_type].append(func)
async def run_handlers(self, event_type: Event, *args, **kwargs) -> None:
for handler in self.event_handlers[event_type]:
if asyncio.iscoroutinefunction(handler):
await handler(*args, **kwargs)
else:
handler(*args, **kwargs)
def _add_login_route(
router: UserRouter, user_db: BaseUserDatabase, auth_backend: BaseAuthentication
router: EventHandlersRouter,
user_db: BaseUserDatabase,
auth_backend: BaseAuthentication,
):
@router.post(f"/login/{auth_backend.name}")
async def login(
@@ -73,9 +44,9 @@ def get_user_router(
authenticator: Authenticator,
reset_password_token_secret: str,
reset_password_token_lifetime_seconds: int = 3600,
) -> UserRouter:
) -> EventHandlersRouter:
"""Generate a router with the authentication routes."""
router = UserRouter()
router = EventHandlersRouter()
reset_password_token_audience = "fastapi-users:reset"

1
mkdocs.yml
View File

@@ -39,6 +39,7 @@ nav:
- configuration/authentication/cookie.md
- configuration/router.md
- configuration/full_example.md
- configuration/oauth.md
- Usage:
- usage/routes.md
- usage/dependency-callables.md

90
oauth_test.py Normal file
View File

@@ -0,0 +1,90 @@
import databases
import sqlalchemy
from fastapi import FastAPI
from fastapi_users import FastAPIUsers, models
from fastapi_users.authentication import JWTAuthentication
from fastapi_users.db import (
SQLAlchemyBaseOAuthAccountTable,
SQLAlchemyBaseUserTable,
SQLAlchemyUserDatabase,
)
from httpx_oauth.clients.google import GoogleOAuth2
from sqlalchemy.ext.declarative import DeclarativeMeta, declarative_base
DATABASE_URL = "sqlite:///./test.db"
SECRET = "SECRET"
google_oauth_client = GoogleOAuth2("CLIENT_ID", "CLIENT_SECRET")
class User(models.BaseUser, models.BaseOAuthAccountMixin):
pass
class UserCreate(User, models.BaseUserCreate):
pass
class UserUpdate(User, models.BaseUserUpdate):
pass
class UserDB(User, models.BaseUserDB):
pass
database = databases.Database(DATABASE_URL)
Base: DeclarativeMeta = declarative_base()
class UserTable(Base, SQLAlchemyBaseUserTable):
pass
class OAuthAccount(SQLAlchemyBaseOAuthAccountTable, Base):
pass
engine = sqlalchemy.create_engine(
DATABASE_URL, connect_args={"check_same_thread": False}
)
Base.metadata.create_all(engine)
users = UserTable.__table__
oauth_accounts = OAuthAccount.__table__
user_db = SQLAlchemyUserDatabase(UserDB, database, users, oauth_accounts)
auth_backends = [
JWTAuthentication(secret=SECRET, lifetime_seconds=3600),
]
app = FastAPI()
fastapi_users = FastAPIUsers(
user_db, auth_backends, User, UserCreate, UserUpdate, UserDB, SECRET,
)
app.include_router(fastapi_users.router, prefix="/users", tags=["users"])
google_oauth_router = fastapi_users.get_oauth_router(google_oauth_client, SECRET)
app.include_router(google_oauth_router, prefix="/google-oauth", tags=["users"])
@fastapi_users.on_after_register()
def on_after_register(user: User):
print(f"User {user.id} has registered.")
@fastapi_users.on_after_forgot_password()
def on_after_forgot_password(user: User, token: str):
print(f"User {user.id} has forgot their password. Reset token: {token}")
@app.on_event("startup")
async def startup():
await database.connect()
@app.on_event("shutdown")
async def shutdown():
await database.disconnect()

3
pyproject.toml
View File

@@ -39,6 +39,9 @@ mongodb = [
tortoise-orm = [
"tortoise-orm ==0.15.7"
]
oauth = [
"httpx-oauth ==0.2.0"
]
[tool.flit.metadata.urls]
Documentation = "https://frankie567.github.io/fastapi-users/"

4
setup.cfg
View File

@@ -21,9 +21,9 @@ known_standard_library = types
known_third_party = pytest,_pytest
[tool:pytest]
markers =
markers =
authentication
db
fastapi_users
oauth
router

152
tests/conftest.py
View File

@@ -4,6 +4,7 @@ import http.cookies
import pytest
from fastapi import Depends, FastAPI
from fastapi.security import OAuth2PasswordBearer
from httpx_oauth.oauth2 import OAuth2
from starlette.requests import Request
from starlette.responses import Response
from starlette.testclient import TestClient
@@ -11,12 +12,13 @@ from starlette.testclient import TestClient
from fastapi_users import models
from fastapi_users.authentication import Authenticator, BaseAuthentication
from fastapi_users.db import BaseUserDatabase
from fastapi_users.models import BaseUserDB
from fastapi_users.models import BaseOAuthAccount, BaseOAuthAccountMixin, BaseUserDB
from fastapi_users.password import get_password_hash
guinevere_password_hash = get_password_hash("guinevere")
angharad_password_hash = get_password_hash("angharad")
viviane_password_hash = get_password_hash("viviane")
lancelot_password_hash = get_password_hash("lancelot")
class User(models.BaseUser):
@@ -35,6 +37,14 @@ class UserDB(User, models.BaseUserDB):
pass
class UserOAuth(User, BaseOAuthAccountMixin):
pass
class UserDBOAuth(UserOAuth, UserDB):
pass
@pytest.fixture
def user() -> UserDB:
return UserDB(
@@ -44,6 +54,16 @@ def user() -> UserDB:
)
@pytest.fixture
def user_oauth(oauth_account1, oauth_account2) -> UserDBOAuth:
return UserDBOAuth(
id="aaa",
email="king.arthur@camelot.bt",
hashed_password=guinevere_password_hash,
oauth_accounts=[oauth_account1, oauth_account2],
)
@pytest.fixture
def inactive_user() -> UserDB:
return UserDB(
@@ -54,6 +74,17 @@ def inactive_user() -> UserDB:
)
@pytest.fixture
def inactive_user_oauth(oauth_account3) -> UserDBOAuth:
return UserDBOAuth(
id="bbb",
email="percival@camelot.bt",
hashed_password=angharad_password_hash,
is_active=False,
oauth_accounts=[oauth_account3],
)
@pytest.fixture
def superuser() -> UserDB:
return UserDB(
@@ -64,6 +95,53 @@ def superuser() -> UserDB:
)
@pytest.fixture
def superuser_oauth() -> UserDBOAuth:
return UserDBOAuth(
id="ccc",
email="merlin@camelot.bt",
hashed_password=viviane_password_hash,
is_superuser=True,
oauth_accounts=[],
)
@pytest.fixture
def oauth_account1() -> BaseOAuthAccount:
return BaseOAuthAccount(
id="aaa",
oauth_name="service1",
access_token="TOKEN",
expires_at=1579000751,
account_id="user_oauth1",
account_email="king.arthur@camelot.bt",
)
@pytest.fixture
def oauth_account2() -> BaseOAuthAccount:
return BaseOAuthAccount(
id="bbb",
oauth_name="service2",
access_token="TOKEN",
expires_at=1579000751,
account_id="user_oauth2",
account_email="king.arthur@camelot.bt",
)
@pytest.fixture
def oauth_account3() -> BaseOAuthAccount:
return BaseOAuthAccount(
id="ccc",
oauth_name="service3",
access_token="TOKEN",
expires_at=1579000751,
account_id="inactive_user_oauth1",
account_email="percival@camelot.bt",
)
@pytest.fixture
def mock_user_db(user, inactive_user, superuser) -> BaseUserDatabase:
class MockUserDatabase(BaseUserDatabase[UserDB]):
@@ -100,6 +178,62 @@ def mock_user_db(user, inactive_user, superuser) -> BaseUserDatabase:
return MockUserDatabase(UserDB)
@pytest.fixture
def mock_user_db_oauth(
user_oauth, inactive_user_oauth, superuser_oauth
) -> BaseUserDatabase:
class MockUserDatabase(BaseUserDatabase[UserDBOAuth]):
async def list(self) -> List[UserDBOAuth]:
return [user_oauth, inactive_user_oauth, superuser_oauth]
async def get(self, id: str) -> Optional[UserDBOAuth]:
if id == user_oauth.id:
return user_oauth
if id == inactive_user_oauth.id:
return inactive_user_oauth
if id == superuser_oauth.id:
return superuser_oauth
return None
async def get_by_email(self, email: str) -> Optional[UserDBOAuth]:
if email == user_oauth.email:
return user_oauth
if email == inactive_user_oauth.email:
return inactive_user_oauth
if email == superuser_oauth.email:
return superuser_oauth
return None
async def get_by_oauth_account(
self, oauth: str, account_id: str
) -> Optional[UserDBOAuth]:
user_oauth_account = user_oauth.oauth_accounts[0]
if (
user_oauth_account.oauth_name == oauth
and user_oauth_account.account_id == account_id
):
return user_oauth
inactive_user_oauth_account = inactive_user_oauth.oauth_accounts[0]
if (
inactive_user_oauth_account.oauth_name == oauth
and inactive_user_oauth_account.account_id == account_id
):
return inactive_user_oauth
return None
async def create(self, user: UserDBOAuth) -> UserDBOAuth:
return user_oauth
async def update(self, user: UserDBOAuth) -> UserDBOAuth:
return user_oauth
async def delete(self, user: UserDBOAuth) -> None:
pass
return MockUserDatabase(UserDBOAuth)
class MockAuthentication(BaseAuthentication):
def __init__(self, name: str = "mock"):
super().__init__(name)
@@ -174,3 +308,19 @@ def get_test_auth_client(mock_user_db):
return TestClient(app)
return _get_test_auth_client
@pytest.fixture()
def oauth_client() -> OAuth2:
CLIENT_ID = "CLIENT_ID"
CLIENT_SECRET = "CLIENT_SECRET"
AUTHORIZE_ENDPOINT = "https://www.camelot.bt/authorize"
ACCESS_TOKEN_ENDPOINT = "https://www.camelot.bt/access-token"
return OAuth2(
CLIENT_ID,
CLIENT_SECRET,
AUTHORIZE_ENDPOINT,
ACCESS_TOKEN_ENDPOINT,
name="service1",
)

3
tests/test_db_base.py
View File

@@ -27,6 +27,9 @@ async def test_not_implemented_methods(user):
with pytest.raises(NotImplementedError):
await base_user_db.get_by_email("lancelot@camelot.bt")
with pytest.raises(NotImplementedError):
await base_user_db.get_by_oauth_account("google", "user_oauth1")
with pytest.raises(NotImplementedError):
await base_user_db.create(user)

100
tests/test_db_mongodb.py
View File

@@ -6,27 +6,44 @@ import pymongo.errors
from fastapi_users.db.mongodb import MongoDBUserDatabase
from fastapi_users.password import get_password_hash
from tests.conftest import UserDB
from tests.conftest import UserDB, UserDBOAuth
@pytest.fixture
async def mongodb_user_db() -> AsyncGenerator[MongoDBUserDatabase, None]:
client = motor.motor_asyncio.AsyncIOMotorClient(
"mongodb://localhost:27017", serverSelectionTimeoutMS=100
)
def get_mongodb_user_db():
async def _get_mongodb_user_db(
user_model,
) -> AsyncGenerator[MongoDBUserDatabase, None]:
client = motor.motor_asyncio.AsyncIOMotorClient(
"mongodb://localhost:27017", serverSelectionTimeoutMS=100
)
try:
await client.server_info()
except pymongo.errors.ServerSelectionTimeoutError:
pytest.skip("MongoDB not available", allow_module_level=True)
return
try:
await client.server_info()
except pymongo.errors.ServerSelectionTimeoutError:
pytest.skip("MongoDB not available", allow_module_level=True)
return
db = client["test_database"]
collection = db["users"]
db = client["test_database"]
collection = db["users"]
yield MongoDBUserDatabase(UserDB, collection)
yield MongoDBUserDatabase(user_model, collection)
await collection.drop()
await collection.drop()
return _get_mongodb_user_db
@pytest.fixture
async def mongodb_user_db(get_mongodb_user_db):
async for u in get_mongodb_user_db(UserDB):
yield u
@pytest.fixture
async def mongodb_user_db_oauth(get_mongodb_user_db):
async for u in get_mongodb_user_db(UserDBOAuth):
yield u
@pytest.mark.asyncio
@@ -96,3 +113,58 @@ async def test_queries_custom_fields(mongodb_user_db: MongoDBUserDatabase[UserDB
assert id_user is not None
assert id_user.id == user.id
assert id_user.first_name == user.first_name
@pytest.mark.asyncio
@pytest.mark.db
async def test_queries_oauth(
mongodb_user_db_oauth: MongoDBUserDatabase[UserDBOAuth],
oauth_account1,
oauth_account2,
):
user = UserDBOAuth(
id="111",
email="lancelot@camelot.bt",
hashed_password=get_password_hash("guinevere"),
oauth_accounts=[oauth_account1, oauth_account2],
)
# Create
user_db = await mongodb_user_db_oauth.create(user)
assert user_db.id is not None
assert hasattr(user_db, "oauth_accounts")
assert len(user_db.oauth_accounts) == 2
# Update
user_db.oauth_accounts[0].access_token = "NEW_TOKEN"
await mongodb_user_db_oauth.update(user_db)
# Get by id
id_user = await mongodb_user_db_oauth.get(user.id)
assert id_user is not None
assert id_user.id == user_db.id
assert id_user.oauth_accounts[0].access_token == "NEW_TOKEN"
# Get by email
email_user = await mongodb_user_db_oauth.get_by_email(str(user.email))
assert email_user is not None
assert email_user.id == user_db.id
assert len(email_user.oauth_accounts) == 2
# List
users = await mongodb_user_db_oauth.list()
assert len(users) == 1
first_user = users[0]
assert first_user.id == user_db.id
assert len(first_user.oauth_accounts) == 2
# Get by OAuth account
oauth_user = await mongodb_user_db_oauth.get_by_oauth_account(
oauth_account1.oauth_name, oauth_account1.account_id
)
assert oauth_user is not None
assert oauth_user.id == user.id
# Unknown OAuth account
unknown_oauth_user = await mongodb_user_db_oauth.get_by_oauth_account("foo", "bar")
assert unknown_oauth_user is None

110
tests/test_db_sqlalchemy.py
View File

@@ -7,9 +7,14 @@ from databases import Database
from sqlalchemy import Column, String
from sqlalchemy.ext.declarative import DeclarativeMeta, declarative_base
from fastapi_users.db.sqlalchemy import SQLAlchemyBaseUserTable, SQLAlchemyUserDatabase
from fastapi_users.db.sqlalchemy import (
NotSetOAuthAccountTableError,
SQLAlchemyBaseOAuthAccountTable,
SQLAlchemyBaseUserTable,
SQLAlchemyUserDatabase,
)
from fastapi_users.password import get_password_hash
from tests.conftest import UserDB
from tests.conftest import UserDB, UserDBOAuth
@pytest.fixture
@@ -19,7 +24,7 @@ async def sqlalchemy_user_db() -> AsyncGenerator[SQLAlchemyUserDatabase, None]:
class User(SQLAlchemyBaseUserTable, Base):
first_name = Column(String, nullable=True)
DATABASE_URL = "sqlite:///./test.db"
DATABASE_URL = "sqlite:///./test-sqlalchemy-user.db"
database = Database(DATABASE_URL)
engine = sqlalchemy.create_engine(
@@ -34,6 +39,33 @@ async def sqlalchemy_user_db() -> AsyncGenerator[SQLAlchemyUserDatabase, None]:
Base.metadata.drop_all(engine)
@pytest.fixture
async def sqlalchemy_user_db_oauth() -> AsyncGenerator[SQLAlchemyUserDatabase, None]:
Base: DeclarativeMeta = declarative_base()
class User(SQLAlchemyBaseUserTable, Base):
first_name = Column(String, nullable=True)
class OAuthAccount(SQLAlchemyBaseOAuthAccountTable, Base):
pass
DATABASE_URL = "sqlite:///./test-sqlalchemy-user-oauth.db"
database = Database(DATABASE_URL)
engine = sqlalchemy.create_engine(
DATABASE_URL, connect_args={"check_same_thread": False}
)
Base.metadata.create_all(engine)
await database.connect()
yield SQLAlchemyUserDatabase(
UserDBOAuth, database, User.__table__, OAuthAccount.__table__
)
Base.metadata.drop_all(engine)
@pytest.mark.asyncio
@pytest.mark.db
async def test_queries(sqlalchemy_user_db: SQLAlchemyUserDatabase[UserDB]):
@@ -89,6 +121,21 @@ async def test_queries(sqlalchemy_user_db: SQLAlchemyUserDatabase[UserDB]):
deleted_user = await sqlalchemy_user_db.get(user.id)
assert deleted_user is None
# Exception when creating/updating a OAuth user
user_oauth = UserDBOAuth(
id="222",
email="lancelot@camelot.bt",
hashed_password=get_password_hash("guinevere"),
)
with pytest.raises(NotSetOAuthAccountTableError):
await sqlalchemy_user_db.create(user_oauth)
with pytest.raises(NotSetOAuthAccountTableError):
await sqlalchemy_user_db.update(user_oauth)
# Exception when trying to get by OAuth account
with pytest.raises(NotSetOAuthAccountTableError):
await sqlalchemy_user_db.get_by_oauth_account("foo", "bar")
@pytest.mark.asyncio
@pytest.mark.db
@@ -108,3 +155,60 @@ async def test_queries_custom_fields(
assert id_user is not None
assert id_user.id == user.id
assert id_user.first_name == user.first_name
@pytest.mark.asyncio
@pytest.mark.db
async def test_queries_oauth(
sqlalchemy_user_db_oauth: SQLAlchemyUserDatabase[UserDBOAuth],
oauth_account1,
oauth_account2,
):
user = UserDBOAuth(
id="111",
email="lancelot@camelot.bt",
hashed_password=get_password_hash("guinevere"),
oauth_accounts=[oauth_account1, oauth_account2],
)
# Create
user_db = await sqlalchemy_user_db_oauth.create(user)
assert user_db.id is not None
assert hasattr(user_db, "oauth_accounts")
assert len(user_db.oauth_accounts) == 2
# Update
user_db.oauth_accounts[0].access_token = "NEW_TOKEN"
await sqlalchemy_user_db_oauth.update(user_db)
# Get by id
id_user = await sqlalchemy_user_db_oauth.get(user.id)
assert id_user is not None
assert id_user.id == user_db.id
assert id_user.oauth_accounts[0].access_token == "NEW_TOKEN"
# Get by email
email_user = await sqlalchemy_user_db_oauth.get_by_email(str(user.email))
assert email_user is not None
assert email_user.id == user_db.id
assert len(email_user.oauth_accounts) == 2
# List
users = await sqlalchemy_user_db_oauth.list()
assert len(users) == 1
first_user = users[0]
assert first_user.id == user_db.id
assert len(first_user.oauth_accounts) == 2
# Get by OAuth account
oauth_user = await sqlalchemy_user_db_oauth.get_by_oauth_account(
oauth_account1.oauth_name, oauth_account1.account_id
)
assert oauth_user is not None
assert oauth_user.id == user.id
# Unknown OAuth account
unknown_oauth_user = await sqlalchemy_user_db_oauth.get_by_oauth_account(
"foo", "bar"
)
assert unknown_oauth_user is None

84
tests/test_db_tortoise.py
View File

@@ -4,18 +4,26 @@ import pytest
from tortoise.exceptions import IntegrityError
from tortoise import Tortoise, fields
from fastapi_users.db.tortoise import TortoiseUserDatabase, TortoiseBaseUserModel
from fastapi_users.db.tortoise import (
TortoiseBaseOAuthAccountModel,
TortoiseBaseUserModel,
TortoiseUserDatabase,
)
from fastapi_users.password import get_password_hash
from tests.conftest import UserDB
from tests.conftest import UserDB, UserDBOAuth
class User(TortoiseBaseUserModel):
first_name = fields.CharField(null=True, max_length=255)
class OAuthAccount(TortoiseBaseOAuthAccountModel):
user = fields.ForeignKeyField("models.User", related_name="oauth_accounts")
@pytest.fixture
async def tortoise_user_db() -> AsyncGenerator[TortoiseUserDatabase, None]:
DATABASE_URL = "sqlite://./test.db"
DATABASE_URL = "sqlite://./test-tortoise-user.db"
await Tortoise.init(
db_url=DATABASE_URL, modules={"models": ["tests.test_db_tortoise"]}
@@ -28,6 +36,21 @@ async def tortoise_user_db() -> AsyncGenerator[TortoiseUserDatabase, None]:
await Tortoise.close_connections()
@pytest.fixture
async def tortoise_user_db_oauth() -> AsyncGenerator[TortoiseUserDatabase, None]:
DATABASE_URL = "sqlite://./test-tortoise-user-oauth.db"
await Tortoise.init(
db_url=DATABASE_URL, modules={"models": ["tests.test_db_tortoise"]}
)
await Tortoise.generate_schemas()
yield TortoiseUserDatabase(UserDBOAuth, User, OAuthAccount)
await User.all().delete()
await Tortoise.close_connections()
@pytest.mark.asyncio
@pytest.mark.db
async def test_queries(tortoise_user_db: TortoiseUserDatabase[UserDB]):
@@ -100,3 +123,58 @@ async def test_queries_custom_fields(tortoise_user_db: TortoiseUserDatabase[User
assert id_user is not None
assert id_user.id == user.id
assert id_user.first_name == user.first_name
@pytest.mark.asyncio
@pytest.mark.db
async def test_queries_oauth(
tortoise_user_db_oauth: TortoiseUserDatabase[UserDBOAuth],
oauth_account1,
oauth_account2,
):
user = UserDBOAuth(
id="111",
email="lancelot@camelot.bt",
hashed_password=get_password_hash("guinevere"),
oauth_accounts=[oauth_account1, oauth_account2],
)
# Create
user_db = await tortoise_user_db_oauth.create(user)
assert user_db.id is not None
assert hasattr(user_db, "oauth_accounts")
assert len(user_db.oauth_accounts) == 2
# Update
user_db.oauth_accounts[0].access_token = "NEW_TOKEN"
await tortoise_user_db_oauth.update(user_db)
# Get by id
id_user = await tortoise_user_db_oauth.get(user.id)
assert id_user is not None
assert id_user.id == user_db.id
assert id_user.oauth_accounts[0].access_token == "NEW_TOKEN"
# Get by email
email_user = await tortoise_user_db_oauth.get_by_email(str(user.email))
assert email_user is not None
assert email_user.id == user_db.id
assert len(email_user.oauth_accounts) == 2
# List
users = await tortoise_user_db_oauth.list()
assert len(users) == 1
first_user = users[0]
assert first_user.id == user_db.id
assert len(first_user.oauth_accounts) == 2
# Get by OAuth account
oauth_user = await tortoise_user_db_oauth.get_by_oauth_account(
oauth_account1.oauth_name, oauth_account1.account_id
)
assert oauth_user is not None
assert oauth_user.id == user.id
# Unknown OAuth account
unknown_oauth_user = await tortoise_user_db_oauth.get_by_oauth_account("foo", "bar")
assert unknown_oauth_user is None

27
tests/test_fastapi_users.py
View File

@@ -1,10 +1,11 @@
import pytest
from fastapi import Depends, FastAPI
from httpx_oauth.oauth2 import OAuth2
from starlette import status
from starlette.testclient import TestClient
from fastapi_users import FastAPIUsers
from fastapi_users.router import Event
from fastapi_users.router import Event, EventHandlersRouter
from tests.conftest import User, UserCreate, UserUpdate, UserDB
@@ -17,7 +18,9 @@ async def async_event_handler():
@pytest.fixture(params=[sync_event_handler, async_event_handler])
def fastapi_users(request, mock_user_db, mock_authentication) -> FastAPIUsers:
def fastapi_users(
request, mock_user_db, mock_authentication, oauth_client
) -> FastAPIUsers:
fastapi_users = FastAPIUsers(
mock_user_db,
[mock_authentication],
@@ -28,6 +31,8 @@ def fastapi_users(request, mock_user_db, mock_authentication) -> FastAPIUsers:
"SECRET",
)
fastapi_users.get_oauth_router(oauth_client, "SECRET")
@fastapi_users.on_after_register()
def on_after_register():
return request.param()
@@ -165,3 +170,21 @@ class TestGetCurrentSuperuser:
"/current-superuser", headers={"Authorization": f"Bearer {superuser.id}"}
)
assert response.status_code == status.HTTP_200_OK
@pytest.mark.fastapi_users
def test_get_oauth_router(mocker, fastapi_users: FastAPIUsers, oauth_client: OAuth2):
# Check that existing OAuth router declared
# before the handlers decorators is correctly binded
existing_oauth_router = fastapi_users.oauth_routers[0]
event_handlers = existing_oauth_router.event_handlers
assert len(event_handlers[Event.ON_AFTER_REGISTER]) == 1
assert len(event_handlers[Event.ON_AFTER_FORGOT_PASSWORD]) == 1
# Check that OAuth router declared
# after the handlers decorators is correctly binded
oauth_router = fastapi_users.get_oauth_router(oauth_client, "SECRET")
assert isinstance(oauth_router, EventHandlersRouter)
event_handlers = oauth_router.event_handlers
assert len(event_handlers[Event.ON_AFTER_REGISTER]) == 1
assert len(event_handlers[Event.ON_AFTER_FORGOT_PASSWORD]) == 1

315
tests/test_router_oauth.py Normal file
View File

@@ -0,0 +1,315 @@
from unittest.mock import MagicMock
import asynctest
import pytest
from fastapi import FastAPI
from starlette import status
from starlette.testclient import TestClient
from fastapi_users.authentication import Authenticator
from fastapi_users.router.common import ErrorCode, Event
from fastapi_users.router.oauth import generate_state_token, get_oauth_router
from tests.conftest import MockAuthentication, UserDB
SECRET = "SECRET"
def event_handler_sync():
return MagicMock(return_value=None)
def event_handler_async():
return asynctest.CoroutineMock(return_value=None)
@pytest.fixture(params=[event_handler_sync, event_handler_async])
def event_handler(request):
return request.param()
@pytest.fixture()
def get_test_app_client(
mock_user_db_oauth, mock_authentication, oauth_client, event_handler
):
def _get_test_app_client(redirect_url: str = None) -> TestClient:
mock_authentication_bis = MockAuthentication(name="mock-bis")
authenticator = Authenticator(
[mock_authentication, mock_authentication_bis], mock_user_db_oauth
)
oauth_router = get_oauth_router(
oauth_client,
mock_user_db_oauth,
UserDB,
authenticator,
SECRET,
redirect_url,
)
oauth_router.add_event_handler(Event.ON_AFTER_REGISTER, event_handler)
app = FastAPI()
app.include_router(oauth_router)
return TestClient(app)
return _get_test_app_client
@pytest.fixture()
def test_app_client(get_test_app_client):
return get_test_app_client()
@pytest.fixture()
def test_app_client_redirect_url(get_test_app_client):
return get_test_app_client("http://www.tintagel.bt/callback")
@pytest.mark.router
@pytest.mark.oauth
class TestAuthorize:
def test_missing_authentication_backend(
self, test_app_client: TestClient, oauth_client
):
with asynctest.patch.object(oauth_client, "get_authorization_url") as mock:
mock.return_value = "AUTHORIZATION_URL"
response = test_app_client.get(
"/authorize", params={"scopes": ["scope1", "scope2"]},
)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
def test_wrong_authentication_backend(
self, test_app_client: TestClient, oauth_client
):
with asynctest.patch.object(oauth_client, "get_authorization_url") as mock:
mock.return_value = "AUTHORIZATION_URL"
response = test_app_client.get(
"/authorize",
params={
"authentication_backend": "foo",
"scopes": ["scope1", "scope2"],
},
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
def test_success(self, test_app_client: TestClient, oauth_client):
with asynctest.patch.object(oauth_client, "get_authorization_url") as mock:
mock.return_value = "AUTHORIZATION_URL"
response = test_app_client.get(
"/authorize",
params={
"authentication_backend": "mock",
"scopes": ["scope1", "scope2"],
},
)
assert response.status_code == status.HTTP_200_OK
mock.assert_awaited_once()
data = response.json()
assert "authorization_url" in data
@pytest.mark.router
@pytest.mark.oauth
class TestCallback:
def test_invalid_state(
self, test_app_client: TestClient, oauth_client, user_oauth, event_handler
):
with asynctest.patch.object(
oauth_client, "get_access_token"
) as get_access_token_mock:
get_access_token_mock.return_value = {
"access_token": "TOKEN",
"expires_at": 1579179542,
}
with asynctest.patch.object(
oauth_client, "get_id_email"
) as get_id_email_mock:
get_id_email_mock.return_value = ("user_oauth1", user_oauth.email)
response = test_app_client.get(
"/callback", params={"code": "CODE", "state": "STATE"},
)
get_id_email_mock.assert_awaited_once_with("TOKEN")
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert event_handler.called is False
def test_existing_user_with_oauth(
self,
mock_user_db_oauth,
test_app_client: TestClient,
oauth_client,
user_oauth,
event_handler,
):
state_jwt = generate_state_token({"authentication_backend": "mock"}, "SECRET")
with asynctest.patch.object(
oauth_client, "get_access_token"
) as get_access_token_mock:
get_access_token_mock.return_value = {
"access_token": "TOKEN",
"expires_at": 1579179542,
}
with asynctest.patch.object(
oauth_client, "get_id_email"
) as get_id_email_mock:
with asynctest.patch.object(
mock_user_db_oauth, "update"
) as user_update_mock:
get_id_email_mock.return_value = ("user_oauth1", user_oauth.email)
response = test_app_client.get(
"/callback", params={"code": "CODE", "state": state_jwt},
)
get_id_email_mock.assert_awaited_once_with("TOKEN")
user_update_mock.assert_awaited_once()
data = response.json()
assert data["token"] == user_oauth.id
assert event_handler.called is False
def test_existing_user_without_oauth(
self,
mock_user_db_oauth,
test_app_client: TestClient,
oauth_client,
superuser_oauth,
event_handler,
):
state_jwt = generate_state_token({"authentication_backend": "mock"}, "SECRET")
with asynctest.patch.object(
oauth_client, "get_access_token"
) as get_access_token_mock:
get_access_token_mock.return_value = {
"access_token": "TOKEN",
"expires_at": 1579179542,
}
with asynctest.patch.object(
oauth_client, "get_id_email"
) as get_id_email_mock:
with asynctest.patch.object(
mock_user_db_oauth, "update"
) as user_update_mock:
get_id_email_mock.return_value = (
"superuser_oauth1",
superuser_oauth.email,
)
response = test_app_client.get(
"/callback", params={"code": "CODE", "state": state_jwt},
)
get_id_email_mock.assert_awaited_once_with("TOKEN")
user_update_mock.assert_awaited_once()
data = response.json()
assert data["token"] == superuser_oauth.id
assert event_handler.called is False
def test_unknown_user(
self,
mock_user_db_oauth,
test_app_client: TestClient,
oauth_client,
event_handler,
):
state_jwt = generate_state_token({"authentication_backend": "mock"}, "SECRET")
with asynctest.patch.object(
oauth_client, "get_access_token"
) as get_access_token_mock:
get_access_token_mock.return_value = {
"access_token": "TOKEN",
"expires_at": 1579179542,
}
with asynctest.patch.object(
oauth_client, "get_id_email"
) as get_id_email_mock:
with asynctest.patch.object(
mock_user_db_oauth, "create"
) as user_create_mock:
get_id_email_mock.return_value = (
"unknown_user_oauth1",
"galahad@camelot.bt",
)
response = test_app_client.get(
"/callback", params={"code": "CODE", "state": state_jwt},
)
get_id_email_mock.assert_awaited_once_with("TOKEN")
user_create_mock.assert_awaited_once()
data = response.json()
assert "token" in data
assert event_handler.called is True
actual_user = event_handler.call_args[0][0]
assert actual_user.id == data["token"]
def test_inactive_user(
self,
mock_user_db_oauth,
test_app_client: TestClient,
oauth_client,
inactive_user_oauth,
event_handler,
):
state_jwt = generate_state_token({"authentication_backend": "mock"}, "SECRET")
with asynctest.patch.object(
oauth_client, "get_access_token"
) as get_access_token_mock:
get_access_token_mock.return_value = {
"access_token": "TOKEN",
"expires_at": 1579179542,
}
with asynctest.patch.object(
oauth_client, "get_id_email"
) as get_id_email_mock:
get_id_email_mock.return_value = (
"inactive_user_oauth1",
inactive_user_oauth.email,
)
response = test_app_client.get(
"/callback", params={"code": "CODE", "state": state_jwt},
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert response.json()["detail"] == ErrorCode.LOGIN_BAD_CREDENTIALS
assert event_handler.called is False
def test_redirect_url_router(
self,
mock_user_db_oauth,
test_app_client_redirect_url: TestClient,
oauth_client,
user_oauth,
):
state_jwt = generate_state_token({"authentication_backend": "mock"}, "SECRET")
with asynctest.patch.object(
oauth_client, "get_access_token"
) as get_access_token_mock:
get_access_token_mock.return_value = {
"access_token": "TOKEN",
"expires_at": 1579179542,
}
with asynctest.patch.object(
oauth_client, "get_id_email"
) as get_id_email_mock:
get_id_email_mock.return_value = ("user_oauth1", user_oauth.email)
response = test_app_client_redirect_url.get(
"/callback", params={"code": "CODE", "state": state_jwt},
)
get_access_token_mock.assert_awaited_once_with(
"CODE", "http://www.tintagel.bt/callback"
)
data = response.json()
assert data["token"] == user_oauth.id

8
tests/test_router.py → tests/test_router_users.py
View File

@@ -47,7 +47,7 @@ def test_app_client(mock_user_db, mock_authentication, event_handler) -> TestCli
[mock_authentication, mock_authentication_bis], mock_user_db
)
userRouter = get_user_router(
user_router = get_user_router(
mock_user_db,
User,
UserCreate,
@@ -58,11 +58,11 @@ def test_app_client(mock_user_db, mock_authentication, event_handler) -> TestCli
LIFETIME,
)
userRouter.add_event_handler(Event.ON_AFTER_REGISTER, event_handler)
userRouter.add_event_handler(Event.ON_AFTER_FORGOT_PASSWORD, event_handler)
user_router.add_event_handler(Event.ON_AFTER_REGISTER, event_handler)
user_router.add_event_handler(Event.ON_AFTER_FORGOT_PASSWORD, event_handler)
app = FastAPI()
app.include_router(userRouter)
app.include_router(user_router)
return TestClient(app)