"""超时硬化 / 有界等待 / 僵尸线程清理 的单测(修永久卡死)。""" 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") # 300 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