xiaoxia
0301370dd8
CI/CD Pipeline / Check if frontend-only change (push) Has been skipped
CI/CD Pipeline / Frontend Lint (push) Has been skipped
CI/CD Pipeline / PR Build API Image (push) Has been skipped
CI/CD Pipeline / PR Build Web Image (push) Has been skipped
CI/CD Pipeline / PR Build Worker Image (push) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 1m32s
CI/CD Pipeline / Validate - Type Check (mypy) (push) Successful in 1m45s
CI/CD Pipeline / Validate - Migration (alembic) (push) Successful in 2m20s
CI/CD Pipeline / Build Staging Web Image (push) Successful in 2m22s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 5m2s
CI/CD Pipeline / Validate - Code Quality (push) Successful in 6m19s
CI/CD Pipeline / Integration Tests (push) Successful in 2m0s
CI/CD Pipeline / Unit Tests (push) Successful in 9m9s
CI/CD Pipeline / Build Production API Image (push) Has been skipped
CI/CD Pipeline / Build Production Web Image (push) Has been skipped
CI/CD Pipeline / Build Production Worker Image (push) Has been skipped
CI/CD Pipeline / CI Gate (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Build Staging API Image (push) Successful in 20m32s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 1m2s
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 41s
CI/CD Pipeline / Staging E2E Tests (push) Successful in 1m53s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 2m50s
CI/CD Pipeline / Canary Release to Production (push) Has been skipped
Co-authored-by: xiaoxia <dev@xiaoxiajianji.com> Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>
461 lines
17 KiB
Python
Executable File
461 lines
17 KiB
Python
Executable File
"""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 result["image_url"].startswith("data:image/svg+xml,")
|
|
assert "手动选帧" in result["image_url"]
|
|
|
|
def test_cover_type_ai_frame_raises_without_mediakit(self):
|
|
"""ai_frame mode raises RuntimeError when MediaKit is unavailable."""
|
|
with pytest.raises(RuntimeError, match="封面数据不可用"):
|
|
_call_ai_cover_service("plan1", ["a1"], "ai_frame")
|
|
|
|
def test_cover_type_ai_regenerate_raises_without_mediakit(self):
|
|
"""ai_regenerate mode raises RuntimeError when MediaKit is unavailable."""
|
|
with pytest.raises(RuntimeError, match="封面数据不可用"):
|
|
_call_ai_cover_service("plan1", ["a1"], "ai_regenerate")
|
|
|
|
def test_cover_type_manual_still_works(self):
|
|
"""manual mode does not require MediaKit and still returns stub."""
|
|
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
|
|
|
|
def test_manual_stub_returns_svg_data_uri(self):
|
|
"""manual 模式返回 SVG data URI 占位图."""
|
|
with patch("shared.ai_service.time.sleep"):
|
|
result = _call_ai_cover_service("plan1", ["a1"], "manual", frame_time=3.0)
|
|
assert result["image_url"].startswith("data:image/svg+xml,")
|
|
assert "/api/v1/" not in result["image_url"]
|
|
assert "手动选帧" in result["image_url"]
|