Compare commits
5 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 7cdf305042 | |||
| 25514d072e | |||
| cafe063859 | |||
| d92ff7aa36 | |||
| cb85b33ee9 |
+413
-699
File diff suppressed because it is too large
Load Diff
Executable
+453
@@ -0,0 +1,453 @@
|
||||
"""shared.ai_service 单元测试.
|
||||
|
||||
主要测试纯逻辑部分:_parse_recommend_response / _fallback_recommend_clips / _call_ai_cover_service.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from shared.ai_service import (
|
||||
_call_ai_cover_service,
|
||||
_fallback_recommend_clips,
|
||||
_parse_recommend_response,
|
||||
)
|
||||
|
||||
# ── _parse_recommend_response 测试 ────────────────────────────────────────
|
||||
|
||||
|
||||
class TestParseRecommendResponseBasic:
|
||||
"""基础解析测试."""
|
||||
|
||||
def test_parse_valid_json(self):
|
||||
content = json.dumps(
|
||||
{
|
||||
"clips": [
|
||||
{
|
||||
"clip_type": "intro",
|
||||
"order": 0,
|
||||
"text_content": "开场",
|
||||
"duration": 3.0,
|
||||
"transition_effect": "fade",
|
||||
"asset_id": "asset1",
|
||||
"start_time": 0.0,
|
||||
"config": {},
|
||||
},
|
||||
{
|
||||
"clip_type": "outro",
|
||||
"order": 1,
|
||||
"text_content": "结尾",
|
||||
"duration": 2.0,
|
||||
"transition_effect": "fade",
|
||||
"asset_id": "",
|
||||
"start_time": 0.0,
|
||||
"config": {},
|
||||
},
|
||||
],
|
||||
"title": "测试视频",
|
||||
"confidence": 0.85,
|
||||
}
|
||||
)
|
||||
result = _parse_recommend_response(content, ["asset1"], 30.0)
|
||||
assert result is not None
|
||||
assert len(result["clips"]) == 2
|
||||
assert result["confidence"] == 0.85
|
||||
assert result["total_duration"] == 5.0
|
||||
assert result["config"]["title"]["text"] == "测试视频"
|
||||
assert result["config"]["title"]["ai_auto"] is True
|
||||
|
||||
def test_parse_none_returns_none(self):
|
||||
result = _parse_recommend_response(None, ["a1"], 30.0) # type: ignore[arg-type]
|
||||
assert result is None
|
||||
|
||||
def test_parse_empty_string_returns_none(self):
|
||||
result = _parse_recommend_response("", ["a1"], 30.0)
|
||||
assert result is None
|
||||
|
||||
def test_parse_whitespace_only_returns_none(self):
|
||||
result = _parse_recommend_response(" ", ["a1"], 30.0)
|
||||
assert result is None
|
||||
|
||||
def test_parse_invalid_json_returns_none(self):
|
||||
result = _parse_recommend_response("not json", ["a1"], 30.0)
|
||||
assert result is None
|
||||
|
||||
def test_parse_non_dict_json_returns_none(self):
|
||||
result = _parse_recommend_response("[1, 2, 3]", ["a1"], 30.0)
|
||||
assert result is None
|
||||
|
||||
|
||||
class TestParseRecommendResponseClips:
|
||||
"""clips 解析测试."""
|
||||
|
||||
def test_parse_no_clips_returns_none(self):
|
||||
content = json.dumps({"title": "test", "clips": []})
|
||||
result = _parse_recommend_response(content, ["a1"], 30.0)
|
||||
assert result is None
|
||||
|
||||
def test_parse_clips_not_list_returns_none(self):
|
||||
content = json.dumps({"clips": "not a list"})
|
||||
result = _parse_recommend_response(content, ["a1"], 30.0)
|
||||
assert result is None
|
||||
|
||||
def test_parse_clips_sorted_by_order(self):
|
||||
content = json.dumps(
|
||||
{
|
||||
"clips": [
|
||||
{"clip_type": "outro", "order": 2, "duration": 2, "asset_id": "a1"},
|
||||
{"clip_type": "intro", "order": 0, "duration": 3, "asset_id": "a1"},
|
||||
{"clip_type": "showcase", "order": 1, "duration": 5, "asset_id": "a1"},
|
||||
],
|
||||
}
|
||||
)
|
||||
result = _parse_recommend_response(content, ["a1"], 30.0)
|
||||
assert result is not None
|
||||
assert len(result["clips"]) == 3
|
||||
assert result["clips"][0]["clip_type"] == "intro"
|
||||
assert result["clips"][1]["clip_type"] == "showcase"
|
||||
assert result["clips"][2]["clip_type"] == "outro"
|
||||
|
||||
def test_parse_clips_renumbered_continuously(self):
|
||||
content = json.dumps(
|
||||
{
|
||||
"clips": [
|
||||
{"clip_type": "intro", "order": 10, "duration": 2, "asset_id": "a1"},
|
||||
{"clip_type": "outro", "order": 20, "duration": 2, "asset_id": "a1"},
|
||||
],
|
||||
}
|
||||
)
|
||||
result = _parse_recommend_response(content, ["a1"], 30.0)
|
||||
assert result is not None
|
||||
assert result["clips"][0]["order"] == 0
|
||||
assert result["clips"][1]["order"] == 1
|
||||
|
||||
def test_parse_skips_invalid_clip_dicts(self):
|
||||
content = json.dumps(
|
||||
{
|
||||
"clips": [
|
||||
{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"},
|
||||
"not a dict",
|
||||
{"clip_type": "outro", "order": 2, "duration": 2, "asset_id": "a1"},
|
||||
],
|
||||
}
|
||||
)
|
||||
result = _parse_recommend_response(content, ["a1"], 30.0)
|
||||
assert result is not None
|
||||
assert len(result["clips"]) == 2
|
||||
|
||||
|
||||
class TestParseRecommendResponseFields:
|
||||
"""各字段解析与边界测试."""
|
||||
|
||||
def test_parse_duration_clamped_min(self):
|
||||
content = json.dumps(
|
||||
{
|
||||
"clips": [
|
||||
{"clip_type": "intro", "order": 0, "duration": 0.5, "asset_id": "a1"},
|
||||
],
|
||||
}
|
||||
)
|
||||
result = _parse_recommend_response(content, ["a1"], 30.0)
|
||||
assert result is not None
|
||||
assert result["clips"][0]["duration"] == 1.0
|
||||
|
||||
def test_parse_duration_clamped_max(self):
|
||||
content = json.dumps(
|
||||
{
|
||||
"clips": [
|
||||
{"clip_type": "intro", "order": 0, "duration": 100, "asset_id": "a1"},
|
||||
],
|
||||
}
|
||||
)
|
||||
result = _parse_recommend_response(content, ["a1"], 30.0)
|
||||
assert result is not None
|
||||
assert result["clips"][0]["duration"] == 30.0
|
||||
|
||||
def test_parse_start_time_clamped_min(self):
|
||||
content = json.dumps(
|
||||
{
|
||||
"clips": [
|
||||
{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1", "start_time": -5.0},
|
||||
],
|
||||
}
|
||||
)
|
||||
result = _parse_recommend_response(content, ["a1"], 30.0)
|
||||
assert result is not None
|
||||
assert result["clips"][0]["start_time"] == 0.0
|
||||
|
||||
def test_parse_asset_id_not_in_list_empty(self):
|
||||
content = json.dumps(
|
||||
{
|
||||
"clips": [
|
||||
{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "unknown_asset"},
|
||||
],
|
||||
}
|
||||
)
|
||||
result = _parse_recommend_response(content, ["a1", "a2"], 30.0)
|
||||
assert result is not None
|
||||
assert result["clips"][0]["asset_id"] == ""
|
||||
|
||||
def test_parse_asset_id_in_list_kept(self):
|
||||
content = json.dumps(
|
||||
{
|
||||
"clips": [
|
||||
{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a2"},
|
||||
],
|
||||
}
|
||||
)
|
||||
result = _parse_recommend_response(content, ["a1", "a2"], 30.0)
|
||||
assert result is not None
|
||||
assert result["clips"][0]["asset_id"] == "a2"
|
||||
|
||||
def test_parse_default_values(self):
|
||||
content = json.dumps(
|
||||
{
|
||||
"clips": [
|
||||
{"order": 0},
|
||||
],
|
||||
}
|
||||
)
|
||||
result = _parse_recommend_response(content, ["a1"], 30.0)
|
||||
assert result is not None
|
||||
clip = result["clips"][0]
|
||||
assert clip["clip_type"] == "showcase"
|
||||
assert clip["text_content"] == ""
|
||||
assert clip["duration"] == 3.0
|
||||
assert clip["transition_effect"] == "cut"
|
||||
assert clip["asset_id"] == ""
|
||||
assert clip["start_time"] == 0.0
|
||||
assert clip["config"] == {}
|
||||
|
||||
|
||||
class TestParseRecommendResponseMarkdown:
|
||||
"""Markdown 代码块包裹的 JSON 测试."""
|
||||
|
||||
def test_parse_markdown_json(self):
|
||||
content = (
|
||||
"```json\n"
|
||||
+ json.dumps(
|
||||
{
|
||||
"clips": [{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"}],
|
||||
"title": "md test",
|
||||
}
|
||||
)
|
||||
+ "\n```"
|
||||
)
|
||||
result = _parse_recommend_response(content, ["a1"], 30.0)
|
||||
assert result is not None
|
||||
assert len(result["clips"]) == 1
|
||||
assert result["config"]["title"]["text"] == "md test"
|
||||
|
||||
def test_parse_backticks_no_language(self):
|
||||
content = (
|
||||
"```\n"
|
||||
+ json.dumps(
|
||||
{
|
||||
"clips": [{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"}],
|
||||
}
|
||||
)
|
||||
+ "\n```"
|
||||
)
|
||||
result = _parse_recommend_response(content, ["a1"], 30.0)
|
||||
assert result is not None
|
||||
assert len(result["clips"]) == 1
|
||||
|
||||
|
||||
class TestParseRecommendResponseConfidence:
|
||||
"""confidence 解析测试."""
|
||||
|
||||
def test_parse_confidence_normal(self):
|
||||
content = json.dumps(
|
||||
{
|
||||
"clips": [{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"}],
|
||||
"confidence": 0.85,
|
||||
}
|
||||
)
|
||||
result = _parse_recommend_response(content, ["a1"], 30.0)
|
||||
assert result is not None
|
||||
assert result["confidence"] == 0.85
|
||||
|
||||
def test_parse_confidence_clamped_min(self):
|
||||
content = json.dumps(
|
||||
{
|
||||
"clips": [{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"}],
|
||||
"confidence": -0.5,
|
||||
}
|
||||
)
|
||||
result = _parse_recommend_response(content, ["a1"], 30.0)
|
||||
assert result is not None
|
||||
assert result["confidence"] == 0.0
|
||||
|
||||
def test_parse_confidence_clamped_max(self):
|
||||
content = json.dumps(
|
||||
{
|
||||
"clips": [{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"}],
|
||||
"confidence": 1.5,
|
||||
}
|
||||
)
|
||||
result = _parse_recommend_response(content, ["a1"], 30.0)
|
||||
assert result is not None
|
||||
assert result["confidence"] == 1.0
|
||||
|
||||
def test_parse_confidence_default(self):
|
||||
content = json.dumps(
|
||||
{
|
||||
"clips": [{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"}],
|
||||
}
|
||||
)
|
||||
result = _parse_recommend_response(content, ["a1"], 30.0)
|
||||
assert result is not None
|
||||
assert result["confidence"] == 0.7
|
||||
|
||||
|
||||
class TestParseRecommendResponseConfig:
|
||||
"""config 生成测试."""
|
||||
|
||||
def test_parse_no_title_no_ai_auto(self):
|
||||
content = json.dumps(
|
||||
{
|
||||
"clips": [{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"}],
|
||||
}
|
||||
)
|
||||
result = _parse_recommend_response(content, ["a1"], 30.0)
|
||||
assert result is not None
|
||||
# 没有 title 时,config 的 title.text 保持默认(DEFAULT_EDIT_PLAN_CONFIG 中的值)
|
||||
assert "title" in result["config"]
|
||||
|
||||
def test_parse_config_is_deep_copy(self):
|
||||
content = json.dumps(
|
||||
{
|
||||
"clips": [{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"}],
|
||||
"title": "test",
|
||||
}
|
||||
)
|
||||
result1 = _parse_recommend_response(content, ["a1"], 30.0)
|
||||
result2 = _parse_recommend_response(content, ["a1"], 30.0)
|
||||
# 修改其中一个不影响另一个
|
||||
result1["config"]["title"]["text"] = "modified"
|
||||
assert result2["config"]["title"]["text"] != "modified"
|
||||
|
||||
|
||||
class TestParseRecommendResponseTotalDuration:
|
||||
"""total_duration 计算测试."""
|
||||
|
||||
def test_parse_total_duration_sum(self):
|
||||
content = json.dumps(
|
||||
{
|
||||
"clips": [
|
||||
{"clip_type": "intro", "order": 0, "duration": 3.5, "asset_id": "a1"},
|
||||
{"clip_type": "showcase", "order": 1, "duration": 5.2, "asset_id": "a1"},
|
||||
{"clip_type": "outro", "order": 2, "duration": 2.0, "asset_id": "a1"},
|
||||
],
|
||||
}
|
||||
)
|
||||
result = _parse_recommend_response(content, ["a1"], 30.0)
|
||||
assert result is not None
|
||||
assert result["total_duration"] == pytest.approx(10.7, abs=0.01)
|
||||
|
||||
|
||||
# ── _fallback_recommend_clips 测试 ────────────────────────────────────────
|
||||
|
||||
|
||||
class TestFallbackRecommendClips:
|
||||
"""本地降级推荐方案测试."""
|
||||
|
||||
def test_fallback_returns_dict_with_clips(self):
|
||||
with patch("shared.ai_service.time.sleep"):
|
||||
result = _fallback_recommend_clips("plan1", "tmpl1", ["a1", "a2"], "one_take", 30.0)
|
||||
assert "clips" in result
|
||||
assert "config" in result
|
||||
assert "total_duration" in result
|
||||
assert "confidence" in result
|
||||
|
||||
def test_fallback_has_intro_and_outro(self):
|
||||
with patch("shared.ai_service.time.sleep"):
|
||||
result = _fallback_recommend_clips("plan1", "tmpl1", ["a1", "a2"], "one_take", 30.0)
|
||||
clips = result["clips"]
|
||||
assert clips[0]["clip_type"] == "intro"
|
||||
assert clips[-1]["clip_type"] == "outro"
|
||||
|
||||
def test_fallback_showcase_count_matches_assets(self):
|
||||
with patch("shared.ai_service.time.sleep"):
|
||||
result = _fallback_recommend_clips("plan1", "tmpl1", ["a1", "a2", "a3"], "one_take", 30.0)
|
||||
showcase_clips = [c for c in result["clips"] if c["clip_type"] == "showcase"]
|
||||
assert len(showcase_clips) == 3
|
||||
|
||||
def test_fallback_no_assets_still_works(self):
|
||||
with patch("shared.ai_service.time.sleep"):
|
||||
result = _fallback_recommend_clips("plan1", "tmpl1", [], "one_take", 30.0)
|
||||
assert len(result["clips"]) >= 2 # 至少有intro和outro
|
||||
|
||||
def test_fallback_intro_uses_first_asset(self):
|
||||
with patch("shared.ai_service.time.sleep"):
|
||||
result = _fallback_recommend_clips("plan1", "tmpl1", ["a1", "a2"], "one_take", 30.0)
|
||||
assert result["clips"][0]["asset_id"] == "a1"
|
||||
|
||||
def test_fallback_outro_has_empty_asset(self):
|
||||
with patch("shared.ai_service.time.sleep"):
|
||||
result = _fallback_recommend_clips("plan1", "tmpl1", ["a1"], "one_take", 30.0)
|
||||
assert result["clips"][-1]["asset_id"] == ""
|
||||
|
||||
def test_fallback_confidence_in_range(self):
|
||||
with patch("shared.ai_service.time.sleep"):
|
||||
result = _fallback_recommend_clips("plan1", "tmpl1", ["a1"], "one_take", 30.0)
|
||||
assert 0.75 <= result["confidence"] <= 0.95
|
||||
|
||||
def test_fallback_title_contains_asset_count(self):
|
||||
with patch("shared.ai_service.time.sleep"):
|
||||
result = _fallback_recommend_clips("plan1", "tmpl1", ["a1", "a2", "a3"], "one_take", 30.0)
|
||||
assert "3" in result["config"]["title"]["text"]
|
||||
assert result["config"]["title"]["ai_auto"] is True
|
||||
|
||||
def test_fallback_total_duration_matches(self):
|
||||
with patch("shared.ai_service.time.sleep"):
|
||||
result = _fallback_recommend_clips("plan1", "tmpl1", ["a1", "a2"], "one_take", 30.0)
|
||||
total = sum(c["duration"] for c in result["clips"])
|
||||
assert result["total_duration"] == round(total, 1)
|
||||
|
||||
def test_fallback_orders_are_sequential(self):
|
||||
with patch("shared.ai_service.time.sleep"):
|
||||
result = _fallback_recommend_clips("plan1", "tmpl1", ["a1", "a2", "a3"], "one_take", 30.0)
|
||||
orders = [c["order"] for c in result["clips"]]
|
||||
assert orders == list(range(len(result["clips"])))
|
||||
|
||||
|
||||
# ── _call_ai_cover_service 测试 ───────────────────────────────────────────
|
||||
|
||||
|
||||
class TestAiCoverService:
|
||||
"""AI封面生成服务测试."""
|
||||
|
||||
def test_cover_type_upload(self):
|
||||
with patch("shared.ai_service.time.sleep"):
|
||||
result = _call_ai_cover_service("plan1", ["a1"], "upload")
|
||||
assert result["type"] == "upload"
|
||||
assert result["image_url"] == ""
|
||||
|
||||
def test_cover_type_manual_with_frame_time(self):
|
||||
with patch("shared.ai_service.time.sleep"):
|
||||
result = _call_ai_cover_service("plan1", ["a1"], "manual", frame_time=5.5)
|
||||
assert result["type"] == "manual"
|
||||
assert result["frame_time"] == 5.5
|
||||
assert "5.5" in result["image_url"]
|
||||
|
||||
def test_cover_type_ai_frame(self):
|
||||
with patch("shared.ai_service.time.sleep"):
|
||||
with patch("shared.ai_service.random.uniform", side_effect=[5.0, 0.9]):
|
||||
result = _call_ai_cover_service("plan1", ["a1"], "ai_frame")
|
||||
assert result["type"] == "ai_frame"
|
||||
assert result["frame_time"] == 5.0
|
||||
assert result["confidence"] == 0.9
|
||||
assert "plan1" in result["image_url"]
|
||||
|
||||
def test_cover_type_ai_regenerate(self):
|
||||
with patch("shared.ai_service.time.sleep"):
|
||||
result = _call_ai_cover_service("plan1", ["a1"], "ai_regenerate")
|
||||
assert result["type"] == "ai_frame"
|
||||
|
||||
def test_cover_frame_time_in_range(self):
|
||||
with patch("shared.ai_service.time.sleep"):
|
||||
result = _call_ai_cover_service("plan1", ["a1"], "ai_frame")
|
||||
assert 1.0 <= result["frame_time"] <= 10.0
|
||||
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user