Fehler 401 bei der Authentifizierung mit Bearer bei fastapi_users ⇐ Python

[Anonymous]

Ich verstehe nicht, wo das Problem liegt. Alles scheint gut zu funktionieren. Ich habe den gesamten Code unten eingefügt.
Meine Codestruktur sieht folgendermaßen aus:
service.py

Code: Select all

import uuid
from typing import Optional

from fastapi import Depends, Request
from fastapi_users import BaseUserManager, UUIDIDMixin
from fastapi_users.authentication import (
AuthenticationBackend,
BearerTransport,
JWTStrategy,
)
from fastapi_users.password import PasswordHelper
from pwdlib import PasswordHash, exceptions
from pwdlib.hashers.argon2 import Argon2Hasher

from app import config, db

from . import crud

class SecurityConfig:
__slots__ = ()
PASSWORD_HASH = PasswordHash((Argon2Hasher(),))
SECRET = config.UserSecret().secret
PASSWORD_HELPER = PasswordHelper(PASSWORD_HASH)
BEARER_TRANSPORT = BearerTransport(tokenUrl="auth/jwt/login")

class UserManager(UUIDIDMixin, BaseUserManager[db.models.User, uuid.UUID]):
reset_password_token_secret = SecurityConfig().SECRET
verification_token_secret = SecurityConfig().SECRET

async def on_after_register(
self, user: db.models.User, request: Optional[Request] = None
):
print(f"User {user.id} has registered.")

async def on_after_forgot_password(
self, user: db.models.User, token: str, request: Optional[Request] = None
):
print(f"User {user.id} has forgot their password. Reset token: {token}")

async def on_after_request_verify(
self, user: db.models.User, token: str, request: Optional[Request] = None
):

print(f"Verification requested for user {user.id}. Verification token: {token}")

async def get_user_manager(user_db=Depends(crud.get_user_db)):
"""
Yields a UserManager instance with the given user_db.

The UserManager is an instance of BaseUserManager that is used to
manage users in the application. It provides methods for registering,
authenticating, and verifying users.

Args:
user_db: The user database to use for the UserManager.
"""
yield UserManager(user_db, SecurityConfig().PASSWORD_HELPER)

def get_jwt_strategy() -> JWTStrategy:
"""
Returns the JWT strategy for the application.

The strategy is used by the auth backend to authenticate users.
It uses the SECRET key to sign the JWT and has a lifetime of 3600*24*7 seconds (7 days).

Returns:
JWTStrategy: The JWT strategy for the application.
"""
return JWTStrategy(secret=SecurityConfig().SECRET, lifetime_seconds=3600 * 24 * 7)

AUTH_BACKEND = AuthenticationBackend(
name="jwt",
transport=SecurityConfig().BEARER_TRANSPORT,
get_strategy=get_jwt_strategy,
)

get_user_db für fastapi_users:

Code: Select all

async def get_user_db(session: db.AsyncSession = Depends(db.get_session_depend)):
"""
Yields a SQLAlchemyUserDatabase instance with the given session and User model.

The SQLAlchemyUserDatabase is an instance of BaseUserDatabase that is used to
manage users in the application.  It provides methods for registering,
authenticating, and verifying users.

Args:
session: The database session to use for the UserManager.
model: The SQLAlchemy model to use for the UserManager.
"""
yield SQLAlchemyUserDatabase(session, db.models.User)

Tests:

Code: Select all

from typing import AsyncGenerator

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

from app import app, db

BASE_URL = "/users"

@pytest.fixture(scope="module")
def db_session_mock():
uri_db = "user:user_password@localhost:3306/testdb"
engine_async = create_async_engine(
f"mysql+aiomysql://{uri_db}", pool_pre_ping=True, pool_recycle=1800
)
async_session_maker = sessionmaker(
bind=engine_async, class_=AsyncSession, expire_on_commit=False
)

engine_sync = create_engine(f"mysql+pymysql://{uri_db}", echo=False, future=True)
db.models.Base.metadata.create_all(engine_sync)

async def override_get_session() -> AsyncGenerator[AsyncSession, None]:
async with async_session_maker() as session:
yield session

yield override_get_session

db.models.Base.metadata.drop_all(engine_sync)

@pytest.fixture(scope="module")
def client(db_session_mock):
app.dependency_overrides[db.get_session_depend] = db_session_mock
with TestClient(app) as client:
yield client

@pytest.fixture(scope="module")
def test_user():
return {"username": "[email protected]", "password": "testpass"}

def test_register_user(client, test_user):
response = client.post(
BASE_URL + "/auth/register",
json={
"email": test_user["username"],
"password": test_user["password"],
"full_name": "alireza",
},
)
assert response.status_code == 201
data = response.json()
assert "id" in data
assert data["email"] == test_user["username"]

def test_login_user(client, test_user):
response = client.post(
BASE_URL + "/auth/jwt/login",
data=test_user,
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
assert response.status_code == 200
data = response.json()
assert "access_token" in data
assert data["token_type"] == "bearer"
assert isinstance(data["access_token"], str)

def test_access_protected_route(client, test_user):
response = client.post(
BASE_URL + "/auth/jwt/login",
data=test_user,
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
assert response.status_code == 200
data = response.json()
assert "access_token"  in data
assert data["token_type"] == "bearer"
assert isinstance(data["access_token"], str)
token = data["access_token"]

response = client.get(
"/users/me",
headers={"Authorization": f"Bearer {token}"},
)
assert response.status_code == 200
data = response.json()
assert data["email"] == test_user["username"]

Code: Select all

========================================= test session starts ==========================================
platform linux -- Python 3.10.16, pytest-8.4.2, pluggy-1.6.0
rootdir: /mnt/Files/Codes/Python/Projects/Freelanse/ChatG/backend
plugins: anyio-4.11.0, asyncio-1.2.0
asyncio: mode=strict, debug=False, asyncio_default_fixture_loop_scope=None, asyncio_default_test_loop_scope=function
collected 3 items

app/tests/test_user_auth.py User 0a3d78ea-168b-4840-afcf-3ff9ebd444fc has registered.
..F

=============================================== FAILURES ===============================================
_____________________________________ test_access_protected_route ______________________________________

client = 
test_user = {'password': 'testpass', 'username': '[email protected]'}

def test_access_protected_route(client, test_user):
response = client.post(
BASE_URL + "/auth/jwt/login",
data=test_user,
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
assert response.status_code == 200
data = response.json()
assert "access_token" in data
assert data["token_type"] == "bearer"
assert isinstance(data["access_token"], str)
token = data["access_token"]

response = client.get(
"/users/me",
headers={"Authorization": f"Bearer {token}"},
)
>       assert response.status_code == 200
E       assert 401 == 200
E        +  where 401 = .status_code

app/tests/test_user_auth.py:94: AssertionError
======================================= short test summary info ========================================
FAILED app/tests/test_user_auth.py::test_access_protected_route - assert 401 == 200
===================================== 1 failed, 2 passed in 5.99s ======================================

FAILED app/tests/test_user_auth.py::test_access_protected_route - Assert 401 == 200

Response [401 Unauthorized]
Seltsamerweise ist Heather nicht hier.??
(Pdb) Response.request.headers

Headers({'host': 'testserver', 'accept': '*/*', 'accept-encoding': 'gzip, deflate', 'connection': 'keep-alive', 'user-agent': 'testclient', 'authorization': '[secure]'})