test: fix unit tests for #1341 — remove _sign_video_url tests, update stub fallback to expect RuntimeError

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

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 单元测试。"""
 

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:

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 占位图."""