"""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"]