xiaoxia-saas/tests/unit/test_mediakit_retry.py

"""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)