284 lines
10 KiB
Python
284 lines
10 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
测试 draft_content 解析修复
|
||
验证字符串类型的 draft_content 能正确解析为字典
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
import json
|
||
import uuid
|
||
from datetime import datetime
|
||
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
from python_core.database.template_postgres import template_table
|
||
from python_core.database.types import TemplateInfo
|
||
|
||
def test_draft_content_parsing():
|
||
"""测试 draft_content 解析功能"""
|
||
print("=== 测试 draft_content 解析修复 ===\n")
|
||
|
||
try:
|
||
# 创建测试数据
|
||
test_draft = {
|
||
"canvas_config": {"width": 1920, "height": 1080, "fps": 30},
|
||
"tracks": [
|
||
{"id": "track1", "type": "video", "name": "视频轨道"}
|
||
],
|
||
"materials": [
|
||
{"id": "mat1", "name": "test.mp4", "type": "video"}
|
||
],
|
||
"duration": 10000
|
||
}
|
||
|
||
template_id = str(uuid.uuid4())
|
||
|
||
# 1. 测试正常的字典类型 draft_content
|
||
print("1. 测试字典类型 draft_content...")
|
||
template_info = TemplateInfo(
|
||
id=template_id,
|
||
name=f"测试模板-字典-{template_id[:8]}",
|
||
description="测试字典类型的 draft_content",
|
||
thumbnail_path="",
|
||
draft_content_path="",
|
||
resources_path="",
|
||
created_at=datetime.now().isoformat(),
|
||
updated_at=datetime.now().isoformat(),
|
||
canvas_config=test_draft.get('canvas_config', {}),
|
||
duration=test_draft.get('duration', 0),
|
||
material_count=len(test_draft.get('materials', [])),
|
||
track_count=len(test_draft.get('tracks', [])),
|
||
tags=["测试"],
|
||
user_id="test_user",
|
||
draft_content=test_draft # 字典类型
|
||
)
|
||
|
||
# 存储到数据库
|
||
created_id = template_table.create_template(template_info)
|
||
print(f" ✅ 模板创建成功: {created_id}")
|
||
|
||
# 读取并验证
|
||
stored_template = template_table.get_template_by_id(created_id)
|
||
if stored_template and isinstance(stored_template.draft_content, dict):
|
||
print(f" ✅ draft_content 类型正确: {type(stored_template.draft_content)}")
|
||
print(f" ✅ draft_content 键数量: {len(stored_template.draft_content)}")
|
||
else:
|
||
print(f" ❌ draft_content 类型错误: {type(stored_template.draft_content) if stored_template else 'None'}")
|
||
return False
|
||
|
||
# 清理
|
||
template_table.delete_template(created_id)
|
||
|
||
# 2. 测试直接在数据库中插入 JSON 字符串
|
||
print("\n2. 测试 JSON 字符串类型 draft_content...")
|
||
|
||
# 直接操作数据库，插入 JSON 字符串
|
||
import psycopg2.extras
|
||
from python_core.database.template_postgres import TemplateTablePostgres
|
||
|
||
template_id2 = str(uuid.uuid4())
|
||
draft_json_str = json.dumps(test_draft)
|
||
|
||
with template_table._get_connection() as conn:
|
||
with conn.cursor() as cursor:
|
||
insert_sql = """
|
||
INSERT INTO templates (
|
||
id, name, description, thumbnail_path, draft_content_path,
|
||
draft_content, resources_path, canvas_config, duration, material_count,
|
||
track_count, tags, is_cloud, user_id, created_at, updated_at
|
||
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
|
||
"""
|
||
|
||
now = datetime.now()
|
||
cursor.execute(insert_sql, (
|
||
template_id2,
|
||
f"测试模板-字符串-{template_id2[:8]}",
|
||
"测试 JSON 字符串类型的 draft_content",
|
||
"",
|
||
"",
|
||
draft_json_str, # 直接插入 JSON 字符串
|
||
"",
|
||
json.dumps({}),
|
||
10000,
|
||
1,
|
||
1,
|
||
json.dumps(["测试"]),
|
||
False,
|
||
"test_user",
|
||
now,
|
||
now
|
||
))
|
||
conn.commit()
|
||
|
||
print(f" ✅ 直接插入 JSON 字符串成功: {template_id2}")
|
||
|
||
# 读取并验证解析
|
||
stored_template2 = template_table.get_template_by_id(template_id2)
|
||
if stored_template2:
|
||
print(f" draft_content 类型: {type(stored_template2.draft_content)}")
|
||
if isinstance(stored_template2.draft_content, dict):
|
||
print(f" ✅ JSON 字符串正确解析为字典")
|
||
print(f" ✅ draft_content 键数量: {len(stored_template2.draft_content)}")
|
||
print(f" ✅ 包含 tracks: {'tracks' in stored_template2.draft_content}")
|
||
print(f" ✅ 包含 materials: {'materials' in stored_template2.draft_content}")
|
||
else:
|
||
print(f" ❌ JSON 字符串解析失败")
|
||
return False
|
||
else:
|
||
print(f" ❌ 模板读取失败")
|
||
return False
|
||
|
||
# 测试 get_draft_content 方法
|
||
print("\n3. 测试 get_draft_content 方法...")
|
||
draft_content = template_table.get_draft_content(template_id2)
|
||
if draft_content is not None and isinstance(draft_content, dict):
|
||
print(f" ✅ get_draft_content 返回字典类型")
|
||
print(f" ✅ 键数量: {len(draft_content)}")
|
||
else:
|
||
print(f" ❌ get_draft_content 返回类型错误: {type(draft_content)}")
|
||
return False
|
||
|
||
# 清理
|
||
template_table.delete_template(template_id2)
|
||
|
||
print("\n✅ 所有测试通过！")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ 测试失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
def test_cli_command():
|
||
"""测试 CLI 命令"""
|
||
print("\n=== 测试 CLI 命令 ===\n")
|
||
|
||
try:
|
||
# 创建一个测试模板
|
||
test_draft = {
|
||
"canvas_config": {"width": 1920, "height": 1080, "fps": 30},
|
||
"tracks": [
|
||
{
|
||
"id": "track1",
|
||
"type": "video",
|
||
"name": "视频轨道",
|
||
"segments": [
|
||
{
|
||
"id": "seg1",
|
||
"start": 0,
|
||
"end": 5000,
|
||
"material_id": "mat1"
|
||
}
|
||
]
|
||
}
|
||
],
|
||
"materials": [
|
||
{
|
||
"id": "mat1",
|
||
"name": "test.mp4",
|
||
"type": "video",
|
||
"path": "test.mp4"
|
||
}
|
||
],
|
||
"duration": 5000
|
||
}
|
||
|
||
template_id = str(uuid.uuid4())
|
||
template_info = TemplateInfo(
|
||
id=template_id,
|
||
name=f"CLI测试模板-{template_id[:8]}",
|
||
description="用于测试 CLI 命令的模板",
|
||
thumbnail_path="",
|
||
draft_content_path="",
|
||
resources_path="",
|
||
created_at=datetime.now().isoformat(),
|
||
updated_at=datetime.now().isoformat(),
|
||
canvas_config=test_draft.get('canvas_config', {}),
|
||
duration=test_draft.get('duration', 0),
|
||
material_count=len(test_draft.get('materials', [])),
|
||
track_count=len(test_draft.get('tracks', [])),
|
||
tags=["CLI测试"],
|
||
user_id="test_user",
|
||
draft_content=test_draft
|
||
)
|
||
|
||
# 创建模板
|
||
created_id = template_table.create_template(template_info)
|
||
print(f"✅ 创建测试模板: {created_id}")
|
||
|
||
# 测试 CLI 命令
|
||
import subprocess
|
||
|
||
cmd = [
|
||
"python3", "-m", "python_core.cli",
|
||
"template", "detail", created_id,
|
||
"--json"
|
||
]
|
||
|
||
print(f"执行命令: {' '.join(cmd)}")
|
||
|
||
result = subprocess.run(
|
||
cmd,
|
||
capture_output=True,
|
||
text=True,
|
||
cwd="/root/projects/mixvideo_v2"
|
||
)
|
||
|
||
print(f"返回码: {result.returncode}")
|
||
if result.stdout:
|
||
print(f"标准输出: {result.stdout}")
|
||
if result.stderr:
|
||
print(f"标准错误: {result.stderr}")
|
||
|
||
if result.returncode == 0:
|
||
# 检查输出是否包含 JSONRPC 成功响应
|
||
if "JSONRPC:" in result.stdout and '"error"' not in result.stdout:
|
||
print("✅ CLI 命令执行成功，没有错误")
|
||
success = True
|
||
else:
|
||
print("❌ CLI 命令返回错误")
|
||
success = False
|
||
else:
|
||
print("❌ CLI 命令执行失败")
|
||
success = False
|
||
|
||
# 清理
|
||
template_table.delete_template(created_id)
|
||
print(f"✅ 清理测试模板: {created_id}")
|
||
|
||
return success
|
||
|
||
except Exception as e:
|
||
print(f"❌ CLI 测试失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
def main():
|
||
"""主测试函数"""
|
||
print("开始测试 draft_content 解析修复...\n")
|
||
|
||
# 测试1：draft_content 解析
|
||
test1_success = test_draft_content_parsing()
|
||
|
||
# 测试2：CLI 命令
|
||
test2_success = test_cli_command()
|
||
|
||
print(f"\n=== 测试结果 ===")
|
||
print(f"测试1 (draft_content解析): {'✅ 通过' if test1_success else '❌ 失败'}")
|
||
print(f"测试2 (CLI命令): {'✅ 通过' if test2_success else '❌ 失败'}")
|
||
|
||
if test1_success and test2_success:
|
||
print("\n🎉 所有测试通过！draft_content 解析修复成功！")
|
||
print("\n修复内容:")
|
||
print("1. ✅ 修复了数据库中 JSON 字符串的解析问题")
|
||
print("2. ✅ 改进了 CLI 命令的类型检查")
|
||
print("3. ✅ 确保 draft_content 始终为字典类型")
|
||
print("4. ✅ 解决了 'str' object has no attribute 'get' 错误")
|
||
else:
|
||
print("\n❌ 部分测试失败，需要进一步检查")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|