diff --git a/tests/unit/test_mediakit_retry.py b/tests/unit/test_mediakit_retry.py new file mode 100644 index 000000000..79e4dcabb --- /dev/null +++ b/tests/unit/test_mediakit_retry.py @@ -0,0 +1,401 @@ +"""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)