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

"""V4-M3: GeminiVideoClient.analyze relevance-only contract."""

from __future__ import annotations

import httpx

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


class FakeResponse:
    def __init__(self, content, *, status_code=200, json_payload=None, headers=None):
        self._content = content
        self._json_payload = json_payload
        self.status_code = status_code
        self.headers = headers or {"content-type": "application/json"}
        self.content = (
            content.encode("utf-8")
            if isinstance(content, str)
            else content or b""
        )
        self.request = httpx.Request("POST", "https://openrouter.test/chat/completions")

    def raise_for_status(self):
        if self.status_code >= 400:
            raise httpx.HTTPStatusError("bad", request=self.request, response=self)

    def json(self):
        if self._json_payload is not None:
            return self._json_payload
        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",
    "matched_search_queries": ["中医养生"],
}
_MEDIA = {"play_url": "http://v/x"}
_CTX = {"ext_data": {"evidence_pack": {"seed_terms": ["中医养生"]}}}


def test_analyze_returns_v4_relevance_fields():
    body = '{"query_relevance_score": 83, "query_relevance_reason": "贴切"}'

    result = _client(body).analyze(_ITEM, _MEDIA, _CTX)

    assert result["schema_version"] == "v4_gemini_query_relevance.v1"
    assert result["query_text"] == "中医养生"
    assert result["query_relevance_score"] == 83.0
    assert result["query_relevance_reason"] == "贴切"
    assert result["final_status"] == "ok"
    assert result["retry_count"] == 0
    assert result["timing_metrics"]["video_fetch"]["total_duration_ms"] >= 0
    assert result["timing_metrics"]["gemini_request"]["attempts"][0]["status"] == "ok"


def test_analyze_prompt_does_not_contain_legacy_50_plus_fields():
    seen = {}

    def post(*args, **kwargs):
        seen.update(kwargs)
        return FakeResponse('{"query_relevance_score": 70, "query_relevance_reason": "x"}')

    _client(post=post).analyze(_ITEM, _MEDIA, _CTX)

    prompt = seen["json"]["messages"][1]["content"][0]["text"]
    assert "fit_senior_50plus" not in prompt
    assert "fit_confidence" not in prompt
    assert '"relevance_score"' not in prompt
    assert "query_relevance_score" in prompt


def test_analyze_parses_json_in_markdown_fence_and_clamps_score():
    body = '```json\n{"query_relevance_score": 180, "query_relevance_reason": "x"}\n```'

    result = _client(body).analyze(_ITEM, _MEDIA, _CTX)

    assert result["query_relevance_score"] == 100.0


def test_analyze_does_not_upload_video_to_oss_by_default():
    body = '{"query_relevance_score": 80, "query_relevance_reason": "x"}'
    seen = {}

    def fetch(play_url, platform):
        seen["fetch"] = {"play_url": play_url, "platform": platform}
        return "data:video/mp4;base64,AAAA"

    client = GeminiVideoClient(
        api_key="k",
        fetch_fn=fetch,
        http_post=lambda *a, **k: FakeResponse(body),
    )
    item = {**_ITEM, "run_id": "run_1"}
    result = client.analyze(item, _MEDIA, _CTX)

    assert seen["fetch"] == {"play_url": "http://v/x", "platform": "douyin"}
    assert result["query_relevance_score"] == 80.0
    assert "media_storage_update" not in result
    assert result["timing_metrics"]["video_fetch"]["gemini_video_source"] == "play_url"


def test_analyze_prefers_oss_url_for_video_fetch():
    body = '{"query_relevance_score": 80, "query_relevance_reason": "x"}'
    seen = {}

    def fetch(play_url, platform):
        seen["fetch"] = {"play_url": play_url, "platform": platform}
        return "data:video/mp4;base64,AAAA"

    media = {**_MEDIA, "oss_url": "https://res.example/video.mp4"}
    result = _client(body, fetch=fetch).analyze(_ITEM, media, _CTX)

    assert seen["fetch"] == {"play_url": "https://res.example/video.mp4", "platform": "douyin"}
    assert result["query_relevance_score"] == 80.0
    assert result["timing_metrics"]["video_fetch"]["gemini_video_source"] == "oss_url"


def test_analyze_falls_back_to_play_url_when_oss_url_fetch_fails():
    body = '{"query_relevance_score": 80, "query_relevance_reason": "x"}'
    calls = []

    def fetch(play_url, platform):
        calls.append(play_url)
        if play_url == "https://res.example/video.mp4":
            raise RuntimeError("cdn failed")
        return "data:video/mp4;base64,AAAA"

    media = {**_MEDIA, "oss_url": "https://res.example/video.mp4"}
    result = _client(body, fetch=fetch).analyze(_ITEM, media, _CTX)

    assert calls == ["https://res.example/video.mp4", "http://v/x"]
    assert result["query_relevance_score"] == 80.0
    assert result["timing_metrics"]["video_fetch"]["gemini_video_source"] == "play_url"
    assert result["timing_metrics"]["video_fetch"]["attempts"][0]["gemini_video_source"] == "oss_url"
    assert result["timing_metrics"]["video_fetch"]["attempts"][0]["error_message"] == "cdn failed"


def test_analyze_video_fetch_failure_after_oss_and_play_url_returns_v4_fail():
    def fetch(play_url, platform):
        raise RuntimeError(f"dl {play_url}")

    media = {**_MEDIA, "oss_url": "https://res.example/video.mp4"}
    result = _client("{}", fetch=fetch).analyze(_ITEM, media, _CTX)

    assert result["final_status"] == "failed"
    assert result["failure_type"] == "video_fetch_failed"
    assert result["exception_type"] == "RuntimeError"
    assert result["error_message"] == "dl http://v/x"
    assert result["timing_metrics"]["video_fetch"]["gemini_video_source"] == "play_url"
    assert [item["gemini_video_source"] for item in result["timing_metrics"]["video_fetch"]["attempts"]] == [
        "oss_url",
        "play_url",
    ]


def test_analyze_timeout_returns_client_timeout_failure():
    body = '{"query_relevance_score": 80, "query_relevance_reason": "x"}'
    calls = []

    def post(*a, **k):
        calls.append(1)
        raise httpx.ReadTimeout("slow")

    client = GeminiVideoClient(
        api_key="k",
        fetch_fn=lambda play_url, platform: "data:video/mp4;base64,AAAA",
        http_post=post,
    )

    result = client.analyze(_ITEM, _MEDIA, _CTX)

    assert len(calls) == 2
    assert result["failure_type"] == "gemini_client_timeout"
    assert result["exception_type"] == "ReadTimeout"
    assert result["error_message"] == "slow"
    assert result["retry_count"] == 2


def test_analyze_no_play_url_returns_v4_fail():
    result = _client("{}").analyze(_ITEM, {}, _CTX)

    assert result["final_status"] == "failed"
    assert result["failure_type"] == "no_play_url"
    assert result["retry_count"] == 1


def test_analyze_no_valid_play_url_preserves_media_failure_reason():
    result = _client("{}").analyze(
        _ITEM,
        {
            "failure_reason": "no_valid_play_url",
            "raw_payload": {"no_valid_play_url": True},
        },
        _CTX,
    )

    assert result["final_status"] == "failed"
    assert result["failure_type"] == "no_valid_play_url"
    assert result["media_storage_update"]["failure_reason"] == "no_valid_play_url"


def test_analyze_video_fetch_failure_returns_v4_fail():
    def boom(play_url, platform):
        raise RuntimeError("dl")

    result = _client("{}", fetch=boom).analyze(_ITEM, _MEDIA, _CTX)

    assert result["final_status"] == "failed"
    assert result["failure_type"] == "video_fetch_failed"
    assert result["exception_type"] == "RuntimeError"
    assert result["error_message"] == "dl"


def test_analyze_retryable_http_error_retries_once_then_fails():
    calls = []

    def post(*a, **k):
        calls.append(1)
        return FakeResponse("{}", status_code=500)

    result = _client(post=post).analyze(_ITEM, _MEDIA, _CTX)

    assert len(calls) == 2
    assert result["failure_type"] == "gemini_http_error"
    assert result["http_status_code"] == 500
    assert result["error_message"] == "bad"
    assert result["retry_count"] == 2


def test_analyze_http_error_keeps_response_body_summary():
    calls = []

    def post(*a, **k):
        calls.append(1)
        return FakeResponse(
            '{"error":{"message":"upstream overloaded"}}',
            status_code=502,
            json_payload={"error": {"message": "upstream overloaded"}, "provider": "x"},
        )

    result = _client(post=post).analyze(_ITEM, _MEDIA, _CTX)

    assert len(calls) == 2
    assert result["failure_type"] == "gemini_http_error"
    assert result["response_body_summary"]["http_status_code"] == 502
    assert result["response_body_summary"]["json_top_level_keys"] == ["error", "provider"]
    assert result["response_body_summary"]["text_excerpt"] == "upstream overloaded"
    attempts = result["timing_metrics"]["gemini_request"]["attempts"]
    assert attempts[-1]["response_body_summary"]["text_excerpt"] == "upstream overloaded"


def test_analyze_response_body_summary_redacts_sensitive_text():
    def post(*a, **k):
        return FakeResponse(
            '{"error":{"message":"failed https://x.test/video.mp4?token=secret Bearer abc.def sk-testsecret1234567890"}}',
            status_code=502,
            json_payload={
                "error": {
                    "message": "failed https://x.test/video.mp4?token=secret Bearer abc.def sk-testsecret1234567890"
                }
            },
        )

    result = _client(post=post).analyze(_ITEM, _MEDIA, _CTX)

    excerpt = result["response_body_summary"]["text_excerpt"]
    assert "token=secret" not in excerpt
    assert "abc.def" not in excerpt
    assert "sk-testsecret" not in excerpt
    assert "https://x.test/video.mp4?[REDACTED]" in excerpt
    assert "Bearer [REDACTED]" in excerpt
    assert "sk-[REDACTED]" in excerpt


def test_analyze_bad_json_retries_once_then_fails():
    calls = []

    def post(*a, **k):
        calls.append(1)
        return FakeResponse("not-json")

    result = _client(post=post).analyze(_ITEM, _MEDIA, _CTX)

    assert len(calls) == 2
    assert result["failure_type"] == "gemini_response_invalid"
    assert result["retry_count"] == 2


def test_analyze_missing_choices_keeps_response_body_summary():
    calls = []

    def post(*a, **k):
        calls.append(1)
        return FakeResponse(
            '{"error":{"message":"missing choices body"}}',
            json_payload={"error": {"message": "missing choices body"}, "id": "r1"},
        )

    result = _client(post=post).analyze(_ITEM, _MEDIA, _CTX)

    assert len(calls) == 2
    assert result["failure_type"] == "gemini_response_invalid"
    assert result["exception_type"] == "KeyError"
    assert result["response_body_summary"]["json_top_level_keys"] == ["error", "id"]
    assert result["response_body_summary"]["text_excerpt"] == "missing choices body"


def test_from_env_missing_key_returns_missing_client():
    client = GeminiVideoClient.from_env({})

    assert isinstance(client, MissingGeminiVideoClient)
    result = client.analyze(_ITEM, _MEDIA, _CTX)
    assert result["final_status"] == "failed"
    assert result["failure_type"] == "gemini_config_missing"