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:
2026-07-31 21:29:02 +08:00
parent 2ab8b56a01
commit 92b062c048
24 changed files with 2771 additions and 350 deletions
+37 -14
View File
@@ -24,7 +24,9 @@ QMDSearch/
│ ├── config.py # 配置管理
│ ├── api/ # API 路由层
│ │ ├── response.py # 统一响应格式 (ok/error/ApiError)
│ │ ├── deps.py # 会话鉴权依赖 (Bearer → session → 用户记录)
│ │ └── v1/ # API v1 版本
│ │ ├── auth.py # 认证与用户管理接口
│ │ ├── search.py # 检索接口
│ │ ├── document.py # 文档入库/管理接口
│ │ └── knowledge.py # 知识库接口
@@ -38,6 +40,8 @@ QMDSearch/
│ │ ├── headings.py # 原生标题树解析
│ │ ├── classifier.py # 文档分类 (主类+标签+置信度)
│ │ ├── chunker.py # 标题树感知 chunk 切分
│ │ ├── users.py # 用户存储与密码哈希 (PBKDF2Redis/内存降级)
│ │ ├── sessions.py # 会话签发与校验 (Redis TTL 12h/内存降级)
│ │ └── ingestion.py # 文档入库 (总结→分类→写入)
│ ├── models/ # 数据模型
│ │ ├── search.py # 检索请求/响应模型
@@ -80,25 +84,44 @@ QMDSearch/
## API 清单
| 方法 | 路径 | 说明 |
|------|------|------|
| GET | `/api/v1/health` | 健康检查 |
| POST | `/api/v1/search` | 分层检索 |
| POST | `/api/v1/documents` | 文档入库(202 异步入库,返回 task_id |
| POST | `/api/v1/documents/upload` | multipart 文件上传入库(202 异步,支持 .txt/.md/.html/.htm/.pdf/.docx |
| GET | `/api/v1/documents/tasks/{task_id}` | 入库任务状态查询(done 附 resultfailed 附 error |
| GET | `/api/v1/knowledge/categories` | 知识分类类目集 |
| GET | `/api/v1/knowledge/stats` | 统计(四层点数 + 类目分布 + uncategorized 数) |
| GET | `/api/v1/documents` | 文档列表(limit/offset 分页) |
| GET | `/api/v1/documents/{doc_id}` | 文档详情 |
| DELETE | `/api/v1/documents/{doc_id}` | 删除文档(幂等) |
| GET | `/admin` | 管理页面 |
| 方法 | 路径 | 说明 | 鉴权 |
|------|------|------|------|
| GET | `/api/v1/health` | 健康检查 | 免登录 |
| POST | `/api/v1/search` | 分层检索 | 免登录 |
| POST | `/api/v1/documents` | 文档入库(202 异步入库,返回 task_id | Bearer |
| POST | `/api/v1/documents/upload` | multipart 文件上传入库(202 异步,支持 .txt/.md/.html/.htm/.pdf/.docx | Bearer |
| GET | `/api/v1/documents/tasks/{task_id}` | 入库任务状态查询(done 附 resultfailed 附 error | 免登录 |
| GET | `/api/v1/knowledge/categories` | 知识分类类目集 | 免登录 |
| GET | `/api/v1/knowledge/stats` | 统计(四层点数 + 类目分布 + uncategorized 数) | 免登录 |
| GET | `/api/v1/documents` | 文档列表(limit/offset 分页) | 免登录 |
| GET | `/api/v1/documents/{doc_id}` | 文档详情 | 免登录 |
| DELETE | `/api/v1/documents/{doc_id}` | 删除文档(幂等) | Bearer |
| POST | `/api/v1/auth/login` | 用户名密码登录,签发 session tokenTTL 12h | 免登录 |
| POST | `/api/v1/auth/logout` | 退出登录(删除当前 session | Bearer |
| POST | `/api/v1/auth/password` | 修改自己的密码(must_change_password 用户唯一可用接口) | Bearer |
| GET | `/api/v1/auth/me` | 当前登录用户信息(脱敏) | Bearer |
| GET | `/api/v1/auth/users` | 用户列表(脱敏) | Bearer + admin |
| POST | `/api/v1/auth/users` | 创建用户(重名/非法用户名/弱密码 1001 | Bearer + admin |
| POST | `/api/v1/auth/users/{username}/password` | 重置指定用户密码(成功后清除其全部 session | Bearer + admin |
| DELETE | `/api/v1/auth/users/{username}` | 删除用户(清除其 session;禁删自己/最后一个 admin | Bearer + admin |
| GET | `/admin` | 管理页面 | 页面登录门禁 |
「Bearer」指请求头 `Authorization: Bearer <token>`token 经 `/api/v1/auth/login` 获取;变更类文档端点(POST /documents、POST /documents/upload、DELETE /documents/{id})需 Bearer tokenadmin/user 角色均可),查询类端点免登录。
入库任务状态持久化在 Redis(key: `ingest_task:{task_id}`):进行中与 done 保留 24hfailed 保留 7 天;Redis 不可用时降级为纯内存。
## 管理页面
浏览器访问 `/admin`单页面含概览、文档管理(列表/详情/删除)、文档入库、检索测试台、类目列表五个区块
浏览器访问 `/admin`页面带登录门禁(未登录/token 失效自动回登录卡片;must_change_password 用户先强制改密后方可进入)。单页面含七个区块:概览、文档管理(列表/详情/删除)、文档入库(文本 + 文件上传)、检索测试台、类目列表、API 指南(端点清单 + 在线测试台)、用户管理(仅 admin 角色挂载,含创建/重置密码/删除)
## 认证与用户
会话制认证:登录签发 session tokenRedis 持久化,TTL 12h;Redis 不可用时降级为进程内存,重启失效),请求经 `Authorization: Bearer <token>` 携带,鉴权依赖见 `app/api/deps.py`
- **角色与权限边界**: `admin` 拥有全部权限(含 /auth/users* 用户管理);`user` 可登录并调用变更类文档端点(入库/上传/删除),访问用户管理端点返回 1006
- **初始 admin 引导**: 空库启动时 `bootstrap_admin` 自动创建 admin 账号,随机明文密码仅在启动日志中打印一次(must_change_password=true),首次登录后须先经 POST /auth/password 改密,改密前访问其他端点返回 1006
- **免登录端点**: 查询类端点(POST /search、GET /documents*、GET /knowledge/*、GET /health)不需要 token
- **相关错误码**: 1005 未认证或凭证无效,1006 权限不足/首次登录须先改密
## 编码规范
+2 -2
View File
@@ -16,8 +16,8 @@ FROM python:3.12-slim AS base
WORKDIR /app
# 安装 uv
COPY --from=ghcr.io/astral-sh/uv:latest /uv /uv/bin/uv
# 安装 uv(放到 PATH 中,供 uv sync 与 CMD 的 uv run 使用)
COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
# 依赖层
COPY pyproject.toml uv.lock ./
+95
View File
@@ -0,0 +1,95 @@
"""API 鉴权依赖:Bearer token → session → 用户记录
UserStore/SessionStore 为模块级懒加载单例:Redis 客户端创建失败时构造传 None,
降级为进程级内存存储(仅开发/兜底,重启失效)。存储后端读写异常统一转 2001。
测试可 monkeypatch _user_store/_session_store 单例完成注入。
"""
import structlog
from fastapi import Depends, Header
from redis import asyncio as redis_async
from app.api.response import ApiError
from app.config import settings
from app.core.sessions import SessionStore, SessionStoreError
from app.core.users import UserRecord, UserStore, UserStoreError
logger = structlog.get_logger()
# 模块级懒加载单例
_user_store: UserStore | None = None
_session_store: SessionStore | None = None
def _create_redis_client() -> redis_async.Redis | None:
"""创建 redis.asyncio 客户端(decode_responses=True);失败返回 None 走内存降级"""
try:
return redis_async.from_url(settings.redis_url, decode_responses=True)
except Exception:
logger.warning("Redis 客户端创建失败,认证存储降级为内存模式", exc_info=True)
return None
def _get_user_store() -> UserStore:
"""用户存储懒加载单例"""
global _user_store
if _user_store is None:
_user_store = UserStore(_create_redis_client())
return _user_store
def _get_session_store() -> SessionStore:
"""会话存储懒加载单例"""
global _session_store
if _session_store is None:
_session_store = SessionStore(_create_redis_client())
return _session_store
def _parse_bearer(authorization: str | None) -> str | None:
"""解析 Authorization 头中的 Bearer token;缺失或格式非法返回 None"""
if not authorization:
return None
scheme, _, token = authorization.partition(" ")
if scheme.lower() != "bearer" or not token.strip():
return None
return token.strip()
async def _resolve_user(authorization: str | None) -> tuple[UserRecord, str]:
"""Bearer token → session → 用户记录(不拦截 must_change_password
凭证缺失/无效/用户不存在抛 ApiError(1005);存储后端异常抛 ApiError(2001)。
供 get_current_user 与改密/退出端点共用(后者须对 must_change_password 用户放行)。
"""
token = _parse_bearer(authorization)
if token is None:
raise ApiError(1005, "未认证或凭证无效")
try:
session = await _get_session_store().get(token)
if session is None:
raise ApiError(1005, "未认证或凭证无效")
user = await _get_user_store().get(session["username"])
except (SessionStoreError, UserStoreError) as exc:
raise ApiError(2001, "认证服务暂不可用") from exc
if user is None:
raise ApiError(1005, "未认证或凭证无效")
return user, token
async def get_current_user(authorization: str | None = Header(None)) -> UserRecord:
"""鉴权依赖:校验 Bearer token 并返回当前用户记录
must_change_password 用户被拦截(1006),须先经 POST /auth/password 改密。
"""
user, _ = await _resolve_user(authorization)
if user.must_change_password:
raise ApiError(1006, "首次登录须先修改密码")
return user
async def require_admin(user: UserRecord = Depends(get_current_user)) -> UserRecord:
"""鉴权依赖:在 get_current_user 之上要求 admin 角色"""
if user.role != "admin":
raise ApiError(1006, "权限不足")
return user
+172 -42
View File
@@ -1,62 +1,192 @@
"""认证 APIPOST /auth/login、POST /auth/register、GET /auth/me"""
"""认证与用户管理 API/api/v1/auth 下的登录/退出/改密与用户管理端点
from typing import Any
会话制认证:登录签发 session token(存储见 app/core/sessions.pyTTL 12h),
请求经 Authorization: Bearer <token> 携带,鉴权依赖见 app/api/deps.py。
用户管理端点(/users*)全部要求 admin 角色;/logout 与 /password 对
must_change_password 用户放行(改密是被拦截用户唯一能用的接口)。
"""
from typing import Any, Literal
import structlog
from fastapi import APIRouter, Depends
from fastapi import APIRouter, Depends, Header
from pydantic import BaseModel
from app.api import deps
from app.api.response import ApiError, ok
from app.config import settings
from app.core.auth import (
ERR_REGISTER_DISABLED,
UserStore,
create_access_token,
get_current_user,
get_user_store,
)
from app.models.auth import (
AuthUser,
LoginRequest,
RegisterRequest,
TokenResponse,
)
from app.core.sessions import SessionStoreError
from app.core.users import UserExistsError, UserRecord, UserStoreError
logger = structlog.get_logger()
router = APIRouter(prefix="/api/v1", tags=["auth"])
router = APIRouter(prefix="/api/v1/auth", tags=["auth"])
def _build_token_response(user, store: UserStore) -> dict[str, Any]:
"""签发 token 并构造统一响应 data"""
token, expires_in = create_access_token(user.username, user.role)
auth_user = AuthUser(username=user.username, role=user.role, created_at=user.created_at)
resp = TokenResponse(access_token=token, expires_in=expires_in, user=auth_user)
return resp.model_dump(mode="json")
class LoginRequest(BaseModel):
"""登录请求"""
username: str
password: str
@router.post("/auth/login")
class PasswordChangeRequest(BaseModel):
"""修改自己的密码"""
old_password: str
new_password: str
class UserCreateRequest(BaseModel):
"""管理员创建用户"""
username: str
password: str
role: Literal["admin", "user"] = "user"
class PasswordResetRequest(BaseModel):
"""管理员重置他人密码"""
new_password: str
def _public_user(user: UserRecord) -> dict[str, Any]:
"""用户记录脱敏:剔除 password_hash/salt"""
return {
"username": user.username,
"role": user.role,
"must_change_password": user.must_change_password,
"created_at": user.created_at,
}
@router.post("/login")
async def login(req: LoginRequest) -> dict[str, Any]:
"""用户名密码登录,返回 JWT access_token"""
store = get_user_store()
user = await store.authenticate(req.username, req.password)
"""用户名密码登录:成功签发 session tokenmust_change_password 用户也可登录)"""
try:
user = await deps._get_user_store().verify_password(req.username, req.password)
except UserStoreError as exc:
raise ApiError(2001, "认证服务暂不可用") from exc
if user is None:
raise ApiError(1005, "用户名或密码错误")
try:
token = await deps._get_session_store().create(user.username, user.role)
except SessionStoreError as exc:
raise ApiError(2001, "认证服务暂不可用") from exc
logger.info("用户登录成功", username=user.username)
return ok(_build_token_response(user, store))
return ok(
{
"token": token,
"username": user.username,
"role": user.role,
"must_change_password": user.must_change_password,
}
)
@router.post("/auth/register")
async def register(req: RegisterRequest) -> dict[str, Any]:
"""注册新用户(role=user),注册后自动签发 token
@router.get("/me")
async def me(user: UserRecord = Depends(deps.get_current_user)) -> dict[str, Any]:
"""返回当前登录用户信息(脱敏),供管理页面校验登录态"""
return ok(_public_user(user))
受 settings.auth_register_enabled 控制,关闭时返回 ERR_REGISTER_DISABLED。
@router.post("/logout")
async def logout(authorization: str | None = Header(None)) -> dict[str, Any]:
"""退出登录:删除当前 session(自解析 Bearer,不拦截 must_change_password"""
_, token = await deps._resolve_user(authorization)
try:
await deps._get_session_store().delete(token)
except SessionStoreError as exc:
raise ApiError(2001, "认证服务暂不可用") from exc
return ok({"logged_out": True})
@router.post("/password")
async def change_password(
req: PasswordChangeRequest, authorization: str | None = Header(None)
) -> dict[str, Any]:
"""修改自己的密码(自解析 Bearer,不拦截 must_change_password——这是被拦截用户唯一可用接口)
旧密码错误 1005;新密码不足 8 位 1001。改密不清除既有 sessionTask1 语义)。
"""
if not settings.auth_register_enabled:
raise ApiError(ERR_REGISTER_DISABLED, "注册已关闭")
store = get_user_store()
user = await store.create(req.username, req.password, role="user")
return ok(_build_token_response(user, store))
user, _ = await deps._resolve_user(authorization)
store = deps._get_user_store()
try:
verified = await store.verify_password(user.username, req.old_password)
if verified is None:
raise ApiError(1005, "旧密码错误")
await store.set_password(user.username, req.new_password)
except (UserStoreError, SessionStoreError) as exc:
raise ApiError(2001, "认证服务暂不可用") from exc
except ValueError as exc:
raise ApiError(1001, str(exc)) from exc
logger.info("用户修改密码", username=user.username)
return ok({"username": user.username})
@router.get("/auth/me")
async def me(user: AuthUser = Depends(get_current_user)) -> dict[str, Any]:
"""返回当前登录用户信息"""
return ok(user.model_dump(mode="json"))
@router.get("/users")
async def list_users(admin: UserRecord = Depends(deps.require_admin)) -> dict[str, Any]:
"""列出全部用户(脱敏:不含 password_hash/salt"""
try:
users = await deps._get_user_store().list()
except UserStoreError as exc:
raise ApiError(2001, "认证服务暂不可用") from exc
return ok([_public_user(user) for user in users])
@router.post("/users")
async def create_user(
req: UserCreateRequest, admin: UserRecord = Depends(deps.require_admin)
) -> dict[str, Any]:
"""创建用户:重名/非法用户名/弱密码 1001;成功返回脱敏后的创建记录"""
try:
user = await deps._get_user_store().create(req.username, req.password, role=req.role)
except UserExistsError as exc:
raise ApiError(1001, str(exc)) from exc
except ValueError as exc:
raise ApiError(1001, str(exc)) from exc
except UserStoreError as exc:
raise ApiError(2001, "认证服务暂不可用") from exc
logger.info("管理员创建用户", username=user.username, role=user.role, operator=admin.username)
return ok(_public_user(user))
@router.post("/users/{username}/password")
async def reset_user_password(
username: str, req: PasswordResetRequest, admin: UserRecord = Depends(deps.require_admin)
) -> dict[str, Any]:
"""重置指定用户密码:用户不存在 1004;成功后清除该用户全部 session"""
store = deps._get_user_store()
try:
if await store.get(username) is None:
raise ApiError(1004, "用户不存在")
await store.set_password(username, req.new_password)
await deps._get_session_store().delete_by_username(username)
except (UserStoreError, SessionStoreError) as exc:
raise ApiError(2001, "认证服务暂不可用") from exc
except ValueError as exc:
raise ApiError(1001, str(exc)) from exc
logger.info("管理员重置用户密码", username=username, operator=admin.username)
return ok({"username": username})
@router.delete("/users/{username}")
async def delete_user(
username: str, admin: UserRecord = Depends(deps.require_admin)
) -> dict[str, Any]:
"""删除用户:不存在 1004;删除自己 1001;最后一个 admin 禁止删除 1001;成功后清除其 session"""
store = deps._get_user_store()
try:
target = await store.get(username)
if target is None:
raise ApiError(1004, "用户不存在")
if target.username == admin.username:
raise ApiError(1001, "不能删除当前登录账号")
if target.role == "admin" and await store.count_admins() <= 1:
raise ApiError(1001, "禁止删除最后一个管理员")
await store.delete(username)
await deps._get_session_store().delete_by_username(username)
except (UserStoreError, SessionStoreError) as exc:
raise ApiError(2001, "认证服务暂不可用") from exc
logger.info("管理员删除用户", username=username, operator=admin.username)
return ok({"username": username})
+7 -11
View File
@@ -10,12 +10,13 @@ import structlog
from fastapi import APIRouter, Depends, File, Form, Query, UploadFile
from fastapi.responses import JSONResponse
from app.api.deps import get_current_user
from app.api.response import ApiError, ok
from app.config import Settings, settings
from app.core.auth import AuthUser, get_current_user, require_admin
from app.core.file_parser import parse_file
from app.core.ingest_tasks import IngestTaskManager
from app.core.ingestion import Ingester
from app.core.users import UserRecord
from app.models.document import DocumentInput
from app.services.qdrant import QdrantService
from app.services.redis import RedisCache, get_cache
@@ -74,7 +75,7 @@ def _allowed_extensions() -> set[str]:
@router.post("/documents")
async def ingest_document(
doc: DocumentInput, user: AuthUser = Depends(get_current_user)
doc: DocumentInput, user: UserRecord = Depends(get_current_user)
) -> JSONResponse:
"""文档入库入口:登记异步任务并返回 202 + task_id,入库结果经任务查询端点获取"""
if not doc.text.strip():
@@ -95,7 +96,7 @@ async def upload_document(
metadata: str = Form(
default="", description='可选元数据 JSON 字符串,如 \'{"author":"x"}\''
),
user: AuthUser = Depends(get_current_user),
user: UserRecord = Depends(get_current_user),
) -> JSONResponse:
"""文件上传入库入口:校验 → 提取文本 → 落盘 → 提交异步入库流水线
@@ -183,9 +184,7 @@ async def upload_document(
@router.get("/documents/tasks/{task_id}")
async def get_ingest_task(
task_id: str, user: AuthUser = Depends(get_current_user)
) -> dict[str, Any]:
async def get_ingest_task(task_id: str) -> dict[str, Any]:
"""查询入库任务状态:含 task_id/status/created_at/updated_atdone 附 resultfailed 附 error"""
task = await _get_task_manager().get(task_id)
if task is None:
@@ -197,7 +196,6 @@ async def get_ingest_task(
async def list_documents(
limit: int = Query(default=20, ge=1, le=100),
offset: str | None = None,
user: AuthUser = Depends(get_current_user),
) -> dict[str, Any]:
"""分页列出文档(L1 摘要),返回 items 与下一页游标 next_offset"""
try:
@@ -209,9 +207,7 @@ async def list_documents(
@router.get("/documents/{doc_id}")
async def get_document(
doc_id: str, user: AuthUser = Depends(get_current_user)
) -> dict[str, Any]:
async def get_document(doc_id: str) -> dict[str, Any]:
"""获取文档详情:L1 记录 + L2/L3 节点 + chunks 数量"""
try:
detail = await _get_qdrant().get_doc_detail(doc_id)
@@ -225,7 +221,7 @@ async def get_document(
@router.delete("/documents/{doc_id}")
async def delete_document(
doc_id: str, user: AuthUser = Depends(require_admin)
doc_id: str, user: UserRecord = Depends(get_current_user)
) -> dict[str, Any]:
"""删除文档:四层集合中该 doc_id 的所有点;幂等,不存在也返回成功(删除数全 0)"""
try:
+100
View File
@@ -0,0 +1,100 @@
"""Session 核心:登录会话的签发与校验
存储后端为 Rediskey: session:{token}JSON 序列化,TTL 12h);
构造时传入 None 降级为进程级内存 dict(带过期时间戳,惰性清理过期项,重启失效)。
所有 Redis 读写异常统一包装为 SessionStoreError 抛出(不静默)。
"""
import json
import secrets
import time
from redis import asyncio as redis_async
# Redis 键前缀
_SESSION_KEY_PREFIX = "session:"
class SessionStoreError(Exception):
"""Session 存储后端读写异常"""
class SessionStore:
"""会话存储:Redis 持久化(带 TTL),构造传 None 降级为内存 dict"""
TTL: int = 43200 # 会话有效期(秒),12h
def __init__(self, redis_client: redis_async.Redis | None) -> None:
# redis.asyncio 客户端(需 decode_responses=True);None → 内存降级
self._redis = redis_client
# 内存降级:token -> (payload, 过期时间戳)
self._memory: dict[str, tuple[dict[str, str], float]] = {}
@staticmethod
def _key(token: str) -> str:
return f"{_SESSION_KEY_PREFIX}{token}"
async def create(self, username: str, role: str) -> str:
"""签发会话:生成 64 位 hex token,存 session:{token} = {username, role}"""
token = secrets.token_hex(32)
payload = {"username": username, "role": role}
if self._redis is None:
self._purge_expired()
self._memory[token] = (payload, time.time() + self.TTL)
return token
try:
await self._redis.setex(self._key(token), self.TTL, json.dumps(payload, ensure_ascii=False))
except Exception as e:
raise SessionStoreError("写入会话失败") from e
return token
async def get(self, token: str) -> dict[str, str] | None:
"""校验会话:无效或已过期返回 None,有效返回 {username, role}"""
if self._redis is None:
item = self._memory.get(token)
if item is None:
return None
payload, expire_at = item
if expire_at <= time.time():
# 惰性清理过期项
del self._memory[token]
return None
return dict(payload)
try:
raw = await self._redis.get(self._key(token))
except Exception as e:
raise SessionStoreError("读取会话失败") from e
if raw is None:
return None
return json.loads(raw)
async def delete(self, token: str) -> None:
"""删除单个会话(幂等)"""
if self._redis is None:
self._memory.pop(token, None)
return
try:
await self._redis.delete(self._key(token))
except Exception as e:
raise SessionStoreError("删除会话失败") from e
async def delete_by_username(self, username: str) -> None:
"""删除指定用户的全部会话(删除用户时清理其登录态)"""
if self._redis is None:
matched = [t for t, (payload, _) in self._memory.items() if payload["username"] == username]
for token in matched:
del self._memory[token]
return
try:
async for key in self._redis.scan_iter(match=f"{_SESSION_KEY_PREFIX}*"):
raw = await self._redis.get(key)
if raw is not None and json.loads(raw).get("username") == username:
await self._redis.delete(key)
except Exception as e:
raise SessionStoreError(f"按用户删除会话失败: {username}") from e
def _purge_expired(self) -> None:
"""惰性清理内存降级模式下的过期会话"""
now = time.time()
for token in [t for t, (_, expire_at) in self._memory.items() if expire_at <= now]:
del self._memory[token]
+191
View File
@@ -0,0 +1,191 @@
"""用户核心:用户记录存储与密码哈希
存储后端为 Rediskey: user:{username},JSON 序列化,无 TTL 持久化);
构造时传入 None 降级为进程级内存 dict(重启丢失,仅用于测试与 Redis 故障兜底)。
所有 Redis 读写异常统一包装为 UserStoreError 抛出(不静默),由上层转 2001。
密码哈希使用标准库 hashlib.pbkdf2_hmacSHA256100_000 迭代 + 16 字节随机 salt)。
"""
import hashlib
import hmac
import json
import re
import secrets
from dataclasses import asdict, dataclass
from datetime import UTC, datetime
from redis import asyncio as redis_async
from structlog.typing import FilteringBoundLogger
# 用户名规则:字母/数字/下划线/连字符,2~32 位
USERNAME_PATTERN = re.compile(r"^[a-zA-Z0-9_-]{2,32}$")
# 密码最小长度
MIN_PASSWORD_LENGTH = 8
# PBKDF2 迭代次数
_HASH_ITERATIONS = 100_000
# salt 字节数(hex 序列化后 32 字符)
_SALT_BYTES = 16
# Redis 键前缀
_USER_KEY_PREFIX = "user:"
class UserExistsError(Exception):
"""用户名已存在"""
class UserStoreError(Exception):
"""用户存储后端读写异常"""
@dataclass
class UserRecord:
"""用户记录(不含明文密码)"""
username: str
role: str # "admin" | "user"
password_hash: str # hex
salt: str # hex, 16 字节
must_change_password: bool
created_at: str # UTC ISO8601
class UserStore:
"""用户存储:Redis 持久化,构造传 None 降级为内存 dict"""
def __init__(self, redis_client: redis_async.Redis | None) -> None:
# redis.asyncio 客户端(需 decode_responses=True);None → 内存降级
self._redis = redis_client
# 内存降级:username -> JSON 字符串(与 Redis 路径同构)
self._memory: dict[str, str] = {}
@staticmethod
def _key(username: str) -> str:
return f"{_USER_KEY_PREFIX}{username}"
def hash_password(self, password: str, salt: bytes) -> str:
"""PBKDF2-HMAC-SHA256100_000 迭代),返回 hex 摘要"""
return hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt, _HASH_ITERATIONS).hex()
async def create(
self,
username: str,
password: str,
role: str = "user",
must_change_password: bool = False,
) -> UserRecord:
"""创建用户
重名抛 UserExistsError;用户名不符合 USERNAME_PATTERN 或密码过短抛 ValueError。
"""
if not USERNAME_PATTERN.match(username):
raise ValueError(f"用户名非法: {username!r}(须为 2~32 位字母/数字/_/-)")
if len(password) < MIN_PASSWORD_LENGTH:
raise ValueError(f"密码长度不得少于 {MIN_PASSWORD_LENGTH}")
if await self.get(username) is not None:
raise UserExistsError(f"用户已存在: {username}")
salt = secrets.token_bytes(_SALT_BYTES)
record = UserRecord(
username=username,
role=role,
password_hash=self.hash_password(password, salt),
salt=salt.hex(),
must_change_password=must_change_password,
created_at=datetime.now(UTC).isoformat(),
)
await self._write(record)
return record
async def get(self, username: str) -> UserRecord | None:
"""按用户名查询,不存在返回 None"""
raw = await self._read_raw(username)
if raw is None:
return None
return UserRecord(**json.loads(raw))
async def list(self) -> list[UserRecord]:
"""列出全部用户(扫描 user:* 键)"""
if self._redis is None:
raws: list[str | None] = list(self._memory.values())
else:
try:
raws = [await self._redis.get(key) async for key in self._redis.scan_iter(match=f"{_USER_KEY_PREFIX}*")]
except Exception as e:
raise UserStoreError("列出用户失败") from e
return [UserRecord(**json.loads(raw)) for raw in raws if raw is not None]
async def delete(self, username: str) -> bool:
"""删除用户,返回是否删除成功(幂等:不存在返回 False)"""
if self._redis is None:
return self._memory.pop(username, None) is not None
try:
return bool(await self._redis.delete(self._key(username)))
except Exception as e:
raise UserStoreError(f"删除用户失败: {username}") from e
async def verify_password(self, username: str, password: str) -> UserRecord | None:
"""校验密码:成功返回用户记录,用户不存在或密码错误返回 None"""
record = await self.get(username)
if record is None:
return None
candidate = self.hash_password(password, bytes.fromhex(record.salt))
# 常量时间比较,防时序侧信道
if not hmac.compare_digest(candidate, record.password_hash):
return None
return record
async def set_password(self, username: str, new_password: str) -> bool:
"""重置密码:重新生成 salt 并哈希,同时清除 must_change_password 标记
用户不存在返回 False;新密码过短抛 ValueError。
"""
if len(new_password) < MIN_PASSWORD_LENGTH:
raise ValueError(f"密码长度不得少于 {MIN_PASSWORD_LENGTH}")
record = await self.get(username)
if record is None:
return False
salt = secrets.token_bytes(_SALT_BYTES)
record.salt = salt.hex()
record.password_hash = self.hash_password(new_password, salt)
record.must_change_password = False
await self._write(record)
return True
async def count_admins(self) -> int:
"""统计 admin 角色用户数"""
return sum(1 for record in await self.list() if record.role == "admin")
async def _read_raw(self, username: str) -> str | None:
if self._redis is None:
return self._memory.get(username)
try:
return await self._redis.get(self._key(username))
except Exception as e:
raise UserStoreError(f"读取用户失败: {username}") from e
async def _write(self, record: UserRecord) -> None:
raw = json.dumps(asdict(record), ensure_ascii=False)
if self._redis is None:
self._memory[record.username] = raw
return
try:
await self._redis.set(self._key(record.username), raw)
except Exception as e:
raise UserStoreError(f"写入用户失败: {record.username}") from e
async def bootstrap_admin(store: UserStore, logger: FilteringBoundLogger) -> str | None:
"""空库引导:无任何用户时创建随机密码的 admin 账号
明文密码通过 logger.warning 打印一次并返回(仅此一次机会);
已有用户时不做任何事,返回 None。
"""
if await store.list():
return None
password = secrets.token_urlsafe(12)
await store.create("admin", password, role="admin", must_change_password=True)
logger.warning("已创建默认管理员,请立即登录并修改密码", username="admin", password=password)
return password
+9 -2
View File
@@ -6,6 +6,7 @@ import structlog
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import FileResponse, JSONResponse, RedirectResponse
from redis import asyncio as redis_async
from app.api.response import ApiError, error
from app.api.v1.auth import router as auth_router
@@ -14,7 +15,7 @@ from app.api.v1.knowledge import router as knowledge_router
from app.api.v1.search import router as search_router
from app.api.v1.settings import router as settings_router
from app.config import settings
from app.core.auth import ensure_default_admin
from app.core.users import UserStore, bootstrap_admin
from app.services.qdrant import QdrantService
logger = structlog.get_logger()
@@ -32,7 +33,13 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
except Exception:
logger.error("Qdrant 集合初始化失败,跳过初始化继续启动")
try:
await ensure_default_admin()
# Redis 客户端创建失败(如 URL 非法)时传 None,UserStore 降级为内存模式
redis_client = redis_async.from_url(settings.redis_url, decode_responses=True)
except Exception:
redis_client = None
try:
# 空库引导默认管理员;明文密码由 bootstrap_admin 内部 warning 打印一次
await bootstrap_admin(UserStore(redis_client), logger)
except Exception:
logger.warning("默认管理员初始化失败,跳过", exc_info=True)
try:
+691 -23
View File
@@ -79,7 +79,7 @@
button.primary:disabled { background: #9ca3af; cursor: not-allowed; }
form .field { margin-bottom: 12px; }
form label { display: block; font-size: 13px; margin-bottom: 4px; color: #374151; }
input[type="text"], input[type="number"], textarea {
input[type="text"], input[type="number"], input[type="password"], textarea, select {
width: 100%; border: 1px solid #d1d5db; border-radius: 4px;
padding: 8px; font-size: 13px; font-family: inherit;
}
@@ -118,6 +118,12 @@
header { position: relative; }
.user-area { position: absolute; top: 14px; right: 24px; display: flex; align-items: center; gap: 8px; }
.user-area .action { padding: 4px 10px; }
.user-area #user-name { color: #f9fafb; font-size: 13px; }
.role-badge {
display: inline-block; border-radius: 3px; padding: 1px 8px; font-size: 12px;
background: #eff6ff; color: #1d4ed8; border: 1px solid #93c5fd;
}
.role-badge.role-admin { background: #fef3c7; color: #b45309; border-color: #fcd34d; }
.login-overlay {
position: fixed; inset: 0; background: rgba(0,0,0,0.45);
display: flex; align-items: center; justify-content: center; z-index: 100;
@@ -139,6 +145,45 @@
padding: 12px 14px; margin-bottom: 12px; font-size: 13px; white-space: pre-wrap;
}
.summary-box .sum-title { font-weight: 600; margin-bottom: 6px; color: #15803d; }
.method-badge {
display: inline-block; border-radius: 3px; padding: 1px 8px;
font-size: 12px; font-weight: 600; color: #fff; margin-right: 8px;
}
.method-get { background: #16a34a; }
.method-post { background: #2563eb; }
.method-delete { background: #dc2626; }
.auth-badge {
display: inline-block; border-radius: 3px; padding: 1px 8px;
font-size: 12px; margin-left: 8px; border: 1px solid transparent;
}
.auth-badge.auth-none { background: #f0fdf4; color: #15803d; border-color: #86efac; }
.auth-badge.auth-bearer { background: #eff6ff; color: #1d4ed8; border-color: #93c5fd; }
.auth-badge.auth-admin { background: #fef3c7; color: #b45309; border-color: #fcd34d; }
.api-item { border: 1px solid #e5e7eb; border-radius: 6px; padding: 12px 14px; margin-bottom: 12px; }
.api-item .api-head { display: flex; align-items: center; flex-wrap: wrap; gap: 4px; margin-bottom: 6px; }
.api-item code { background: #f3f4f6; border-radius: 3px; padding: 1px 6px; font-size: 13px; }
.api-item .api-desc { font-size: 13px; color: #374151; margin-bottom: 8px; }
.api-item h4 { font-size: 13px; margin: 10px 0 6px; color: #374151; }
.json-pre {
background: #f9fafb; border: 1px solid #e5e7eb; border-radius: 4px;
padding: 8px; white-space: pre-wrap; word-break: break-word;
font-size: 12px; max-height: 260px; overflow: auto;
}
.curl-box { position: relative; }
.curl-box .curl-copy { position: absolute; top: 6px; right: 6px; }
.curl-pre {
background: #1f2937; color: #e5e7eb; border-radius: 4px;
padding: 8px; white-space: pre-wrap; word-break: break-all; font-size: 12px;
}
.try-panel { margin-top: 10px; border-top: 1px dashed #e5e7eb; padding-top: 10px; }
.try-panel .field { margin-bottom: 10px; }
.try-panel label { display: block; font-size: 13px; margin-bottom: 4px; color: #374151; }
.try-panel textarea { min-height: 100px; font-family: ui-monospace, monospace; }
.try-result pre {
background: #f9fafb; border: 1px solid #e5e7eb; border-radius: 4px;
padding: 8px; white-space: pre-wrap; word-break: break-word;
font-size: 12px; max-height: 320px; overflow: auto;
}
</style>
</head>
<body>
@@ -159,11 +204,36 @@
</form>
</div>
</div>
<div id="password-overlay" class="login-overlay hidden">
<div class="login-box">
<h2 id="password-title">修改密码</h2>
<div class="error-bar hidden" id="error-password"></div>
<div class="fallback-flag hidden" id="password-forced-tip">首次登录须先修改密码,完成后方可进入管理后台</div>
<form id="password-form">
<div class="field">
<label for="password-old">旧密码</label>
<input type="password" id="password-old" name="old_password" required autocomplete="current-password">
</div>
<div class="field">
<label for="password-new">新密码(至少 8 位)</label>
<input type="password" id="password-new" name="new_password" required autocomplete="new-password">
</div>
<div class="field">
<label for="password-confirm">确认新密码</label>
<input type="password" id="password-confirm" name="confirm_password" required autocomplete="new-password">
</div>
<button type="submit" class="primary">确认修改</button>
<button type="button" class="action" id="btn-password-cancel">取消</button>
</form>
</div>
</div>
<header>
<h1>知识库管理后台</h1>
<div id="user-area" class="user-area hidden">
<span class="muted" id="user-name"></span>
<button type="button" class="action" id="btn-logout">登出</button>
<span id="user-name"></span>
<span id="user-role" class="role-badge"></span>
<button type="button" class="action" id="btn-change-password">修改密码</button>
<button type="button" class="action" id="btn-logout">退出登录</button>
</div>
<nav id="nav">
<button type="button" data-target="section-overview" class="active">概览</button>
@@ -171,6 +241,7 @@
<button type="button" data-target="section-ingest">文档入库</button>
<button type="button" data-target="section-search">检索测试台</button>
<button type="button" data-target="section-categories">类目列表</button>
<button type="button" data-target="section-api-guide">API 指南</button>
</nav>
</header>
<main>
@@ -268,8 +339,49 @@
<tbody id="categories-tbody"></tbody>
</table>
</section>
<section id="section-api-guide" class="hidden">
<h2>API 指南</h2>
<div class="muted" style="margin-bottom:12px;">全部端点清单与在线测试台;base URL 取当前站点(location.origin),需鉴权端点发送时自动附带当前登录 token。仅可从清单选择端点,不支持自定义 URL。</div>
<div id="api-guide-list"></div>
</section>
</main>
<template id="tpl-section-users">
<section id="section-users" class="hidden">
<h2>用户管理</h2>
<div class="error-bar hidden" id="error-users"></div>
<div class="toolbar">
<button type="button" class="action" id="btn-refresh-users">刷新</button>
</div>
<table>
<thead>
<tr><th>用户名</th><th>角色</th><th>须改密</th><th>创建时间</th><th>操作</th></tr>
</thead>
<tbody id="users-tbody"></tbody>
</table>
<h3 style="font-size:14px; margin-top:24px;">创建用户</h3>
<form id="user-create-form">
<div class="field">
<label for="user-new-name">用户名</label>
<input type="text" id="user-new-name" name="username" required autocomplete="off">
</div>
<div class="field">
<label for="user-new-password">初始密码(至少 8 位)</label>
<input type="password" id="user-new-password" name="password" required autocomplete="new-password">
</div>
<div class="field">
<label for="user-new-role">角色</label>
<select id="user-new-role" name="role">
<option value="user" selected>user</option>
<option value="admin">admin</option>
</select>
</div>
<button type="submit" class="primary">创建用户</button>
</form>
</section>
</template>
<script>
"use strict";
@@ -299,25 +411,68 @@ function hideError(boxId) {
document.getElementById(boxId).classList.add("hidden");
}
/* ---------- 认证 token 管理 ---------- */
function getToken() { return localStorage.getItem("qmd_token") || ""; }
function setToken(t) { localStorage.setItem("qmd_token", t); }
function clearToken() { localStorage.removeItem("qmd_token"); }
/* ---------- 认证状态管理 ---------- */
var TOKEN_KEY = "qmd_token";
var USER_KEY = "qmd_user";
var passwordForced = false;
function getToken() { return localStorage.getItem(TOKEN_KEY) || ""; }
function getStoredUser() {
try {
return JSON.parse(localStorage.getItem(USER_KEY) || "null");
} catch (e) {
return null;
}
}
function saveAuth(token, user) {
localStorage.setItem(TOKEN_KEY, token);
localStorage.setItem(USER_KEY, JSON.stringify({
username: user.username,
role: user.role,
must_change_password: !!user.must_change_password
}));
}
function clearAuth() {
localStorage.removeItem(TOKEN_KEY);
localStorage.removeItem(USER_KEY);
}
function showLogin() {
document.getElementById("login-overlay").classList.remove("hidden");
document.getElementById("password-overlay").classList.add("hidden");
document.getElementById("user-area").classList.add("hidden");
loadedOnce = {};
unmountUsersSection();
}
function afterLogin(user) {
/* 渲染主界面:顶栏用户区 + 角色门禁(仅 admin 挂载用户管理区块) */
function enterApp(user) {
document.getElementById("login-overlay").classList.add("hidden");
document.getElementById("password-overlay").classList.add("hidden");
document.getElementById("user-area").classList.remove("hidden");
document.getElementById("user-name").textContent = user.username + " (" + user.role + ")";
document.getElementById("user-name").textContent = user.username;
var badge = document.getElementById("user-role");
badge.textContent = user.role;
badge.className = "role-badge" + (user.role === "admin" ? " role-admin" : "");
loadedOnce = {};
if (user.role === "admin") {
mountUsersSection();
} else {
unmountUsersSection();
}
activateSection("section-overview");
}
/* 登录/会话校验成功后的统一入口:must_change_password 用户先强制改密 */
function afterAuth(user) {
if (user.must_change_password) {
document.getElementById("login-overlay").classList.add("hidden");
openPasswordModal(true);
return;
}
enterApp(user);
}
document.getElementById("login-form").addEventListener("submit", function (event) {
event.preventDefault();
hideError("error-login");
@@ -325,6 +480,7 @@ document.getElementById("login-form").addEventListener("submit", function (event
username: document.getElementById("login-username").value,
password: document.getElementById("login-password").value
};
/* 登录接口不携带 Authorization 头 */
fetch("/api/v1/auth/login", {
method: "POST",
headers: { "Content-Type": "application/json" },
@@ -334,19 +490,84 @@ document.getElementById("login-form").addEventListener("submit", function (event
showError("error-login", { code: body.code, message: body.message });
return;
}
setToken(body.data.access_token);
afterLogin(body.data.user);
var data = body.data;
saveAuth(data.token, data);
afterAuth(data);
}).catch(function (err) {
showError("error-login", { code: "NETWORK", message: "登录请求失败: " + (err && err.message ? err.message : err) });
});
});
document.getElementById("btn-logout").addEventListener("click", function () {
clearToken();
showLogin();
api("/api/v1/auth/logout", { method: "POST" }).catch(function () {
/* 服务端退出失败不阻塞本地登出 */
}).finally(function () {
clearAuth();
showLogin();
});
});
/* 统一 API 封装:自动注入 token,code !== 0 抛错,未认证跳登录 */
/* ---------- 修改密码 ---------- */
function openPasswordModal(forced) {
passwordForced = !!forced;
document.getElementById("password-title").textContent = forced ? "首次登录须修改密码" : "修改密码";
document.getElementById("password-forced-tip").classList.toggle("hidden", !forced);
/* 强制改密层无取消按钮、不可跳过 */
document.getElementById("btn-password-cancel").classList.toggle("hidden", forced);
if (forced) {
document.getElementById("login-overlay").classList.add("hidden");
}
hideError("error-password");
document.getElementById("password-form").reset();
document.getElementById("password-overlay").classList.remove("hidden");
}
document.getElementById("btn-change-password").addEventListener("click", function () {
openPasswordModal(false);
});
document.getElementById("btn-password-cancel").addEventListener("click", function () {
if (passwordForced) { return; }
document.getElementById("password-overlay").classList.add("hidden");
});
document.getElementById("password-form").addEventListener("submit", function (event) {
event.preventDefault();
hideError("error-password");
var newPwd = document.getElementById("password-new").value;
if (newPwd !== document.getElementById("password-confirm").value) {
showError("error-password", { code: "VALIDATION", message: "两次输入的新密码不一致" });
return;
}
api("/api/v1/auth/password", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
old_password: document.getElementById("password-old").value,
new_password: newPwd
})
}).then(function () {
document.getElementById("password-overlay").classList.add("hidden");
if (passwordForced) {
/* 强制改密成功后才进入主界面 */
passwordForced = false;
var user = getStoredUser();
if (user) {
user.must_change_password = false;
localStorage.setItem(USER_KEY, JSON.stringify(user));
enterApp(user);
}
} else {
alert("密码修改成功");
}
}).catch(function (err) {
showError("error-password", err);
});
});
/* 统一 API 封装:自动注入 Authorization: Bearer <token>(登录接口除外),code !== 0 抛错;
1005 未认证/凭证无效 → 清除本地凭证并回登录卡片;1006 权限不足 → 抛给调用方在错误条展示 */
function api(path, options) {
options = options || {};
options.headers = Object.assign({}, options.headers || {});
@@ -355,18 +576,13 @@ function api(path, options) {
options.headers["Authorization"] = "Bearer " + token;
}
return fetch(path, options).then(function (resp) {
if (resp.status === 401) {
clearToken();
showLogin();
throw { code: 1003, message: "未认证或登录已过期,请重新登录" };
}
return resp.json().catch(function () {
throw { code: "HTTP " + resp.status, message: "响应解析失败" };
});
}).then(function (body) {
if (body.code !== 0) {
if (body.code === 1003 || body.code === 1005) {
clearToken();
if (body.code === 1005) {
clearAuth();
showLogin();
}
throw { code: body.code, message: body.message };
@@ -402,6 +618,8 @@ function activateSection(targetId) {
if (targetId === "section-overview") { loadOverview(); }
if (targetId === "section-docs") { loadDocuments(true); }
if (targetId === "section-categories") { loadCategories(); }
if (targetId === "section-users") { loadUsers(); }
if (targetId === "section-api-guide") { renderApiGuide(); }
}
}
@@ -870,6 +1088,449 @@ function loadCategories() {
}).catch(function (err) { showError("error-categories", err); });
}
/* ---------- 6. 用户管理(仅 admin 渲染) ---------- */
/* admin 登录时将用户管理区块从 template 挂载进 DOM;非 admin 完全不渲染 */
function mountUsersSection() {
if (document.getElementById("section-users")) { return; }
var tpl = document.getElementById("tpl-section-users");
document.querySelector("main").appendChild(tpl.content.cloneNode(true));
var navBtn = el("button", "用户管理");
navBtn.type = "button";
navBtn.id = "nav-users";
navBtn.setAttribute("data-target", "section-users");
document.getElementById("nav").appendChild(navBtn);
document.getElementById("btn-refresh-users").addEventListener("click", loadUsers);
document.getElementById("user-create-form").addEventListener("submit", createUser);
}
function unmountUsersSection() {
var sec = document.getElementById("section-users");
if (sec) { sec.parentNode.removeChild(sec); }
var navBtn = document.getElementById("nav-users");
if (navBtn) { navBtn.parentNode.removeChild(navBtn); }
}
function loadUsers() {
hideError("error-users");
api("/api/v1/auth/users").then(function (users) {
renderUsers(users || []);
}).catch(function (err) { showError("error-users", err); });
}
function renderUsers(users) {
var tbody = document.getElementById("users-tbody");
clearChildren(tbody);
users.forEach(function (user) {
var tr = el("tr");
tr.appendChild(el("td", user.username));
var roleTd = el("td");
roleTd.appendChild(el("span", user.role, "role-badge" + (user.role === "admin" ? " role-admin" : "")));
tr.appendChild(roleTd);
tr.appendChild(el("td", user.must_change_password ? "是" : "否"));
tr.appendChild(el("td", user.created_at || ""));
var opsTd = el("td");
var resetBtn = el("button", "重置密码", "action");
resetBtn.type = "button";
resetBtn.addEventListener("click", function () { resetUserPassword(user.username); });
var deleteBtn = el("button", "删除", "action danger");
deleteBtn.type = "button";
deleteBtn.addEventListener("click", function () { deleteUser(user.username, user.role); });
opsTd.appendChild(resetBtn);
opsTd.appendChild(deleteBtn);
tr.appendChild(opsTd);
tbody.appendChild(tr);
});
}
function createUser(event) {
event.preventDefault();
hideError("error-users");
var payload = {
username: document.getElementById("user-new-name").value,
password: document.getElementById("user-new-password").value,
role: document.getElementById("user-new-role").value
};
api("/api/v1/auth/users", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(payload)
}).then(function () {
document.getElementById("user-create-form").reset();
loadUsers();
}).catch(function (err) { showError("error-users", err); });
}
function resetUserPassword(username) {
var newPwd = prompt("为用户「" + username + "」设置新密码(至少 8 位):");
if (newPwd === null) { return; }
if (!newPwd) {
showError("error-users", { code: "VALIDATION", message: "新密码不能为空" });
return;
}
hideError("error-users");
api("/api/v1/auth/users/" + encodeURIComponent(username) + "/password", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ new_password: newPwd })
}).then(function () {
loadUsers();
}).catch(function (err) { showError("error-users", err); });
}
function deleteUser(username, role) {
if (!confirm("确定删除用户「" + username + "」(角色 " + role + ")吗?该操作不可恢复。")) {
return;
}
hideError("error-users");
api("/api/v1/auth/users/" + encodeURIComponent(username), { method: "DELETE" }).then(function () {
loadUsers();
}).catch(function (err) { showError("error-users", err); });
}
/* ---------- 7. API 指南 ---------- */
/* 展示用静态维护清单:与后端路由表无自动同步,
测试 tests/test_admin_page.py 校验每个 method+path 均在真实路由集合内防止漂移。
auth 取值:none=免登录 / bearer=需登录 / admin=仅 admin */
var API_GUIDE = [
{ method: "GET", path: "/api/v1/health", desc: "健康检查", auth: "none",
query: [], pathParams: [], form: [], body: null, upload: false },
{ method: "POST", path: "/api/v1/search", desc: "分层检索(query 路由 → L1→L2→L3 剪枝 → chunk hybrid 检索)", auth: "none",
query: [], pathParams: [], form: [],
body: { query: "NAS 上如何部署服务?", top_k: 5, summarize: false }, upload: false },
{ method: "POST", path: "/api/v1/documents", desc: "文档入库(202 异步入库,返回 task_id", auth: "bearer",
query: [], pathParams: [], form: [],
body: { title: "示例文档", source: "manual", text: "文档正文内容……" }, upload: false },
{ method: "POST", path: "/api/v1/documents/upload", desc: "multipart 文件上传入库(.txt/.md/.html/.htm/.pdf/.docx202 异步)", auth: "bearer",
query: [], pathParams: [],
form: [
{ name: "title", desc: "可选标题,默认取文件名去扩展", def: "" },
{ name: "source", desc: "可选来源标识,默认 file:原文件名", def: "" }
],
body: null, upload: true },
{ method: "GET", path: "/api/v1/documents/tasks/{task_id}", desc: "入库任务状态查询(done 附 resultfailed 附 error", auth: "none",
query: [],
pathParams: [{ name: "task_id", desc: "入库任务 ID", def: "" }],
form: [], body: null, upload: false },
{ method: "GET", path: "/api/v1/documents", desc: "文档列表(limit/offset 分页)", auth: "none",
query: [
{ name: "limit", desc: "每页条数(1-100", def: "20" },
{ name: "offset", desc: "分页游标(上一页返回的 next_offset", def: "" }
],
pathParams: [], form: [], body: null, upload: false },
{ method: "GET", path: "/api/v1/documents/{doc_id}", desc: "文档详情(L1 记录 + L2/L3 节点 + chunks 数量)", auth: "none",
query: [],
pathParams: [{ name: "doc_id", desc: "文档 ID", def: "" }],
form: [], body: null, upload: false },
{ method: "DELETE", path: "/api/v1/documents/{doc_id}", desc: "删除文档(四层集合全量删除,幂等)", auth: "bearer",
query: [],
pathParams: [{ name: "doc_id", desc: "文档 ID", def: "" }],
form: [], body: null, upload: false },
{ method: "GET", path: "/api/v1/knowledge/categories", desc: "知识分类类目集", auth: "none",
query: [], pathParams: [], form: [], body: null, upload: false },
{ method: "GET", path: "/api/v1/knowledge/stats", desc: "统计(四层点数 + 类目分布 + uncategorized 数)", auth: "none",
query: [], pathParams: [], form: [], body: null, upload: false },
{ method: "POST", path: "/api/v1/auth/login", desc: "用户名密码登录,成功签发 Bearer tokensession TTL 12h", auth: "none",
query: [], pathParams: [], form: [],
body: { username: "admin", password: "你的密码" }, upload: false },
{ method: "POST", path: "/api/v1/auth/logout", desc: "退出登录(删除当前 session", auth: "bearer",
query: [], pathParams: [], form: [], body: null, upload: false },
{ method: "POST", path: "/api/v1/auth/password", desc: "修改自己的密码(must_change_password 用户唯一可用接口)", auth: "bearer",
query: [], pathParams: [], form: [],
body: { old_password: "旧密码", new_password: "新密码至少8位" }, upload: false },
{ method: "GET", path: "/api/v1/auth/me", desc: "当前登录用户信息(脱敏)", auth: "bearer",
query: [], pathParams: [], form: [], body: null, upload: false },
{ method: "GET", path: "/api/v1/auth/users", desc: "用户列表(脱敏,不含 password_hash/salt", auth: "admin",
query: [], pathParams: [], form: [], body: null, upload: false },
{ method: "POST", path: "/api/v1/auth/users", desc: "创建用户(重名/弱密码 1001", auth: "admin",
query: [], pathParams: [], form: [],
body: { username: "newuser", password: "初始密码至少8位", role: "user" }, upload: false },
{ method: "POST", path: "/api/v1/auth/users/{username}/password", desc: "重置指定用户密码(成功后清除其全部 session)", auth: "admin",
query: [],
pathParams: [{ name: "username", desc: "目标用户名", def: "" }],
form: [],
body: { new_password: "新密码至少8位" }, upload: false },
{ method: "DELETE", path: "/api/v1/auth/users/{username}", desc: "删除用户并清其 session(不能删自己/最后一个 admin", auth: "admin",
query: [],
pathParams: [{ name: "username", desc: "目标用户名", def: "" }],
form: [], body: null, upload: false }
];
var API_AUTH_LABELS = { none: "免登录", bearer: "需登录", admin: "仅 admin" };
/* path 模板占位替换:values 为空时回退默认值,再回退 {name} 占位(供 curl 展示) */
function substitutePath(ep, values) {
var p = ep.path;
ep.pathParams.forEach(function (pp) {
var v = values[pp.name] !== undefined && values[pp.name] !== "" ? values[pp.name] : (pp.def || "{" + pp.name + "}");
p = p.split("{" + pp.name + "}").join(v);
});
return p;
}
function buildCurl(ep) {
var url = location.origin + substitutePath(ep, {});
var qs = [];
ep.query.forEach(function (q) {
if (q.def) { qs.push(q.name + "=" + encodeURIComponent(q.def)); }
});
if (qs.length) { url += "?" + qs.join("&"); }
var parts = ['curl -X ' + ep.method + ' "' + url + '"'];
if (ep.auth !== "none") {
parts.push('-H "Authorization: Bearer <TOKEN>"');
}
if (ep.upload) {
parts.push('-F "file=@/path/to/file.md"');
} else if (ep.body) {
parts.push('-H "Content-Type: application/json"');
parts.push("-d '" + JSON.stringify(ep.body) + "'");
}
return parts.join(" \\\n ");
}
/* 复制到剪贴板:navigator.clipboard 不可用时降级为 textarea 选中复制 */
function copyText(text, btn) {
function done() {
btn.textContent = "已复制";
setTimeout(function () { btn.textContent = "复制"; }, 1500);
}
function fallback() {
var ta = document.createElement("textarea");
ta.value = text;
document.body.appendChild(ta);
ta.select();
try { document.execCommand("copy"); } catch (e) { /* 复制失败不阻塞 */ }
document.body.removeChild(ta);
done();
}
if (navigator.clipboard && navigator.clipboard.writeText) {
navigator.clipboard.writeText(text).then(done, fallback);
} else {
fallback();
}
}
function renderParamTable(item, title, params) {
if (!params.length) { return; }
item.appendChild(el("h4", title));
var table = el("table");
var headTr = el("tr");
headTr.appendChild(el("th", "名称"));
headTr.appendChild(el("th", "说明"));
headTr.appendChild(el("th", "默认"));
var thead = el("thead");
thead.appendChild(headTr);
table.appendChild(thead);
var tbody = el("tbody");
params.forEach(function (p) {
var tr = el("tr");
tr.appendChild(el("td", p.name));
tr.appendChild(el("td", p.desc || ""));
tr.appendChild(el("td", p.def || "(空)"));
tbody.appendChild(tr);
});
table.appendChild(tbody);
item.appendChild(table);
}
/* 试一下面板内按 kindpath/query/form)渲染参数输入框 */
function renderTryInputs(panel, kind, params) {
params.forEach(function (p) {
var field = el("div", null, "field");
field.appendChild(el("label", p.name + (p.desc ? "" + p.desc + "" : "")));
var input = el("input");
input.type = "text";
input.className = "try-" + kind + "-input";
input.setAttribute("data-param", p.name);
input.value = p.def || "";
input.placeholder = p.desc || p.name;
field.appendChild(input);
panel.appendChild(field);
});
}
function renderTryResult(panel, status, elapsed, text) {
var result = panel.querySelector(".try-result");
result.querySelector(".try-status").textContent = String(status);
result.querySelector(".try-elapsed").textContent = elapsed + " ms";
var pretty = text;
try { pretty = JSON.stringify(JSON.parse(text), null, 2); } catch (e) { /* 非 JSON 响应原样展示 */ }
result.querySelector(".try-result-pre").textContent = pretty;
result.classList.remove("hidden");
}
function sendApiTry(idx, panel) {
var ep = API_GUIDE[idx];
var errBox = panel.querySelector(".try-error");
var sendBtn = panel.querySelector(".try-send");
function fail(message) {
errBox.textContent = message;
errBox.classList.remove("hidden");
}
errBox.classList.add("hidden");
/* path 参数:必填,替换 path 模板占位 */
var pathValues = {};
var pathInputs = panel.querySelectorAll(".try-path-input");
for (var i = 0; i < pathInputs.length; i++) {
var pName = pathInputs[i].getAttribute("data-param");
var pVal = pathInputs[i].value.trim();
if (!pVal) {
fail("请填写 path 参数:" + pName);
return;
}
pathValues[pName] = encodeURIComponent(pVal);
}
var url = substitutePath(ep, pathValues);
/* query 参数:仅拼接非空值 */
var pairs = [];
var queryInputs = panel.querySelectorAll(".try-query-input");
for (var j = 0; j < queryInputs.length; j++) {
var qVal = queryInputs[j].value.trim();
if (qVal) {
pairs.push(queryInputs[j].getAttribute("data-param") + "=" + encodeURIComponent(qVal));
}
}
if (pairs.length) { url += "?" + pairs.join("&"); }
var options = { method: ep.method, headers: {} };
if (ep.upload) {
var fileInput = panel.querySelector(".try-file");
if (!fileInput.files || fileInput.files.length === 0) {
fail("请选择要上传的文件");
return;
}
/* FormData 由浏览器自动生成 multipart 边界,不设置 Content-Type 头 */
var formData = new FormData();
formData.append("file", fileInput.files[0]);
var formInputs = panel.querySelectorAll(".try-form-input");
for (var k = 0; k < formInputs.length; k++) {
var fVal = formInputs[k].value.trim();
if (fVal) { formData.append(formInputs[k].getAttribute("data-param"), fVal); }
}
options.body = formData;
} else if (ep.body) {
var raw = panel.querySelector(".try-body").value;
try {
JSON.parse(raw);
} catch (e) {
fail("Body 不是合法 JSON,未发送请求:" + (e && e.message ? e.message : e));
return;
}
options.headers["Content-Type"] = "application/json";
options.body = raw;
}
if (ep.auth !== "none") {
var token = getToken();
if (token) { options.headers["Authorization"] = "Bearer " + token; }
}
sendBtn.disabled = true;
var started = performance.now();
fetch(url, options).then(function (resp) {
var elapsed = Math.round(performance.now() - started);
return resp.text().then(function (text) {
renderTryResult(panel, resp.status, elapsed, text);
});
}).catch(function (err) {
renderTryResult(panel, "请求失败", Math.round(performance.now() - started),
String(err && err.message ? err.message : err));
}).finally(function () {
sendBtn.disabled = false;
});
}
function renderApiItem(ep, idx) {
var item = el("div", null, "api-item");
var head = el("div", null, "api-head");
head.appendChild(el("span", ep.method, "method-badge method-" + ep.method.toLowerCase()));
var code = el("code", ep.path);
head.appendChild(code);
head.appendChild(el("span", API_AUTH_LABELS[ep.auth] || ep.auth, "auth-badge auth-" + ep.auth));
item.appendChild(head);
item.appendChild(el("div", ep.desc, "api-desc"));
renderParamTable(item, "Path 参数", ep.pathParams);
renderParamTable(item, "Query 参数", ep.query);
renderParamTable(item, "表单字段", ep.form);
if (ep.body) {
item.appendChild(el("h4", "Body 示例(JSON"));
item.appendChild(el("pre", JSON.stringify(ep.body, null, 2), "json-pre"));
}
item.appendChild(el("h4", "curl 示例"));
var curlBox = el("div", null, "curl-box");
var curlPre = el("pre", buildCurl(ep), "curl-pre");
var copyBtn = el("button", "复制", "action curl-copy");
copyBtn.type = "button";
copyBtn.addEventListener("click", function () { copyText(curlPre.textContent, copyBtn); });
curlBox.appendChild(copyBtn);
curlBox.appendChild(curlPre);
item.appendChild(curlBox);
/* 「试一下」面板:只能从清单展开,不提供任意 URL 输入框 */
var toggleBtn = el("button", "试一下", "action try-toggle");
toggleBtn.type = "button";
var panel = el("div", null, "try-panel hidden");
toggleBtn.addEventListener("click", function () { panel.classList.toggle("hidden"); });
item.appendChild(toggleBtn);
renderTryInputs(panel, "path", ep.pathParams);
renderTryInputs(panel, "query", ep.query);
if (ep.upload) {
renderTryInputs(panel, "form", ep.form);
var fileField = el("div", null, "field");
fileField.appendChild(el("label", "文件(.txt/.md/.html/.htm/.pdf/.docx"));
var fileInput = el("input");
fileInput.type = "file";
fileInput.className = "try-file";
fileInput.accept = ".txt,.md,.html,.htm,.pdf,.docx";
fileField.appendChild(fileInput);
panel.appendChild(fileField);
}
if (ep.body) {
var bodyField = el("div", null, "field");
bodyField.appendChild(el("label", "BodyJSON,可编辑)"));
bodyField.appendChild(el("textarea", JSON.stringify(ep.body, null, 2), "try-body"));
panel.appendChild(bodyField);
}
var sendBtn = el("button", "发送", "primary try-send");
sendBtn.type = "button";
sendBtn.addEventListener("click", function () { sendApiTry(idx, panel); });
panel.appendChild(sendBtn);
panel.appendChild(el("div", null, "error-bar hidden try-error"));
var result = el("div", null, "result-box hidden try-result");
var statusKv = el("div", null, "kv");
statusKv.appendChild(el("span", "状态码", "k"));
statusKv.appendChild(el("span", null, "try-status"));
result.appendChild(statusKv);
var elapsedKv = el("div", null, "kv");
elapsedKv.appendChild(el("span", "耗时", "k"));
elapsedKv.appendChild(el("span", null, "try-elapsed"));
result.appendChild(elapsedKv);
result.appendChild(el("pre", null, "try-result-pre"));
panel.appendChild(result);
item.appendChild(panel);
return item;
}
function renderApiGuide() {
var container = document.getElementById("api-guide-list");
clearChildren(container);
API_GUIDE.forEach(function (ep, idx) {
container.appendChild(renderApiItem(ep, idx));
});
}
/* ---------- 初始化 ---------- */
(function init() {
@@ -877,9 +1538,16 @@ function loadCategories() {
showLogin();
return;
}
/* 本地有 token 时先调 /auth/me 验证登录态 */
api("/api/v1/auth/me").then(function (user) {
afterLogin(user);
}).catch(function () {
saveAuth(getToken(), user);
afterAuth(user);
}).catch(function (err) {
if (err && err.code === 1006) {
/* must_change_password 用户被 /me 拦截:强制先改密(后端放行 /auth/password */
openPasswordModal(true);
return;
}
showLogin();
});
})();
+4 -8
View File
@@ -16,7 +16,7 @@ services:
- .env
depends_on:
qdrant:
condition: service_healthy
condition: service_started
redis:
condition: service_healthy
ollama:
@@ -36,11 +36,8 @@ services:
- "${QDRANT_DASHBOARD_PORT:-6334}:6334"
volumes:
- ${NAS_DATA_DIR:-./data}/qdrant:/qdrant/storage
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:6333/healthz"]
interval: 10s
timeout: 5s
retries: 5
# 官方 qdrant 镜像不含 curl/wget,内部 healthcheck 无法执行 HTTP 探测;
# 改用 service_started 依赖(qdrant 启动即就绪),由 app 的 restart 策略兜底。
networks:
- qmdsearch
@@ -48,8 +45,7 @@ services:
image: redis:7-alpine
container_name: qmdsearch-redis
restart: unless-stopped
ports:
- "${REDIS_PORT:-6379}:6379"
# 仅内部网络访问,不对外暴露主机端口(避免与 NAS 已有 redis 的 6379 冲突)
volumes:
- ${NAS_DATA_DIR:-./data}/redis:/data
healthcheck:
+52 -6
View File
@@ -1,16 +1,21 @@
"""pytest 全局夹具:覆盖 JWT 认证依赖,让现有 API 测试默认以 admin 身份运行
业务接口(search/document/knowledge)已加 Depends(get_current_user)、
DELETE /documents 加了 Depends(require_admin)。这里通过 autouse 夹具把两个依赖
统一替换为返回固定 admin AuthUser 的 lambda使现有 API 测试无需改动即可通过认证。
单个测试需要走真实认证逻辑时(如 tests/test_auth.py),可在测试函数内
pop 掉对应 overrideautouse fixture yield 后会统一 clear。
业务接口(search/knowledge/settings)仍使用 app.core.auth 的 JWT 依赖,
这里通过 autouse 夹具把两个依赖统一替换为返回固定 admin AuthUser 的 lambda
使现有 API 测试无需改动即可通过认证。单个测试需要走真实认证逻辑时,
可在测试函数内 pop 掉对应 overrideautouse fixture yield 后会统一 clear。
文档变更类端点(POST /documents、/documents/upload、DELETE /documents/{id}
使用 app.api.deps 的会话认证(UserStore/SessionStore),由 auth_stores /
admin_headers 夹具注入内存存储并签发真实 session token。
另外清理 LLM 客户端 / 解析插件 / 去重策略三处进程级缓存,避免跨测试串扰
(不同测试可能注入不同的 redis/ollama 实例,缓存按配置签名而非实例区分)。
"""
import asyncio
from datetime import UTC, datetime
from typing import Any
import pytest
@@ -23,13 +28,54 @@ TEST_USER = AuthUser(username="testuser", role="admin", created_at=datetime.now(
@pytest.fixture(autouse=True)
def override_auth():
"""所有测试默认以 admin 身份运行;测试结束清理 dependency_overrides"""
"""所有测试默认以 admin 身份运行(旧 JWT 依赖);测试结束清理 dependency_overrides"""
app.dependency_overrides[get_current_user] = lambda: TEST_USER
app.dependency_overrides[require_admin] = lambda: TEST_USER
yield
app.dependency_overrides.clear()
@pytest.fixture
def auth_stores(monkeypatch: pytest.MonkeyPatch):
"""注入内存 UserStore/SessionStore 到 deps 单例,并预置 admin 账号(admin/admin-pass-123
同时关闭 lifespan 的默认管理员引导,避免测试库被写入随机密码账号。
返回 (user_store, session_store),测试可直接操作用户数据。
"""
import app.main as main_module
from app.api import deps
from app.core.sessions import SessionStore
from app.core.users import UserStore
user_store = UserStore(None)
session_store = SessionStore(None)
monkeypatch.setattr(deps, "_user_store", user_store)
monkeypatch.setattr(deps, "_session_store", session_store)
async def _noop_bootstrap(*args: Any, **kwargs: Any) -> None:
return None
monkeypatch.setattr(main_module, "bootstrap_admin", _noop_bootstrap)
async def _seed() -> None:
await user_store.create("admin", "admin-pass-123", role="admin")
asyncio.run(_seed())
return user_store, session_store
@pytest.fixture
def admin_headers(auth_stores) -> dict[str, str]:
"""为内存 admin 签发真实 session,返回 Authorization Bearer 请求头"""
_, session_store = auth_stores
async def _login() -> str:
return await session_store.create("admin", "admin")
token = asyncio.run(_login())
return {"Authorization": f"Bearer {token}"}
@pytest.fixture(autouse=True)
def _invalidate_runtime_caches():
"""每个测试前后清理 LLM/解析插件/去重策略进程级缓存
+5 -3
View File
@@ -166,7 +166,9 @@ async def _seed_documents(service: QdrantService) -> None:
class TestAdminClosedLoop:
"""文档管理 API + stats 的完整闭环(真实内存 Qdrant + TestClient"""
async def test_document_admin_closed_loop(self, monkeypatch: pytest.MonkeyPatch) -> None:
async def test_document_admin_closed_loop(
self, monkeypatch: pytest.MonkeyPatch, admin_headers: dict[str, str]
) -> None:
service = QdrantService(client=AsyncQdrantClient(location=":memory:"))
await service.ensure_collections()
await _seed_documents(service)
@@ -235,7 +237,7 @@ class TestAdminClosedLoop:
}
# 4. 删除文档 Adeleted_total > 0,各集合删除数与写入一致
resp = client.delete(f"/api/v1/documents/{DOC_A}")
resp = client.delete(f"/api/v1/documents/{DOC_A}", headers=admin_headers)
assert resp.status_code == 200
body = resp.json()
assert body["code"] == 0
@@ -282,7 +284,7 @@ class TestAdminClosedLoop:
assert [item["doc_id"] for item in items] == [DOC_B]
# 6. 幂等再删:code=0 且 deleted_total=0,各集合删除数全 0
resp = client.delete(f"/api/v1/documents/{DOC_A}")
resp = client.delete(f"/api/v1/documents/{DOC_A}", headers=admin_headers)
body = resp.json()
assert body["code"] == 0
data = body["data"]
+223 -3
View File
@@ -4,8 +4,15 @@
1. 路由存在性:200 + content-type 为 text/html
2. 五区块可识别标记(文案与 section id)
3. 零外部依赖:无 http(s) 外链资源、无 CDN 引用
4. fetch 调用路径与后端 API 契约一致
4. fetch 调用路径与后端 API 契约一致(含 /api/v1/auth/*
5. 删除操作的 confirm() 二次确认逻辑
6. 登录门禁:登录卡片、localStorage key、/auth/me 验证、/auth/login 路径
7. 顶栏用户区:用户名、角色徽章、修改密码、退出登录
8. 修改密码:旧/新/确认表单、must_change_password 强制改密
9. 用户管理区块:列表/创建表单/角色下拉/重置/删除 confirm、admin 角色门禁
10. 请求拦截:Authorization Bearer 注入、1005 回登录、1006 错误条
11. API 指南区块:导航/section、API_GUIDE 清单与真实路由一致性、试一下面板、
curl 复制、auth 标注、upload 文件选择、禁止自定义 URL
"""
import re
@@ -68,12 +75,17 @@ def test_admin_page_no_external_resources(admin_html: str) -> None:
def test_admin_page_fetch_paths(admin_html: str) -> None:
"""fetch 调用路径与后端 API 契约一致"""
"""fetch 调用路径与后端 API 契约一致(含认证相关端点)"""
for path in (
"/api/v1/search",
"/api/v1/documents",
"/api/v1/knowledge/stats",
"/api/v1/knowledge/categories",
"/api/v1/auth/login",
"/api/v1/auth/me",
"/api/v1/auth/logout",
"/api/v1/auth/password",
"/api/v1/auth/users",
):
assert path in admin_html
@@ -92,9 +104,17 @@ def _collect_route_paths(routes: list) -> set[str]:
def test_admin_page_fetch_paths_in_real_routes(admin_html: str) -> None:
"""页面 api() 调用路径均在真实后端路由集合内(含新增的 tasks 路径)"""
"""页面 api() 调用路径均在真实后端路由集合内(含 tasks 与 /api/v1/auth/* 路径)"""
route_paths = _collect_route_paths(app.routes)
assert "/api/v1/documents/tasks/{task_id}" in route_paths
for auth_path in (
"/api/v1/auth/login",
"/api/v1/auth/me",
"/api/v1/auth/logout",
"/api/v1/auth/password",
"/api/v1/auth/users",
):
assert auth_path in route_paths
prefixes = set(re.findall(r'api\("([^"?]+)', admin_html))
assert prefixes, "页面应包含 api() 调用"
@@ -123,3 +143,203 @@ def test_admin_page_ingest_polling(admin_html: str) -> None:
def test_admin_page_has_confirm(admin_html: str) -> None:
"""删除操作包含 confirm() 二次确认逻辑"""
assert "confirm(" in admin_html
def test_admin_page_login_card(admin_html: str) -> None:
"""登录门禁:登录卡片、localStorage key、/auth/me 验证与 /auth/login 路径"""
assert 'id="login-overlay"' in admin_html
assert 'id="login-form"' in admin_html
assert 'id="login-username"' in admin_html
assert 'id="login-password"' in admin_html
assert "登录" in admin_html
# token 与用户信息持久化 key
assert "qmd_token" in admin_html
assert "qmd_user" in admin_html
assert "localStorage" in admin_html
# 页面加载时经 /auth/me 验证登录态
assert "/api/v1/auth/me" in admin_html
# 登录请求路径
assert "/api/v1/auth/login" in admin_html
def test_admin_page_user_area(admin_html: str) -> None:
"""顶栏用户区:用户名、角色徽章、修改密码与退出登录按钮"""
assert 'id="user-area"' in admin_html
assert 'id="user-name"' in admin_html
assert 'id="user-role"' in admin_html
assert "role-badge" in admin_html
assert 'id="btn-change-password"' in admin_html
assert "修改密码" in admin_html
assert 'id="btn-logout"' in admin_html
assert "退出登录" in admin_html
# 退出登录调用后端 logout 端点并清理 localStorage
assert "/api/v1/auth/logout" in admin_html
assert "clearAuth(" in admin_html
def test_admin_page_password_form(admin_html: str) -> None:
"""修改密码:旧/新/确认表单、改密端点与 must_change_password 强制改密层"""
assert 'id="password-overlay"' in admin_html
assert 'id="password-form"' in admin_html
assert 'id="password-old"' in admin_html
assert 'id="password-new"' in admin_html
assert 'id="password-confirm"' in admin_html
assert "/api/v1/auth/password" in admin_html
# must_change_password 登录后强制改密:提示语 + 取消按钮在强制模式隐藏
assert "must_change_password" in admin_html
assert "首次登录须" in admin_html
assert 'id="password-forced-tip"' in admin_html
assert 'id="btn-password-cancel"' in admin_html
assert "passwordForced" in admin_html
def test_admin_page_users_section(admin_html: str) -> None:
"""用户管理区块:用户列表/创建表单/角色下拉/重置密码/删除 confirm/角色门禁"""
assert 'id="section-users"' in admin_html
assert 'id="users-tbody"' in admin_html
# 表头列:用户名/角色/须改密/创建时间/操作
for col in ("用户名", "角色", "须改密", "创建时间", "操作"):
assert col in admin_html
# 创建用户表单与角色下拉
assert 'id="user-create-form"' in admin_html
assert 'id="user-new-name"' in admin_html
assert 'id="user-new-password"' in admin_html
assert 'id="user-new-role"' in admin_html
assert '<option value="admin"' in admin_html
assert '<option value="user"' in admin_html
assert "创建用户" in admin_html
# 操作列:重置密码(弹输入)与删除(confirm 二次确认)
assert "重置密码" in admin_html
assert "prompt(" in admin_html
assert "确定删除用户" in admin_html
# 用户管理端点
assert "/api/v1/auth/users" in admin_html
# 角色门禁:仅 admin 挂载该区块进 DOM,非 admin 完全不渲染
assert 'role === "admin"' in admin_html
assert "mountUsersSection" in admin_html
assert "unmountUsersSection" in admin_html
def test_admin_page_auth_interceptor(admin_html: str) -> None:
"""请求拦截:统一注入 Authorization Bearer1005 回登录;1006 错误条提示"""
assert 'options.headers["Authorization"] = "Bearer " + token' in admin_html
# 1005 未认证/凭证无效 → 清 localStorage 并回登录卡片
assert "body.code === 1005" in admin_html
assert "clearAuth();" in admin_html
assert "showLogin();" in admin_html
# 1006 权限不足 → 错误码透传给错误条
assert "1006" in admin_html
def test_admin_page_api_guide_section(admin_html: str) -> None:
"""API 指南区块:导航按钮、section、静态清单与鉴权标注(user 角色也可见)"""
assert 'data-target="section-api-guide"' in admin_html
assert 'id="section-api-guide"' in admin_html
assert 'id="api-guide-list"' in admin_html
assert "API 指南" in admin_html
# 静态维护的端点清单
assert "API_GUIDE" in admin_html
# auth 中文标注:免登录 / 需登录 / 仅 admin
for label in ("免登录", "需登录", "仅 admin"):
assert label in admin_html
# 方法颜色徽章
for cls in ("method-get", "method-post", "method-delete"):
assert cls in admin_html
def test_admin_page_api_guide_try_panel(admin_html: str) -> None:
"""试一下面板:发送按钮/状态码/耗时/格式化 JSON 展示/curl 复制/upload 文件选择"""
assert "试一下" in admin_html
assert "try-panel" in admin_html
assert "try-toggle" in admin_html
# 发送按钮与结果区(状态码/耗时/pre 格式化展示)
assert "try-send" in admin_html
assert "发送" in admin_html
assert "状态码" in admin_html
assert "耗时" in admin_html
assert "try-result-pre" in admin_html
assert "JSON.stringify(JSON.parse(text), null, 2)" in admin_html
# 需鉴权端点自动带当前 token
assert 'ep.auth !== "none"' in admin_html
# body 非 JSON 本地报错不发请求
assert "未发送请求" in admin_html
# curl 示例与复制(clipboard + 降级 textarea 选中)
assert "curl 示例" in admin_html
assert "curl-copy" in admin_html
assert "复制" in admin_html
assert "navigator.clipboard" in admin_html
assert "execCommand" in admin_html
# base URL 用 location.origin 动态拼接
assert "location.origin" in admin_html
# upload 端点:file input + FormData(不带 Content-Type 头)
assert "try-file" in admin_html
assert 'fileInput.type = "file"' in admin_html
assert "FormData" in admin_html
def test_admin_page_api_guide_no_custom_url(admin_html: str) -> None:
"""安全约束:只能从清单选择端点,页面无自定义任意 URL 输入"""
assert "api-url" not in admin_html
assert "自定义 URL" not in admin_html.replace("不支持自定义 URL", "")
# 试一下面板只能从清单展开:端点 URL 由清单定义拼接,无自由输入
assert "不提供任意 URL 输入框" in admin_html
_API_GUIDE_ENTRY_RE = re.compile(r'method:\s*"(\w+)",\s*path:\s*"([^"]+)"')
def _extract_api_guide_entries(admin_html: str) -> list[tuple[str, str]]:
"""从页面 JS 提取 API_GUIDE 静态清单中的 (method, path) 列表"""
match = re.search(r"API_GUIDE\s*=\s*\[(.*?)\n\];", admin_html, re.S)
assert match, "页面应包含 API_GUIDE 静态清单"
return _API_GUIDE_ENTRY_RE.findall(match.group(1))
def _collect_route_methods(routes: list) -> set[tuple[str, str]]:
"""递归收集路由 (method, path) 对(FastAPI 0.140+ include_router 包装为 _IncludedRouter"""
pairs: set[tuple[str, str]] = set()
for route in routes:
path = getattr(route, "path", None)
methods = getattr(route, "methods", None)
if path and methods:
for method in methods:
pairs.add((method, path))
sub_router = getattr(route, "original_router", None)
if sub_router is not None:
pairs |= _collect_route_methods(sub_router.routes)
return pairs
def test_admin_page_api_guide_paths_in_real_routes(admin_html: str) -> None:
"""API_GUIDE 每个 method+path 均在真实后端路由集合内,且全部真实端点被覆盖(防漂移)"""
entries = _extract_api_guide_entries(admin_html)
assert entries, "API_GUIDE 应至少包含一个端点"
route_methods = _collect_route_methods(app.routes)
for method, path in entries:
assert (method, path) in route_methods, f"API_GUIDE 端点 {method} {path} 不在后端路由集合内"
# 全部真实端点覆盖:health/search/documents*/knowledge*/auth*
expected = {
("GET", "/api/v1/health"),
("POST", "/api/v1/search"),
("POST", "/api/v1/documents"),
("POST", "/api/v1/documents/upload"),
("GET", "/api/v1/documents/tasks/{task_id}"),
("GET", "/api/v1/documents"),
("GET", "/api/v1/documents/{doc_id}"),
("DELETE", "/api/v1/documents/{doc_id}"),
("GET", "/api/v1/knowledge/categories"),
("GET", "/api/v1/knowledge/stats"),
("POST", "/api/v1/auth/login"),
("POST", "/api/v1/auth/logout"),
("POST", "/api/v1/auth/password"),
("GET", "/api/v1/auth/me"),
("GET", "/api/v1/auth/users"),
("POST", "/api/v1/auth/users"),
("POST", "/api/v1/auth/users/{username}/password"),
("DELETE", "/api/v1/auth/users/{username}"),
}
assert expected <= route_methods, f"后端缺少预期端点: {sorted(expected - route_methods)}"
missing = expected - set(entries)
assert not missing, f"API_GUIDE 缺少端点: {sorted(missing)}"
+6 -221
View File
@@ -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_adminadmin 通过、非 admin 抛 FORBIDDEN(直接测函数,不经 API)"""
+451
View File
@@ -0,0 +1,451 @@
"""认证与用户管理 API 测试(TestClient + 内存 UserStore/SessionStore,不真实联网)
覆盖:
- POST /auth/login:成功 / 密码错误 / 用户不存在 / 缺字段
- POST /auth/logout:成功注销后会话失效;无 token 1005
- POST /auth/password:旧密码错误 1005、弱密码 1001、成功后旧 token 仍可用(Task1 语义:改密不清 session
- must_change_password:登录成功但业务端点 1006,改密后恢复
- /auth/users CRUDadmin 全流程;user 角色 1006;重名/非法用户名/弱密码/非法角色 1001;
删自己/最后 admin 1001;不存在 1004;重置他人密码后旧密码失效且会话被清除
- 文档端点鉴权:POST /documents、DELETE /documents/{id} 无 token 1005、user 角色 token 可调;
GET /documents、POST /search 无 token 正常(免登录回归)
内存存储注入复用 conftest 的 auth_stores / admin_headers 夹具。
"""
import asyncio
from collections.abc import Iterator
from typing import Any
import pytest
from fastapi.testclient import TestClient
from app.api.v1 import document as document_module
from app.api.v1 import search as search_module
from app.main import app
from app.models.document import DocumentInput
from app.models.search import SearchRequest, SearchResponse
from app.services.qdrant import QdrantService
class FakeManager:
"""假入库任务管理器:记录 submit 调用并返回固定 task_id"""
def __init__(self, task_id: str = "task-1") -> None:
self.task_id = task_id
self.submitted: list[DocumentInput] = []
async def submit(self, doc: DocumentInput) -> str:
self.submitted.append(doc)
return self.task_id
class FakeQdrant:
"""假 QdrantService:列表返回空、删除返回全 0"""
async def scroll_l1(self, limit: int = 20, offset: str | None = None) -> tuple[list, str | None]:
return [], None
async def delete_by_doc_id(self, doc_id: str) -> dict[str, int]:
return {"doc_l1": 0, "doc_l2": 0, "doc_l3": 0, "chunks": 0}
class FakeRetriever:
"""假检索器:返回空命中的 SearchResponse"""
async def search(self, request: SearchRequest) -> SearchResponse:
return SearchResponse(query=request.query)
@pytest.fixture
def client(monkeypatch: pytest.MonkeyPatch, auth_stores) -> Iterator[TestClient]:
"""TestClientlifespan 建集合改空操作;auth_stores 注入内存认证存储(含 admin/admin-pass-123"""
async def _noop_ensure_collections(self: QdrantService) -> None:
return None
monkeypatch.setattr(QdrantService, "ensure_collections", _noop_ensure_collections)
with TestClient(app) as test_client:
yield test_client
def _login(client: TestClient, username: str, password: str) -> dict[str, Any]:
"""调登录接口并返回响应体"""
return client.post("/api/v1/auth/login", json={"username": username, "password": password}).json()
def _create_user(user_store, username: str, password: str, *, role: str = "user", must_change: bool = False) -> None:
"""同步包装:向内存 UserStore 写入用户"""
asyncio.run(user_store.create(username, password, role=role, must_change_password=must_change))
def _headers(session_store, username: str, role: str) -> dict[str, str]:
"""同步包装:为指定用户签发 session 并返回 Bearer 请求头"""
token = asyncio.run(session_store.create(username, role))
return {"Authorization": f"Bearer {token}"}
class TestLogin:
"""POST /api/v1/auth/login"""
def test_login_success(self, client: TestClient) -> None:
body = _login(client, "admin", "admin-pass-123")
assert body["code"] == 0
data = body["data"]
assert data["token"]
assert data["username"] == "admin"
assert data["role"] == "admin"
assert data["must_change_password"] is False
def test_login_wrong_password(self, client: TestClient) -> None:
body = _login(client, "admin", "wrong-password")
assert body["code"] == 1005
assert body["data"] is None
def test_login_user_not_found(self, client: TestClient) -> None:
body = _login(client, "nobody", "whatever-123")
assert body["code"] == 1005
assert body["data"] is None
def test_login_missing_field(self, client: TestClient) -> None:
resp = client.post("/api/v1/auth/login", json={"username": "admin"})
body = resp.json()
assert body["code"] == 1001
assert body["data"] is None
class TestMe:
"""GET /api/v1/auth/me"""
def test_me_success(self, client: TestClient, admin_headers: dict[str, str]) -> None:
body = client.get("/api/v1/auth/me", headers=admin_headers).json()
assert body["code"] == 0
data = body["data"]
assert data["username"] == "admin"
assert data["role"] == "admin"
assert "password_hash" not in data
assert "salt" not in data
def test_me_without_token(self, client: TestClient) -> None:
body = client.get("/api/v1/auth/me").json()
assert body["code"] == 1005
class TestLogout:
"""POST /api/v1/auth/logout"""
def test_logout_invalidates_session(self, client: TestClient, admin_headers: dict[str, str]) -> None:
resp = client.post("/api/v1/auth/logout", headers=admin_headers)
assert resp.json()["code"] == 0
# 注销后同一 token 不再可用
body = client.get("/api/v1/auth/users", headers=admin_headers).json()
assert body["code"] == 1005
def test_logout_without_token(self, client: TestClient) -> None:
body = client.post("/api/v1/auth/logout").json()
assert body["code"] == 1005
assert body["data"] is None
class TestChangePassword:
"""POST /api/v1/auth/password"""
def test_wrong_old_password(self, client: TestClient, admin_headers: dict[str, str]) -> None:
body = client.post(
"/api/v1/auth/password",
json={"old_password": "wrong-old", "new_password": "new-pass-456"},
headers=admin_headers,
).json()
assert body["code"] == 1005
assert body["data"] is None
def test_weak_new_password(self, client: TestClient, admin_headers: dict[str, str]) -> None:
body = client.post(
"/api/v1/auth/password",
json={"old_password": "admin-pass-123", "new_password": "short"},
headers=admin_headers,
).json()
assert body["code"] == 1001
assert body["data"] is None
def test_success_keeps_session_and_new_password_works(
self, client: TestClient, admin_headers: dict[str, str]
) -> None:
body = client.post(
"/api/v1/auth/password",
json={"old_password": "admin-pass-123", "new_password": "new-pass-456"},
headers=admin_headers,
).json()
assert body["code"] == 0
# Task1 语义:set_password 不清 session,旧 token 仍可用
assert client.get("/api/v1/auth/users", headers=admin_headers).json()["code"] == 0
# 新密码可登录,旧密码失效
assert _login(client, "admin", "new-pass-456")["code"] == 0
assert _login(client, "admin", "admin-pass-123")["code"] == 1005
def test_change_password_without_token(self, client: TestClient) -> None:
body = client.post(
"/api/v1/auth/password", json={"old_password": "a", "new_password": "new-pass-456"}
).json()
assert body["code"] == 1005
class TestMustChangePassword:
"""must_change_password 用户:登录放行、业务端点拦截、改密后恢复"""
def test_blocked_until_password_change(
self, client: TestClient, auth_stores, monkeypatch: pytest.MonkeyPatch
) -> None:
user_store, session_store = auth_stores
_create_user(user_store, "rookie", "temp-pass-123", role="admin", must_change=True)
monkeypatch.setattr(document_module, "_get_task_manager", lambda: FakeManager())
# 登录成功,响应携带 must_change_password 标记
body = _login(client, "rookie", "temp-pass-123")
assert body["code"] == 0
assert body["data"]["must_change_password"] is True
headers = {"Authorization": f"Bearer {body['data']['token']}"}
# 业务端点(users 列表 / 文档入库)被拦截
assert client.get("/api/v1/auth/users", headers=headers).json()["code"] == 1006
resp = client.post("/api/v1/documents", json={"text": "正文"}, headers=headers)
assert resp.json()["code"] == 1006
# 改密(被拦截用户唯一可用接口)后恢复
change = client.post(
"/api/v1/auth/password",
json={"old_password": "temp-pass-123", "new_password": "new-pass-456"},
headers=headers,
).json()
assert change["code"] == 0
assert client.get("/api/v1/auth/users", headers=headers).json()["code"] == 0
resp = client.post("/api/v1/documents", json={"text": "正文"}, headers=headers)
assert resp.status_code == 202
class TestUsersCrud:
"""用户管理端点(/auth/users*,全部要求 admin"""
def test_list_users_sanitized(self, client: TestClient, admin_headers: dict[str, str]) -> None:
body = client.get("/api/v1/auth/users", headers=admin_headers).json()
assert body["code"] == 0
users = body["data"]
assert len(users) == 1
admin = users[0]
assert set(admin.keys()) == {"username", "role", "must_change_password", "created_at"}
assert admin["username"] == "admin"
assert "password_hash" not in admin
assert "salt" not in admin
def test_users_endpoints_forbidden_for_user_role(self, client: TestClient, auth_stores) -> None:
user_store, session_store = auth_stores
_create_user(user_store, "bob", "bob-pass-123", role="user")
headers = _headers(session_store, "bob", "user")
assert client.get("/api/v1/auth/users", headers=headers).json()["code"] == 1006
body = client.post(
"/api/v1/auth/users", json={"username": "carol", "password": "carol-pass-123"}, headers=headers
).json()
assert body["code"] == 1006
assert client.delete("/api/v1/auth/users/admin", headers=headers).json()["code"] == 1006
def test_create_user_success(self, client: TestClient, admin_headers: dict[str, str]) -> None:
body = client.post(
"/api/v1/auth/users",
json={"username": "carol", "password": "carol-pass-123", "role": "user"},
headers=admin_headers,
).json()
assert body["code"] == 0
data = body["data"]
assert set(data.keys()) == {"username", "role", "must_change_password", "created_at"}
assert data["username"] == "carol"
assert data["role"] == "user"
assert data["must_change_password"] is False
# 新用户可登录
assert _login(client, "carol", "carol-pass-123")["code"] == 0
def test_create_user_duplicate(self, client: TestClient, admin_headers: dict[str, str]) -> None:
body = client.post(
"/api/v1/auth/users",
json={"username": "admin", "password": "another-pass-123"},
headers=admin_headers,
).json()
assert body["code"] == 1001
assert body["data"] is None
def test_create_user_invalid_name(self, client: TestClient, admin_headers: dict[str, str]) -> None:
body = client.post(
"/api/v1/auth/users",
json={"username": "x", "password": "valid-pass-123"},
headers=admin_headers,
).json()
assert body["code"] == 1001
def test_create_user_weak_password(self, client: TestClient, admin_headers: dict[str, str]) -> None:
body = client.post(
"/api/v1/auth/users",
json={"username": "dave", "password": "short"},
headers=admin_headers,
).json()
assert body["code"] == 1001
def test_create_user_invalid_role(self, client: TestClient, admin_headers: dict[str, str]) -> None:
body = client.post(
"/api/v1/auth/users",
json={"username": "dave", "password": "dave-pass-123", "role": "superuser"},
headers=admin_headers,
).json()
assert body["code"] == 1001
def test_reset_password(self, client: TestClient, auth_stores, admin_headers: dict[str, str]) -> None:
user_store, session_store = auth_stores
_create_user(user_store, "erin", "erin-pass-123", role="user")
old_headers = _headers(session_store, "erin", "user")
body = client.post(
"/api/v1/auth/users/erin/password",
json={"new_password": "erin-new-456"},
headers=admin_headers,
).json()
assert body["code"] == 0
# 旧密码失效、新密码可登录;该用户既有 session 被清除
assert _login(client, "erin", "erin-pass-123")["code"] == 1005
assert _login(client, "erin", "erin-new-456")["code"] == 0
assert client.post("/api/v1/auth/logout", headers=old_headers).json()["code"] == 1005
def test_reset_password_user_not_found(self, client: TestClient, admin_headers: dict[str, str]) -> None:
body = client.post(
"/api/v1/auth/users/ghost/password",
json={"new_password": "ghost-pass-123"},
headers=admin_headers,
).json()
assert body["code"] == 1004
def test_reset_password_weak(self, client: TestClient, admin_headers: dict[str, str]) -> None:
body = client.post(
"/api/v1/auth/users/admin/password",
json={"new_password": "short"},
headers=admin_headers,
).json()
assert body["code"] == 1001
def test_delete_user_success_clears_sessions(
self, client: TestClient, auth_stores, admin_headers: dict[str, str]
) -> None:
user_store, session_store = auth_stores
_create_user(user_store, "admin2", "admin2-pass-123", role="admin")
admin2_headers = _headers(session_store, "admin2", "admin")
body = client.delete("/api/v1/auth/users/admin2", headers=admin_headers).json()
assert body["code"] == 0
# 用户消失:登录 1005;其 session 已清除
assert _login(client, "admin2", "admin2-pass-123")["code"] == 1005
assert client.get("/api/v1/auth/users", headers=admin2_headers).json()["code"] == 1005
def test_delete_user_not_found(self, client: TestClient, admin_headers: dict[str, str]) -> None:
body = client.delete("/api/v1/auth/users/ghost", headers=admin_headers).json()
assert body["code"] == 1004
def test_delete_self_forbidden(self, client: TestClient, auth_stores) -> None:
user_store, session_store = auth_stores
# 存在另一个 admin,排除“最后 admin”规则干扰,单独验证删自己
_create_user(user_store, "admin2", "admin2-pass-123", role="admin")
admin2_headers = _headers(session_store, "admin2", "admin")
body = client.delete("/api/v1/auth/users/admin2", headers=admin2_headers).json()
assert body["code"] == 1001
def test_delete_last_admin_forbidden(self, client: TestClient, admin_headers: dict[str, str]) -> None:
# 库中唯一 admin 即当前登录者,删除被拒绝
body = client.delete("/api/v1/auth/users/admin", headers=admin_headers).json()
assert body["code"] == 1001
class TestDocumentAuth:
"""文档端点鉴权:变更类需登录,GET 系列免登录"""
def test_post_documents_requires_auth(self, client: TestClient, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(document_module, "_get_task_manager", lambda: FakeManager())
body = client.post("/api/v1/documents", json={"text": "正文"}).json()
assert body["code"] == 1005
assert body["data"] is None
def test_post_documents_with_user_role(
self, client: TestClient, auth_stores, monkeypatch: pytest.MonkeyPatch
) -> None:
user_store, session_store = auth_stores
_create_user(user_store, "bob", "bob-pass-123", role="user")
manager = FakeManager(task_id="task-bob")
monkeypatch.setattr(document_module, "_get_task_manager", lambda: manager)
resp = client.post(
"/api/v1/documents", json={"text": "正文"}, headers=_headers(session_store, "bob", "user")
)
assert resp.status_code == 202
assert resp.json()["data"]["task_id"] == "task-bob"
assert len(manager.submitted) == 1
def test_delete_document_requires_auth(self, client: TestClient, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(document_module, "_get_qdrant", lambda: FakeQdrant())
body = client.delete("/api/v1/documents/doc-1").json()
assert body["code"] == 1005
def test_delete_document_with_user_role(
self, client: TestClient, auth_stores, monkeypatch: pytest.MonkeyPatch
) -> None:
user_store, session_store = auth_stores
_create_user(user_store, "bob", "bob-pass-123", role="user")
monkeypatch.setattr(document_module, "_get_qdrant", lambda: FakeQdrant())
body = client.delete("/api/v1/documents/doc-1", headers=_headers(session_store, "bob", "user")).json()
assert body["code"] == 0
assert body["data"]["deleted_total"] == 0
def test_get_documents_no_token(self, client: TestClient, monkeypatch: pytest.MonkeyPatch) -> None:
"""GET /documents 免登录回归"""
monkeypatch.setattr(document_module, "_get_qdrant", lambda: FakeQdrant())
body = client.get("/api/v1/documents").json()
assert body["code"] == 0
assert body["data"] == {"items": [], "next_offset": None}
def test_search_no_token(self, client: TestClient, monkeypatch: pytest.MonkeyPatch) -> None:
"""POST /search 免登录回归(检索鉴权由 conftest 覆盖旧 JWT 依赖放行)"""
monkeypatch.setattr(search_module, "_retriever", FakeRetriever())
body = client.post("/api/v1/search", json={"query": "任意查询"}).json()
assert body["code"] == 0
assert body["data"]["hits"] == []
+219
View File
@@ -0,0 +1,219 @@
"""用户体系集成验证:内存环境全链路
内存 UserStore/SessionStore 注入 app.api.deps 单例(不跑 lifespan,无需真实 Redis);
文档闭环复用 test_e2e_integration 的内存 Qdrant + FakeOllama 环境(真实入库→删除)。
覆盖:
- 空库 bootstrap_admin 引导 → 登录 → must_change_password 拦截(1006) → 改密 → 恢复(200)
- 多角色权限:admin 创建 user,user 真实入库/删除文档成功,用户管理端点 1006
- 未认证与免登录边界:变更类文档端点无 token 1005;查询类端点/health 免登录
- admin 用户管理闭环:重置密码(旧失效/新可登)、删除用户(session 清除 → me 1005
- 引导幂等:已有用户时 bootstrap_admin 返回 None 不重复创建
"""
from typing import Any
import pytest
import structlog
from fastapi.testclient import TestClient
from app.api import deps
from app.api.v1 import document as document_module
from app.api.v1 import search as search_module
from app.config import Settings
from app.core.ingest_tasks import IngestTaskManager, IngestTaskStatus
from app.core.sessions import SessionStore
from app.core.users import UserStore, bootstrap_admin
from app.main import app
from tests.test_e2e_integration import FakeCache, _make_env, _structured_doc
@pytest.fixture
def empty_auth_stores(monkeypatch: pytest.MonkeyPatch) -> tuple[UserStore, SessionStore]:
"""注入空的内存 UserStore/SessionStore 到 deps 单例(不预置账号,用于引导场景)"""
user_store = UserStore(None)
session_store = SessionStore(None)
monkeypatch.setattr(deps, "_user_store", user_store)
monkeypatch.setattr(deps, "_session_store", session_store)
return user_store, session_store
def _login(client: TestClient, username: str, password: str) -> dict[str, Any]:
"""调登录接口并返回响应体"""
return client.post("/api/v1/auth/login", json={"username": username, "password": password}).json()
def _bearer(token: str) -> dict[str, str]:
"""构造 Authorization Bearer 请求头"""
return {"Authorization": f"Bearer {token}"}
class TestBootstrapAndForcedChange:
"""空库引导 + must_change_password 强制改密全链路"""
async def test_bootstrap_login_forced_change(self, empty_auth_stores: tuple[UserStore, SessionStore]) -> None:
user_store, _ = empty_auth_stores
# 空库引导:创建 must_change_password=true 的 admin,明文密码仅本次返回
password = await bootstrap_admin(user_store, structlog.get_logger())
assert password is not None
record = await user_store.get("admin")
assert record is not None
assert record.role == "admin"
assert record.must_change_password is True
# 不进上下文:跳过 lifespan,无需真实 Qdrant/Redis
client = TestClient(app)
# 引导密码可登录,响应携带 must_change_password 标记
login = _login(client, "admin", password)
assert login["code"] == 0
assert login["data"]["must_change_password"] is True
headers = _bearer(login["data"]["token"])
# 改密前:业务端点(用户列表)被 1006 拦截
assert client.get("/api/v1/auth/users", headers=headers).json()["code"] == 1006
# 改密(被拦截用户唯一可用接口)
change = client.post(
"/api/v1/auth/password",
json={"old_password": password, "new_password": "admin-new-pass-1"},
headers=headers,
).json()
assert change["code"] == 0
# 改密后:同一 token 恢复访问
resp = client.get("/api/v1/auth/users", headers=headers)
assert resp.status_code == 200
assert resp.json()["code"] == 0
async def test_bootstrap_idempotent(self, empty_auth_stores: tuple[UserStore, SessionStore]) -> None:
user_store, _ = empty_auth_stores
logger = structlog.get_logger()
first = await bootstrap_admin(user_store, logger)
second = await bootstrap_admin(user_store, logger)
assert first is not None
assert second is None # 已有用户时不重复创建
assert len(await user_store.list()) == 1
class TestRolesAndDocumentChain:
"""多角色权限 + 文档入库/删除真实内存链路"""
async def test_user_role_document_chain(self, auth_stores, monkeypatch: pytest.MonkeyPatch) -> None:
# 文档端点模块级单例替换为内存链路实例(内存 Qdrant + FakeOllama
env = await _make_env()
manager = IngestTaskManager(env.ingester, None, Settings())
monkeypatch.setattr(document_module, "_task_manager", manager)
monkeypatch.setattr(document_module, "_qdrant", env.qdrant)
client = TestClient(app)
# admin 登录并创建 user 角色账号
admin_login = _login(client, "admin", "admin-pass-123")
assert admin_login["code"] == 0
admin_headers = _bearer(admin_login["data"]["token"])
created = client.post(
"/api/v1/auth/users",
json={"username": "alice", "password": "alice-pass-123", "role": "user"},
headers=admin_headers,
).json()
assert created["code"] == 0
assert created["data"]["role"] == "user"
# user 登录
user_login = _login(client, "alice", "alice-pass-123")
assert user_login["code"] == 0
user_headers = _bearer(user_login["data"]["token"])
# user 提交文档入库(202),等待任务跑完写入四层集合
doc = _structured_doc()
resp = client.post(
"/api/v1/documents", json={"text": doc.text, "title": doc.title}, headers=user_headers
)
assert resp.status_code == 202
task_id = resp.json()["data"]["task_id"]
final = await manager.wait_done(task_id)
assert final["status"] == IngestTaskStatus.DONE
doc_id = final["result"]["document_id"]
# user 删除自己入库的文档(四层集合真实清除)
deleted = client.delete(f"/api/v1/documents/{doc_id}", headers=user_headers).json()
assert deleted["code"] == 0
assert deleted["data"]["doc_id"] == doc_id
assert deleted["data"]["deleted_total"] > 0
# user 角色访问用户管理端点 → 1006
assert client.get("/api/v1/auth/users", headers=user_headers).json()["code"] == 1006
forbidden = client.post(
"/api/v1/auth/users",
json={"username": "mallory", "password": "mallory-pass-123"},
headers=user_headers,
).json()
assert forbidden["code"] == 1006
class TestUnauthenticatedBoundaries:
"""未认证拦截与免登录端点回归"""
async def test_boundaries(self, monkeypatch: pytest.MonkeyPatch) -> None:
# 查询类端点挂内存链路,避免触达真实 Qdrant/Redis
env = await _make_env()
monkeypatch.setattr(document_module, "_task_manager", IngestTaskManager(env.ingester, None, Settings()))
monkeypatch.setattr(document_module, "_qdrant", env.qdrant)
monkeypatch.setattr(search_module, "_retriever", env.retriever)
monkeypatch.setattr(search_module, "get_cache", lambda: FakeCache())
client = TestClient(app)
# 变更类文档端点:无 token → 1005
assert client.post("/api/v1/documents", json={"text": "正文"}).json()["code"] == 1005
assert client.delete("/api/v1/documents/some-doc-id").json()["code"] == 1005
# 查询类端点免登录(检索鉴权由 conftest 覆盖旧 JWT 依赖放行)
search = client.post("/api/v1/search", json={"query": "安装步骤有哪些注意事项?"}).json()
assert search["code"] == 0
listing = client.get("/api/v1/documents").json()
assert listing["code"] == 0
health = client.get("/api/v1/health")
assert health.status_code == 200
assert health.json()["status"] == "ok"
class TestAdminUserManagement:
"""admin 重置密码 / 删除用户闭环"""
def test_reset_and_delete_user(self, auth_stores) -> None:
client = TestClient(app)
admin_login = _login(client, "admin", "admin-pass-123")
assert admin_login["code"] == 0
admin_headers = _bearer(admin_login["data"]["token"])
# admin 创建 user
created = client.post(
"/api/v1/auth/users",
json={"username": "bob", "password": "bob-pass-123"},
headers=admin_headers,
).json()
assert created["code"] == 0
assert created["data"]["role"] == "user"
# 重置 bob 密码:旧密码登录 1005,新密码登录成功
reset = client.post(
"/api/v1/auth/users/bob/password",
json={"new_password": "bob-new-pass-456"},
headers=admin_headers,
).json()
assert reset["code"] == 0
assert _login(client, "bob", "bob-pass-123")["code"] == 1005
bob_login = _login(client, "bob", "bob-new-pass-456")
assert bob_login["code"] == 0
bob_headers = _bearer(bob_login["data"]["token"])
# admin 删除 bob:其 session 一并清除,me → 1005
deleted = client.delete("/api/v1/auth/users/bob", headers=admin_headers).json()
assert deleted["code"] == 0
assert client.get("/api/v1/auth/me", headers=bob_headers).json()["code"] == 1005
+3 -2
View File
@@ -43,14 +43,15 @@ class FakeQdrant:
@pytest.fixture
def client(monkeypatch: pytest.MonkeyPatch) -> Iterator[TestClient]:
"""TestClientlifespan 中的 Qdrant 集合初始化替换为空操作"""
def client(monkeypatch: pytest.MonkeyPatch, admin_headers: dict[str, str]) -> Iterator[TestClient]:
"""TestClientlifespan 中的 Qdrant 集合初始化替换为空操作;默认携带 admin 认证头"""
async def _noop_ensure_collections(self: QdrantService) -> None:
return None
monkeypatch.setattr(QdrantService, "ensure_collections", _noop_ensure_collections)
with TestClient(app) as test_client:
test_client.headers.update(admin_headers)
yield test_client
+3 -2
View File
@@ -24,14 +24,15 @@ class FakeManager:
@pytest.fixture
def client(monkeypatch: pytest.MonkeyPatch) -> Iterator[TestClient]:
"""TestClientlifespan 中的 Qdrant 集合初始化替换为空操作"""
def client(monkeypatch: pytest.MonkeyPatch, admin_headers: dict[str, str]) -> Iterator[TestClient]:
"""TestClientlifespan 中的 Qdrant 集合初始化替换为空操作;默认携带 admin 认证头"""
async def _noop_ensure_collections(self: QdrantService) -> None:
return None
monkeypatch.setattr(QdrantService, "ensure_collections", _noop_ensure_collections)
with TestClient(app) as test_client:
test_client.headers.update(admin_headers)
yield test_client
+5 -2
View File
@@ -29,7 +29,9 @@ class FakeManager:
@pytest.fixture
def client(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> Iterator[TestClient]:
def client(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path, admin_headers: dict[str, str]
) -> Iterator[TestClient]:
async def _noop_ensure_collections(self: QdrantService) -> None:
return None
@@ -38,6 +40,7 @@ def client(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> Iterator[TestClie
document_module.settings, "upload_dir", str(tmp_path / "uploads")
)
with TestClient(app) as test_client:
test_client.headers.update(admin_headers)
yield test_client
@@ -91,7 +94,7 @@ def test_upload_md_returns_202_and_saves_file(
manager = FakeManager(task_id="abc-upload")
_inject_manager(monkeypatch, manager)
content = "# Hello\n\nThis is a markdown file.".encode("utf-8")
content = b"# Hello\n\nThis is a markdown file."
resp = client.post(
"/api/v1/documents/upload",
files={"file": ("notes.md", content, "text/markdown")},
+5 -3
View File
@@ -281,7 +281,7 @@ class TestSearchIntegration:
class TestApiIntegration:
"""API 层冒烟:documents → search → knowledge/categories,统一响应格式 code=0"""
async def test_api_smoke(self, monkeypatch: pytest.MonkeyPatch):
async def test_api_smoke(self, monkeypatch: pytest.MonkeyPatch, admin_headers: dict[str, str]):
env = await _make_env()
doc = _structured_doc()
# 入库走异步任务:测试自建纯内存任务管理器(内存 Qdrant + FakeOllama 的 Ingester
@@ -298,8 +298,10 @@ class TestApiIntegration:
monkeypatch.setattr(search_module, "get_cache", lambda: FakeCache())
with TestClient(app) as client:
# 入库:202 拿 task_id,等待任务终态后断言入库结果
resp_doc = client.post("/api/v1/documents", json={"text": doc.text, "title": doc.title})
# 入库:202 拿 task_id,等待任务终态后断言入库结果(变更类端点需 admin 认证头)
resp_doc = client.post(
"/api/v1/documents", json={"text": doc.text, "title": doc.title}, headers=admin_headers
)
assert resp_doc.status_code == 202
body_doc = resp_doc.json()
assert body_doc["code"] == 0
+6 -4
View File
@@ -137,7 +137,7 @@ def _patch_app(
class TestIngestAsyncFullLoop:
"""全链路闭环:异步入库 → 任务查询 → 文档管理 → 检索 → 删除"""
async def test_full_loop(self, monkeypatch: pytest.MonkeyPatch):
async def test_full_loop(self, monkeypatch: pytest.MonkeyPatch, admin_headers: dict[str, str]):
qdrant, ingester, retriever = await _make_env(FakeOllama())
recording = RecordingIngester(ingester)
manager = IngestTaskManager(recording, None, Settings()) # type: ignore[arg-type]
@@ -145,8 +145,10 @@ class TestIngestAsyncFullLoop:
doc = _structured_doc()
with TestClient(app) as client:
# 1. 提交入库:202 + task_id
resp_post = client.post("/api/v1/documents", json={"text": doc.text, "title": doc.title})
# 1. 提交入库:202 + task_id(变更类端点需 admin 认证头)
resp_post = client.post(
"/api/v1/documents", json={"text": doc.text, "title": doc.title}, headers=admin_headers
)
assert resp_post.status_code == 202
body_post = resp_post.json()
assert body_post["code"] == 0
@@ -185,7 +187,7 @@ class TestIngestAsyncFullLoop:
assert any(hit["doc_id"] == document_id for hit in hits)
# 6. 删除:四层集合中该文档全部清除
resp_delete = client.delete(f"/api/v1/documents/{document_id}")
resp_delete = client.delete(f"/api/v1/documents/{document_id}", headers=admin_headers)
body_delete = resp_delete.json()
assert body_delete["code"] == 0
assert body_delete["data"]["deleted_total"] > 0
+3 -2
View File
@@ -43,14 +43,15 @@ def _task(task_id: str, status: str, **extra: Any) -> dict[str, Any]:
@pytest.fixture
def client(monkeypatch: pytest.MonkeyPatch) -> Iterator[TestClient]:
"""TestClientlifespan 中的 Qdrant 集合初始化替换为空操作"""
def client(monkeypatch: pytest.MonkeyPatch, admin_headers: dict[str, str]) -> Iterator[TestClient]:
"""TestClientlifespan 中的 Qdrant 集合初始化替换为空操作;默认携带 admin 认证头"""
async def _noop_ensure_collections(self: QdrantService) -> None:
return None
monkeypatch.setattr(QdrantService, "ensure_collections", _noop_ensure_collections)
with TestClient(app) as test_client:
test_client.headers.update(admin_headers)
yield test_client
+171
View File
@@ -0,0 +1,171 @@
"""SessionStore 单元测试(FakeRedis 与内存降级,不连真实 Redis)
覆盖:
- create/get/delete 生命周期;伪造 token 返回 None
- TTL=43200 传入 Redissetex
- delete_by_username 清除指定用户全部会话
- 内存降级(构造传 None)全功能可用 + 过期惰性清理
- Redis 异常包装为 SessionStoreError
"""
import fnmatch
import time
import pytest
from app.core.sessions import SessionStore, SessionStoreError
class FakeRedis:
"""最小内存版 redis.asyncio 客户端:get/set/setex/delete/scan_iter/keys + TTL 记录
fail=True 时所有操作抛 ConnectionError,用于验证异常包装语义。
"""
def __init__(self, fail: bool = False) -> None:
self.fail = fail
self.store: dict[str, str] = {}
self.ttls: dict[str, int] = {}
def _check(self) -> None:
if self.fail:
raise ConnectionError("redis down")
async def get(self, key: str) -> str | None:
self._check()
return self.store.get(key)
async def set(self, key: str, value: str) -> bool:
self._check()
self.store[key] = value
return True
async def setex(self, key: str, ttl: int, value: str) -> bool:
self._check()
self.store[key] = value
self.ttls[key] = ttl
return True
async def delete(self, *keys: str) -> int:
self._check()
deleted = 0
for key in keys:
if key in self.store:
del self.store[key]
deleted += 1
return deleted
async def scan_iter(self, match: str = "*"):
self._check()
for key in list(self.store):
if fnmatch.fnmatch(key, match):
yield key
async def keys(self, pattern: str = "*") -> list[str]:
self._check()
return [key for key in self.store if fnmatch.fnmatch(key, pattern)]
class TestLifecycle:
"""会话签发 / 校验 / 删除"""
async def test_create_get_delete(self):
store = SessionStore(FakeRedis())
token = await store.create("alice", "admin")
assert isinstance(token, str) and len(token) == 64 # token_hex(32)
assert await store.get(token) == {"username": "alice", "role": "admin"}
await store.delete(token)
assert await store.get(token) is None
async def test_forged_token_returns_none(self):
assert await SessionStore(FakeRedis()).get("f" * 64) is None
async def test_tokens_unique(self):
store = SessionStore(FakeRedis())
assert await store.create("alice", "user") != await store.create("alice", "user")
async def test_delete_is_idempotent(self):
store = SessionStore(FakeRedis())
token = await store.create("alice", "user")
await store.delete(token)
await store.delete(token) # 二次删除不报错
assert await store.get(token) is None
class TestTtl:
"""TTL 透传 Redis"""
async def test_ttl_passed_to_setex(self):
redis = FakeRedis()
store = SessionStore(redis)
token = await store.create("alice", "user")
assert SessionStore.TTL == 43200
assert redis.ttls[f"session:{token}"] == 43200
class TestDeleteByUsername:
"""按用户清除全部会话"""
async def test_clears_only_target_user(self):
store = SessionStore(FakeRedis())
t1 = await store.create("alice", "user")
t2 = await store.create("alice", "user")
t3 = await store.create("bob", "user")
await store.delete_by_username("alice")
assert await store.get(t1) is None
assert await store.get(t2) is None
assert await store.get(t3) is not None
async def test_unknown_username_noop(self):
store = SessionStore(FakeRedis())
token = await store.create("alice", "user")
await store.delete_by_username("nobody")
assert await store.get(token) is not None
class TestMemoryFallback:
"""构造传 None:内存 dict 降级 + 过期惰性清理"""
async def test_full_lifecycle(self):
store = SessionStore(None)
token = await store.create("alice", "admin")
assert await store.get(token) == {"username": "alice", "role": "admin"}
await store.delete_by_username("alice")
assert await store.get(token) is None
async def test_expired_session_returns_none_and_purged(self):
store = SessionStore(None)
token = await store.create("alice", "user")
# 手动把过期时间戳拨到过去,模拟 TTL 到期
payload, _ = store._memory[token]
store._memory[token] = (payload, time.time() - 1)
assert await store.get(token) is None
assert token not in store._memory # 惰性清理
async def test_create_purges_expired_entries(self):
store = SessionStore(None)
expired_token = await store.create("old", "user")
payload, _ = store._memory[expired_token]
store._memory[expired_token] = (payload, time.time() - 1)
await store.create("new", "user") # 签发时顺手清理过期项
assert expired_token not in store._memory
class TestRedisErrors:
"""Redis 读写异常统一包装为 SessionStoreError(不静默)"""
async def test_create_raises_store_error(self):
with pytest.raises(SessionStoreError):
await SessionStore(FakeRedis(fail=True)).create("alice", "user")
async def test_get_raises_store_error(self):
with pytest.raises(SessionStoreError):
await SessionStore(FakeRedis(fail=True)).get("t" * 64)
async def test_delete_raises_store_error(self):
with pytest.raises(SessionStoreError):
await SessionStore(FakeRedis(fail=True)).delete("t" * 64)
async def test_delete_by_username_raises_store_error(self):
with pytest.raises(SessionStoreError):
await SessionStore(FakeRedis(fail=True)).delete_by_username("alice")
+311
View File
@@ -0,0 +1,311 @@
"""UserStore / bootstrap_admin 单元测试(FakeRedis 与内存降级,不连真实 Redis)
覆盖:
- 哈希:同密码同 salt 一致;不同 salt 不同
- create:重名 UserExistsError;非法用户名/弱密码 ValueError;记录不含明文密码
- get/list/delete/verify_password/set_password/count_admins
- bootstrap_admin:空库创建 admin 并打印/返回密码;非空库不重复创建
- 内存降级(构造传 None)全功能可用
- Redis 异常包装为 UserStoreError
"""
import fnmatch
import json
from datetime import datetime
import pytest
from app.core.users import (
UserExistsError,
UserStore,
UserStoreError,
bootstrap_admin,
)
class FakeRedis:
"""最小内存版 redis.asyncio 客户端:get/set/setex/delete/scan_iter/keys + TTL 记录
fail=True 时所有操作抛 ConnectionError,用于验证异常包装语义。
"""
def __init__(self, fail: bool = False) -> None:
self.fail = fail
self.store: dict[str, str] = {}
self.ttls: dict[str, int] = {}
def _check(self) -> None:
if self.fail:
raise ConnectionError("redis down")
async def get(self, key: str) -> str | None:
self._check()
return self.store.get(key)
async def set(self, key: str, value: str) -> bool:
self._check()
self.store[key] = value
return True
async def setex(self, key: str, ttl: int, value: str) -> bool:
self._check()
self.store[key] = value
self.ttls[key] = ttl
return True
async def delete(self, *keys: str) -> int:
self._check()
deleted = 0
for key in keys:
if key in self.store:
del self.store[key]
deleted += 1
return deleted
async def scan_iter(self, match: str = "*"):
self._check()
for key in list(self.store):
if fnmatch.fnmatch(key, match):
yield key
async def keys(self, pattern: str = "*") -> list[str]:
self._check()
return [key for key in self.store if fnmatch.fnmatch(key, pattern)]
class FakeLogger:
"""捕获 warning 调用的假 loggerstructlog 风格:event + kwargs"""
def __init__(self) -> None:
self.warnings: list[tuple[str, dict]] = []
def warning(self, event: str, **kwargs) -> None:
self.warnings.append((event, kwargs))
class TestHashPassword:
"""PBKDF2 哈希确定性"""
def test_same_password_same_salt_consistent(self):
store = UserStore(None)
salt = b"0123456789abcdef"
assert store.hash_password("password123", salt) == store.hash_password("password123", salt)
def test_different_salt_different_hash(self):
store = UserStore(None)
assert store.hash_password("password123", b"a" * 16) != store.hash_password("password123", b"b" * 16)
class TestCreate:
"""创建用户:校验、重名、无明文"""
async def test_create_success(self):
store = UserStore(FakeRedis())
record = await store.create("alice", "password123")
assert record.username == "alice"
assert record.role == "user"
assert record.must_change_password is False
# created_at 为可解析的 ISO8601
datetime.fromisoformat(record.created_at)
async def test_record_contains_no_plaintext(self):
redis = FakeRedis()
store = UserStore(redis)
record = await store.create("alice", "password123")
assert record.password_hash != "password123"
assert "password123" not in record.password_hash
# salt 为 16 字节 hex
assert len(bytes.fromhex(record.salt)) == 16
# 落库 JSON 同样不含明文密码
raw = redis.store["user:alice"]
assert "password123" not in raw
assert json.loads(raw)["username"] == "alice"
async def test_duplicate_raises_user_exists(self):
store = UserStore(FakeRedis())
await store.create("alice", "password123")
with pytest.raises(UserExistsError):
await store.create("alice", "another-password")
@pytest.mark.parametrize("username", ["a", "x" * 33, "bad name", "bad!name", "中文名", ""])
async def test_invalid_username_raises_value_error(self, username: str):
with pytest.raises(ValueError):
await UserStore(FakeRedis()).create(username, "password123")
@pytest.mark.parametrize("username", ["ab", "a" * 32, "A-Z_0-9"])
async def test_valid_username_boundary(self, username: str):
record = await UserStore(FakeRedis()).create(username, "password123")
assert record.username == username
@pytest.mark.parametrize("password", ["", "short", "1234567"])
async def test_short_password_raises_value_error(self, password: str):
with pytest.raises(ValueError):
await UserStore(FakeRedis()).create("alice", password)
async def test_exact_min_password_length_ok(self):
record = await UserStore(FakeRedis()).create("alice", "12345678")
assert record.username == "alice"
class TestGetListDelete:
"""查询 / 列表 / 删除"""
async def test_get_roundtrip_and_miss(self):
store = UserStore(FakeRedis())
created = await store.create("alice", "password123")
assert await store.get("alice") == created
assert await store.get("nobody") is None
async def test_list_scans_only_user_keys(self):
redis = FakeRedis()
store = UserStore(redis)
await store.create("alice", "password123")
await store.create("bob", "password456", role="admin")
redis.store["other:key"] = "not-a-user" # 非 user: 前缀不参与
names = {r.username for r in await store.list()}
assert names == {"alice", "bob"}
async def test_delete_idempotent(self):
store = UserStore(FakeRedis())
await store.create("alice", "password123")
assert await store.delete("alice") is True
assert await store.get("alice") is None
assert await store.delete("alice") is False
class TestVerifyPassword:
"""密码校验"""
async def test_success_returns_record(self):
store = UserStore(FakeRedis())
created = await store.create("alice", "password123")
assert await store.verify_password("alice", "password123") == created
async def test_wrong_password_returns_none(self):
store = UserStore(FakeRedis())
await store.create("alice", "password123")
assert await store.verify_password("alice", "wrong-password") is None
async def test_unknown_user_returns_none(self):
assert await UserStore(FakeRedis()).verify_password("nobody", "password123") is None
class TestSetPassword:
"""重置密码"""
async def test_reset_clears_flag_and_invalidates_old(self):
store = UserStore(FakeRedis())
await store.create("alice", "old-password", must_change_password=True)
assert await store.set_password("alice", "new-password") is True
record = await store.get("alice")
assert record is not None
assert record.must_change_password is False
assert await store.verify_password("alice", "old-password") is None
assert await store.verify_password("alice", "new-password") is not None
async def test_unknown_user_returns_false(self):
assert await UserStore(FakeRedis()).set_password("nobody", "password123") is False
async def test_short_password_raises_value_error(self):
store = UserStore(FakeRedis())
await store.create("alice", "password123")
with pytest.raises(ValueError):
await store.set_password("alice", "short")
class TestCountAdmins:
async def test_counts_only_admin_role(self):
store = UserStore(FakeRedis())
assert await store.count_admins() == 0
await store.create("admin1", "password123", role="admin")
await store.create("admin2", "password123", role="admin")
await store.create("user1", "password123")
assert await store.count_admins() == 2
await store.delete("admin1")
assert await store.count_admins() == 1
class TestBootstrapAdmin:
"""空库引导创建默认管理员"""
async def test_empty_store_creates_admin_and_returns_password(self):
store = UserStore(FakeRedis())
logger = FakeLogger()
password = await bootstrap_admin(store, logger) # type: ignore[arg-type]
assert isinstance(password, str) and len(password) > 0
record = await store.get("admin")
assert record is not None
assert record.role == "admin"
assert record.must_change_password is True
# 返回的明文密码可直接通过校验
assert await store.verify_password("admin", password) is not None
# 明文密码仅在日志中打印一次
assert len(logger.warnings) == 1
assert logger.warnings[0][1].get("password") == password
async def test_non_empty_store_returns_none_without_changes(self):
store = UserStore(FakeRedis())
await store.create("alice", "password123")
logger = FakeLogger()
assert await bootstrap_admin(store, logger) is None # type: ignore[arg-type]
assert [r.username for r in await store.list()] == ["alice"]
assert logger.warnings == []
async def test_second_call_returns_none(self):
store = UserStore(FakeRedis())
logger = FakeLogger()
assert await bootstrap_admin(store, logger) is not None # type: ignore[arg-type]
assert await bootstrap_admin(store, logger) is None # type: ignore[arg-type]
assert len(await store.list()) == 1
class TestMemoryFallback:
"""构造传 None:纯内存 dict 降级,全功能可用"""
async def test_full_flow_without_redis(self):
store = UserStore(None)
await store.create("alice", "password123", role="admin")
await store.create("bob", "password456")
assert await store.count_admins() == 1
assert {r.username for r in await store.list()} == {"alice", "bob"}
assert await store.verify_password("alice", "password123") is not None
assert await store.verify_password("alice", "wrong-password") is None
assert await store.set_password("bob", "new-password") is True
assert await store.verify_password("bob", "new-password") is not None
assert await store.delete("bob") is True
assert await store.get("bob") is None
async def test_bootstrap_admin_memory(self):
store = UserStore(None)
password = await bootstrap_admin(store, FakeLogger()) # type: ignore[arg-type]
assert password is not None
record = await store.get("admin")
assert record is not None and record.role == "admin"
class TestRedisErrors:
"""Redis 读写异常统一包装为 UserStoreError(不静默)"""
async def test_create_raises_store_error(self):
with pytest.raises(UserStoreError):
await UserStore(FakeRedis(fail=True)).create("alice", "password123")
async def test_get_raises_store_error(self):
with pytest.raises(UserStoreError):
await UserStore(FakeRedis(fail=True)).get("alice")
async def test_list_raises_store_error(self):
with pytest.raises(UserStoreError):
await UserStore(FakeRedis(fail=True)).list()
async def test_delete_raises_store_error(self):
with pytest.raises(UserStoreError):
await UserStore(FakeRedis(fail=True)).delete("alice")
async def test_set_password_raises_store_error(self):
with pytest.raises(UserStoreError):
await UserStore(FakeRedis(fail=True)).set_password("alice", "password123")
async def test_count_admins_raises_store_error(self):
with pytest.raises(UserStoreError):
await UserStore(FakeRedis(fail=True)).count_admins()