- 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()
490 lines
16 KiB
Python
490 lines
16 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Resolution Parameter Integration Test
|
||
|
||
集成测试 resolution 参数的完整处理流程,包括:
|
||
1. 加载实际的服务配置
|
||
2. 验证 resolution + aspect_ratio -> size 的转换
|
||
3. 测试不同 provider 的配置差异
|
||
|
||
运行方式:
|
||
cd /Users/cillin/workspeace/pixel/backend
|
||
python tests/test_resolution_integration.py
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import sys
|
||
from pathlib import Path
|
||
from typing import Dict, Optional, Tuple
|
||
from dataclasses import dataclass
|
||
|
||
# Add src to path
|
||
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
||
|
||
|
||
@dataclass
|
||
class ResolutionTestCase:
|
||
"""测试用例"""
|
||
provider: str
|
||
model: str
|
||
task_type: str # "image" or "video"
|
||
resolution: str
|
||
aspect_ratio: str
|
||
expected_size: Optional[str] = None
|
||
description: str = ""
|
||
|
||
|
||
class ResolutionConfigLoader:
|
||
"""加载服务配置"""
|
||
|
||
CONFIG_DIR = Path(__file__).parent.parent / "src" / "config" / "services"
|
||
|
||
@classmethod
|
||
def load_config(cls, provider: str, task_type: str) -> Dict:
|
||
"""加载指定 provider 和任务类型的配置"""
|
||
config_path = cls.CONFIG_DIR / provider / f"{task_type}.json"
|
||
|
||
if not config_path.exists():
|
||
return {}
|
||
|
||
with open(config_path, 'r', encoding='utf-8') as f:
|
||
return json.load(f)
|
||
|
||
@classmethod
|
||
def get_model_config(cls, provider: str, task_type: str, model_key: str) -> Dict:
|
||
"""获取特定模型的配置"""
|
||
config = cls.load_config(provider, task_type)
|
||
return config.get(model_key, {})
|
||
|
||
|
||
class ResolutionResolver:
|
||
"""
|
||
模拟控制器的 resolution 解析逻辑
|
||
"""
|
||
|
||
# 图片默认回退值
|
||
IMAGE_FALLBACKS = {
|
||
"1K": {
|
||
"16:9": "1280*720",
|
||
"9:16": "720*1280",
|
||
"1:1": "1024*1024",
|
||
"4:3": "1280*960",
|
||
"3:4": "960*1280",
|
||
"2.35:1": "1280*544"
|
||
},
|
||
"2K": {
|
||
"16:9": "2560*1440",
|
||
"9:16": "1440*2560",
|
||
"1:1": "2048*2048",
|
||
"4:3": "2560*1920",
|
||
"3:4": "1920*2560"
|
||
},
|
||
"4K": {
|
||
"16:9": "3840*2160",
|
||
"9:16": "2160*3840",
|
||
"1:1": "4096*4096",
|
||
"4:3": "3840*2880",
|
||
"3:4": "2880*3840"
|
||
}
|
||
}
|
||
|
||
# 视频默认回退值
|
||
VIDEO_FALLBACKS = {
|
||
"16:9": "1280*720",
|
||
"9:16": "720*1280",
|
||
"1:1": "1280*1280",
|
||
"4:3": "1280*960",
|
||
"3:4": "960*1280"
|
||
}
|
||
|
||
@classmethod
|
||
def resolve_image_size(
|
||
cls,
|
||
aspect_ratio: Optional[str],
|
||
resolution: Optional[str],
|
||
service_config: Dict
|
||
) -> Optional[str]:
|
||
"""
|
||
模拟图片控制器的 size 解析逻辑
|
||
参考: backend/src/controllers/generations/image.py:56-106
|
||
"""
|
||
if not aspect_ratio:
|
||
# 如果没有 aspect_ratio 但有 resolution,直接使用 resolution 作为 size
|
||
if resolution and ('*' in resolution or 'x' in resolution):
|
||
return resolution
|
||
return None
|
||
|
||
model_config = service_config or {}
|
||
resolutions_config = model_config.get("resolutions", {})
|
||
|
||
# 使用提供的 resolution 或默认 "1K"
|
||
res_level = resolution or "1K"
|
||
|
||
# 尝试嵌套结构: resolutions.1K.16:9
|
||
if resolutions_config and res_level in resolutions_config:
|
||
ratio_map = resolutions_config[res_level]
|
||
if isinstance(ratio_map, dict) and aspect_ratio in ratio_map:
|
||
return ratio_map[aspect_ratio]
|
||
|
||
# 尝试扁平结构回退: resolutions.16:9
|
||
if resolutions_config and aspect_ratio in resolutions_config:
|
||
return resolutions_config[aspect_ratio]
|
||
|
||
# 使用硬编码默认值
|
||
if res_level in cls.IMAGE_FALLBACKS:
|
||
return cls.IMAGE_FALLBACKS[res_level].get(aspect_ratio)
|
||
|
||
# 终极回退
|
||
return "1024*1024"
|
||
|
||
@classmethod
|
||
def resolve_video_size(
|
||
cls,
|
||
aspect_ratio: Optional[str],
|
||
resolution: Optional[str],
|
||
service_config: Dict
|
||
) -> Optional[str]:
|
||
"""
|
||
模拟视频控制器的 size 解析逻辑
|
||
参考: backend/src/controllers/generations/video.py:53-81
|
||
"""
|
||
if not aspect_ratio:
|
||
return None
|
||
|
||
model_config = service_config or {}
|
||
resolutions_config = model_config.get("resolutions", {})
|
||
|
||
# 使用提供的 resolution 或默认 "720P"
|
||
res_level = resolution or "720P"
|
||
|
||
# 尝试嵌套结构
|
||
if resolutions_config and res_level in resolutions_config:
|
||
ratio_map = resolutions_config[res_level]
|
||
if isinstance(ratio_map, dict) and aspect_ratio in ratio_map:
|
||
return ratio_map[aspect_ratio]
|
||
|
||
# 尝试扁平结构回退
|
||
if resolutions_config and aspect_ratio in resolutions_config:
|
||
return resolutions_config[aspect_ratio]
|
||
|
||
# 使用最小回退
|
||
return cls.VIDEO_FALLBACKS.get(aspect_ratio)
|
||
|
||
|
||
class ResolutionTester:
|
||
"""运行测试"""
|
||
|
||
def __init__(self):
|
||
self.passed = 0
|
||
self.failed = 0
|
||
self.errors = []
|
||
|
||
def test(self, name: str, condition: bool, details: str = ""):
|
||
"""运行单个测试"""
|
||
if condition:
|
||
self.passed += 1
|
||
print(f" ✓ {name}")
|
||
else:
|
||
self.failed += 1
|
||
msg = f" ✗ {name}"
|
||
if details:
|
||
msg += f" - {details}"
|
||
print(msg)
|
||
self.errors.append((name, details))
|
||
|
||
def run_all_tests(self):
|
||
"""运行所有测试"""
|
||
print("=" * 70)
|
||
print("Resolution Parameter Integration Test")
|
||
print("=" * 70)
|
||
|
||
# 1. 测试图片分辨率解析
|
||
print("\n📷 Image Resolution Tests")
|
||
print("-" * 70)
|
||
self._test_image_resolutions()
|
||
|
||
# 2. 测试视频分辨率解析
|
||
print("\n🎬 Video Resolution Tests")
|
||
print("-" * 70)
|
||
self._test_video_resolutions()
|
||
|
||
# 3. 测试实际配置文件
|
||
print("\n📂 Real Config File Tests")
|
||
print("-" * 70)
|
||
self._test_real_configs()
|
||
|
||
# 4. 测试边界情况
|
||
print("\n🔍 Edge Case Tests")
|
||
print("-" * 70)
|
||
self._test_edge_cases()
|
||
|
||
# 5. 测试前后端不一致问题
|
||
print("\n⚠️ Frontend-Backend Consistency Tests")
|
||
print("-" * 70)
|
||
self._test_consistency_issues()
|
||
|
||
# 汇总结果
|
||
print("\n" + "=" * 70)
|
||
print("Test Summary")
|
||
print("=" * 70)
|
||
total = self.passed + self.failed
|
||
print(f"Total: {total} | Passed: {self.passed} | Failed: {self.failed}")
|
||
|
||
if self.failed > 0:
|
||
print("\nFailed Tests:")
|
||
for name, details in self.errors:
|
||
print(f" - {name}: {details}")
|
||
return 1
|
||
else:
|
||
print("\n✅ All tests passed!")
|
||
return 0
|
||
|
||
def _test_image_resolutions(self):
|
||
"""测试图片分辨率解析"""
|
||
# DashScope 图片配置
|
||
dashscope_config = {
|
||
"resolutions": {
|
||
"1K": {
|
||
"16:9": "1280*720",
|
||
"9:16": "720*1280",
|
||
"1:1": "1280*1280"
|
||
},
|
||
"2K": {
|
||
"16:9": "2560*1440",
|
||
"9:16": "1440*2560",
|
||
"1:1": "2048*2048"
|
||
}
|
||
}
|
||
}
|
||
|
||
resolver = ResolutionResolver()
|
||
|
||
# 测试 1K 16:9
|
||
size = resolver.resolve_image_size("16:9", "1K", dashscope_config)
|
||
self.test("Image: 1K + 16:9 = 1280*720", size == "1280*720", f"got {size}")
|
||
|
||
# 测试 2K 16:9
|
||
size = resolver.resolve_image_size("16:9", "2K", dashscope_config)
|
||
self.test("Image: 2K + 16:9 = 2560*1440", size == "2560*1440", f"got {size}")
|
||
|
||
# 测试 1K 9:16
|
||
size = resolver.resolve_image_size("9:16", "1K", dashscope_config)
|
||
self.test("Image: 1K + 9:16 = 720*1280", size == "720*1280", f"got {size}")
|
||
|
||
# 测试默认 resolution (1K)
|
||
size = resolver.resolve_image_size("16:9", None, dashscope_config)
|
||
self.test("Image: default resolution = 1K", size == "1280*720", f"got {size}")
|
||
|
||
# 测试无配置时的回退
|
||
size = resolver.resolve_image_size("16:9", "1K", {})
|
||
self.test("Image: fallback 1K 16:9", size == "1280*720", f"got {size}")
|
||
|
||
# 测试 4K (硬编码回退)
|
||
size = resolver.resolve_image_size("16:9", "4K", {})
|
||
self.test("Image: 4K fallback 16:9", size == "3840*2160", f"got {size}")
|
||
|
||
def _test_video_resolutions(self):
|
||
"""测试视频分辨率解析"""
|
||
# Kling 视频配置
|
||
kling_config = {
|
||
"resolutions": {
|
||
"720P": {
|
||
"16:9": "1280*720",
|
||
"9:16": "720*1280"
|
||
},
|
||
"1080P": {
|
||
"16:9": "1920*1080",
|
||
"9:16": "1080*1920"
|
||
}
|
||
}
|
||
}
|
||
|
||
resolver = ResolutionResolver()
|
||
|
||
# 测试 720P 16:9
|
||
size = resolver.resolve_video_size("16:9", "720P", kling_config)
|
||
self.test("Video: 720P + 16:9 = 1280*720", size == "1280*720", f"got {size}")
|
||
|
||
# 测试 1080P 16:9
|
||
size = resolver.resolve_video_size("16:9", "1080P", kling_config)
|
||
self.test("Video: 1080P + 16:9 = 1920*1080", size == "1920*1080", f"got {size}")
|
||
|
||
# 测试 720P 9:16
|
||
size = resolver.resolve_video_size("9:16", "720P", kling_config)
|
||
self.test("Video: 720P + 9:16 = 720*1280", size == "720*1280", f"got {size}")
|
||
|
||
# 测试默认 resolution (720P)
|
||
size = resolver.resolve_video_size("16:9", None, kling_config)
|
||
self.test("Video: default resolution = 720P", size == "1280*720", f"got {size}")
|
||
|
||
# 测试无配置时的回退
|
||
size = resolver.resolve_video_size("16:9", "720P", {})
|
||
self.test("Video: fallback 720P 16:9", size == "1280*720", f"got {size}")
|
||
|
||
def _test_real_configs(self):
|
||
"""测试实际配置文件"""
|
||
loader = ResolutionConfigLoader()
|
||
|
||
# 测试 DashScope 图片配置
|
||
config = loader.load_config("dashscope", "image")
|
||
if config:
|
||
self.test("Config: dashscope/image.json exists", True)
|
||
|
||
# 检查 qwen-image 配置
|
||
qwen_config = config.get("qwen-image", {})
|
||
if "resolutions" in qwen_config:
|
||
resolutions = qwen_config["resolutions"]
|
||
has_1k = "1K" in resolutions
|
||
has_2k = "2K" in resolutions
|
||
self.test("Config: qwen-image has 1K", has_1k)
|
||
self.test("Config: qwen-image has 2K", has_2k)
|
||
|
||
if has_1k:
|
||
ratio_map = resolutions["1K"]
|
||
self.test("Config: 1K has 16:9", "16:9" in ratio_map)
|
||
else:
|
||
self.test("Config: dashscope/image.json", False, "File not found")
|
||
|
||
# 测试 Kling 视频配置
|
||
config = loader.load_config("kling", "video")
|
||
if config:
|
||
self.test("Config: kling/video.json exists", True)
|
||
else:
|
||
self.test("Config: kling/video.json exists", False, "File not found")
|
||
|
||
def _test_edge_cases(self):
|
||
"""测试边界情况"""
|
||
resolver = ResolutionResolver()
|
||
|
||
# 测试没有 aspect_ratio
|
||
size = resolver.resolve_image_size(None, "1K", {})
|
||
self.test("Edge: no aspect_ratio", size is None, f"got {size}")
|
||
|
||
# 测试没有 aspect_ratio 但有像素格式的 resolution
|
||
size = resolver.resolve_image_size(None, "1920*1080", {})
|
||
self.test("Edge: resolution as explicit size", size == "1920*1080", f"got {size}")
|
||
|
||
# 测试未知 resolution level
|
||
size = resolver.resolve_image_size("16:9", "8K", {})
|
||
# 应该使用硬编码回退
|
||
self.test("Edge: unknown resolution level uses fallback",
|
||
size == "1280*720", f"got {size}")
|
||
|
||
# 测试未知 aspect_ratio
|
||
size = resolver.resolve_image_size("999:1", "1K", {})
|
||
# 应该使用终极回退
|
||
self.test("Edge: unknown aspect_ratio uses ultimate fallback",
|
||
size == "1024*1024", f"got {size}")
|
||
|
||
# 测试空配置
|
||
size = resolver.resolve_video_size("16:9", "720P", None)
|
||
self.test("Edge: None config uses fallback", size == "1280*720", f"got {size}")
|
||
|
||
def _test_consistency_issues(self):
|
||
"""测试前后端一致性问题"""
|
||
import re
|
||
|
||
# 修复后的前端验证逻辑 (frontend/src/lib/utils/generationValidation.ts)
|
||
# 支持两种格式:
|
||
# 1. 质量级别: "1K", "2K", "4K", "720P", "1080P"
|
||
# 2. 像素格式: "1024*1024", "1920x1080" (向后兼容)
|
||
|
||
quality_pattern = r'^(1K|2K|4K|720P|1080P|480P|360P)$'
|
||
pixel_pattern = r'^\d+[x*]\d+$'
|
||
|
||
# 后端控制器逻辑期望格式: "1K", "2K", "4K"
|
||
backend_resolution_levels = ["1K", "2K", "4K"]
|
||
|
||
# 测试前端验证接受质量级别
|
||
for val in backend_resolution_levels:
|
||
match = bool(re.match(quality_pattern, val, re.IGNORECASE))
|
||
self.test(f"✅ Frontend accepts '{val}' (quality level)", match)
|
||
|
||
# 测试前端验证接受像素格式 (向后兼容)
|
||
pixel_formats = ["1024*1024", "1920x1080", "2560*1440"]
|
||
for val in pixel_formats:
|
||
match = bool(re.match(pixel_pattern, val))
|
||
self.test(f"✅ Frontend accepts '{val}' (pixel format)", match)
|
||
|
||
# 总结
|
||
print("\n ✅ Consistency Fixed:")
|
||
print(" - Frontend accepts: quality level or pixel format")
|
||
print(" - Backend expects: quality level")
|
||
print(" - Result: Parameters flow correctly!")
|
||
|
||
|
||
def print_usage_guide():
|
||
"""打印使用指南"""
|
||
print("""
|
||
================================================================================
|
||
Resolution Parameter Usage Guide
|
||
================================================================================
|
||
|
||
📷 IMAGE GENERATION
|
||
-------------------
|
||
Valid resolution values: "1K", "2K", "4K"
|
||
Valid aspect_ratio values: "16:9", "9:16", "1:1", "4:3", "3:4", "2.35:1"
|
||
|
||
Example Request:
|
||
{
|
||
"prompt": "A beautiful sunset",
|
||
"model": "dashscope/qwen-image",
|
||
"resolution": "2K",
|
||
"aspectRatio": "16:9"
|
||
}
|
||
|
||
Resolution Mapping (1K):
|
||
16:9 -> 1280*720
|
||
9:16 -> 720*1280
|
||
1:1 -> 1024*1024
|
||
4:3 -> 1280*960
|
||
3:4 -> 960*1280
|
||
|
||
🎬 VIDEO GENERATION
|
||
-------------------
|
||
Valid resolution values: "720P", "1080P"
|
||
Valid aspect_ratio values: "16:9", "9:16", "1:1", "4:3", "3:4"
|
||
|
||
Example Request:
|
||
{
|
||
"prompt": "A dancing figure",
|
||
"model": "kling/kling-video",
|
||
"resolution": "1080P",
|
||
"aspectRatio": "16:9",
|
||
"duration": 5
|
||
}
|
||
|
||
Resolution Mapping (720P):
|
||
16:9 -> 1280*720
|
||
9:16 -> 720*1280
|
||
1:1 -> 1280*1280
|
||
|
||
⚠️ KNOWN ISSUES
|
||
----------------
|
||
1. Frontend validation (generationValidation.ts:74) expects pixel format
|
||
like '1024*1024', but backend controllers expect quality levels like '1K'.
|
||
|
||
2. This inconsistency means resolution parameter may be rejected by frontend
|
||
validation before reaching the backend.
|
||
|
||
3. Workaround: Frontend should skip format validation for resolution,
|
||
or accept both formats.
|
||
|
||
================================================================================
|
||
""")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
if len(sys.argv) > 1 and sys.argv[1] == "--help":
|
||
print_usage_guide()
|
||
sys.exit(0)
|
||
|
||
tester = ResolutionTester()
|
||
exit_code = tester.run_all_tests()
|
||
|
||
if len(sys.argv) > 1 and sys.argv[1] == "--guide":
|
||
print_usage_guide()
|
||
|
||
sys.exit(exit_code)
|