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

"""V3-M2B: GeminiVideoClient.analyze (mocked fetch + httpx)."""

from __future__ import annotations

import httpx

from content_agent.integrations.gemini_video import (
    GeminiVideoClient,
    MissingGeminiVideoClient,
)


class FakeResponse:
    def __init__(self, content):
        self._content = content

    def raise_for_status(self):
        return None

    def json(self):
        return {"choices": [{"message": {"content": self._content}}]}


def _client(content=None, *, post=None, fetch=None):
    return GeminiVideoClient(
        api_key="k",
        fetch_fn=fetch or (lambda play_url, platform: "data:video/mp4;base64,AAAA"),
        http_post=post or (lambda *a, **k: FakeResponse(content)),
    )


_ITEM = {"platform": "douyin", "platform_content_id": "c1"}
_MEDIA = {"play_url": "http://v/x"}
_CTX = {"ext_data": {"evidence_pack": {"seed_terms": ["中医养生"]}}}


def test_analyze_returns_four_fields():
    body = '{"fit_senior_50plus": true, "fit_confidence": 0.85, "relevance_score": 0.7, "reason": "贴切"}'
    result = _client(body).analyze(_ITEM, _MEDIA, _CTX)
    assert result == {
        "fit_senior_50plus": True,
        "fit_confidence": 0.85,
        "relevance_score": 0.7,
        "reason": "贴切",
    }


def test_analyze_parses_json_in_markdown_fence():
    body = '```json\n{"fit_senior_50plus": false, "fit_confidence": 0.4, "relevance_score": 0.2, "reason": "x"}\n```'
    result = _client(body).analyze(_ITEM, _MEDIA, _CTX)
    assert result["fit_senior_50plus"] is False
    assert result["fit_confidence"] == 0.4


def test_analyze_clamps_out_of_range_scores():
    body = '{"fit_senior_50plus": true, "fit_confidence": 1.7, "relevance_score": -3, "reason": "x"}'
    result = _client(body).analyze(_ITEM, _MEDIA, _CTX)
    assert result["fit_confidence"] == 1.0
    assert result["relevance_score"] == 0.0


def test_analyze_passes_raw_save_path_when_dir_configured(tmp_path):
    # 2026-06-12 拍板: 配置留档目录后,analyze 按 {dir}/{run_id}/{platform_content_id}.mp4
    # 把路径传给 fetch_fn;未配置(默认 None)不传该 kwarg——老签名 lambda 桩零改仍可用。
    body = '{"fit_senior_50plus": true, "fit_confidence": 0.9, "relevance_score": 0.8, "reason": "x"}'
    seen = {}

    def fetch(play_url, platform, **kwargs):
        seen.update(kwargs)
        return "data:video/mp4;base64,AAAA"

    client = GeminiVideoClient(
        api_key="k",
        fetch_fn=fetch,
        http_post=lambda *a, **k: FakeResponse(body),
        raw_video_save_dir=str(tmp_path),
    )
    item = {**_ITEM, "run_id": "run_1"}
    client.analyze(item, _MEDIA, _CTX)
    assert seen["save_raw_to"] == str(tmp_path / "run_1" / "c1.mp4")

    result = _client(body).analyze(item, _MEDIA, _CTX)
    assert result["fit_senior_50plus"] is True


def test_analyze_no_play_url_returns_fail():
    result = _client("{}").analyze(_ITEM, {}, _CTX)
    assert result["status"] == "failed"
    assert result["reason"] == "no_play_url"


def test_analyze_video_fetch_failure_returns_fail():
    def boom(play_url, platform):
        raise RuntimeError("dl")
    result = _client("{}", fetch=boom).analyze(_ITEM, _MEDIA, _CTX)
    assert result["status"] == "failed"
    assert "video_fetch_failed" in result["reason"]


def test_analyze_http_error_returns_fail():
    def post(*a, **k):
        raise httpx.ConnectError("boom")
    result = _client(post=post).analyze(_ITEM, _MEDIA, _CTX)
    assert result["status"] == "failed"
    assert "gemini_http_error" in result["reason"]


def test_analyze_bad_json_returns_fail():
    result = _client("not-json").analyze(_ITEM, _MEDIA, _CTX)
    assert result["status"] == "failed"
    assert "gemini_response_invalid" in result["reason"]


def test_from_env_missing_key_returns_missing_client():
    client = GeminiVideoClient.from_env({})
    assert isinstance(client, MissingGeminiVideoClient)
    assert client.analyze(_ITEM, _MEDIA, _CTX)["status"] == "failed"