"""用户体系集成验证：内存环境全链路 内存 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