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

"""通义千问视频 client:provider 开关 + video_url 取 oss + 解析。"""

import json

import httpx

from content_agent.integrations.qwen_video import QwenVideoClient
from content_agent.run_service import _gemini_video_client_from_env
from content_agent.integrations.gemini_video import GeminiVideoClient
from content_agent.integrations import timeout_config


class _FakeResponse:
    def __init__(self, payload, *, status_code=200, headers=None):
        self._payload = payload
        self.status_code = status_code
        self.headers = headers or {"content-type": "application/json"}
        self.content = json.dumps(payload).encode()
        self.request = httpx.Request("POST", "https://dashscope.test/chat/completions")

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

    def json(self):
        return self._payload


def test_provider_switch_picks_qwen_vs_openrouter():
    qwen = _gemini_video_client_from_env({"CONTENT_AGENT_VIDEO_LLM_PROVIDER": "dashscope", "CONTENT_AGENT_VIDEO_LLM_API_KEY": "k"})
    assert isinstance(qwen, QwenVideoClient)
    openrouter = _gemini_video_client_from_env({"OPENROUTER_API_KEY": "k"})
    assert isinstance(openrouter, GeminiVideoClient)


def test_qwen_analyze_sends_video_url_oss_and_parses():
    captured = {}

    def fake_post(url, headers=None, json=None, timeout=None):
        captured["url"] = url
        captured["json"] = json
        captured["timeout"] = timeout
        return _FakeResponse(
            {"choices": [{"message": {"content": '{"query_relevance_score": 88, "query_relevance_reason": "贴合"}'}}]}
        )

    client = QwenVideoClient(api_key="k", model="qwen3-vl-plus", http_post=fake_post)
    out = client.analyze(
        {"platform": "douyin", "search_query": "拉伸"},
        {"oss_url": "https://res.example.com/v.mp4", "play_url": "https://douyinvod.com/x"},
        {},
    )
    assert out["final_status"] == "ok"
    assert out["query_relevance_score"] == 88
    # 发的是 dashscope chat/completions,且用 video_url 取 oss(不是 play_url、不是 base64)
    assert captured["url"].endswith("/chat/completions")
    content = captured["json"]["messages"][1]["content"]
    video_part = next(p for p in content if p["type"] == "video_url")
    assert video_part["video_url"]["url"] == "https://res.example.com/v.mp4"
    assert captured["timeout"].read == timeout_config.read_timeout("video_llm") == 120.0


def test_content_inspection_split_detection():
    from content_agent.flow_ledger_service import _is_content_inspection_blocked
    # 新数据:独立 failure_type
    assert _is_content_inspection_blocked({"failure_type": "content_inspection_blocked"}) is True
    # 旧数据(改之前跑的 run):从响应体文本兜底识别
    assert _is_content_inspection_blocked(
        {"failure_type": "qwen_http_error", "openrouter": {"response_summary": {"text_excerpt": "<400> InternalError.Algo.DataInspectionFailed: Input video data may contain inappropriate content."}}}
    ) is True
    # 普通技术问题不误判
    assert _is_content_inspection_blocked({"failure_type": "no_valid_play_url"}) is False
    assert _is_content_inspection_blocked(None) is False


def test_qwen_missing_oss_returns_failed():
    client = QwenVideoClient(api_key="k", http_post=lambda *a, **k: None)
    out = client.analyze({"platform": "douyin"}, {"play_url": "https://douyinvod.com/x"}, {})
    assert out["final_status"] == "failed"  # 只有 play_url(模型取不到)→ 视作无源


def test_qwen_429_retries_three_times_and_keeps_retry_after():
    calls = []

    def fake_post(*args, **kwargs):
        calls.append(kwargs)
        return _FakeResponse(
            {"error": {"message": "too many"}},
            status_code=429,
            headers={"content-type": "application/json", "retry-after": "7"},
        )

    client = QwenVideoClient(api_key="k", http_post=fake_post)
    out = client.analyze({"platform": "douyin"}, {"oss_url": "https://res.example.com/v.mp4"}, {})

    assert len(calls) == 3
    assert out["failure_type"] == "qwen_http_error"
    assert out["http_status_code"] == 429
    assert out["qwen_retry_after_seconds"] == 7.0
    assert out["retry_count"] == 3
    assert out["response_body_summary"]["retry_after"] == "7"


def test_qwen_content_inspection_does_not_retry():
    calls = []

    def fake_post(*args, **kwargs):
        calls.append(kwargs)
        return _FakeResponse(
            {"error": {"message": "InternalError.Algo.DataInspectionFailed: inappropriate content"}},
            status_code=400,
        )

    client = QwenVideoClient(api_key="k", http_post=fake_post)
    out = client.analyze({"platform": "douyin"}, {"oss_url": "https://res.example.com/v.mp4"}, {})

    assert len(calls) == 1
    assert out["failure_type"] == "content_inspection_blocked"
    assert out.get("qwen_retryable") is False