f9f4560459
- FastAPI backend with SQLModel, Alembic migrations, AgentScope agents - Next.js 15 frontend with React 19, Tailwind, Zustand, React Flow - Multi-provider AI system (DashScope, Kling, MiniMax, Volcengine, OpenAI, etc.) - All HTTP clients migrated from sync requests to async httpx - Admin-managed API keys via environment variables - SSRF vulnerability fixed in ensure_url()
242 lines
7.3 KiB
Python
242 lines
7.3 KiB
Python
"""
|
||
集成测试
|
||
|
||
测试完整的请求流程和组件集成
|
||
"""
|
||
import pytest
|
||
import sys
|
||
import os
|
||
|
||
# 添加项目根目录到路径
|
||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
|
||
|
||
from fastapi.testclient import TestClient
|
||
from src.main import app
|
||
from src.utils.errors import ErrorCode
|
||
|
||
client = TestClient(app)
|
||
|
||
|
||
def unwrap_response_data(response):
|
||
payload = response.json()
|
||
return payload.get("data", payload)
|
||
|
||
|
||
class TestHealthEndpoints:
|
||
"""健康检查端点测试"""
|
||
|
||
def test_basic_health_check(self):
|
||
"""测试基础健康检查"""
|
||
response = client.get("/health")
|
||
assert response.status_code == 200
|
||
|
||
data = unwrap_response_data(response)
|
||
assert data["status"] == "healthy"
|
||
assert "service" in data
|
||
assert "timestamp" in data
|
||
assert "uptime_seconds" in data
|
||
|
||
def test_detailed_health_check(self):
|
||
"""测试详细健康检查"""
|
||
response = client.get("/health/detailed")
|
||
assert response.status_code == 200
|
||
|
||
data = unwrap_response_data(response)
|
||
assert "status" in data
|
||
assert "components" in data
|
||
assert "database" in data["components"]
|
||
assert "task_manager" in data["components"]
|
||
|
||
def test_liveness_probe(self):
|
||
"""测试存活探针"""
|
||
response = client.get("/health/live")
|
||
assert response.status_code == 200
|
||
|
||
data = unwrap_response_data(response)
|
||
assert data["status"] == "alive"
|
||
|
||
def test_readiness_probe(self):
|
||
"""测试就绪探针"""
|
||
response = client.get("/health/ready")
|
||
# 可能返回 200 或 503，取决于组件状态
|
||
assert response.status_code in [200, 503]
|
||
|
||
def test_metrics_endpoint(self):
|
||
"""测试 Prometheus 指标端点"""
|
||
response = client.get("/metrics")
|
||
assert response.status_code == 200
|
||
assert "text/plain" in response.headers["content-type"]
|
||
|
||
# 检查是否包含一些关键指标
|
||
content = response.text
|
||
assert "task_created_total" in content or "http_request" in content
|
||
|
||
|
||
class TestErrorHandling:
|
||
"""错误处理测试"""
|
||
|
||
def test_404_not_found(self):
|
||
"""测试 404 错误"""
|
||
response = client.get("/api/projects/non-existent-id-12345")
|
||
# 项目不存在会返回 404（HTTPException）
|
||
assert response.status_code == 404
|
||
|
||
data = response.json()
|
||
assert "detail" in data or "message" in data
|
||
|
||
def test_invalid_parameter(self):
|
||
"""测试参数验证错误"""
|
||
# 测试一个需要参数的端点
|
||
response = client.post("/api/v1/generations/image", json={
|
||
# 缺少必需的 prompt 参数
|
||
})
|
||
|
||
# 应该返回 422（Pydantic 验证错误）
|
||
assert response.status_code == 422
|
||
|
||
def test_method_not_allowed(self):
|
||
"""测试方法不允许"""
|
||
response = client.put("/health") # health 只支持 GET
|
||
assert response.status_code == 405
|
||
|
||
|
||
class TestConfigEndpoints:
|
||
"""配置端点测试"""
|
||
|
||
def test_get_system_config(self):
|
||
"""测试获取系统配置"""
|
||
response = client.get("/api/v1/config/system")
|
||
assert response.status_code == 200
|
||
|
||
def test_get_models_config(self):
|
||
"""测试获取模型配置"""
|
||
response = client.get("/api/v1/config/models")
|
||
assert response.status_code == 200
|
||
|
||
payload = unwrap_response_data(response)
|
||
assert isinstance(payload, dict)
|
||
|
||
def test_get_defaults(self):
|
||
"""测试获取默认模型"""
|
||
response = client.get("/api/v1/config/defaults")
|
||
assert response.status_code == 200
|
||
|
||
data = response.json()
|
||
assert "data" in data
|
||
|
||
def test_get_styles(self):
|
||
"""测试获取样式配置"""
|
||
response = client.get("/api/v1/config/styles")
|
||
assert response.status_code == 200
|
||
|
||
data = unwrap_response_data(response)
|
||
assert isinstance(data, dict)
|
||
|
||
|
||
class TestTaskEndpoints:
|
||
"""任务管理端点测试"""
|
||
|
||
def test_get_task_stats_from_new_controller(self):
|
||
"""测试获取任务统计（新的 tasks 控制器）"""
|
||
# 由于路由冲突，旧的 generations.py 的 /tasks 路由会先匹配
|
||
# 我们测试新的 tasks 控制器的功能通过直接导入
|
||
from src.services.task_manager import task_manager
|
||
stats = task_manager.get_stats()
|
||
assert "total_tasks" in stats
|
||
assert "queue_size" in stats
|
||
|
||
def test_get_task_from_old_controller(self):
|
||
"""测试获取任务（旧的 generations 控制器）"""
|
||
# 旧的 generations.py 控制器有 /tasks 路由
|
||
response = client.get("/api/v1/tasks")
|
||
# 应该返回 200（旧控制器的响应）
|
||
assert response.status_code == 200
|
||
|
||
def test_get_nonexistent_task(self):
|
||
"""测试获取不存在的任务"""
|
||
response = client.get("/api/v1/tasks/non-existent-task-id-12345")
|
||
# 可能返回 404 或 200（取决于哪个控制器处理）
|
||
assert response.status_code in [200, 404]
|
||
|
||
|
||
class TestCanvasEndpoints:
|
||
"""画布端点测试"""
|
||
|
||
def test_get_canvas_default(self):
|
||
"""测试获取默认画布"""
|
||
response = client.get("/api/v1/canvas?id=default")
|
||
assert response.status_code == 200
|
||
|
||
data = response.json()
|
||
assert "data" in data
|
||
|
||
def test_save_canvas(self):
|
||
"""测试保存画布"""
|
||
canvas_data = {
|
||
"id": "test-canvas",
|
||
"projectId": "test-project",
|
||
"nodes": [],
|
||
"connections": [],
|
||
"groups": [],
|
||
"history": [],
|
||
"history_index": 0
|
||
}
|
||
|
||
response = client.post("/api/v1/canvas", json=canvas_data)
|
||
assert response.status_code == 200
|
||
|
||
data = response.json()
|
||
assert "data" in data
|
||
|
||
|
||
class TestPerformanceHeaders:
|
||
"""性能监控头测试"""
|
||
|
||
def test_process_time_header(self):
|
||
"""测试处理时间头"""
|
||
response = client.get("/health")
|
||
assert response.status_code == 200
|
||
|
||
# 检查是否有处理时间头
|
||
assert "X-Process-Time" in response.headers
|
||
|
||
# 处理时间应该是一个数字
|
||
process_time = float(response.headers["X-Process-Time"])
|
||
assert process_time >= 0
|
||
|
||
|
||
class TestCORS:
|
||
"""CORS 测试"""
|
||
|
||
def test_cors_headers(self):
|
||
"""测试 CORS 头"""
|
||
response = client.options("/api/v1/config/system", headers={
|
||
"Origin": "http://localhost:3000",
|
||
"Access-Control-Request-Method": "GET"
|
||
})
|
||
|
||
# 检查 CORS 头
|
||
assert "access-control-allow-origin" in response.headers
|
||
|
||
|
||
def test_openapi_schema():
|
||
"""测试 OpenAPI 规范"""
|
||
response = client.get("/openapi.json")
|
||
assert response.status_code == 200
|
||
|
||
schema = response.json()
|
||
assert "openapi" in schema
|
||
assert "info" in schema
|
||
assert "paths" in schema
|
||
|
||
|
||
def test_docs_endpoint():
|
||
"""测试文档端点"""
|
||
response = client.get("/docs")
|
||
assert response.status_code == 200
|
||
assert "text/html" in response.headers["content-type"]
|
||
|
||
|
||
if __name__ == "__main__":
|
||
pytest.main([__file__, "-v", "--tb=short"])
|