63 lines
2.0 KiB
Python
63 lines
2.0 KiB
Python
"""
|
|
FastAPI 依赖注入 get_current_user 单元测试。
|
|
|
|
通过 FastAPI TestClient 验证 Authorization header 处理。
|
|
"""
|
|
|
|
import os
|
|
|
|
os.environ.setdefault("JWT_SECRET_KEY", "test-secret-key-for-unit-tests")
|
|
|
|
import pytest
|
|
from fastapi import Depends, FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.auth.dependencies import CurrentUser, get_current_user
|
|
from app.auth.jwt import create_access_token, create_refresh_token
|
|
|
|
# 构造一个最小 FastAPI 应用用于测试依赖注入
|
|
_test_app = FastAPI()
|
|
|
|
|
|
@_test_app.get("/protected")
|
|
async def protected_route(user: CurrentUser = Depends(get_current_user)):
|
|
return {"user_id": user.user_id, "site_id": user.site_id}
|
|
|
|
|
|
client = TestClient(_test_app)
|
|
|
|
|
|
class TestGetCurrentUser:
|
|
def test_valid_access_token(self):
|
|
token = create_access_token(user_id=10, site_id=100)
|
|
resp = client.get("/protected", headers={"Authorization": f"Bearer {token}"})
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["user_id"] == 10
|
|
assert data["site_id"] == 100
|
|
|
|
def test_missing_auth_header_returns_401(self):
|
|
"""缺少 Authorization header 时返回 401。"""
|
|
resp = client.get("/protected")
|
|
assert resp.status_code in (401, 403)
|
|
|
|
def test_invalid_token_returns_401(self):
|
|
resp = client.get(
|
|
"/protected", headers={"Authorization": "Bearer invalid.token.here"}
|
|
)
|
|
assert resp.status_code == 401
|
|
|
|
def test_refresh_token_rejected(self):
|
|
"""refresh 令牌不能用于访问受保护端点。"""
|
|
token = create_refresh_token(user_id=1, site_id=1)
|
|
resp = client.get("/protected", headers={"Authorization": f"Bearer {token}"})
|
|
assert resp.status_code == 401
|
|
|
|
def test_current_user_is_frozen_dataclass(self):
|
|
"""CurrentUser 是不可变的。"""
|
|
user = CurrentUser(user_id=1, site_id=2)
|
|
assert user.user_id == 1
|
|
assert user.site_id == 2
|
|
with pytest.raises(AttributeError):
|
|
user.user_id = 99 # type: ignore[misc]
|