SupplyAgent/tests/supply_agent/test_qwen_video_analysis.py

"""Regression tests for the bounded Qwen video-analysis pipeline."""

from __future__ import annotations

import asyncio
import json
from pathlib import Path

import httpx
import pytest

from agents.find_agent.support import qwen_video_analysis as video_analysis


@pytest.mark.asyncio
async def test_download_retries_once_then_abandons(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    attempts = 0
    destination = tmp_path / "video.mp4"

    async def fail_download(url: str, dest: Path, timeout: float) -> None:
        nonlocal attempts
        attempts += 1
        dest.write_bytes(b"partial")
        raise httpx.ReadError("connection dropped")

    monkeypatch.setattr(video_analysis, "_download_video_once", fail_download)

    with pytest.raises(video_analysis.VideoDownloadError) as exc_info:
        await video_analysis._download_video(
            "https://source.example/video.mp4",
            destination,
            1.0,
        )

    assert attempts == 2
    assert "已尝试 2 次" in str(exc_info.value)
    assert not destination.exists()


@pytest.mark.asyncio
async def test_download_timeout_is_a_hard_per_attempt_deadline(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    attempts = 0

    async def stalled_download(url: str, dest: Path, timeout: float) -> None:
        nonlocal attempts
        attempts += 1
        await asyncio.sleep(1)

    monkeypatch.setattr(video_analysis, "_download_video_once", stalled_download)

    with pytest.raises(video_analysis.VideoDownloadError) as exc_info:
        await asyncio.wait_for(
            video_analysis._download_video(
                "https://source.example/video.mp4",
                tmp_path / "video.mp4",
                0.01,
            ),
            timeout=0.2,
        )

    assert attempts == 2
    assert exc_info.value.timed_out is True


@pytest.mark.asyncio
async def test_short_video_is_kept_complete_and_converted_to_oss(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    uploaded: dict[str, object] = {}

    async def download(url: str, dest: Path, timeout: float) -> None:
        dest.write_bytes(b"complete-short-video")

    def upload(path: Path, source_url: str, max_duration: float) -> str:
        uploaded["bytes"] = path.read_bytes()
        uploaded["max_duration"] = max_duration
        return "https://ours.example/video-clips/short_30s.mp4"

    monkeypatch.setattr(video_analysis, "_require_ffmpeg", lambda: "ffmpeg")
    monkeypatch.setattr(video_analysis, "_ensure_oss_configured", lambda: None)
    monkeypatch.setattr(video_analysis, "_download_video", download)
    monkeypatch.setattr(video_analysis, "_probe_duration", lambda ffmpeg, path: 12.5)
    monkeypatch.setattr(video_analysis, "_upload_clip", upload)
    monkeypatch.setattr(
        video_analysis,
        "_truncate_video",
        lambda *args: pytest.fail("short video must not be truncated"),
    )

    analysis_url, meta = await video_analysis._prepare_analysis_url(
        "https://source.example/short.mp4",
        None,
        1.0,
    )

    assert analysis_url == "https://ours.example/video-clips/short_30s.mp4"
    assert uploaded == {"bytes": b"complete-short-video", "max_duration": 30.0}
    assert meta["truncated"] is False
    assert meta["converted_to_oss"] is True
    assert meta["original_duration_seconds"] == 12.5
    assert meta["max_duration_seconds"] == 30.0


@pytest.mark.asyncio
async def test_long_video_is_clipped_to_30_seconds_and_converted_to_oss(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    truncated_at: list[float] = []
    uploaded_bytes: list[bytes] = []

    async def download(url: str, dest: Path, timeout: float) -> None:
        dest.write_bytes(b"long-video")

    def truncate(
        ffmpeg: str,
        source: Path,
        output: Path,
        max_duration: float,
    ) -> None:
        truncated_at.append(max_duration)
        output.write_bytes(b"first-30-seconds")

    def upload(path: Path, source_url: str, max_duration: float) -> str:
        uploaded_bytes.append(path.read_bytes())
        return "https://ours.example/video-clips/long_30s.mp4"

    monkeypatch.setattr(video_analysis, "_require_ffmpeg", lambda: "ffmpeg")
    monkeypatch.setattr(video_analysis, "_ensure_oss_configured", lambda: None)
    monkeypatch.setattr(video_analysis, "_download_video", download)
    monkeypatch.setattr(video_analysis, "_probe_duration", lambda ffmpeg, path: 90.0)
    monkeypatch.setattr(video_analysis, "_truncate_video", truncate)
    monkeypatch.setattr(video_analysis, "_upload_clip", upload)

    analysis_url, meta = await video_analysis._prepare_analysis_url(
        "https://source.example/long.mp4",
        180.0,
        1.0,
    )

    assert analysis_url == "https://ours.example/video-clips/long_30s.mp4"
    assert truncated_at == [30.0]
    assert uploaded_bytes == [b"first-30-seconds"]
    assert meta["truncated"] is True
    assert meta["converted_to_oss"] is True
    assert meta["max_duration_seconds"] == 30.0


@pytest.mark.asyncio
async def test_qwen_receives_only_the_converted_url(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    received_urls: list[str] = []

    async def prepare(*args: object, **kwargs: object) -> tuple[str, dict[str, object]]:
        return "https://ours.example/video.mp4", {
            "converted_to_oss": True,
            "original_video_url": "https://source.example/video.mp4",
        }

    def analyze(
        video_url: str,
        prompt: str,
        fps: float,
        model: str,
        timeout: float,
    ) -> str:
        received_urls.append(video_url)
        return "analysis"

    monkeypatch.setattr(video_analysis, "_prepare_analysis_url", prepare)
    monkeypatch.setattr(video_analysis, "_analyze_video_sync", analyze)

    result = json.loads(
        await video_analysis.qwen_video_analyze(
            "https://source.example/video.mp4",
            tool_timeout=1.0,
        )
    )

    assert received_urls == ["https://ours.example/video.mp4"]
    assert result["video_url"] == "https://ours.example/video.mp4"
    assert result["original_video_url"] == "https://source.example/video.mp4"


@pytest.mark.asyncio
async def test_whole_tool_has_a_hard_timeout(monkeypatch: pytest.MonkeyPatch) -> None:
    async def never_finishes(*args: object, **kwargs: object) -> tuple[str, dict[str, object]]:
        await asyncio.Event().wait()
        raise AssertionError("unreachable")

    monkeypatch.setattr(video_analysis, "_prepare_analysis_url", never_finishes)

    result = json.loads(
        await asyncio.wait_for(
            video_analysis.qwen_video_analyze(
                "https://source.example/video.mp4",
                tool_timeout=0.01,
            ),
            timeout=0.2,
        )
    )

    assert result["error_code"] == "tool_timeout"
    assert "视频解析工具超时" in result["error"]