"""Worker AI 任务单元测试. 测试覆盖： - AI推荐（豆包调用成功/失败/降级） - 推荐响应解析（多种格式） - 封面生成降级 """ from __future__ import annotations import json import sys import unittest from unittest.mock import MagicMock, patch sys.path.insert(0, "apps/worker") sys.path.insert(0, "packages") from worker_app.tasks.ai_tasks import ( # noqa: E402 _fallback_recommend_clips, _parse_recommend_response, run_ai_recommend, run_generate_cover, ) class TestFallbackRecommend(unittest.TestCase): """降级推荐方案测试.""" def test_fallback_returns_expected_structure(self): """降级推荐返回正确结构.""" result = _fallback_recommend_clips( plan_id="plan-1", template_id="tpl-1", asset_ids=["a1", "a2", "a3"], editing_mode="one_take", target_duration=30.0, ) self.assertIn("clips", result) self.assertIn("config", result) self.assertIn("total_duration", result) self.assertIn("confidence", result) def test_fallback_clips_structure(self): """每个片段都有必要字段.""" result = _fallback_recommend_clips( plan_id="plan-1", template_id="tpl-1", asset_ids=["a1", "a2"], editing_mode="one_take", target_duration=20.0, ) clips = result["clips"] self.assertTrue(len(clips) >= 3) # intro + showcase + outro for clip in clips: self.assertIn("clip_type", clip) self.assertIn("order", clip) self.assertIn("text_content", clip) self.assertIn("duration", clip) self.assertIn("transition_effect", clip) self.assertIn("asset_id", clip) self.assertIn("start_time", clip) self.assertIn("config", clip) def test_fallback_first_is_intro_last_is_outro(self): """第一个是开场，最后一个是结尾.""" result = _fallback_recommend_clips( plan_id="plan-1", template_id="tpl-1", asset_ids=["a1", "a2", "a3"], editing_mode="one_take", target_duration=30.0, ) clips = result["clips"] self.assertEqual(clips[0]["clip_type"], "intro") self.assertEqual(clips[-1]["clip_type"], "outro") def test_fallback_order_sequential(self): """order 连续递增.""" result = _fallback_recommend_clips( plan_id="plan-1", template_id="tpl-1", asset_ids=["a1", "a2"], editing_mode="one_take", target_duration=30.0, ) for i, clip in enumerate(result["clips"]): self.assertEqual(clip["order"], i) def test_fallback_empty_assets(self): """空素材列表也能生成.""" result = _fallback_recommend_clips( plan_id="plan-1", template_id="tpl-1", asset_ids=[], editing_mode="one_take", target_duration=10.0, ) self.assertTrue(len(result["clips"]) >= 2) def test_fallback_confidence_in_range(self): """置信度在0-1之间.""" result = _fallback_recommend_clips( plan_id="plan-1", template_id="tpl-1", asset_ids=["a1"], editing_mode="one_take", target_duration=10.0, ) self.assertGreaterEqual(result["confidence"], 0.0) self.assertLessEqual(result["confidence"], 1.0) class TestRecommendResponseParsing(unittest.TestCase): """推荐响应解析测试.""" def _asset_ids(self): return ["a1", "a2", "a3"] def test_parse_valid_response(self): """解析正常响应.""" data = { "clips": [ { "clip_type": "intro", "order": 0, "text_content": "开场", "duration": 3.0, "transition_effect": "fade", "asset_id": "a1", "start_time": 0.0, "config": {}, }, { "clip_type": "showcase", "order": 1, "text_content": "展示", "duration": 5.0, "transition_effect": "cut", "asset_id": "a2", "start_time": 1.0, "config": {}, }, { "clip_type": "outro", "order": 2, "text_content": "结尾", "duration": 2.0, "transition_effect": "fade", "asset_id": "", "start_time": 0.0, "config": {}, }, ], "title": "精彩视频", "confidence": 0.85, } result = _parse_recommend_response(json.dumps(data), self._asset_ids(), 30.0) self.assertIsNotNone(result) self.assertEqual(len(result["clips"]), 3) self.assertEqual(result["clips"][0]["clip_type"], "intro") self.assertEqual(result["confidence"], 0.85) self.assertIn("精彩视频", result["config"].get("title", {}).get("text", "")) def test_parse_markdown_code_block(self): """解析markdown代码块.""" data = { "clips": [ { "clip_type": "showcase", "order": 0, "text_content": "t", "duration": 3, "transition_effect": "cut", "asset_id": "a1", "start_time": 0, "config": {}, } ], "confidence": 0.7, } content = "```json\n" + json.dumps(data) + "\n```" result = _parse_recommend_response(content, self._asset_ids(), 30.0) self.assertIsNotNone(result) self.assertEqual(len(result["clips"]), 1) def test_parse_empty_content(self): """空内容返回None.""" result = _parse_recommend_response("", self._asset_ids(), 30.0) self.assertIsNone(result) def test_parse_invalid_json(self): """无效JSON返回None.""" result = _parse_recommend_response("不是json", self._asset_ids(), 30.0) self.assertIsNone(result) def test_parse_no_clips(self): """无clips字段返回None.""" result = _parse_recommend_response(json.dumps({"title": "abc"}), self._asset_ids(), 30.0) self.assertIsNone(result) def test_parse_filters_invalid_asset_ids(self): """过滤不在输入列表中的asset_id.""" data = { "clips": [ { "clip_type": "showcase", "order": 0, "text_content": "t", "duration": 3, "transition_effect": "cut", "asset_id": "fake-id", "start_time": 0, "config": {}, } ], "confidence": 0.7, } result = _parse_recommend_response(json.dumps(data), self._asset_ids(), 30.0) self.assertIsNotNone(result) # 非法asset_id被清空 self.assertEqual(result["clips"][0]["asset_id"], "") def test_parse_clamps_duration(self): """时长被限制在合理范围.""" data = { "clips": [ { "clip_type": "showcase", "order": 0, "text_content": "t", "duration": 100, "transition_effect": "cut", "asset_id": "a1", "start_time": 0, "config": {}, } ] } result = _parse_recommend_response(json.dumps(data), self._asset_ids(), 30.0) self.assertIsNotNone(result) self.assertLessEqual(result["clips"][0]["duration"], 30.0) def test_parse_reorders_clips(self): """clips按order排序并重新编号.""" data = { "clips": [ { "clip_type": "showcase", "order": 5, "text_content": "b", "duration": 3, "transition_effect": "cut", "asset_id": "a2", "start_time": 0, "config": {}, }, { "clip_type": "intro", "order": 0, "text_content": "a", "duration": 3, "transition_effect": "fade", "asset_id": "a1", "start_time": 0, "config": {}, }, ] } result = _parse_recommend_response(json.dumps(data), self._asset_ids(), 30.0) self.assertIsNotNone(result) # 第一个应该是order=0的intro self.assertEqual(result["clips"][0]["clip_type"], "intro") # order被重新编号为连续 self.assertEqual(result["clips"][0]["order"], 0) self.assertEqual(result["clips"][1]["order"], 1) def test_parse_confidence_clamped(self): """confidence被限制在0-1.""" data = { "clips": [ { "clip_type": "showcase", "order": 0, "text_content": "t", "duration": 3, "transition_effect": "cut", "asset_id": "a1", "start_time": 0, "config": {}, } ], "confidence": 2.5, } result = _parse_recommend_response(json.dumps(data), self._asset_ids(), 30.0) self.assertIsNotNone(result) self.assertLessEqual(result["confidence"], 1.0) class TestRunAIRecommend(unittest.TestCase): """run_ai_recommend 集成测试.""" def test_fallback_when_client_unavailable(self): """客户端不可用时走降级.""" mock_client = MagicMock() mock_client.is_available = False mock_client.chat_completion = MagicMock(return_value=None) with patch("packages.shared.ai_service.get_doubao_client", return_value=mock_client): result = run_ai_recommend( plan_id="plan-1", template_id="tpl-1", asset_ids=["a1", "a2"], editing_mode="one_take", target_duration=20.0, ) self.assertIn("clips", result) self.assertIn("total_duration", result) mock_client.chat_completion.assert_not_called() def test_doubao_success(self): """豆包调用成功路径.""" mock_client = MagicMock() mock_client.is_available = True mock_response = { "clips": [ { "clip_type": "intro", "order": 0, "text_content": "开场", "duration": 3.0, "transition_effect": "fade", "asset_id": "a1", "start_time": 0.0, "config": {}, }, { "clip_type": "outro", "order": 1, "text_content": "结尾", "duration": 2.0, "transition_effect": "fade", "asset_id": "a2", "start_time": 0.0, "config": {}, }, ], "title": "AI生成标题", "confidence": 0.9, } mock_client.chat_completion = MagicMock(return_value=json.dumps(mock_response)) with patch("packages.shared.ai_service.get_doubao_client", return_value=mock_client): result = run_ai_recommend( plan_id="plan-1", template_id="tpl-1", asset_ids=["a1", "a2"], editing_mode="one_take", target_duration=30.0, ) self.assertEqual(result["confidence"], 0.9) self.assertEqual(len(result["clips"]), 2) mock_client.chat_completion.assert_called_once() def test_doubao_failure_fallback(self): """豆包调用失败降级.""" mock_client = MagicMock() mock_client.is_available = True mock_client.chat_completion = MagicMock(return_value=None) with patch("packages.shared.ai_service.get_doubao_client", return_value=mock_client): result = run_ai_recommend( plan_id="plan-1", template_id="tpl-1", asset_ids=["a1"], editing_mode="one_take", target_duration=10.0, ) # 降级后有结果 self.assertTrue(len(result["clips"]) >= 2) mock_client.chat_completion.assert_called_once() def test_doubao_unparseable_fallback(self): """豆包返回无法解析时降级.""" mock_client = MagicMock() mock_client.is_available = True mock_client.chat_completion = MagicMock(return_value="一堆废话不是json") with patch("packages.shared.ai_service.get_doubao_client", return_value=mock_client): result = run_ai_recommend( plan_id="plan-1", template_id="tpl-1", asset_ids=["a1"], editing_mode="one_take", target_duration=10.0, ) # 降级后有结果 self.assertTrue(len(result["clips"]) >= 2) class TestGenerateCover(unittest.TestCase): """封面生成测试.""" def test_ai_frame_type_raises_without_mediakit(self): """AI封面模式在MediaKit不可用时抛出RuntimeError.""" with self.assertRaises(RuntimeError): run_generate_cover( plan_id="plan-1", asset_ids=["a1"], cover_type="ai_frame", ) def test_manual_type(self): """手动选帧模式.""" result = run_generate_cover( plan_id="plan-1", asset_ids=["a1"], cover_type="manual", frame_time=5.0, ) self.assertEqual(result["type"], "manual") self.assertEqual(result["frame_time"], 5.0) def test_upload_type(self): """上传封面模式.""" result = run_generate_cover( plan_id="plan-1", asset_ids=["a1"], cover_type="upload", ) self.assertEqual(result["type"], "upload") if __name__ == "__main__": unittest.main()