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


def test_fetch_saves_raw_video_when_requested(monkeypatch, tmp_path):
    # 2026-06-12 拍板: save_raw_to 指定时原片落盘(自动建父目录);默认不落盘。
    _patch_compress(monkeypatch)
    save_path = tmp_path / "run_1" / "c1.mp4"
    fetch_and_compress(
        "http://v/x", "douyin",
        http_client=FakeHttpClient(content=b"rawvideo"),
        ffmpeg_exe="ffmpeg",
        save_raw_to=str(save_path),
    )
    assert save_path.read_bytes() == b"rawvideo"

    fetch_and_compress("http://v/x", "douyin", http_client=FakeHttpClient(), ffmpeg_exe="ffmpeg")
    assert list(tmp_path.iterdir()) == [save_path.parent]