kris/ForcePilot
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

test: 增加路由的集成测试

Browse Source 
- 新增多个集成测试，涵盖认证、聊天、知识库和系统路由的API端点
- 更新测试环境变量配置，确保测试凭据的灵活性
- 移除冗余的测试文件，优化测试结构
This commit is contained in:
Wenjie Zhang 2025-10-11 01:07:45 +08:00
parent d7e0b229c6
commit c72e2cfb90
16 changed files with 430 additions and 1069 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.env.template
View File

@ -37,3 +37,7 @@ NEO4J_URI=
NEO4J_USERNAME=
NEO4J_PASSWORD=
# endregion neo4j
# Servies
YUXI_SUPER_ADMIN_NAME=
YUXI_SUPER_ADMIN_PASSWORD=

13
AGENTS.md
View File

@ -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 来格式化代码。
其他：

9
Makefile
View File

@ -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 .
uv run ruff format --diff .
router-tests:
docker compose exec -T api uv run --group test pytest test/api $(PYTEST_ARGS)

27
docs/changelog/contributing.md
View File

@ -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（容器内）
:::
- 确保所有测试通过
- 添加新功能的测试用例
- 验证现有功能不受影响
- 测试不同环境下的兼容性
<details>
<summary>常用命令</summary>
```bash
# 全量路由测试
make router-tests
# 仅运行知识库相关用例
make router-tests PYTEST_ARGS="-k knowledge_router"
# 不经过 Makefile，直接调用 pytest
uv run --group test pytest test/api -vv
```
</details>
## 许可证
本项目基于 MIT License 开源，贡献的代码将遵循相同的许可证。

4
test/.env.test.example
View File

@ -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

252
test/README.md
View File

@ -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客户端复用
- 异步并发执行
- 智能跳过不可用的测试
- 最小化资源占用

190
test/api/test_auth_api.py
View File

@ -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]

53
test/api/test_auth_router.py Normal file
View File

@ -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

308
test/api/test_chat_api.py
View File

@ -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]

59
test/api/test_chat_router.py Normal file
View File

@ -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

25
test/api/test_dashboard_router.py Normal file
View File

@ -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)

31
test/api/test_graph_router.py Normal file
View File

@ -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)

53
test/api/test_knowledge_router.py Normal file
View File

@ -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

238
test/api/test_system_api.py
View File

@ -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]

42
test/api/test_system_router.py Normal file
View File

@ -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

191
test/conftest.py
View File

@ -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