2a3e18b444
- 新增认证API测试，覆盖登录、权限验证、用户管理等关键流程 - 新增对话API测试，包含智能体调用、线程管理、工具获取及配置查询 - 新增系统API测试，涵盖系统健康检查、信息查询及配置管理权限控制 - 配置pytest支持异步测试、标记分类及详细输出 - 添加测试环境配置示例文件 .env.test.example - 新增测试用例README文档，说明测试架构及运行方法 - 优化Makefile，添加格式化差异检查任务 format_diff - 更新.gitignore，支持忽略测试环境配置文件 .env.test
239 lines
9.6 KiB
Python
239 lines
9.6 KiB
Python
"""
|
|
系统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]
|