diff --git a/tests/unit/test_config_schemas_and_ai_endpoints.py b/tests/unit/test_config_schemas_and_ai_endpoints.py index 314d38541..5accbc0df 100644 --- a/tests/unit/test_config_schemas_and_ai_endpoints.py +++ b/tests/unit/test_config_schemas_and_ai_endpoints.py @@ -231,18 +231,17 @@ class TestAIRunTasks: # 即使没有素材,也应该有 intro + outro assert len(result["clips"]) >= 2 - def test_run_generate_cover_ai_frame(self): + def test_run_generate_cover_ai_frame_raises_without_mediakit(self): + """ai_frame cover raises RuntimeError when MediaKit is unavailable.""" from apps.worker.worker_app.tasks.ai_tasks import run_generate_cover + import pytest - result = run_generate_cover( - plan_id="plan-001", - asset_ids=["asset-1"], - cover_type="ai_frame", - ) - assert result["type"] == "ai_frame" - assert "image_url" in result - assert "frame_time" in result - assert "confidence" in result + with pytest.raises(RuntimeError, match="MediaKit"): + run_generate_cover( + plan_id="plan-001", + asset_ids=["asset-1"], + cover_type="ai_frame", + ) def test_run_generate_cover_manual(self): from apps.worker.worker_app.tasks.ai_tasks import run_generate_cover diff --git a/tests/unit/test_generation_preview.py b/tests/unit/test_generation_preview.py index 1848c92b2..4c48b049b 100644 --- a/tests/unit/test_generation_preview.py +++ b/tests/unit/test_generation_preview.py @@ -630,7 +630,7 @@ class TestToPreviewResponse: assert resp.file_size == 0 def test_completed_task_with_videos(self): - """已完成任务,带视频结果(URL签名后返回)""" + """已完成任务,带视频结果(裸URL直接返回,rendered/*已公开读)""" task = _make_task( status=GenerationTaskStatus.COMPLETED, progress=100.0, @@ -639,12 +639,8 @@ class TestToPreviewResponse: video.file_url = "https://cdn.example.com/preview.mp4" video.duration = 30.5 video.file_size = 1024000 - # Mock storage service to return a signed URL - mock_storage = MagicMock() - mock_storage.get_download_url.return_value = "https://cdn.example.com/preview.mp4?sig=test123" - with patch("app.api.routes.generation_preview.get_storage_service", return_value=mock_storage): - resp = _to_preview_response(task, generated_videos=[video]) - assert resp.video_url == "https://cdn.example.com/preview.mp4?sig=test123" + resp = _to_preview_response(task, generated_videos=[video]) + assert resp.video_url == "https://cdn.example.com/preview.mp4" assert resp.duration == 30.5 assert resp.file_size == 1024000 @@ -986,7 +982,7 @@ class TestGetPreviewRoute: # Mock URL 签名(返回带签名的 URL) mock_storage = MagicMock() - mock_storage.get_download_url.return_value = "https://cdn.example.com/preview_final.mp4?sig=abc123" + mock_storage.get_download_url.return_value = "https://cdn.example.com/preview_final.mp4" with patch("app.api.routes.generation_preview.GetGenerationTaskUseCase") as MockGet: MockGet.return_value.execute.return_value = task @@ -999,7 +995,7 @@ class TestGetPreviewRoute: generation_task_repository=repo, generated_video_repository=vid_repo, ) - assert resp.video_url == "https://cdn.example.com/preview_final.mp4?sig=abc123" + assert resp.video_url == "https://cdn.example.com/preview_final.mp4" assert resp.duration == 25.0 @@ -1088,50 +1084,6 @@ class TestWorkerPreviewResolution: # ═══════════════════════════════════════════════════════════════════════════════ -class TestSignVideoUrl: - """_sign_video_url 预签名 URL 测试。""" - - def test_empty_url_returns_empty(self): - """空 URL 直接返回空字符串。""" - from app.api.routes.generation_preview import _sign_video_url - - assert _sign_video_url("") == "" - - def test_signs_oss_url(self): - """OSS URL 应被签名。""" - from app.api.routes.generation_preview import _sign_video_url - - mock_storage = MagicMock() - mock_storage.get_download_url.return_value = "https://signed.example.com/video.mp4?sig=abc" - - with patch("app.api.routes.generation_preview.get_storage_service", return_value=mock_storage): - result = _sign_video_url("https://bucket.oss.example.com/video.mp4") - - assert result == "https://signed.example.com/video.mp4?sig=abc" - mock_storage.get_download_url.assert_called_once() - - def test_fallback_on_sign_failure(self): - """签名失败时降级返回原始 URL。""" - from app.api.routes.generation_preview import _sign_video_url - - with patch("app.api.routes.generation_preview.get_storage_service", side_effect=RuntimeError("no storage")): - result = _sign_video_url("https://bucket.oss.example.com/video.mp4") - - assert result == "https://bucket.oss.example.com/video.mp4" - - def test_sign_returns_none_fallback(self): - """get_download_url 返回 None 时降级返回原始 URL。""" - from app.api.routes.generation_preview import _sign_video_url - - mock_storage = MagicMock() - mock_storage.get_download_url.return_value = None - - with patch("app.api.routes.generation_preview.get_storage_service", return_value=mock_storage): - result = _sign_video_url("https://bucket.oss.example.com/video.mp4") - - assert result == "https://bucket.oss.example.com/video.mp4" - - class TestInferVideoRatioFromTemplate: """_infer_video_ratio_from_template 单元测试。""" diff --git a/tests/unit/test_mediakit_cover.py b/tests/unit/test_mediakit_cover.py index 22193e67e..7681d8340 100755 --- a/tests/unit/test_mediakit_cover.py +++ b/tests/unit/test_mediakit_cover.py @@ -167,39 +167,36 @@ class TestAICoverService: mock_client.extract_frames.assert_called_once() @patch("packages.shared.mediakit_client.get_mediakit_client") - def test_call_ai_cover_with_mediakit_failure_fallback(self, mock_get_client): - """MediaKit 失败时降级到 stub.""" + def test_call_ai_cover_with_mediakit_failure_raises(self, mock_get_client): + """MediaKit 失败时抛出 RuntimeError(不再降级到 stub).""" mock_client = Mock() mock_client.is_available = True mock_client.extract_frames.side_effect = Exception("API error") mock_get_client.return_value = mock_client from packages.shared.ai_service import _call_ai_cover_service + import pytest - result = _call_ai_cover_service( - plan_id="plan-123", - asset_ids=["asset-1"], - cover_type="ai_frame", - primary_video_url="https://example.com/video.mp4", - ) + with pytest.raises(RuntimeError, match="MediaKit"): + _call_ai_cover_service( + plan_id="plan-123", + asset_ids=["asset-1"], + cover_type="ai_frame", + primary_video_url="https://example.com/video.mp4", + ) - # 应该降级到 stub - assert result["type"] == "ai_frame" - assert result["image_url"].startswith("data:image/svg+xml,") - - def test_call_ai_cover_without_video_url_fallback(self): - """没有视频 URL 时使用 stub.""" + def test_call_ai_cover_without_video_url_raises(self): + """没有视频 URL 时抛出 RuntimeError(不再降级到 stub).""" from packages.shared.ai_service import _call_ai_cover_service + import pytest - result = _call_ai_cover_service( - plan_id="plan-123", - asset_ids=["asset-1"], - cover_type="ai_frame", - primary_video_url=None, - ) - - assert result["type"] == "ai_frame" - assert result["image_url"].startswith("data:image/svg+xml,") + with pytest.raises(RuntimeError, match="MediaKit"): + _call_ai_cover_service( + plan_id="plan-123", + asset_ids=["asset-1"], + cover_type="ai_frame", + primary_video_url=None, + ) def test_call_ai_cover_upload_type(self): """upload 类型直接返回.""" @@ -231,43 +228,41 @@ class TestAICoverService: assert result["frame_time"] == 5.0 @patch("packages.shared.mediakit_client.get_mediakit_client") - def test_call_ai_cover_mediakit_not_available(self, mock_get_client): - """MediaKit 未配置时使用 stub.""" + def test_call_ai_cover_mediakit_not_available_raises(self, mock_get_client): + """MediaKit 未配置时抛出 RuntimeError(不再降级到 stub).""" mock_client = Mock() mock_client.is_available = False mock_get_client.return_value = mock_client from packages.shared.ai_service import _call_ai_cover_service + import pytest - result = _call_ai_cover_service( - plan_id="plan-123", - asset_ids=["asset-1"], - cover_type="ai_frame", - primary_video_url="https://example.com/video.mp4", - ) - - assert result["type"] == "ai_frame" - assert result["image_url"].startswith("data:image/svg+xml,") + with pytest.raises(RuntimeError, match="MediaKit"): + _call_ai_cover_service( + plan_id="plan-123", + asset_ids=["asset-1"], + cover_type="ai_frame", + primary_video_url="https://example.com/video.mp4", + ) @patch("packages.shared.mediakit_client.get_mediakit_client") - def test_call_ai_cover_empty_frames_fallback(self, mock_get_client): - """MediaKit 返回空帧列表时降级.""" + def test_call_ai_cover_empty_frames_raises(self, mock_get_client): + """MediaKit 返回空帧列表时抛出 RuntimeError(不再降级).""" mock_client = Mock() mock_client.is_available = True mock_client.extract_frames.return_value = [] mock_get_client.return_value = mock_client from packages.shared.ai_service import _call_ai_cover_service + import pytest - result = _call_ai_cover_service( - plan_id="plan-123", - asset_ids=["asset-1"], - cover_type="ai_frame", - primary_video_url="https://example.com/video.mp4", - ) - - assert result["type"] == "ai_frame" - assert result["image_url"].startswith("data:image/svg+xml,") + with pytest.raises(RuntimeError, match="MediaKit"): + _call_ai_cover_service( + plan_id="plan-123", + asset_ids=["asset-1"], + cover_type="ai_frame", + primary_video_url="https://example.com/video.mp4", + ) class TestGenerateCover: diff --git a/tests/unit/test_shared_ai_service.py b/tests/unit/test_shared_ai_service.py index 7603a90bb..b9da45a95 100755 --- a/tests/unit/test_shared_ai_service.py +++ b/tests/unit/test_shared_ai_service.py @@ -434,34 +434,22 @@ class TestAiCoverService: assert result["image_url"].startswith("data:image/svg+xml,") assert "手动选帧" 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 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="MediaKit"): + _call_ai_cover_service("plan1", ["a1"], "ai_frame") - 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_type_ai_regenerate_raises_without_mediakit(self): + """ai_regenerate mode raises RuntimeError when MediaKit is unavailable.""" + with pytest.raises(RuntimeError, match="MediaKit"): + _call_ai_cover_service("plan1", ["a1"], "ai_regenerate") - def test_cover_frame_time_in_range(self): + 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"], "ai_frame") - assert 1.0 <= result["frame_time"] <= 10.0 - - def test_stub_returns_svg_data_uri(self): - """stub 降级返回 SVG data URI,不含任何后端 API 路径.""" - with patch("shared.ai_service.time.sleep"): - result = _call_ai_cover_service("plan1", ["a1"], "ai_frame") - assert result["image_url"].startswith("data:image/svg+xml,") - assert "/api/v1/" not in result["image_url"] - assert "1080" in result["image_url"] - assert "1920" in result["image_url"] + 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 占位图."""