kplam
dce9e31bde
- 新增 JWT 认证模块，支持登录/注册/用户管理 - 新增文件上传接口，支持 .txt/.md/.html/.pdf/.docx 等格式解析入库 - 新增检索结果 AI 总结功能 - 新增文本去重缓存机制 - 新增全局认证夹具简化测试 - 新增配置项与环境变量支持 - 完善文档与测试覆盖
294 lines
10 KiB
Python
294 lines
10 KiB
Python
"""认证 API 与核心函数的单元测试（mock UserStore，不依赖真实 Redis）
|
||
|
||
覆盖：
|
||
- POST /auth/login：成功 / 密码错误 / 用户不存在
|
||
- POST /auth/register：成功 / 用户已存在 / 注册关闭
|
||
- GET /auth/me：有效 token / 无 token / 无效 token（需走真实 get_current_user）
|
||
- require_admin：非 admin 抛 FORBIDDEN（直接测函数）
|
||
- create_access_token + decode_token 往返一致
|
||
- hash_password + verify_password 正确 / 错误
|
||
|
||
UserStore 的 authenticate/create 在 FakeUserStore 中 mock，避免依赖真实 Redis。
|
||
"""
|
||
|
||
from datetime import UTC, datetime
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
from app.api.response import ApiError
|
||
from app.api.v1 import auth as auth_module
|
||
from app.config import settings
|
||
from app.core import auth as auth_core
|
||
from app.core.auth import (
|
||
ERR_BAD_CREDENTIALS,
|
||
ERR_FORBIDDEN,
|
||
ERR_REGISTER_DISABLED,
|
||
ERR_TOKEN_INVALID,
|
||
ERR_UNAUTHORIZED,
|
||
ERR_USER_EXISTS,
|
||
create_access_token,
|
||
decode_token,
|
||
get_current_user,
|
||
hash_password,
|
||
require_admin,
|
||
verify_password,
|
||
)
|
||
from app.main import app
|
||
from app.models.auth import AuthUser, StoredUser
|
||
|
||
|
||
def _now() -> datetime:
|
||
return datetime.now(UTC)
|
||
|
||
|
||
class FakeUserStore:
|
||
"""假 UserStore：按预置数据返回结果或抛 ApiError，记录调用
|
||
|
||
authenticate/create 使用预置 hashed_password，不调用真实 hash_password。
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
user: StoredUser | None = None,
|
||
exists: bool = False,
|
||
create_error: ApiError | None = None,
|
||
auth_fail: bool = False,
|
||
) -> None:
|
||
self.user = user
|
||
self._exists = exists
|
||
self.create_error = create_error
|
||
self.auth_fail = auth_fail
|
||
self.create_calls: list[tuple[str, str, str]] = []
|
||
|
||
async def get(self, username: str) -> StoredUser | None:
|
||
if self.user is not None and self.user.username == username:
|
||
return self.user
|
||
return None
|
||
|
||
async def exists(self, username: str) -> bool:
|
||
return self._exists
|
||
|
||
async def create(self, username: str, password: str, role: str = "user") -> StoredUser:
|
||
if self.create_error is not None:
|
||
raise self.create_error
|
||
user = StoredUser(
|
||
username=username,
|
||
role=role,
|
||
created_at=_now(),
|
||
hashed_password="fake-hash",
|
||
)
|
||
self.create_calls.append((username, password, role))
|
||
return user
|
||
|
||
async def authenticate(self, username: str, password: str) -> StoredUser:
|
||
if self.auth_fail or self.user is None or self.user.username != username:
|
||
raise ApiError(ERR_BAD_CREDENTIALS, "用户名或密码错误")
|
||
return self.user
|
||
|
||
|
||
class NullUserStore:
|
||
"""恒返回 None 的空存储：用于 /auth/me 测试，让 get_current_user 降级用 JWT payload"""
|
||
|
||
async def get(self, username: str) -> StoredUser | None:
|
||
return None
|
||
|
||
async def exists(self, username: str) -> bool:
|
||
return False
|
||
|
||
|
||
def _install_store(monkeypatch: pytest.MonkeyPatch, store: FakeUserStore) -> None:
|
||
"""将 auth 路由模块的 get_user_store 替换为假存储"""
|
||
monkeypatch.setattr(auth_module, "get_user_store", lambda: store)
|
||
|
||
|
||
def _make_stored_user(username: str = "alice", role: str = "user") -> StoredUser:
|
||
return StoredUser(username=username, role=role, created_at=_now(), hashed_password="fake-hash")
|
||
|
||
|
||
class TestAuthLoginApi:
|
||
"""POST /auth/login"""
|
||
|
||
def test_login_success(self, monkeypatch: pytest.MonkeyPatch):
|
||
user = _make_stored_user(username="alice", role="admin")
|
||
_install_store(monkeypatch, FakeUserStore(user=user))
|
||
|
||
client = TestClient(app)
|
||
resp = client.post("/api/v1/auth/login", json={"username": "alice", "password": "secret123"})
|
||
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body["code"] == 0
|
||
data = body["data"]
|
||
assert data["access_token"]
|
||
assert data["token_type"] == "bearer"
|
||
assert data["expires_in"] == settings.jwt_expire_minutes * 60
|
||
assert data["user"]["username"] == "alice"
|
||
assert data["user"]["role"] == "admin"
|
||
assert "created_at" in data["user"]
|
||
|
||
def test_login_wrong_password(self, monkeypatch: pytest.MonkeyPatch):
|
||
user = _make_stored_user(username="alice")
|
||
# authenticate 失败（密码不匹配）
|
||
_install_store(monkeypatch, FakeUserStore(user=user, auth_fail=True))
|
||
|
||
client = TestClient(app)
|
||
resp = client.post("/api/v1/auth/login", json={"username": "alice", "password": "WRONG"})
|
||
|
||
body = resp.json()
|
||
assert body["code"] == ERR_BAD_CREDENTIALS
|
||
assert body["data"] is None
|
||
|
||
def test_login_user_not_found(self, monkeypatch: pytest.MonkeyPatch):
|
||
_install_store(monkeypatch, FakeUserStore(user=None))
|
||
|
||
client = TestClient(app)
|
||
resp = client.post("/api/v1/auth/login", json={"username": "nobody", "password": "whatever"})
|
||
|
||
body = resp.json()
|
||
assert body["code"] == ERR_BAD_CREDENTIALS
|
||
|
||
|
||
class TestAuthRegisterApi:
|
||
"""POST /auth/register"""
|
||
|
||
def test_register_success(self, monkeypatch: pytest.MonkeyPatch):
|
||
store = FakeUserStore(user=None, exists=False)
|
||
_install_store(monkeypatch, store)
|
||
|
||
client = TestClient(app)
|
||
resp = client.post(
|
||
"/api/v1/auth/register", json={"username": "newbie", "password": "passwd123"}
|
||
)
|
||
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body["code"] == 0
|
||
data = body["data"]
|
||
assert data["access_token"]
|
||
assert data["user"]["username"] == "newbie"
|
||
assert data["user"]["role"] == "user"
|
||
assert len(store.create_calls) == 1
|
||
assert store.create_calls[0][0] == "newbie"
|
||
assert store.create_calls[0][2] == "user"
|
||
|
||
def test_register_user_exists(self, monkeypatch: pytest.MonkeyPatch):
|
||
store = FakeUserStore(
|
||
exists=True,
|
||
create_error=ApiError(ERR_USER_EXISTS, "用户名已存在: newbie"),
|
||
)
|
||
_install_store(monkeypatch, store)
|
||
|
||
client = TestClient(app)
|
||
resp = client.post(
|
||
"/api/v1/auth/register", json={"username": "newbie", "password": "passwd123"}
|
||
)
|
||
|
||
body = resp.json()
|
||
assert body["code"] == ERR_USER_EXISTS
|
||
assert body["data"] is None
|
||
|
||
def test_register_disabled(self, monkeypatch: pytest.MonkeyPatch):
|
||
monkeypatch.setattr(settings, "auth_register_enabled", False)
|
||
_install_store(monkeypatch, FakeUserStore())
|
||
|
||
client = TestClient(app)
|
||
resp = client.post(
|
||
"/api/v1/auth/register", json={"username": "newbie", "password": "passwd123"}
|
||
)
|
||
|
||
body = resp.json()
|
||
assert body["code"] == ERR_REGISTER_DISABLED
|
||
assert body["data"] is None
|
||
|
||
|
||
class TestAuthMeApi:
|
||
"""GET /auth/me：需走真实 get_current_user，测试内清除 override"""
|
||
|
||
def test_me_with_valid_token(self, monkeypatch: pytest.MonkeyPatch):
|
||
# 清除 override，让 get_current_user 走真实认证
|
||
app.dependency_overrides.pop(get_current_user, None)
|
||
# mock get_user_store 返回空存储（模拟 Redis 无该用户，降级用 JWT payload）
|
||
monkeypatch.setattr(auth_core, "get_user_store", lambda: NullUserStore())
|
||
|
||
token, _ = create_access_token("alice", "admin")
|
||
client = TestClient(app)
|
||
resp = client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {token}"})
|
||
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body["code"] == 0
|
||
assert body["data"]["username"] == "alice"
|
||
assert body["data"]["role"] == "admin"
|
||
|
||
def test_me_without_token(self, monkeypatch: pytest.MonkeyPatch):
|
||
app.dependency_overrides.pop(get_current_user, None)
|
||
monkeypatch.setattr(auth_core, "get_user_store", lambda: NullUserStore())
|
||
|
||
client = TestClient(app)
|
||
resp = client.get("/api/v1/auth/me")
|
||
|
||
body = resp.json()
|
||
assert body["code"] == ERR_UNAUTHORIZED
|
||
assert body["data"] is None
|
||
|
||
def test_me_with_invalid_token(self, monkeypatch: pytest.MonkeyPatch):
|
||
app.dependency_overrides.pop(get_current_user, None)
|
||
monkeypatch.setattr(auth_core, "get_user_store", lambda: NullUserStore())
|
||
|
||
client = TestClient(app)
|
||
resp = client.get("/api/v1/auth/me", headers={"Authorization": "Bearer not-a-jwt"})
|
||
|
||
body = resp.json()
|
||
assert body["code"] == ERR_TOKEN_INVALID
|
||
assert body["data"] is None
|
||
|
||
|
||
class TestRequireAdmin:
|
||
"""require_admin：admin 通过、非 admin 抛 FORBIDDEN（直接测函数，不经 API）"""
|
||
|
||
async def test_admin_passes(self):
|
||
admin = AuthUser(username="alice", role="admin", created_at=_now())
|
||
result = await require_admin(user=admin)
|
||
assert result.role == "admin"
|
||
|
||
async def test_non_admin_raises_forbidden(self):
|
||
normal = AuthUser(username="bob", role="user", created_at=_now())
|
||
with pytest.raises(ApiError) as exc:
|
||
await require_admin(user=normal)
|
||
assert exc.value.code == ERR_FORBIDDEN
|
||
|
||
|
||
class TestAuthCoreFunctions:
|
||
"""核心函数：token 往返 / 密码哈希"""
|
||
|
||
def test_create_and_decode_token_roundtrip(self):
|
||
token, expires_in = create_access_token("alice", "admin")
|
||
assert expires_in == settings.jwt_expire_minutes * 60
|
||
payload = decode_token(token)
|
||
assert payload["sub"] == "alice"
|
||
assert payload["role"] == "admin"
|
||
assert "iat" in payload and "exp" in payload
|
||
|
||
def test_decode_invalid_token_raises(self):
|
||
with pytest.raises(ApiError) as exc:
|
||
decode_token("not-a-jwt")
|
||
assert exc.value.code == ERR_TOKEN_INVALID
|
||
|
||
def test_hash_password_not_plaintext(self):
|
||
hashed = hash_password("my-secret")
|
||
assert hashed != "my-secret"
|
||
assert hashed.startswith("$2") # bcrypt 哈希前缀
|
||
|
||
def test_verify_password_correct(self):
|
||
hashed = hash_password("my-secret")
|
||
assert verify_password("my-secret", hashed) is True
|
||
|
||
def test_verify_password_wrong(self):
|
||
hashed = hash_password("my-secret")
|
||
assert verify_password("wrong-password", hashed) is False
|
||
|
||
def test_verify_password_garbage_hash_returns_false(self):
|
||
# 非法 hash 会被 bcrypt 拒绝，verify_password 捕获异常返回 False
|
||
assert verify_password("any", "not-a-valid-hash") is False
|