mirror of
https://github.com/fastapi-users/fastapi-users.git
synced 2025-11-01 18:48:14 +08:00
Implement user-facing API
This commit is contained in:
@ -1,7 +1,8 @@
|
||||
from typing import Optional
|
||||
|
||||
import pytest
|
||||
from fastapi import HTTPException
|
||||
from fastapi import Depends, HTTPException
|
||||
from fastapi.security import OAuth2PasswordBearer
|
||||
from starlette import status
|
||||
from starlette.responses import Response
|
||||
|
||||
@ -62,13 +63,21 @@ class MockAuthentication(BaseAuthentication):
|
||||
async def get_login_response(self, user: BaseUserDB, response: Response):
|
||||
return {"token": user.id}
|
||||
|
||||
async def authenticate(self, token: str) -> BaseUserDB:
|
||||
user = await self.user_db.get(token)
|
||||
if user is None or not user.is_active:
|
||||
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
|
||||
return user
|
||||
def get_authentication_method(self, user_db: BaseUserDatabase):
|
||||
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/login")
|
||||
|
||||
async def authentication_method(token: str = Depends(oauth2_scheme)):
|
||||
credentials_exception = HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED
|
||||
)
|
||||
user = await user_db.get(token)
|
||||
if user is None or not user.is_active:
|
||||
raise credentials_exception
|
||||
return user
|
||||
|
||||
return authentication_method
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_authentication(mock_user_db) -> MockAuthentication:
|
||||
return MockAuthentication(mock_user_db)
|
||||
def mock_authentication() -> MockAuthentication:
|
||||
return MockAuthentication()
|
||||
|
||||
@ -14,8 +14,8 @@ LIFETIME = 3600
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def jwt_authentication(mock_user_db):
|
||||
return JWTAuthentication(SECRET, LIFETIME, mock_user_db)
|
||||
def jwt_authentication():
|
||||
return JWTAuthentication(SECRET, LIFETIME)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@ -28,12 +28,14 @@ def token():
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_auth_client(jwt_authentication):
|
||||
def test_auth_client(jwt_authentication, mock_user_db):
|
||||
app = FastAPI()
|
||||
|
||||
@app.get("/test-auth")
|
||||
def test_auth(
|
||||
user: BaseUserDB = Depends(jwt_authentication.get_authentication_method())
|
||||
user: BaseUserDB = Depends(
|
||||
jwt_authentication.get_authentication_method(mock_user_db)
|
||||
)
|
||||
):
|
||||
return user
|
||||
|
||||
|
||||
54
tests/test_fastapi_users.py
Normal file
54
tests/test_fastapi_users.py
Normal file
@ -0,0 +1,54 @@
|
||||
import pytest
|
||||
from fastapi import Depends, FastAPI
|
||||
from starlette import status
|
||||
from starlette.testclient import TestClient
|
||||
|
||||
from fastapi_users import FastAPIUsers
|
||||
from fastapi_users.models import BaseUser, BaseUserDB
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fastapi_users(mock_user_db, mock_authentication) -> FastAPIUsers:
|
||||
class User(BaseUser):
|
||||
pass
|
||||
|
||||
return FastAPIUsers(mock_user_db, mock_authentication, User)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_app_client(fastapi_users: FastAPIUsers) -> TestClient:
|
||||
app = FastAPI()
|
||||
app.include_router(fastapi_users.router)
|
||||
|
||||
@app.get("/authenticated")
|
||||
def authenticated(user=Depends(fastapi_users.get_current_user)):
|
||||
return user
|
||||
|
||||
return TestClient(app)
|
||||
|
||||
|
||||
class TestRouter:
|
||||
def test_routes_exist(self, test_app_client: TestClient):
|
||||
response = test_app_client.post("/register")
|
||||
assert response.status_code != status.HTTP_404_NOT_FOUND
|
||||
|
||||
response = test_app_client.post("/login")
|
||||
assert response.status_code != status.HTTP_404_NOT_FOUND
|
||||
|
||||
|
||||
class TestGetCurrentUser:
|
||||
def test_missing_token(self, test_app_client: TestClient):
|
||||
response = test_app_client.get("/authenticated")
|
||||
assert response.status_code == status.HTTP_401_UNAUTHORIZED
|
||||
|
||||
def test_invalid_token(self, test_app_client: TestClient):
|
||||
response = test_app_client.get(
|
||||
"/authenticated", headers={"Authorization": "Bearer foo"}
|
||||
)
|
||||
assert response.status_code == status.HTTP_401_UNAUTHORIZED
|
||||
|
||||
def test_valid_token(self, test_app_client: TestClient, user: BaseUserDB):
|
||||
response = test_app_client.get(
|
||||
"/authenticated", headers={"Authorization": f"Bearer {user.id}"}
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
Reference in New Issue
Block a user