kplam
6c6f690788
- 移除冗余依赖包 - 新增账号禁用校验与用户管理能力 - 新增文档下载与管理页面文件展示 - 新增API文档页面与用户管理前端页面 - 重构时区处理与docker-compose部署配置 - 完善测试用例与项目文档
618 lines
24 KiB
Python
618 lines
24 KiB
Python
"""认证与用户管理 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 CRUD：admin 全流程；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]:
|
||
"""TestClient：lifespan 建集合改空操作；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", "enabled", "created_at"}
|
||
assert admin["username"] == "admin"
|
||
assert admin["enabled"] is True
|
||
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", "enabled", "created_at"}
|
||
assert data["username"] == "carol"
|
||
assert data["role"] == "user"
|
||
assert data["must_change_password"] is False
|
||
assert data["enabled"] is True
|
||
# 新用户可登录
|
||
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 TestUpdateUser:
|
||
"""PATCH /api/v1/auth/users/{username}：更新角色/启用状态"""
|
||
|
||
def test_update_role_success(self, client: TestClient, admin_headers: dict[str, str]) -> None:
|
||
client.post(
|
||
"/api/v1/auth/users",
|
||
json={"username": "carol", "password": "carol-pass-123"},
|
||
headers=admin_headers,
|
||
)
|
||
body = client.patch(
|
||
"/api/v1/auth/users/carol", json={"role": "admin"}, headers=admin_headers
|
||
).json()
|
||
|
||
assert body["code"] == 0
|
||
assert body["data"]["role"] == "admin"
|
||
assert body["data"]["enabled"] is True
|
||
|
||
def test_update_enabled_success(
|
||
self, client: TestClient, auth_stores, admin_headers: dict[str, str]
|
||
) -> None:
|
||
user_store, _ = auth_stores
|
||
_create_user(user_store, "dave", "dave-pass-123", role="user")
|
||
body = client.patch(
|
||
"/api/v1/auth/users/dave", json={"enabled": False}, headers=admin_headers
|
||
).json()
|
||
|
||
assert body["code"] == 0
|
||
assert body["data"]["enabled"] is False
|
||
assert body["data"]["role"] == "user"
|
||
|
||
def test_update_both_role_and_enabled(
|
||
self, client: TestClient, auth_stores, admin_headers: dict[str, str]
|
||
) -> None:
|
||
user_store, _ = auth_stores
|
||
_create_user(user_store, "erin", "erin-pass-123", role="user")
|
||
body = client.patch(
|
||
"/api/v1/auth/users/erin",
|
||
json={"role": "admin", "enabled": False},
|
||
headers=admin_headers,
|
||
).json()
|
||
|
||
assert body["code"] == 0
|
||
assert body["data"]["role"] == "admin"
|
||
assert body["data"]["enabled"] is False
|
||
|
||
def test_at_least_one_field_required(self, client: TestClient, admin_headers: dict[str, str]) -> None:
|
||
body = client.patch(
|
||
"/api/v1/auth/users/admin", json={}, headers=admin_headers
|
||
).json()
|
||
|
||
assert body["code"] == 1001
|
||
assert body["data"] is None
|
||
|
||
def test_user_not_found(self, client: TestClient, admin_headers: dict[str, str]) -> None:
|
||
body = client.patch(
|
||
"/api/v1/auth/users/ghost", json={"role": "user"}, headers=admin_headers
|
||
).json()
|
||
|
||
assert body["code"] == 1004
|
||
assert body["data"] is None
|
||
|
||
def test_invalid_role(self, client: TestClient, admin_headers: dict[str, str]) -> None:
|
||
body = client.patch(
|
||
"/api/v1/auth/users/admin", json={"role": "superuser"}, headers=admin_headers
|
||
).json()
|
||
|
||
assert body["code"] == 1001
|
||
|
||
def test_last_admin_demote_forbidden(self, client: TestClient, admin_headers: dict[str, str]) -> None:
|
||
# 唯一 admin 降级为 user → 1001
|
||
body = client.patch(
|
||
"/api/v1/auth/users/admin", json={"role": "user"}, headers=admin_headers
|
||
).json()
|
||
|
||
assert body["code"] == 1001
|
||
# 未生效
|
||
assert client.get("/api/v1/auth/users", headers=admin_headers).json()["code"] == 0
|
||
|
||
def test_last_admin_disable_forbidden(self, client: TestClient, admin_headers: dict[str, str]) -> None:
|
||
# 唯一 admin 禁用 → 1001
|
||
body = client.patch(
|
||
"/api/v1/auth/users/admin", json={"enabled": False}, headers=admin_headers
|
||
).json()
|
||
|
||
assert body["code"] == 1001
|
||
|
||
def test_non_admin_forbidden(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")
|
||
|
||
body = client.patch(
|
||
"/api/v1/auth/users/bob", json={"role": "admin"}, headers=headers
|
||
).json()
|
||
|
||
assert body["code"] == 1006
|
||
|
||
def test_demote_when_multiple_admins_ok(
|
||
self, client: TestClient, auth_stores, admin_headers: dict[str, str]
|
||
) -> None:
|
||
# 存在第二个 admin 时，可降级其中一个
|
||
user_store, _ = auth_stores
|
||
_create_user(user_store, "admin2", "admin2-pass-123", role="admin")
|
||
body = client.patch(
|
||
"/api/v1/auth/users/admin2", json={"role": "user"}, headers=admin_headers
|
||
).json()
|
||
|
||
assert body["code"] == 0
|
||
assert body["data"]["role"] == "user"
|
||
|
||
|
||
class TestDisabledUser:
|
||
"""禁用用户：登录拦截 + 旧 token 失效 + 业务端点拦截"""
|
||
|
||
def test_disabled_user_login_blocked(
|
||
self, client: TestClient, auth_stores, admin_headers: dict[str, str]
|
||
) -> None:
|
||
user_store, _ = auth_stores
|
||
_create_user(user_store, "bob", "bob-pass-123", role="user")
|
||
# 禁用 bob
|
||
client.patch(
|
||
"/api/v1/auth/users/bob", json={"enabled": False}, headers=admin_headers
|
||
)
|
||
# 登录 → 1005 账号已禁用
|
||
body = _login(client, "bob", "bob-pass-123")
|
||
|
||
assert body["code"] == 1005
|
||
assert body["data"] is None
|
||
|
||
def test_disabled_user_old_token_invalidated(
|
||
self, client: TestClient, auth_stores, admin_headers: dict[str, str]
|
||
) -> None:
|
||
user_store, _ = auth_stores
|
||
_create_user(user_store, "bob", "bob-pass-123", role="user")
|
||
# bob 登录拿到 token
|
||
login_body = _login(client, "bob", "bob-pass-123")
|
||
assert login_body["code"] == 0
|
||
bob_headers = {"Authorization": f"Bearer {login_body['data']['token']}"}
|
||
# 禁用 bob → session 清除
|
||
client.patch(
|
||
"/api/v1/auth/users/bob", json={"enabled": False}, headers=admin_headers
|
||
)
|
||
# 旧 token 调业务端点 → 1005（session 已清）
|
||
body = client.get("/api/v1/auth/me", headers=bob_headers).json()
|
||
|
||
assert body["code"] == 1005
|
||
|
||
def test_disabled_user_blocked_even_with_valid_session(
|
||
self, client: TestClient, auth_stores, admin_headers: dict[str, str]
|
||
) -> None:
|
||
"""禁用用户即使持有效 session（直接签发绕过登录），业务端点仍拦截 1005"""
|
||
user_store, session_store = auth_stores
|
||
_create_user(user_store, "bob", "bob-pass-123", role="user")
|
||
# 禁用 bob → session 清除
|
||
client.patch(
|
||
"/api/v1/auth/users/bob", json={"enabled": False}, headers=admin_headers
|
||
)
|
||
# 直接为 bob 签发新 session 绕过登录与清理，验证 get_current_user 的 enabled 拦截
|
||
bob_headers = _headers(session_store, "bob", "user")
|
||
body = client.get("/api/v1/auth/me", headers=bob_headers).json()
|
||
|
||
assert body["code"] == 1005
|
||
|
||
|
||
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"] == []
|