content-find-agent-new/tests/test_video_fetch.py

"""V3-M2A: video_fetch download + compress + data URL (all mocked)."""

from __future__ import annotations

import base64

import httpx
import pytest

from content_agent.integrations import video_fetch
from content_agent.integrations.video_fetch import VideoFetchError, fetch_and_compress


class FakeHttpClient:
    def __init__(self, *, content=b"rawvideo", status=200, error=None):
        self.content = content
        self.status = status
        self.error = error
        self.calls = []

    def get(self, url, *, headers, follow_redirects, timeout):
        self.calls.append({"url": url, "headers": headers})
        if self.error:
            raise self.error
        return httpx.Response(
            self.status, content=self.content, request=httpx.Request("GET", url)
        )


def _patch_compress(monkeypatch, out=b"\x00\x00\x00\x18ftypmp42compressed"):
    monkeypatch.setattr(video_fetch, "_compress", lambda raw, exe, **kwargs: out)


def test_fetch_builds_data_url(monkeypatch):
    _patch_compress(monkeypatch)
    out = fetch_and_compress("http://v/x", "douyin", http_client=FakeHttpClient(), ffmpeg_exe="ffmpeg")
    assert out.startswith("data:video/mp4;base64,")
    decoded = base64.b64decode(out.split(",", 1)[1])
    assert decoded.startswith(b"\x00\x00\x00\x18ftyp")


def test_fetch_uses_platform_headers(monkeypatch):
    _patch_compress(monkeypatch)
    for platform, ua_token, referer in [
        ("douyin", "iPhone", "https://www.douyin.com/"),
        ("shipinhao", "Windows", "https://channels.weixin.qq.com/"),
    ]:
        client = FakeHttpClient()
        fetch_and_compress("http://v/x", platform, http_client=client, ffmpeg_exe="ffmpeg")
        headers = client.calls[0]["headers"]
        assert ua_token in headers["User-Agent"]
        assert headers["Referer"] == referer


def test_fetch_missing_play_url_raises():
    with pytest.raises(VideoFetchError, match="missing play_url"):
        fetch_and_compress("", "douyin", http_client=FakeHttpClient())


def test_fetch_download_failure_raises():
    client = FakeHttpClient(error=httpx.ConnectError("boom"))
    with pytest.raises(VideoFetchError, match="download failed"):
        fetch_and_compress("http://v/x", "douyin", http_client=client, ffmpeg_exe="ffmpeg")


def test_fetch_enforces_total_download_timeout(monkeypatch):
    _patch_compress(monkeypatch)
    ticks = iter([0.0, 0.0, 31.0, 31.0])

    with pytest.raises(VideoFetchError, match="download timeout") as exc_info:
        fetch_and_compress(
            "http://v/x",
            "douyin",
            http_client=FakeHttpClient(),
            ffmpeg_exe="ffmpeg",
            timeout_seconds=30.0,
            clock=lambda: next(ticks),
        )
    assert exc_info.value.metrics["total_duration_ms"] == 31000


def test_fetch_oversize_raises(monkeypatch):
    _patch_compress(monkeypatch, out=b"x" * (video_fetch.MAX_INLINE_BYTES + 1))
    with pytest.raises(VideoFetchError, match="oversize"):
        fetch_and_compress("http://v/x", "douyin", http_client=FakeHttpClient(), ffmpeg_exe="ffmpeg")