diff --git a/.env.template b/.env.template index b6dd1a65..76fdeef3 100644 --- a/.env.template +++ b/.env.template @@ -37,3 +37,7 @@ NEO4J_URI= NEO4J_USERNAME= NEO4J_PASSWORD= # endregion neo4j + +# Servies +YUXI_SUPER_ADMIN_NAME= +YUXI_SUPER_ADMIN_PASSWORD= diff --git a/AGENTS.md b/AGENTS.md index e50f30b5..e7bd452e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -5,17 +5,6 @@ Yuxi-Know 是一个基于知识图谱和向量数据库的智能知识库系统 文档中心在 `docs` 文件夹下面。 - -# 核心服务与容器 (Core Services & Containers) - -项目使用 Docker Compose 管理多个服务，主要容器名称如下： -- `api-dev` - FastAPI 后端服务 -- `web-dev` - Vue.js 前端开发服务器 -- `graph` - Neo4j 图数据库 -- `milvus` - 向量数据库，包含 etcd 和 MinIO 依赖 -- `mineru` - 可选的 MinerU OCR 服务（需要 GPU） -- `paddlex` - 可选的 PaddleX OCR 服务（需要 GPU） - ## 开发与调试工作流 (Development & Debugging Workflow) 本项目完全通过 Docker Compose 进行管理。所有开发和调试都应在运行的容器环境中进行。使用 `docker compose up -d` 命令进行构建和启动。 @@ -32,7 +21,7 @@ Yuxi-Know 是一个基于知识图谱和向量数据库的智能知识库系统 后端开发规范： -- 项目使用 uv 来管理依赖 +- 项目使用 uv 来管理依赖，所以需要使用 uv run 来调试。 - Python 代码要符合 Python 的规范，尽量使用较新的语法，避免使用旧版本的语法（版本兼容到 3.12+），使用 make lint 检查 lint。使用 make format 来格式化代码。 其他： diff --git a/Makefile b/Makefile index 02868e55..19cf1dea 100644 --- a/Makefile +++ b/Makefile @@ -1,5 +1,7 @@ -.PHONY: start logs +.PHONY: start stop logs lint format format_diff router-tests + +PYTEST_ARGS ?= start: docker compose up -d @@ -27,4 +29,7 @@ format: uv run ruff check . --fix format_diff: - uv run ruff format --diff . \ No newline at end of file + uv run ruff format --diff . + +router-tests: + docker compose exec -T api uv run --group test pytest test/api $(PYTEST_ARGS) diff --git a/docs/changelog/contributing.md b/docs/changelog/contributing.md index 1aa97c17..fced7c3b 100644 --- a/docs/changelog/contributing.md +++ b/docs/changelog/contributing.md @@ -60,16 +60,29 @@ chore: 构建过程或辅助工具的变动 ### 测试要求 -::: info - 部分测试脚本待补充 +::: tip 测试 +- `make lint` / `make format` 保持代码整洁 +- `cp test/.env.test.example test/.env.test` 配置测试凭据 +- `make router-tests` 运行集成路由测试，支持 `PYTEST_ARGS="-k chat_router"` +- `uv run --group test pytest test/api` 可直接运行 pytest（容器内） ::: -- 确保所有测试通过 -- 添加新功能的测试用例 -- 验证现有功能不受影响 -- 测试不同环境下的兼容性 +
+常用命令 + +```bash +# 全量路由测试 +make router-tests + +# 仅运行知识库相关用例 +make router-tests PYTEST_ARGS="-k knowledge_router" + +# 不经过 Makefile，直接调用 pytest +uv run --group test pytest test/api -vv +``` + +
## 许可证 本项目基于 MIT License 开源，贡献的代码将遵循相同的许可证。 - diff --git a/server/routers/auth_router.py b/server/routers/auth_router.py index d6288982..7b840a87 100644 --- a/server/routers/auth_router.py +++ b/server/routers/auth_router.py @@ -7,7 +7,7 @@ from pydantic import BaseModel from sqlalchemy.orm import Session from src.storage.db.manager import db_manager -from src.storage.db.models import User +from src.storage.db.models import User, OperationLog from server.utils.auth_middleware import get_admin_user, get_current_user, get_db, get_required_user from server.utils.auth_utils import AuthUtils from server.utils.user_utils import generate_unique_user_id, validate_username, is_valid_phone_number @@ -503,15 +503,18 @@ async def delete_user( detail="不能删除自己的账户", ) - # 记录操作 - log_operation( - db, current_user.id, "删除用户", f"删除用户: {user.username}, ID: {user.id}, 角色: {user.role}", request - ) + deletion_detail = f"删除用户: {user.username}, ID: {user.id}, 角色: {user.role}" + + # 清理关联的操作日志，避免外键约束报错 + db.query(OperationLog).filter(OperationLog.user_id == user.id).delete(synchronize_session=False) # 删除用户 db.delete(user) db.commit() + # 记录操作 + log_operation(db, current_user.id, "删除用户", deletion_detail, request) + return {"success": True, "message": "用户已删除"} diff --git a/src/storage/db/models.py b/src/storage/db/models.py index 15f413e3..29f84e3a 100644 --- a/src/storage/db/models.py +++ b/src/storage/db/models.py @@ -176,7 +176,7 @@ class User(Base): login_locked_until = Column(DateTime, nullable=True) # 锁定到什么时候 # 关联操作日志 - operation_logs = relationship("OperationLog", back_populates="user") + operation_logs = relationship("OperationLog", back_populates="user", cascade="all, delete-orphan") def to_dict(self, include_password=False): result = { diff --git a/test/.env.test.example b/test/.env.test.example index 7f41f8a7..f69467e2 100644 --- a/test/.env.test.example +++ b/test/.env.test.example @@ -5,8 +5,8 @@ TEST_BASE_URL=http://localhost:5050 # 测试用户凭据 -TEST_USERNAME=zwj -TEST_PASSWORD=zwj12138 +TEST_USERNAME= +TEST_PASSWORD= # 数据库连接（如果需要） # DATABASE_URL=sqlite:///test.db diff --git a/test/README.md b/test/README.md deleted file mode 100644 index 51fcbee3..00000000 --- a/test/README.md +++ /dev/null @@ -1,252 +0,0 @@ -# Yuxi-Know API 测试脚本 - -## 概述 - -本测试套件为Yuxi-Know项目提供了全面的API接口测试，涵盖认证、系统管理、对话等核心功能模块。 - -## 测试架构 - -### 测试工具栈 -- **pytest**: Python测试框架 -- **pytest-asyncio**: 异步测试支持 -- **httpx**: 现代HTTP客户端 -- **pytest-cov**: 测试覆盖率分析 - -### 测试模块结构 - -``` -test/ -├── api/ # API接口测试 -│ ├── test_auth_api.py # 认证API测试 -│ ├── test_system_api.py # 系统API测试 -│ └── test_chat_api.py # 对话API测试 -├── conftest.py # pytest配置和公共fixtures -├── run_tests.sh # 测试运行脚本 -├── .env.test.example # 测试环境配置示例 -└── README.md # 测试文档 -``` - -## 快速开始 - -### 1. 准备环境 - -```bash -# 复制测试配置文件 -cp test/.env.test.example test/.env.test - -# 编辑配置文件，设置测试服务器地址和认证信息 -vim test/.env.test -``` - -### 2. 运行测试 - -```bash -# 使用测试脚本（推荐） -./test/run_tests.sh all - -# 或直接使用pytest -uv run pytest test/api/ -v -``` - -## 测试类型 - -### 认证测试 (test_auth_api.py) -- ✅ 健康检查（无需认证） -- ✅ 首次运行状态检查 -- ✅ 登录凭据验证 -- ✅ 用户信息获取 -- ✅ 用户管理权限 -- ✅ 无效令牌处理 - -### 系统测试 (test_system_api.py) -- ✅ 系统健康检查 -- ✅ 系统信息获取 -- ✅ 配置管理（获取/更新） -- ✅ 批量配置更新 -- ✅ 系统日志获取 -- ✅ 权限控制验证 - -### 对话测试 (test_chat_api.py) -- ✅ 智能体列表获取 -- ✅ 简单对话调用 -- ✅ 流式对话测试 -- ✅ 线程管理（创建/更新/删除） -- ✅ 工具列表获取 -- ✅ 智能体配置管理 - -## 运行命令 - -### 使用测试脚本 - -```bash -# 所有测试 -./test/run_tests.sh all - -# 认证测试 -./test/run_tests.sh auth - -# 系统测试 -./test/run_tests.sh system - -# 对话测试 -./test/run_tests.sh chat - -# 快速测试（排除慢速测试） -./test/run_tests.sh quick - -# 检查服务器状态 -./test/run_tests.sh check -``` - -### 使用pytest直接运行 - -```bash -# 运行所有API测试 -uv run pytest test/api/ -v - -# 运行特定测试文件 -uv run pytest test/api/test_auth_api.py -v - -# 运行特定测试方法 -uv run pytest test/api/test_auth_api.py::TestAuthAPI::test_login_valid_credentials -v - -# 运行带标记的测试 -uv run pytest test/api/ -v -m "auth" -uv run pytest test/api/ -v -m "not slow" - -# 生成覆盖率报告 -uv run pytest test/api/ --cov=server --cov=src --cov-report=html -``` - -## 测试标记 - -测试使用pytest标记进行分类： - -- `@pytest.mark.auth`: 需要认证的测试 -- `@pytest.mark.slow`: 慢速测试（如流式对话） -- `@pytest.mark.integration`: 集成测试 - -## 配置说明 - -### 环境变量 (test/.env.test) - -```bash -# 测试服务器地址 -TEST_BASE_URL=http://localhost:5050 - -# 测试用户凭据 -TEST_USERNAME=zwj -TEST_PASSWORD=zwj12138 -``` - -### pytest配置 (pyproject.toml) - -```toml -[tool.pytest.ini_options] -addopts = "-v --tb=short" -testpaths = ["test"] -markers = [ - "auth: marks tests that require authentication", - "slow: marks tests as slow", - "integration: marks tests as integration tests" -] -asyncio_mode = "auto" -asyncio_default_fixture_loop_scope = "session" -``` - -## 测试特性 - -### 智能错误处理 -- 自动检测服务器可用性 -- 优雅处理认证失败 -- 权限不足时跳过相关测试 - -### 异步支持 -- 完整的异步测试支持 -- 高效的并发测试执行 -- 会话级别的资源管理 - -### 灵活配置 -- 环境变量配置 -- 可配置的服务器地址 -- 支持不同环境切换 - -## 故障排除 - -### 常见问题 - -1. **连接失败** - ```bash - # 检查服务器是否运行 - curl http://localhost:5050/api/system/health - - # 检查Docker服务 - docker-compose ps - ``` - -2. **认证失败** - - 检查 `test/.env.test` 中的用户凭据 - - 确认用户存在且密码正确 - - 验证用户权限级别 - -3. **依赖问题** - ```bash - # 重新安装测试依赖 - uv add --group test pytest pytest-asyncio pytest-httpx pytest-cov - ``` - -### 调试选项 - -```bash -# 详细输出 -uv run pytest test/api/ -v -s - -# 显示完整错误信息 -uv run pytest test/api/ --tb=long - -# 停在第一个失败 -uv run pytest test/api/ -x - -# 运行最后失败的测试 -uv run pytest test/api/ --lf -``` - -## 扩展开发 - -### 添加新的测试 - -1. 在 `test/api/` 目录下创建新的测试文件 -2. 继承测试基类或直接编写测试函数 -3. 使用现有的fixtures (test_client, auth_headers) -4. 添加适当的pytest标记 - -### 自定义fixtures - -在 `test/conftest.py` 中添加新的fixtures： - -```python -@pytest_asyncio.fixture -async def custom_fixture(): - # 初始化资源 - yield resource - # 清理资源 -``` - -## 持续集成 - -测试脚本可以轻松集成到CI/CD流水线： - -```yaml -# GitHub Actions示例 -- name: Run API Tests - run: | - cp test/.env.test.example test/.env.test - ./test/run_tests.sh all -``` - -## 性能考虑 - -- 会话级别的HTTP客户端复用 -- 异步并发执行 -- 智能跳过不可用的测试 -- 最小化资源占用 \ No newline at end of file diff --git a/test/api/test_auth_api.py b/test/api/test_auth_api.py deleted file mode 100644 index 92a843a3..00000000 --- a/test/api/test_auth_api.py +++ /dev/null @@ -1,190 +0,0 @@ -""" -认证API测试 -测试用户认证、用户管理等相关接口 -""" - -import pytest -import httpx - - -class TestAuthAPI: - """认证API测试类""" - - @pytest.mark.asyncio - async def test_health_check_no_auth(self, test_client: httpx.AsyncClient): - """测试健康检查接口（无需认证）""" - response = await test_client.get("/api/system/health") - assert response.status_code == 200 - data = response.json() - assert data["status"] == "ok" - assert "message" in data - - @pytest.mark.asyncio - async def test_check_first_run(self, test_client: httpx.AsyncClient): - """测试检查首次运行状态""" - response = await test_client.get("/api/auth/check-first-run") - assert response.status_code == 200 - data = response.json() - assert "first_run" in data - assert isinstance(data["first_run"], bool) - - @pytest.mark.asyncio - async def test_login_invalid_credentials(self, test_client: httpx.AsyncClient): - """测试无效凭据登录""" - login_data = {"username": "invalid_user", "password": "invalid_password"} - response = await test_client.post("/api/auth/token", data=login_data) - # 期望返回401未授权错误 - assert response.status_code == 401 - data = response.json() - assert "detail" in data - - @pytest.mark.asyncio - async def test_login_valid_credentials(self, test_client: httpx.AsyncClient): - """测试有效凭据登录（如果存在管理员用户）""" - # 首先检查是否是首次运行 - first_run_response = await test_client.get("/api/auth/check-first-run") - first_run_data = first_run_response.json() - - if first_run_data.get("first_run", False): - # 如果是首次运行，先初始化管理员 - init_data = {"username": "admin", "password": "admin123"} - init_response = await test_client.post("/api/auth/initialize", json=init_data) - - if init_response.status_code == 200: - init_result = init_response.json() - assert "access_token" in init_result - assert init_result["token_type"] == "bearer" - assert init_result["username"] == "admin" - assert init_result["role"] == "superadmin" - - # 测试用新创建的管理员账户登录 - login_data = {"username": "admin", "password": "admin123"} - login_response = await test_client.post("/api/auth/token", data=login_data) - assert login_response.status_code == 200 - - login_result = login_response.json() - assert "access_token" in login_result - assert login_result["token_type"] == "bearer" - assert login_result["username"] == "admin" - return - - # 如果不是首次运行，尝试用默认凭据登录 - login_data = {"username": "zwj", "password": "zwj12138"} - response = await test_client.post("/api/auth/token", data=login_data) - - # 如果登录成功 - if response.status_code == 200: - data = response.json() - assert "access_token" in data - assert data["token_type"] == "bearer" - assert "username" in data - assert "role" in data - else: - # 如果登录失败，记录信息但不失败测试（可能没有预设用户） - print(f"Login failed with status {response.status_code}: {response.json()}") - - @pytest.mark.asyncio - @pytest.mark.auth - async def test_get_current_user(self, test_client: httpx.AsyncClient, auth_headers: dict): - """测试获取当前用户信息（需要认证）""" - if not auth_headers: - pytest.skip("No auth token available, skipping authenticated test") - - response = await test_client.get("/api/auth/me", headers=auth_headers) - - if response.status_code == 200: - data = response.json() - assert "id" in data - assert "username" in data - assert "role" in data - assert "created_at" in data - else: - # 如果认证失败，记录但不使测试失败 - print(f"Auth test failed with status {response.status_code}: {response.json()}") - - @pytest.mark.asyncio - @pytest.mark.auth - async def test_get_users_list(self, test_client: httpx.AsyncClient, auth_headers: dict): - """测试获取用户列表（需要管理员权限）""" - if not auth_headers: - pytest.skip("No auth token available, skipping authenticated test") - - response = await test_client.get("/api/auth/users", headers=auth_headers) - - if response.status_code == 200: - data = response.json() - assert isinstance(data, list) - # 检查每个用户对象的基本字段 - if len(data) > 0: - user = data[0] - assert "id" in user - assert "username" in user - assert "role" in user - elif response.status_code == 403: - # 权限不足，这是预期的情况之一 - print("User does not have admin privileges") - else: - print(f"Get users test failed with status {response.status_code}: {response.json()}") - - @pytest.mark.asyncio - async def test_unauthorized_access(self, test_client: httpx.AsyncClient): - """测试未授权访问受保护的端点""" - # 尝试不带认证访问受保护的端点 - response = await test_client.get("/api/auth/me") - # 注意：根据实际API行为，可能返回500而不是401 - assert response.status_code in [401, 500] # 接受两种情况 - - response = await test_client.get("/api/auth/users") - assert response.status_code in [401, 500] # 接受两种情况 - - @pytest.mark.asyncio - async def test_invalid_token_access(self, test_client: httpx.AsyncClient): - """测试使用无效token访问受保护的端点""" - invalid_headers = {"Authorization": "Bearer invalid_token_here"} - - response = await test_client.get("/api/auth/me", headers=invalid_headers) - assert response.status_code == 401 - - response = await test_client.get("/api/auth/users", headers=invalid_headers) - assert response.status_code == 401 - - @pytest.mark.asyncio - @pytest.mark.auth - async def test_create_user_permission(self, test_client: httpx.AsyncClient, auth_headers: dict): - """测试创建用户权限（需要管理员权限）""" - if not auth_headers: - pytest.skip("No auth token available, skipping authenticated test") - - # 尝试创建一个测试用户 - user_data = { - "username": f"test_user_{id(self)}", # 使用对象ID作为唯一标识 - "password": "test_password_123", - "role": "user", - } - - response = await test_client.post("/api/auth/users", json=user_data, headers=auth_headers) - - if response.status_code == 200: - data = response.json() - assert data["username"] == user_data["username"] - assert data["role"] == user_data["role"] - assert "id" in data - - # 清理：删除创建的测试用户 - user_id = data["id"] - delete_response = await test_client.delete(f"/api/auth/users/{user_id}", headers=auth_headers) - # 删除可能成功也可能失败，取决于权限，不强制断言 - assert delete_response.status_code in [200, 404] - - elif response.status_code == 400: - # 用户已存在或其他业务逻辑错误 - print(f"Create user failed due to business logic: {response.json()}") - elif response.status_code == 403: - # 权限不足 - print("User does not have permission to create users") - else: - print(f"Create user test failed with status {response.status_code}: {response.json()}") - - -# 添加一些集成测试标记 -pytestmark = [pytest.mark.integration] diff --git a/test/api/test_auth_router.py b/test/api/test_auth_router.py new file mode 100644 index 00000000..4348b8bf --- /dev/null +++ b/test/api/test_auth_router.py @@ -0,0 +1,53 @@ +""" +Integration tests for authentication-related API routes. +""" + +from __future__ import annotations + +import uuid + +import pytest + +pytestmark = [pytest.mark.asyncio, pytest.mark.integration] + + +async def test_login_with_invalid_credentials(test_client): + response = await test_client.post("/api/auth/token", data={"username": "invalid", "password": "invalid"}) + assert response.status_code == 401 + assert "detail" in response.json() + + +async def test_admin_can_login_and_fetch_profile(test_client, admin_headers): + profile_response = await test_client.get("/api/auth/me", headers=admin_headers) + assert profile_response.status_code == 200 + data = profile_response.json() + assert data["role"] in {"admin", "superadmin"} + assert data["username"] + assert data["user_id"] + + +async def test_admin_can_create_and_delete_user(test_client, admin_headers): + suffix = uuid.uuid4().hex[:8] + payload = { + "username": f"rtu_{suffix}", + "password": "routerTest123!", + "role": "user", + } + create_response = await test_client.post("/api/auth/users", json=payload, headers=admin_headers) + assert create_response.status_code == 200, create_response.text + + created_user = create_response.json() + assert created_user["username"] == payload["username"] + assert created_user["role"] == payload["role"] + + delete_response = await test_client.delete(f"/api/auth/users/{created_user['id']}", headers=admin_headers) + assert delete_response.status_code == 200, delete_response.text + delete_payload = delete_response.json() + assert delete_payload["success"] is True + assert delete_payload["message"] == "用户已删除" + + +async def test_invalid_token_is_rejected(test_client): + headers = {"Authorization": "Bearer not-a-real-token"} + response = await test_client.get("/api/auth/me", headers=headers) + assert response.status_code == 401 diff --git a/test/api/test_chat_api.py b/test/api/test_chat_api.py deleted file mode 100644 index c2614dc6..00000000 --- a/test/api/test_chat_api.py +++ /dev/null @@ -1,308 +0,0 @@ -""" -对话API测试 -测试智能体对话、线程管理等相关接口 -""" - -import pytest -import httpx -import json - - -class TestChatAPI: - """对话API测试类""" - - @pytest.mark.asyncio - @pytest.mark.auth - async def test_get_default_agent(self, test_client: httpx.AsyncClient, auth_headers: dict): - """测试获取默认智能体""" - if not auth_headers: - pytest.skip("No auth token available, skipping authenticated test") - - response = await test_client.get("/api/chat/default_agent", headers=auth_headers) - - if response.status_code == 200: - data = response.json() - assert "default_agent_id" in data - print(f"Default agent ID: {data['default_agent_id']}") - else: - print(f"Get default agent failed with status {response.status_code}: {response.json()}") - - @pytest.mark.asyncio - @pytest.mark.auth - async def test_get_agents(self, test_client: httpx.AsyncClient, auth_headers: dict): - """测试获取智能体列表""" - if not auth_headers: - pytest.skip("No auth token available, skipping authenticated test") - - response = await test_client.get("/api/chat/agent", headers=auth_headers) - - if response.status_code == 200: - data = response.json() - assert "agents" in data - assert isinstance(data["agents"], list) - print(f"Found {len(data['agents'])} agents") - if data["agents"]: - agent = data["agents"][0] - print(f"First agent: {agent}") - else: - print(f"Get agents failed with status {response.status_code}: {response.json()}") - - @pytest.mark.asyncio - @pytest.mark.auth - async def test_simple_call(self, test_client: httpx.AsyncClient, auth_headers: dict, test_query: str): - """测试简单对话调用""" - if not auth_headers: - pytest.skip("No auth token available, skipping authenticated test") - - call_data = {"query": test_query, "meta": {"test": True}} - - response = await test_client.post("/api/chat/call", json=call_data, headers=auth_headers) - - if response.status_code == 200: - data = response.json() - assert "response" in data - assert isinstance(data["response"], str) - print(f"Chat response: {data['response'][:100]}...") - else: - # 安全地处理响应 - try: - error_data = response.json() - print(f"Simple call failed with status {response.status_code}: {error_data}") - except Exception: - print(f"Simple call failed with status {response.status_code}: {response.text[:200]}...") - - @pytest.mark.asyncio - @pytest.mark.auth - async def test_unauthorized_chat_access(self, test_client: httpx.AsyncClient): - """测试未授权访问对话接口""" - # 测试简单调用 - response = await test_client.post("/api/chat/call", json={"query": "test"}) - assert response.status_code == 401 - - # 测试获取智能体 - response = await test_client.get("/api/chat/agent") - assert response.status_code == 401 - - # 测试获取默认智能体 - response = await test_client.get("/api/chat/default_agent") - assert response.status_code == 401 - - @pytest.mark.asyncio - @pytest.mark.auth - @pytest.mark.slow - async def test_agent_chat_streaming(self, test_client: httpx.AsyncClient, auth_headers: dict): - """测试智能体流式对话""" - if not auth_headers: - pytest.skip("No auth token available, skipping authenticated test") - - # 首先获取可用的智能体 - agents_response = await test_client.get("/api/chat/agent", headers=auth_headers) - - if agents_response.status_code != 200: - pytest.skip("Cannot get agents list, skipping agent chat test") - - agents_data = agents_response.json() - if not agents_data.get("agents"): - pytest.skip("No agents available, skipping agent chat test") - - agent_id = agents_data["agents"][0].get("id") - if not agent_id: - pytest.skip("No valid agent ID found, skipping agent chat test") - - # 测试智能体对话 - chat_data = {"query": "Hello, this is a test message", "config": {"test": True}, "meta": {"test": True}} - - response = await test_client.post(f"/api/chat/agent/{agent_id}", json=chat_data, headers=auth_headers) - - if response.status_code == 200: - # 处理流式响应 - content = await response.aread() - lines = content.decode("utf-8").strip().split("\n") - - # 验证流式响应格式 - for i, line in enumerate(lines[:3]): # 只检查前几行 - try: - json_data = json.loads(line) - assert "status" in json_data - print(f"Stream line {i}: status={json_data.get('status')}") - except json.JSONDecodeError: - print(f"Invalid JSON in line {i}: {line[:50]}...") - - print(f"Received {len(lines)} stream lines from agent {agent_id}") - else: - print(f"Agent chat failed with status {response.status_code}: {response.json()}") - - @pytest.mark.asyncio - @pytest.mark.auth - async def test_create_thread(self, test_client: httpx.AsyncClient, auth_headers: dict): - """测试创建对话线程""" - if not auth_headers: - pytest.skip("No auth token available, skipping authenticated test") - - # 获取可用的智能体 - agents_response = await test_client.get("/api/chat/agent", headers=auth_headers) - - if agents_response.status_code != 200: - pytest.skip("Cannot get agents list, skipping thread test") - - agents_data = agents_response.json() - if not agents_data.get("agents"): - pytest.skip("No agents available, skipping thread test") - - agent_id = agents_data["agents"][0].get("id") - if not agent_id: - pytest.skip("No valid agent ID found, skipping thread test") - - # 创建线程 - thread_data = {"title": "测试对话线程", "agent_id": agent_id, "description": "这是一个测试创建的对话线程"} - - response = await test_client.post("/api/chat/thread", json=thread_data, headers=auth_headers) - - if response.status_code == 200: - data = response.json() - assert "id" in data - assert data["agent_id"] == agent_id - assert data["title"] == thread_data["title"] - print(f"Created thread: {data['id']}") - - # 测试获取线程列表 - threads_response = await test_client.get(f"/api/chat/threads?agent_id={agent_id}", headers=auth_headers) - - if threads_response.status_code == 200: - threads_data = threads_response.json() - assert isinstance(threads_data, list) - # 验证我们创建的线程在列表中 - thread_ids = [t["id"] for t in threads_data] - assert data["id"] in thread_ids - print(f"Found {len(threads_data)} threads for agent {agent_id}") - - # 清理：删除创建的线程 - delete_response = await test_client.delete(f"/api/chat/thread/{data['id']}", headers=auth_headers) - if delete_response.status_code == 200: - print(f"Successfully deleted thread {data['id']}") - - else: - print(f"Create thread failed with status {response.status_code}: {response.json()}") - - @pytest.mark.asyncio - @pytest.mark.auth - async def test_get_tools(self, test_client: httpx.AsyncClient, auth_headers: dict): - """测试获取工具列表""" - if not auth_headers: - pytest.skip("No auth token available, skipping authenticated test") - - # 获取可用的智能体 - agents_response = await test_client.get("/api/chat/agent", headers=auth_headers) - - if agents_response.status_code != 200: - pytest.skip("Cannot get agents list, skipping tools test") - - agents_data = agents_response.json() - if not agents_data.get("agents"): - pytest.skip("No agents available, skipping tools test") - - agent_id = agents_data["agents"][0].get("id") - if not agent_id: - pytest.skip("No valid agent ID found, skipping tools test") - - response = await test_client.get(f"/api/chat/tools?agent_id={agent_id}", headers=auth_headers) - - if response.status_code == 200: - data = response.json() - assert "tools" in data - assert isinstance(data["tools"], dict) - print(f"Found {len(data['tools'])} tools for agent {agent_id}") - if data["tools"]: - tool_names = list(data["tools"].keys()) - print(f"Available tools: {tool_names[:5]}...") # 显示前5个工具名 - else: - print(f"Get tools failed with status {response.status_code}: {response.json()}") - - @pytest.mark.asyncio - @pytest.mark.auth - async def test_agent_config(self, test_client: httpx.AsyncClient, auth_headers: dict): - """测试智能体配置获取""" - if not auth_headers: - pytest.skip("No auth token available, skipping authenticated test") - - # 获取可用的智能体 - agents_response = await test_client.get("/api/chat/agent", headers=auth_headers) - - if agents_response.status_code != 200: - pytest.skip("Cannot get agents list, skipping config test") - - agents_data = agents_response.json() - if not agents_data.get("agents"): - pytest.skip("No agents available, skipping config test") - - agent_id = agents_data["agents"][0].get("id") - if not agent_id: - pytest.skip("No valid agent ID found, skipping config test") - - response = await test_client.get(f"/api/chat/agent/{agent_id}/config", headers=auth_headers) - - if response.status_code == 200: - data = response.json() - assert "success" in data - assert "config" in data - print(f"Agent {agent_id} config keys: {list(data['config'].keys()) if data['config'] else 'None'}") - else: - print(f"Get agent config failed with status {response.status_code}: {response.json()}") - - @pytest.mark.asyncio - @pytest.mark.auth - async def test_thread_management_full_cycle(self, test_client: httpx.AsyncClient, auth_headers: dict): - """测试完整的线程管理周期：创建、更新、删除""" - if not auth_headers: - pytest.skip("No auth token available, skipping authenticated test") - - # 获取智能体 - agents_response = await test_client.get("/api/chat/agent", headers=auth_headers) - if agents_response.status_code != 200: - pytest.skip("Cannot get agents list") - - agents_data = agents_response.json() - if not agents_data.get("agents"): - pytest.skip("No agents available") - - agent_id = agents_data["agents"][0].get("id") - if not agent_id: - pytest.skip("No valid agent ID found") - - # 1. 创建线程 - thread_data = {"title": "完整测试线程", "agent_id": agent_id, "description": "用于完整周期测试的线程"} - - create_response = await test_client.post("/api/chat/thread", json=thread_data, headers=auth_headers) - assert create_response.status_code == 200 - thread = create_response.json() - thread_id = thread["id"] - - # 2. 更新线程 - update_data = {"title": "更新后的标题", "description": "更新后的描述"} - - update_response = await test_client.put(f"/api/chat/thread/{thread_id}", json=update_data, headers=auth_headers) - - if update_response.status_code == 200: - updated_thread = update_response.json() - assert updated_thread["title"] == update_data["title"] - assert updated_thread["description"] == update_data["description"] - print(f"Successfully updated thread {thread_id}") - - # 3. 删除线程 - delete_response = await test_client.delete(f"/api/chat/thread/{thread_id}", headers=auth_headers) - - if delete_response.status_code == 200: - print(f"Successfully deleted thread {thread_id}") - - # 验证线程确实被删除（不应该再出现在列表中） - threads_response = await test_client.get(f"/api/chat/threads?agent_id={agent_id}", headers=auth_headers) - - if threads_response.status_code == 200: - threads_data = threads_response.json() - thread_ids = [t["id"] for t in threads_data] - assert thread_id not in thread_ids - print("Confirmed thread deletion") - - -# 标记为集成测试 -pytestmark = [pytest.mark.integration] diff --git a/test/api/test_chat_router.py b/test/api/test_chat_router.py new file mode 100644 index 00000000..aabebab5 --- /dev/null +++ b/test/api/test_chat_router.py @@ -0,0 +1,59 @@ +""" +Integration tests for chat router endpoints. +""" + +from __future__ import annotations + +import pytest + +pytestmark = [pytest.mark.asyncio, pytest.mark.integration] + + +async def test_chat_endpoints_require_authentication(test_client): + assert (await test_client.get("/api/chat/default_agent")).status_code == 401 + assert (await test_client.get("/api/chat/agent")).status_code == 401 + + +async def test_admin_can_list_agents(test_client, admin_headers): + response = await test_client.get("/api/chat/agent", headers=admin_headers) + assert response.status_code == 200, response.text + payload = response.json() + assert isinstance(payload["agents"], list) + assert "metadata" in payload + + +async def test_admin_can_read_default_agent(test_client, admin_headers): + response = await test_client.get("/api/chat/default_agent", headers=admin_headers) + assert response.status_code == 200, response.text + assert "default_agent_id" in response.json() + + +async def test_setting_default_agent_requires_admin(test_client, admin_headers, standard_user): + # Attempt as admin first to obtain a candidate agent id. + agents_response = await test_client.get("/api/chat/agent", headers=admin_headers) + assert agents_response.status_code == 200, agents_response.text + agents = agents_response.json().get("agents", []) + + if not agents: + pytest.skip("No agents are registered in the system.") + + candidate_agent_id = agents[0].get("id") + if not candidate_agent_id: + pytest.skip("Agent payload missing id field.") + + # Normal user should not be able to update the default agent. + forbidden_response = await test_client.post( + "/api/chat/set_default_agent", + json={"agent_id": candidate_agent_id}, + headers=standard_user["headers"], + ) + assert forbidden_response.status_code == 403 + + # Admin should succeed. + update_response = await test_client.post( + "/api/chat/set_default_agent", + json={"agent_id": candidate_agent_id}, + headers=admin_headers, + ) + assert update_response.status_code == 200, update_response.text + assert update_response.json()["default_agent_id"] == candidate_agent_id diff --git a/test/api/test_dashboard_router.py b/test/api/test_dashboard_router.py new file mode 100644 index 00000000..d0557862 --- /dev/null +++ b/test/api/test_dashboard_router.py @@ -0,0 +1,25 @@ +""" +Integration tests for dashboard router endpoints. +""" + +from __future__ import annotations + +import pytest + +pytestmark = [pytest.mark.asyncio, pytest.mark.integration] + + +async def test_dashboard_requires_authentication(test_client): + response = await test_client.get("/api/dashboard/conversations") + assert response.status_code == 401 + + +async def test_standard_user_is_forbidden(test_client, standard_user): + response = await test_client.get("/api/dashboard/conversations", headers=standard_user["headers"]) + assert response.status_code == 403 + + +async def test_admin_can_fetch_conversations(test_client, admin_headers): + response = await test_client.get("/api/dashboard/conversations", headers=admin_headers) + assert response.status_code == 200, response.text + assert isinstance(response.json(), list) diff --git a/test/api/test_graph_router.py b/test/api/test_graph_router.py new file mode 100644 index 00000000..34b492db --- /dev/null +++ b/test/api/test_graph_router.py @@ -0,0 +1,31 @@ +""" +Integration tests for graph router endpoints. +""" + +from __future__ import annotations + +import pytest + +pytestmark = [pytest.mark.asyncio, pytest.mark.integration] + + +async def test_graph_routes_require_auth(test_client): + response = await test_client.get("/api/graph/lightrag/databases") + assert response.status_code == 401 + + +async def test_standard_user_cannot_access_graph_endpoints(test_client, standard_user): + response = await test_client.get( + "/api/graph/lightrag/databases", headers=standard_user["headers"] + ) + assert response.status_code == 403 + + +async def test_admin_can_list_lightrag_databases(test_client, admin_headers, knowledge_database): + response = await test_client.get("/api/graph/lightrag/databases", headers=admin_headers) + assert response.status_code == 200, response.text + payload = response.json() + assert payload["success"] is True + databases = payload["data"]["databases"] + assert isinstance(databases, list) + assert any(db["db_id"] == knowledge_database["db_id"] for db in databases) diff --git a/test/api/test_knowledge_router.py b/test/api/test_knowledge_router.py new file mode 100644 index 00000000..932153c4 --- /dev/null +++ b/test/api/test_knowledge_router.py @@ -0,0 +1,53 @@ +""" +Integration tests for knowledge router endpoints. +""" + +from __future__ import annotations + +import pytest + +pytestmark = [pytest.mark.asyncio, pytest.mark.integration] + + +async def test_admin_can_manage_knowledge_databases(test_client, admin_headers, knowledge_database): + db_id = knowledge_database["db_id"] + + list_response = await test_client.get("/api/knowledge/databases", headers=admin_headers) + assert list_response.status_code == 200, list_response.text + databases = list_response.json().get("databases", []) + assert any(entry["db_id"] == db_id for entry in databases) + + get_response = await test_client.get(f"/api/knowledge/databases/{db_id}", headers=admin_headers) + assert get_response.status_code == 200, get_response.text + assert get_response.json()["db_id"] == db_id + + update_response = await test_client.put( + f"/api/knowledge/databases/{db_id}", + json={"name": knowledge_database["name"], "description": "Updated by pytest"}, + headers=admin_headers, + ) + assert update_response.status_code == 200, update_response.text + assert update_response.json()["database"]["description"] == "Updated by pytest" + + +async def test_knowledge_routes_enforce_permissions(test_client, standard_user, knowledge_database): + db_id = knowledge_database["db_id"] + + forbidden_create = await test_client.post( + "/api/knowledge/databases", + json={ + "database_name": "unauthorized_db", + "description": "Should not succeed", + "embed_model_name": "siliconflow/BAAI/bge-m3", + }, + headers=standard_user["headers"], + ) + assert forbidden_create.status_code == 403 + + forbidden_list = await test_client.get("/api/knowledge/databases", headers=standard_user["headers"]) + assert forbidden_list.status_code == 403 + + forbidden_get = await test_client.get( + f"/api/knowledge/databases/{db_id}", headers=standard_user["headers"] + ) + assert forbidden_get.status_code == 403 diff --git a/test/api/test_system_api.py b/test/api/test_system_api.py deleted file mode 100644 index fdfac250..00000000 --- a/test/api/test_system_api.py +++ /dev/null @@ -1,238 +0,0 @@ -""" -系统API测试 -测试系统管理、配置管理等相关接口 -""" - -import pytest -import httpx - - -class TestSystemAPI: - """系统API测试类""" - - @pytest.mark.asyncio - async def test_health_check(self, test_client: httpx.AsyncClient): - """测试系统健康检查接口""" - response = await test_client.get("/api/system/health") - assert response.status_code == 200 - - data = response.json() - assert data["status"] == "ok" - assert "message" in data - print(f"Health check response: {data}") - - @pytest.mark.asyncio - async def test_get_system_info(self, test_client: httpx.AsyncClient): - """测试获取系统信息接口（公开接口）""" - response = await test_client.get("/api/system/info") - - if response.status_code == 200: - data = response.json() - assert "success" in data - assert "data" in data - print(f"System info response: {data}") - else: - # 如果接口不存在或有其他问题，记录但不失败 - print(f"System info request failed with status {response.status_code}: {response.json()}") - - @pytest.mark.asyncio - @pytest.mark.auth - async def test_get_config_unauthorized(self, test_client: httpx.AsyncClient): - """测试未授权访问配置接口""" - response = await test_client.get("/api/system/config") - assert response.status_code == 401 - print("Config access properly requires authentication") - - @pytest.mark.asyncio - @pytest.mark.auth - async def test_get_config_authorized(self, test_client: httpx.AsyncClient, auth_headers: dict): - """测试已授权访问配置接口""" - if not auth_headers: - pytest.skip("No auth token available, skipping authenticated test") - - response = await test_client.get("/api/system/config", headers=auth_headers) - - if response.status_code == 200: - data = response.json() - # 验证配置数据结构 - assert isinstance(data, dict) - # 检查一些常见的配置项 - expected_config_keys = [ - "enable_reranker", - "enable_web_search", - "model_provider", - "model_name", - "embed_model", - ] - for key in expected_config_keys: - if key in data: - print(f"Config contains {key}: {data[key]}") - print(f"Config keys: {list(data.keys())}") - elif response.status_code == 403: - print("User does not have admin privileges to access config") - else: - print(f"Get config failed with status {response.status_code}: {response.json()}") - - @pytest.mark.asyncio - @pytest.mark.auth - async def test_update_config_unauthorized(self, test_client: httpx.AsyncClient): - """测试未授权更新配置""" - config_data = {"key": "test_key", "value": "test_value"} - response = await test_client.post("/api/system/config", json=config_data) - assert response.status_code == 401 - - @pytest.mark.asyncio - @pytest.mark.auth - @pytest.mark.slow - async def test_update_single_config(self, test_client: httpx.AsyncClient, auth_headers: dict): - """测试更新单个配置项""" - if not auth_headers: - pytest.skip("No auth token available, skipping authenticated test") - - # 先获取当前配置 - get_response = await test_client.get("/api/system/config", headers=auth_headers) - - if get_response.status_code != 200: - pytest.skip("Cannot access config, skipping update test") - - original_config = get_response.json() - - # 尝试更新一个安全的配置项（如果存在的话） - test_config_updates = [ - {"key": "enable_web_search", "value": True}, - {"key": "enable_reranker", "value": False}, - ] - - for config_update in test_config_updates: - key = config_update["key"] - if key in original_config: - original_value = original_config[key] - new_value = config_update["value"] - - # 更新配置 - response = await test_client.post("/api/system/config", json=config_update, headers=auth_headers) - - if response.status_code == 200: - updated_config = response.json() - assert updated_config[key] == new_value - print(f"Successfully updated {key} from {original_value} to {new_value}") - - # 恢复原值 - restore_data = {"key": key, "value": original_value} - restore_response = await test_client.post( - "/api/system/config", json=restore_data, headers=auth_headers - ) - if restore_response.status_code == 200: - print(f"Successfully restored {key} to {original_value}") - break - elif response.status_code == 403: - print("User does not have permission to update config") - break - else: - print(f"Config update failed with status {response.status_code}: {response.json()}") - - @pytest.mark.asyncio - @pytest.mark.auth - @pytest.mark.slow - async def test_batch_update_config(self, test_client: httpx.AsyncClient, auth_headers: dict): - """测试批量更新配置项""" - if not auth_headers: - pytest.skip("No auth token available, skipping authenticated test") - - # 准备批量更新数据 - batch_updates = {"enable_web_search": True, "enable_reranker": False} - - response = await test_client.post("/api/system/config/update", json=batch_updates, headers=auth_headers) - - if response.status_code == 200: - updated_config = response.json() - for key, value in batch_updates.items(): - if key in updated_config: - assert updated_config[key] == value - print(f"Batch update verified: {key} = {value}") - elif response.status_code == 403: - print("User does not have permission to batch update config") - else: - print(f"Batch config update failed with status {response.status_code}: {response.json()}") - - @pytest.mark.asyncio - @pytest.mark.auth - async def test_get_system_logs(self, test_client: httpx.AsyncClient, auth_headers: dict): - """测试获取系统日志""" - if not auth_headers: - pytest.skip("No auth token available, skipping authenticated test") - - response = await test_client.get("/api/system/logs", headers=auth_headers) - - if response.status_code == 200: - data = response.json() - assert "log" in data - assert "message" in data - assert data["message"] == "success" - assert "log_file" in data - print(f"Log file path: {data.get('log_file')}") - print(f"Log length: {len(data.get('log', ''))}") - elif response.status_code == 403: - print("User does not have permission to access logs") - else: - print(f"Get logs failed with status {response.status_code}: {response.json()}") - - @pytest.mark.asyncio - @pytest.mark.auth - async def test_restart_system_permission(self, test_client: httpx.AsyncClient, auth_headers: dict): - """测试系统重启权限（仅超级管理员）""" - if not auth_headers: - pytest.skip("No auth token available, skipping authenticated test") - - response = await test_client.post("/api/system/restart", headers=auth_headers) - - if response.status_code == 200: - data = response.json() - assert "message" in data - print(f"Restart response: {data}") - elif response.status_code == 403: - print("User does not have superadmin privileges to restart system") - else: - print(f"Restart request failed with status {response.status_code}: {response.json()}") - - @pytest.mark.asyncio - @pytest.mark.auth - async def test_reload_info_config(self, test_client: httpx.AsyncClient, auth_headers: dict): - """测试重新加载信息配置""" - if not auth_headers: - pytest.skip("No auth token available, skipping authenticated test") - - response = await test_client.post("/api/system/info/reload", headers=auth_headers) - - if response.status_code == 200: - data = response.json() - assert "success" in data - assert "message" in data - print(f"Info reload response: {data}") - elif response.status_code == 403: - print("User does not have permission to reload info config") - else: - print(f"Info reload failed with status {response.status_code}: {response.json()}") - - @pytest.mark.asyncio - @pytest.mark.auth - async def test_get_ocr_stats(self, test_client: httpx.AsyncClient, auth_headers: dict): - """测试获取OCR统计信息""" - if not auth_headers: - pytest.skip("No auth token available, skipping authenticated test") - - response = await test_client.get("/api/system/ocr/stats", headers=auth_headers) - - if response.status_code == 200: - data = response.json() - print(f"OCR stats response: {data}") - elif response.status_code == 403: - print("User does not have permission to access OCR stats") - elif response.status_code == 404: - print("OCR stats endpoint not available") - else: - print(f"OCR stats request failed with status {response.status_code}: {response.json()}") - - -# 标记为集成测试 -pytestmark = [pytest.mark.integration] diff --git a/test/api/test_system_router.py b/test/api/test_system_router.py new file mode 100644 index 00000000..ae7d19db --- /dev/null +++ b/test/api/test_system_router.py @@ -0,0 +1,42 @@ +""" +Integration tests for system router endpoints. +""" + +from __future__ import annotations + +import pytest + +pytestmark = [pytest.mark.asyncio, pytest.mark.integration] + + +async def test_health_endpoint_is_public(test_client): + response = await test_client.get("/api/system/health") + assert response.status_code == 200 + assert response.json()["status"] == "ok" + + +async def test_info_endpoint_is_public(test_client): + response = await test_client.get("/api/system/info") + assert response.status_code == 200 + payload = response.json() + assert payload["success"] is True + assert "data" in payload + + +async def test_config_endpoints_require_admin(test_client, standard_user): + assert (await test_client.get("/api/system/config")).status_code == 401 + assert ( + await test_client.get("/api/system/config", headers=standard_user["headers"]) + ).status_code == 403 + + +async def test_admin_can_fetch_config_and_reload_info(test_client, admin_headers): + config_response = await test_client.get("/api/system/config", headers=admin_headers) + assert config_response.status_code == 200, config_response.text + assert isinstance(config_response.json(), dict) + + reload_response = await test_client.post("/api/system/info/reload", headers=admin_headers) + assert reload_response.status_code == 200, reload_response.text + reload_payload = reload_response.json() + assert reload_payload["success"] is True + assert "data" in reload_payload diff --git a/test/conftest.py b/test/conftest.py index e4a26c49..e160bca0 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -1,84 +1,159 @@ """ -pytest配置文件和公共fixtures -提供测试环境配置和公共的测试工具 +Shared pytest fixtures for exercising FastAPI routers over the running API service. """ +from __future__ import annotations + import os +import uuid +from collections.abc import AsyncGenerator + +import httpx import pytest import pytest_asyncio -import httpx -from collections.abc import AsyncGenerator from dotenv import load_dotenv -# 加载测试环境变量 -load_dotenv("test/.env.test") +# Load project and test specific environment variables. +load_dotenv(".env", override=False) +load_dotenv("test/.env.test", override=False) -# 测试配置 -TEST_BASE_URL = os.getenv("TEST_BASE_URL", "http://localhost:5050") -TEST_USERNAME = os.getenv("TEST_USERNAME", "zwj") -TEST_PASSWORD = os.getenv("TEST_PASSWORD", "zwj12138") +API_BASE_URL = os.getenv("TEST_BASE_URL", "http://localhost:5050").rstrip("/") +ADMIN_LOGIN = os.getenv("TEST_USERNAME") +ADMIN_PASSWORD = os.getenv("TEST_PASSWORD") + +assert ADMIN_LOGIN, "TEST_USERNAME is not set" +assert ADMIN_PASSWORD, "TEST_PASSWORD is not set" + +_ADMIN_TOKEN_CACHE: str | None = None +HTTP_TIMEOUT = httpx.Timeout(30.0, connect=5.0) @pytest_asyncio.fixture(scope="function") async def test_client() -> AsyncGenerator[httpx.AsyncClient, None]: - """创建异步HTTP客户端""" - timeout = httpx.Timeout(30.0, connect=5.0) - async with httpx.AsyncClient(base_url=TEST_BASE_URL, timeout=timeout, follow_redirects=True) as client: + """Async HTTP client bound to the live API base URL.""" + async with httpx.AsyncClient(base_url=API_BASE_URL, timeout=HTTP_TIMEOUT, follow_redirects=True) as client: yield client @pytest_asyncio.fixture(scope="function") -async def auth_token(test_client: httpx.AsyncClient) -> str: - """获取认证令牌用于需要认证的测试""" +async def admin_token() -> str: + """Authenticate with super admin credentials and cache the bearer token.""" + global _ADMIN_TOKEN_CACHE + + if _ADMIN_TOKEN_CACHE: + return _ADMIN_TOKEN_CACHE + + if not ADMIN_LOGIN or not ADMIN_PASSWORD: + pytest.skip("Admin credentials are not configured via environment variables.") + + async with httpx.AsyncClient( + base_url=API_BASE_URL, timeout=HTTP_TIMEOUT, follow_redirects=True + ) as bootstrap_client: + response = await bootstrap_client.post( + "/api/auth/token", data={"username": ADMIN_LOGIN, "password": ADMIN_PASSWORD} + ) + + if response.status_code == 401: + first_run_response = await bootstrap_client.get("/api/auth/check-first-run") + if first_run_response.status_code == 200 and first_run_response.json().get("first_run", False): + pytest.fail( + "Super admin account has not been initialized. Please complete `/api/auth/initialize` " + "before running router tests." + ) + + if response.status_code != 200: + pytest.fail(f"Failed to authenticate as admin (status={response.status_code}): {response.text}") + + token = response.json().get("access_token") + if not token: + pytest.fail("Admin authentication did not return an access token.") + + _ADMIN_TOKEN_CACHE = token + return token + + +@pytest.fixture(scope="function") +def admin_headers(admin_token: str) -> dict[str, str]: + """Authorization headers for the super admin user.""" + return {"Authorization": f"Bearer {admin_token}"} + + +@pytest_asyncio.fixture(scope="function") +async def standard_user(test_client: httpx.AsyncClient, admin_headers: dict[str, str]) -> dict: + """ + Provision a temporary standard user for permission checks. + + Yields a dictionary with `user`, `password`, and `headers` keys. + """ + username = f"pytest_user_{uuid.uuid4().hex[:8]}" + password = f"Pw!{uuid.uuid4().hex[:8]}" + + response = await test_client.post( + "/api/auth/users", + json={"username": username, "password": password, "role": "user"}, + headers=admin_headers, + ) + if response.status_code != 200: + pytest.fail(f"Failed to create standard user (status={response.status_code}): {response.text}") + + user_payload = response.json() + login_response = await test_client.post( + "/api/auth/token", + data={"username": user_payload["user_id"], "password": password}, + ) + if login_response.status_code != 200: + pytest.fail( + f"Failed to authenticate as standard user (status={login_response.status_code}): {login_response.text}" + ) + + access_token = login_response.json().get("access_token") + if not access_token: + pytest.fail("Standard user login succeeded but no access token was returned.") + try: - # 尝试登录获取token - login_data = {"username": TEST_USERNAME, "password": TEST_PASSWORD} - response = await test_client.post("/api/auth/token", data=login_data) - - if response.status_code == 200: - data = response.json() - return data.get("access_token", "") - else: - # 如果登录失败，返回空token，某些测试可能不需要认证 - print(f"Login failed with status {response.status_code}, continuing without auth token") - return "" - except Exception as e: - print(f"Auth setup failed: {e}, continuing without auth token") - return "" + yield { + "user": user_payload, + "password": password, + "headers": {"Authorization": f"Bearer {access_token}"}, + } + finally: + response = await test_client.delete(f"/api/auth/users/{user_payload['id']}", headers=admin_headers) + assert response.status_code == 200, f"Failed to cleanup test user {user_payload['user_id']}: {response.text}" -@pytest_asyncio.fixture -async def auth_headers(auth_token: str) -> dict: - """返回包含认证信息的headers""" - if auth_token: - return {"Authorization": f"Bearer {auth_token}"} - return {} +@pytest_asyncio.fixture(scope="function") +async def knowledge_database(test_client: httpx.AsyncClient, admin_headers: dict[str, str]) -> dict: + """ + Create a temporary knowledge database for tests that need LightRAG metadata. + """ + db_name = f"pytest_kb_{uuid.uuid4().hex[:6]}" + create_response = await test_client.post( + "/api/knowledge/databases", + json={ + "database_name": db_name, + "description": "Pytest managed knowledge base", + "embed_model_name": "siliconflow/BAAI/bge-m3", + "kb_type": "lightrag", + "additional_params": {}, + }, + headers=admin_headers, + ) + if create_response.status_code != 200: + pytest.fail(f"Failed to create knowledge database (status={create_response.status_code}): {create_response.text}") + + db_payload = create_response.json() + db_id = db_payload["db_id"] + + try: + yield db_payload + finally: + await test_client.delete(f"/api/knowledge/databases/{db_id}", headers=admin_headers) -@pytest.fixture -def test_query(): - """测试用的简单查询""" - return "你好，这是一个测试查询" - - -@pytest.fixture -def test_chat_payload(test_query: str): - """测试对话的请求负载""" - return {"query": test_query, "meta": {"test": True}} - - -# pytest配置选项 -def pytest_configure(config): - """pytest配置""" +def pytest_configure(config: pytest.Config) -> None: + """Register commonly used custom markers.""" config.addinivalue_line("markers", "auth: marks tests that require authentication") - config.addinivalue_line("markers", "slow: marks tests as slow") - config.addinivalue_line("markers", "integration: marks tests as integration tests") + config.addinivalue_line("markers", "integration: marks tests that hit the live API service") -# 设置异步模式 pytest_plugins = ["pytest_asyncio"] - -# 测试标记 -pytest.mark.auth = pytest.mark.auth -pytest.mark.slow = pytest.mark.slow -pytest.mark.integration = pytest.mark.integration