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

"""超时硬化 / 有界等待 / 僵尸线程清理 的单测(修永久卡死)。"""

from __future__ import annotations

import threading
import time

import httpx

from content_agent.business_modules.content_discovery import pattern_recall
from content_agent.business_modules.content_discovery.pattern_recall import recall_decision
from content_agent.integrations import (
    crawapi_http,
    oss_upload,
    timeout_config,
    video_fetch,
)
from content_agent.integrations.bounded_pool import DaemonThreadPoolExecutor, run_bounded
from content_agent.integrations.runtime_files import LocalRuntimeFileStore
from content_agent import flow_ledger_service as fls
from tests.gemini_helpers import FakeGeminiVideoClient, fake_gemini_pool


# ---------- timeout_config ----------

def test_total_timeout_defaults_match_user_caps():
    env = {}
    assert timeout_config.total_timeout("oss", env=env) == 300.0
    assert timeout_config.total_timeout("video_download", env=env) == 600.0
    assert timeout_config.total_timeout("video_llm", env=env) == 600.0
    assert timeout_config.total_timeout("crawapi", env=env) == 180.0
    assert timeout_config.total_timeout("query_llm", env=env) == 120.0
    assert timeout_config.total_timeout("pg", env=env) == 30.0


def test_total_timeout_env_override_and_hard_ceiling():
    assert timeout_config.total_timeout("oss", env={"CONTENT_AGENT_OSS_TIMEOUT_SECONDS": "120"}) == 120.0
    # env 想配 9999 也被硬上限钳到 600,杜绝再现 3600。
    assert timeout_config.total_timeout("oss", env={"CONTENT_AGENT_OSS_TIMEOUT_SECONDS": "9999"}) == 600.0
    # 坏值忽略,回默认。
    assert timeout_config.total_timeout("oss", env={"CONTENT_AGENT_OSS_TIMEOUT_SECONDS": "abc"}) == 300.0


def test_httpx_timeout_is_segmented_with_short_read():
    t = timeout_config.httpx_timeout("video_download", env={})
    assert isinstance(t, httpx.Timeout)
    assert t.read == 120.0          # read 短,停吐字节即抛
    assert t.write == 600.0         # write 承载总时长
    assert t.connect == timeout_config.CONNECT_TIMEOUT_SECONDS
    video_llm = timeout_config.httpx_timeout("video_llm", env={})
    assert video_llm.read == 120.0  # Qwen/Gemini 停吐字节 120s 即抛


def test_as_httpx_timeout_read_capped_by_total():
    t = timeout_config.as_httpx_timeout(5.0, read=60.0)
    assert t.read == 5.0            # read 不超过总时长
    assert t.write == 5.0


# ---------- bounded_pool ----------

def test_run_bounded_results_aligned_by_offset():
    items = [1, 2, 3, 4]
    out = run_bounded(items, lambda x: x * 10, max_workers=3, per_future_timeout=5.0, on_timeout=lambda i, o: -1)
    assert out == [10, 20, 30, 40]


def test_run_bounded_single_timeout_skips_and_does_not_hang():
    started = time.monotonic()

    def work(x):
        if x == "slow":
            time.sleep(2.0)  # 远超 per_future_timeout;daemon 线程,被放弃
        return f"ok:{x}"

    out = run_bounded(
        ["a", "slow", "b"],
        work,
        max_workers=3,
        per_future_timeout=0.1,
        on_timeout=lambda item, offset: f"timeout:{item}",
    )
    elapsed = time.monotonic() - started
    assert out[0] == "ok:a"
    assert out[1] == "timeout:slow"   # 单条超时记占位
    assert out[2] == "ok:b"           # 其余正常
    assert elapsed < 1.5              # 主线程不被卡死 worker 拖住(不等满 2s)


def test_run_bounded_worker_exception_becomes_placeholder():
    def work(x):
        if x == "boom":
            raise RuntimeError("worker exploded")
        return f"ok:{x}"

    out = run_bounded(
        ["a", "boom"],
        work,
        max_workers=2,
        per_future_timeout=5.0,
        on_timeout=lambda item, offset: f"failed:{item}",
    )
    assert out == ["ok:a", "failed:boom"]


def test_daemon_thread_pool_executor_threads_are_daemon():
    with DaemonThreadPoolExecutor(max_workers=1, thread_name_prefix="t") as pool:
        is_daemon = pool.submit(lambda: threading.current_thread().daemon).result(timeout=5)
    assert is_daemon is True


# ---------- recall_decision: 单条判定超时跳过、run 不中止 ----------

class _SlowForOneClient(FakeGeminiVideoClient):
    def __init__(self, slow_id: str, sleep_s: float = 2.0):
        super().__init__()
        self.slow_id = slow_id
        self.sleep_s = sleep_s

    def analyze(self, content, media, source_context):
        if str(content.get("platform_content_id")) == self.slow_id:
            time.sleep(self.sleep_s)
        return super().analyze(content, media, source_context)


def test_one_slow_video_judge_times_out_and_run_continues(tmp_path, monkeypatch):
    monkeypatch.setattr(recall_decision, "_resolve_max_workers", lambda: 4)
    monkeypatch.setattr(recall_decision, "JUDGE_WORKER_RESULT_TIMEOUT_SECONDS", 0.1)
    runtime = LocalRuntimeFileStore(tmp_path)
    runtime.prepare_run("run_001")
    ids = ["content_000", "content_001", "content_002"]
    items = [{"platform_content_id": cid, "platform": "douyin"} for cid in ids]
    media = [{"platform_content_id": cid} for cid in ids]
    bundles = [{"content": {"platform_content_id": cid}} for cid in ids]

    started = time.monotonic()
    recalled = pattern_recall.run(
        "run_001", "policy_run_001", items, media, bundles, {}, runtime,
        _SlowForOneClient("content_001", sleep_s=2.0),
    )
    elapsed = time.monotonic() - started

    by_id = {row["platform_content_id"]: row for row in recalled["pattern_recall_evidence"]}
    assert by_id["content_001"]["evidence_summary"]["final_status"] == "failed"
    assert by_id["content_001"]["evidence_summary"]["failure_type"] == "video_judge_timeout"
    assert by_id["content_000"]["evidence_summary"]["final_status"] == "ok"
    assert by_id["content_002"]["evidence_summary"]["final_status"] == "ok"
    assert elapsed < 1.5  # 不等满那条 2s 的慢 worker


# ---------- flow_ledger 新失败类型展示登记 ----------

def test_flow_ledger_registers_new_timeout_failure_types():
    assert fls._technical_retry_stage("video_judge_timeout") == "video_judge"
    assert fls._technical_retry_stage("oss_worker_timeout") == "oss"  # startswith oss_
    assert fls._technical_retry_stage_label("video_judge_timeout") == "视频判定调度"
    assert fls._technical_retry_failure_label("video_judge_timeout") == "视频判定调度超时"
    assert fls._technical_retry_failure_label("oss_worker_timeout") == "OSS 归档 worker 超时"
    assert "超时" in fls._technical_retry_brief_reason("video_judge_timeout", {}, {})
    assert "超时" in fls._technical_retry_brief_reason("oss_worker_timeout", {}, {})


# ---------- 各 client 的 httpx.Timeout 真生效(代表性 2 处) ----------

def test_oss_upload_passes_segmented_timeout():
    captured = {}

    def fake_post(url, *, json, timeout):
        captured["timeout"] = timeout
        return httpx.Response(200, json={"oss_object": {"cdn_url": "x"}}, request=httpx.Request("POST", url))

    oss_upload.upload_video_from_url("http://v/1.mp4", http_post=fake_post)
    assert isinstance(captured["timeout"], httpx.Timeout)
    assert captured["timeout"].read == timeout_config.read_timeout("oss")   # 60
    assert captured["timeout"].write == 300.0


def test_crawapi_post_passes_segmented_timeout():
    captured = {}

    class FakeClient:
        def post(self, url, *, json, headers, timeout):
            captured["timeout"] = timeout
            return httpx.Response(200, json={"code": 0, "data": {}}, request=httpx.Request("POST", url))

    crawapi_http.post_crawapi_json(
        http_client=FakeClient(),
        base_url="https://crawler.example/",
        path="search",
        payload={},
        operation="search",
        timeout_seconds=180.0,
        business_codes=set(),
    )
    assert isinstance(captured["timeout"], httpx.Timeout)
    assert captured["timeout"].read == timeout_config.read_timeout("crawapi")  # 60
    assert captured["timeout"].write == 180.0


def test_video_download_default_timeout_lowered():
    assert video_fetch.DOWNLOAD_TIMEOUT_SECONDS == 600.0