"""MediaKit 重试逻辑 + 错误日志增强测试. 覆盖范围： 1. _is_retryable_error — 可重试错误判断 2. extract_frames 重试 — signal: killed / InternalError 自动重试一次 3. analyze_videos 重试 — 同上 4. 不可重试错误不触发重试 5. 默认抽帧策略为 TimeInterval """ from __future__ import annotations from unittest.mock import MagicMock, Mock, patch import pytest from packages.shared.mediakit_client import ( MediaKitClient, _is_retryable_error, ) # ── _is_retryable_error ────────────────────────────────────────────────────── class TestIsRetryableError: """可重试错误判断.""" def test_signal_killed(self): """signal: killed 可重试.""" assert _is_retryable_error("signal: killed") is True def test_internal_error(self): """InternalError 可重试.""" assert _is_retryable_error("InternalError: something went wrong") is True def test_oom(self): """OOM 可重试.""" assert _is_retryable_error("ExtractFrames OOM") is True def test_out_of_memory(self): """out of memory 可重试.""" assert _is_retryable_error("process out of memory") is True def test_model_timeout_not_retryable(self): """model timeout 不可重试.""" assert _is_retryable_error("model timeout") is False def test_network_error_not_retryable(self): """普通网络错误不可重试.""" assert _is_retryable_error("Connection refused") is False def test_empty_string(self): """空字符串不可重试.""" assert _is_retryable_error("") is False def test_case_insensitive(self): """大小写不敏感.""" assert _is_retryable_error("SIGNAL: KILLED") is True assert _is_retryable_error("internalerror") is True # ── extract_frames 重试逻辑 ────────────────────────────────────────────────── class TestExtractFramesRetry: """extract_frames 重试逻辑测试.""" def _make_client(self, api_key: str = "test-key") -> MediaKitClient: with patch("packages.shared.mediakit_client.get_shared_settings") as mock_settings: mock_settings.return_value = MagicMock( mediakit_api_key=api_key, mediakit_base_url="https://mock.mediakit.com/api/v1", mediakit_timeout=30, ) return MediaKitClient() @patch("packages.shared.mediakit_client.time.sleep") @patch("packages.shared.mediakit_client.httpx.get") @patch("packages.shared.mediakit_client.httpx.post") def test_retry_on_signal_killed_then_success(self, mock_post, mock_get, mock_sleep): """signal: killed 后重试成功.""" # 第一次提交成功 mock_post.return_value = MagicMock( status_code=200, json=lambda: {"task_id": "task-retry-1"}, raise_for_status=MagicMock(), ) # 第一次轮询返回 failed(signal: killed)，第二次返回 success call_count = [0] def get_side_effect(*args, **kwargs): call_count[0] += 1 if call_count[0] == 1: # 第一次轮询：失败 return MagicMock( status_code=200, json=lambda: {"status": "failed", "error": "signal: killed"}, raise_for_status=MagicMock(), ) else: # 第二次轮询（重试后）：成功 return MagicMock( status_code=200, json=lambda: { "status": "success", "result": {"snapshots": [{"image_url": "https://example.com/frame.jpg", "timestamp": 1.0}]}, }, raise_for_status=MagicMock(), ) mock_get.side_effect = get_side_effect client = self._make_client() frames = client.extract_frames( video_url="https://example.com/video.mp4", strategy="TimeInterval", max_frames=5, poll_interval=0.01, max_retries=1, ) assert frames is not None assert len(frames) == 1 assert frames[0]["image_url"] == "https://example.com/frame.jpg" # 验证 sleep 被调用（重试间隔） mock_sleep.assert_called() @patch("packages.shared.mediakit_client.time.sleep") @patch("packages.shared.mediakit_client.httpx.get") @patch("packages.shared.mediakit_client.httpx.post") def test_no_retry_on_non_retryable_error(self, mock_post, mock_get, mock_sleep): """不可重试错误直接返回 None，不重试.""" mock_post.return_value = MagicMock( status_code=200, json=lambda: {"task_id": "task-fail"}, raise_for_status=MagicMock(), ) mock_get.return_value = MagicMock( status_code=200, json=lambda: {"status": "failed", "error": "model timeout"}, raise_for_status=MagicMock(), ) client = self._make_client() frames = client.extract_frames( video_url="https://example.com/video.mp4", strategy="TimeInterval", max_frames=5, poll_interval=0.01, max_retries=1, ) assert frames is None # 不可重试错误不应该调用 sleep（不重试） mock_sleep.assert_not_called() @patch("packages.shared.mediakit_client.time.sleep") @patch("packages.shared.mediakit_client.httpx.get") @patch("packages.shared.mediakit_client.httpx.post") def test_retry_exhausted_returns_none(self, mock_post, mock_get, mock_sleep): """重试次数用尽后返回 None.""" mock_post.return_value = MagicMock( status_code=200, json=lambda: {"task_id": "task-fail-retry"}, raise_for_status=MagicMock(), ) # 每次都返回 signal: killed mock_get.return_value = MagicMock( status_code=200, json=lambda: {"status": "failed", "error": "signal: killed"}, raise_for_status=MagicMock(), ) client = self._make_client() frames = client.extract_frames( video_url="https://example.com/video.mp4", strategy="TimeInterval", max_frames=5, poll_interval=0.01, max_retries=1, # 最多重试 1 次 ) assert frames is None # post 被调用 2 次（原始 + 1 次重试） assert mock_post.call_count == 2 def test_default_strategy_is_time_interval(self): """默认抽帧策略为 TimeInterval（不是 SceneChange）.""" with patch("packages.shared.mediakit_client.httpx.post") as mock_post: with patch("packages.shared.mediakit_client.httpx.get") as mock_get: mock_post.return_value = MagicMock( status_code=200, json=lambda: {"task_id": "task-default"}, raise_for_status=MagicMock(), ) mock_get.return_value = MagicMock( status_code=200, json=lambda: { "status": "success", "result": {"snapshots": [{"image_url": "url", "timestamp": 0.0}]}, }, raise_for_status=MagicMock(), ) client = self._make_client() # 不传 strategy，使用默认值 client.extract_frames( video_url="https://example.com/video.mp4", poll_interval=0.01, ) # 验证提交时使用了 TimeInterval call_args = mock_post.call_args payload = call_args.kwargs["json"] assert payload["strategy"] == "TimeInterval" # ── analyze_videos 重试逻辑 ────────────────────────────────────────────────── class TestAnalyzeVideosRetry: """analyze_videos 重试逻辑测试.""" def _make_client(self, api_key: str = "test-key") -> MediaKitClient: with patch("packages.shared.mediakit_client.get_shared_settings") as mock_settings: mock_settings.return_value = MagicMock( mediakit_api_key=api_key, mediakit_base_url="https://mock.mediakit.com/api/v1", mediakit_timeout=30, ) return MediaKitClient() @patch("packages.shared.mediakit_client.time.sleep") @patch("packages.shared.mediakit_client.httpx.get") @patch("packages.shared.mediakit_client.httpx.post") def test_retry_on_internal_error_then_success(self, mock_post, mock_get, mock_sleep): """InternalError 后重试成功.""" mock_post.return_value = MagicMock( status_code=200, json=lambda: {"task_id": "task-retry-vu"}, raise_for_status=MagicMock(), ) call_count = [0] def get_side_effect(*args, **kwargs): call_count[0] += 1 if call_count[0] == 1: return MagicMock( status_code=200, json=lambda: {"status": "failed", "error": "InternalError: service busy"}, raise_for_status=MagicMock(), ) else: return MagicMock( status_code=200, json=lambda: { "status": "completed", "result": {"contents": ["视频展示了一只猫在沙发上睡觉"]}, }, raise_for_status=MagicMock(), ) mock_get.side_effect = get_side_effect client = self._make_client() result = client.analyze_videos( video_urls=["https://example.com/cat.mp4"], prompt="描述视频内容", poll_interval=0.01, max_retries=1, ) assert result is not None assert len(result) == 1 assert "猫" in result[0] mock_sleep.assert_called() @patch("packages.shared.mediakit_client.time.sleep") @patch("packages.shared.mediakit_client.httpx.get") @patch("packages.shared.mediakit_client.httpx.post") def test_no_retry_on_unknown_error(self, mock_post, mock_get, mock_sleep): """未知错误不重试.""" mock_post.return_value = MagicMock( status_code=200, json=lambda: {"task_id": "task-unknown-fail"}, raise_for_status=MagicMock(), ) mock_get.return_value = MagicMock( status_code=200, json=lambda: {"status": "failed", "error": "unknown error"}, raise_for_status=MagicMock(), ) client = self._make_client() result = client.analyze_videos( video_urls=["https://example.com/v.mp4"], prompt="describe", poll_interval=0.01, max_retries=1, ) assert result is None mock_sleep.assert_not_called() # post 只调用 1 次（不重试） assert mock_post.call_count == 1 # ── 错误日志增强 ───────────────────────────────────────────────────────────── class TestErrorLogging: """错误日志包含完整信息.""" def _make_client(self, api_key: str = "test-key") -> MediaKitClient: with patch("packages.shared.mediakit_client.get_shared_settings") as mock_settings: mock_settings.return_value = MagicMock( mediakit_api_key=api_key, mediakit_base_url="https://mock.mediakit.com/api/v1", mediakit_timeout=30, ) return MediaKitClient() @patch("packages.shared.mediakit_client.httpx.get") @patch("packages.shared.mediakit_client.httpx.post") def test_extract_frames_error_includes_task_id(self, mock_post, mock_get, caplog): """抽帧失败日志包含 task_id 和完整 error.""" import logging mock_post.return_value = MagicMock( status_code=200, json=lambda: {"task_id": "task-log-test"}, raise_for_status=MagicMock(), ) mock_get.return_value = MagicMock( status_code=200, json=lambda: { "status": "failed", "error": "signal: killed", "error_node": "ExtractFrames", }, raise_for_status=MagicMock(), ) client = self._make_client() with caplog.at_level(logging.ERROR): client.extract_frames( video_url="https://example.com/video.mp4", strategy="TimeInterval", max_frames=5, poll_interval=0.01, max_retries=0, # 不重试，直接失败 ) # 验证日志中包含 task_id 和 error 信息 error_logs = [r.message for r in caplog.records if r.levelno >= logging.ERROR] assert any("task-log-test" in msg for msg in error_logs) assert any("signal: killed" in msg for msg in error_logs) @patch("packages.shared.mediakit_client.httpx.get") @patch("packages.shared.mediakit_client.httpx.post") def test_analyze_videos_error_includes_full_error(self, mock_post, mock_get, caplog): """视频理解失败日志包含完整 error 和 error_node.""" import logging mock_post.return_value = MagicMock( status_code=200, json=lambda: {"task_id": "task-vu-log"}, raise_for_status=MagicMock(), ) mock_get.return_value = MagicMock( status_code=200, json=lambda: { "status": "failed", "error": "signal: killed in ExtractFrames", "error_node": "ExtractFrames", }, raise_for_status=MagicMock(), ) client = self._make_client() with caplog.at_level(logging.ERROR): client.analyze_videos( video_urls=["https://example.com/v.mp4"], prompt="describe", poll_interval=0.01, max_retries=0, ) error_logs = [r.message for r in caplog.records if r.levelno >= logging.ERROR] assert any("task-vu-log" in msg for msg in error_logs) assert any("ExtractFrames" in msg for msg in error_logs)