feat: 新增用户管理（用户增删改查、密码重置、角色权限、会话认证）与 API 指南
- 新增 app/api/deps.py、app/core/users.py、app/core/sessions.py：会话鉴权依赖、 用户存储（PBKDF2-HMAC-SHA256 + 随机 salt，Redis/内存降级）、会话签发与校验（TTL 12h） - auth.py 新增用户管理端点（列表/创建/重置密码/删除）与 admin/user 角色权限边界， user 访问用户管理返回 1006，禁删自己与最后一个 admin - admin.html 新增用户管理面板（仅 admin 挂载）与 API 指南在线测试台 - Dockerfile 将 uv 放入 PATH；docker-compose 调整 qdrant 依赖为 service_started 并移除依赖 curl 的 healthcheck（官方镜像不含 curl） - 新增用户管理测试（users/sessions/auth_api/auth_integration），全量 461 项测试通过 Co-Authored-By: WorkBuddy <workbuddy@tencent.com>
This commit is contained in:
+6
-221
@@ -1,249 +1,34 @@
|
||||
"""认证 API 与核心函数的单元测试（mock UserStore，不依赖真实 Redis）
|
||||
"""旧 JWT 认证核心（app.core.auth）的单元测试
|
||||
|
||||
覆盖：
|
||||
- POST /auth/login：成功 / 密码错误 / 用户不存在
|
||||
- POST /auth/register：成功 / 用户已存在 / 注册关闭
|
||||
- GET /auth/me：有效 token / 无 token / 无效 token（需走真实 get_current_user）
|
||||
API 层（/api/v1/auth/*）已切换为会话制认证，端点契约测试见 tests/test_auth_api.py；
|
||||
本文件仅保留 app.core.auth 的函数级覆盖：
|
||||
- require_admin：非 admin 抛 FORBIDDEN（直接测函数）
|
||||
- create_access_token + decode_token 往返一致
|
||||
- hash_password + verify_password 正确 / 错误
|
||||
|
||||
UserStore 的 authenticate/create 在 FakeUserStore 中 mock，避免依赖真实 Redis。
|
||||
- create_access_token + decode_token 往返一致 / 非法 token 拒绝
|
||||
- hash_password + verify_password 正确 / 错误 / 非法哈希
|
||||
"""
|
||||
|
||||
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
|
||||
from app.models.auth import AuthUser
|
||||
|
||||
|
||||
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）"""
|
||||
|
||||
|
||||
Reference in New Issue
Block a user