searxng-use-cli/tests/test_adaptive_throttle.py

"""Tests for v2.0.0 AdaptiveThrottle state machine.

Covers: initial state, consecutive failure escalation (delay doubling +
concurrency halving), consecutive success recovery, 429 global pause,
thread safety (stats snapshot), fetch_top_results integration with
adaptive throttling.
"""
import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "scripts"))

import time
from unittest.mock import patch, MagicMock
from fetch import FetchResult
import search as search_mod
from search import AdaptiveThrottle, fetch_top_results


# ----- Initial state -----

def test_throttle_initial_state():
    t = AdaptiveThrottle(initial_delay=0.3, initial_concurrency=5)
    assert t.delay == 0.3
    assert t.concurrency == 5
    s = t.stats()
    assert s["current_delay"] == 0.3
    assert s["current_concurrency"] == 5
    assert s["consecutive_failures"] == 0
    assert s["consecutive_successes"] == 0
    assert s["global_paused"] is False


# ----- Failure escalation -----

def test_throttle_failure_escalation_delay():
    """连续 3 次失败 → delay 翻倍。"""
    t = AdaptiveThrottle(initial_delay=0.3, initial_concurrency=5)
    t.report_failure()
    t.report_failure()
    assert t.delay == 0.3  # 还未到 3 次
    t.report_failure()  # 第 3 次
    assert t.delay == 0.6  # 翻倍


def test_throttle_failure_escalation_concurrency():
    """连续 3 次失败 → concurrency 减半。"""
    t = AdaptiveThrottle(initial_delay=0.3, initial_concurrency=4)
    t.report_failure()
    t.report_failure()
    assert t.concurrency == 4  # 还未到 3 次
    t.report_failure()  # 第 3 次
    assert t.concurrency == 2  # 减半


def test_throttle_failure_counter_resets_after_escalation():
    """触发升级后，连续失败计数器重置。"""
    t = AdaptiveThrottle(initial_delay=0.3, initial_concurrency=5)
    for _ in range(3):
        t.report_failure()
    # 升级后计数器重置
    assert t.stats()["consecutive_failures"] == 0
    # 再失败 2 次不会再次升级
    t.report_failure()
    t.report_failure()
    assert t.delay == 0.6  # 没变


def test_throttle_delay_capped_at_10():
    """delay 上限 10s。"""
    t = AdaptiveThrottle(initial_delay=0.3, initial_concurrency=5)
    # 触发多次升级
    for _ in range(10):
        for _ in range(3):
            t.report_failure()
    assert t.delay <= 10.0


def test_throttle_concurrency_floors_at_1():
    """concurrency 下限 1。"""
    t = AdaptiveThrottle(initial_delay=0.3, initial_concurrency=4)
    for _ in range(10):
        for _ in range(3):
            t.report_failure()
    assert t.concurrency >= 1


# ----- Success recovery -----

def test_throttle_success_recovery_delay():
    """连续 5 次成功 → delay 减半（恢复）。"""
    t = AdaptiveThrottle(initial_delay=0.3, initial_concurrency=5)
    # 先升级
    for _ in range(3):
        t.report_failure()
    assert t.delay == 0.6
    # 连续 5 次成功
    for _ in range(5):
        t.report_success()
    assert t.delay == 0.3  # 恢复到初始值


def test_throttle_success_recovery_concurrency():
    """连续 5 次成功 → concurrency 恢复。"""
    t = AdaptiveThrottle(initial_delay=0.3, initial_concurrency=4)
    # 先升级
    for _ in range(3):
        t.report_failure()
    assert t.concurrency == 2
    # 连续 5 次成功
    for _ in range(5):
        t.report_success()
    assert t.concurrency == 4  # 恢复


def test_throttle_success_resets_failure_counter():
    """成功重置连续失败计数器。"""
    t = AdaptiveThrottle(initial_delay=0.3, initial_concurrency=5)
    t.report_failure()
    t.report_failure()
    t.report_success()
    assert t.stats()["consecutive_failures"] == 0


def test_throttle_success_below_initial_no_change():
    """已经处于初始值时，成功不会让 delay 更低。"""
    t = AdaptiveThrottle(initial_delay=0.3, initial_concurrency=5)
    for _ in range(5):
        t.report_success()
    assert t.delay == 0.3  # 不低于初始值


# ----- 429 global pause -----

def test_throttle_429_triggers_global_pause():
    """429 错误触发全局暂停。"""
    t = AdaptiveThrottle(initial_delay=0.3, initial_concurrency=5)
    t.report_failure("HTTP 429 Too Many Requests")
    assert t.stats()["global_paused"] is True


def test_throttle_non_429_no_global_pause():
    """非 429 错误不触发全局暂停。"""
    t = AdaptiveThrottle(initial_delay=0.3, initial_concurrency=5)
    t.report_failure("HTTP 500 Internal Server Error")
    assert t.stats()["global_paused"] is False


def test_throttle_global_pause_expires():
    """全局暂停会随时间过期。"""
    t = AdaptiveThrottle(initial_delay=0.3, initial_concurrency=5)
    # 用极短暂停时间测试（monkeypatch 不行，因为 30s 硬编码）
    # 改为直接等待验证逻辑：暂停时间 30s 太长，改为验证标志位
    t.report_failure("HTTP 429")
    assert t.stats()["global_paused"] is True
    # 不实际等待 30s；改为验证 stats 逻辑正确即可
    # （wait_if_paused 的实际阻塞行为在集成测试中验证）


# ----- Mixed scenarios -----

def test_throttle_alternating_success_failure():
    """交替成功/失败不触发升级（计数器被重置）。"""
    t = AdaptiveThrottle(initial_delay=0.3, initial_concurrency=5)
    for _ in range(10):
        t.report_failure()
        t.report_success()
    assert t.delay == 0.3  # 从未连续 3 次失败
    assert t.concurrency == 5


def test_throttle_partial_recovery_then_failure():
    """部分恢复后再次失败，从当前状态继续升级。"""
    t = AdaptiveThrottle(initial_delay=0.3, initial_concurrency=5)
    # 升级到 0.6
    for _ in range(3):
        t.report_failure()
    assert t.delay == 0.6
    # 部分恢复（3 次成功，不够 5 次）
    for _ in range(3):
        t.report_success()
    assert t.delay == 0.6  # 还没恢复
    # 再次失败 3 次
    for _ in range(3):
        t.report_failure()
    assert t.delay == 1.2  # 从 0.6 继续翻倍


# ----- Thread safety -----

def test_throttle_stats_is_snapshot():
    """stats() 返回快照，修改快照不影响内部状态。"""
    t = AdaptiveThrottle(initial_delay=0.3, initial_concurrency=5)
    s = t.stats()
    s["current_delay"] = 999
    # 内部状态不变
    assert t.delay == 0.3


# ----- fetch_top_results integration -----

def _make_search_results(n=3):
    return {"results": [{"url": f"https://example{i}.com", "title": f"Test {i}"}
                        for i in range(n)]}


def _make_fetch_result_ok(url="https://example.com"):
    return {
        "url": url, "status": "ok", "text": "content", "text_length": 7,
        "truncated": False, "user_agent_used": "TestUA",
        "anti_bot_detected": False, "waf_type": None, "fallback_used": None,
    }


def test_fetch_top_results_uses_adaptive_throttle():
    """fetch_top_results 接受外部 throttle 实例。"""
    throttle = AdaptiveThrottle(0.0, 5)
    with patch("search.fetch_page", return_value=_make_fetch_result_ok()):
        results = fetch_top_results(_make_search_results(3), 3,
                                    request_delay=0.0, throttle=throttle)
    assert len(results) == 3
    # 成功 3 次，但不够 5 次触发恢复
    assert throttle.stats()["consecutive_successes"] == 3


def test_fetch_top_results_creates_throttle_if_none():
    """未传入 throttle 时内部创建。"""
    with patch("search.fetch_page", return_value=_make_fetch_result_ok()):
        results = fetch_top_results(_make_search_results(2), 2,
                                    request_delay=0.0)
    assert len(results) == 2


def test_fetch_top_results_reports_failures_to_throttle():
    """抓取失败反馈给 throttle。"""
    throttle = AdaptiveThrottle(0.3, 5)  # 非零初始值，翻倍后才 >0
    err_result = {"url": "https://x.com", "status": "error",
                  "error": "HTTP 500", "text": "", "text_length": 0,
                  "truncated": False, "anti_bot_detected": False,
                  "waf_type": None, "fallback_used": None}
    with patch("search.fetch_page", return_value=err_result):
        fetch_top_results(_make_search_results(3), 3,
                         request_delay=0.3, throttle=throttle)
    # 3 次失败触发升级
    assert throttle.stats()["consecutive_failures"] == 0  # 升级后重置
    assert throttle.delay > 0.3  # delay 翻倍（0.3 → 0.6）


def test_fetch_top_results_passes_referer():
    """referer 透传给 fetch_page。"""
    with patch("search.fetch_page", return_value=_make_fetch_result_ok()) as mock_fp:
        fetch_top_results(_make_search_results(1), 1,
                         request_delay=0.0, referer="https://instance.com/")
    call_kwargs = mock_fp.call_args[1]
    assert call_kwargs.get("referer") == "https://instance.com/"


def test_fetch_top_results_passes_fallback_enabled():
    """fallback_enabled 透传给 fetch_page。"""
    with patch("search.fetch_page", return_value=_make_fetch_result_ok()) as mock_fp:
        fetch_top_results(_make_search_results(1), 1,
                         request_delay=0.0, fallback_enabled=False)
    call_kwargs = mock_fp.call_args[1]
    assert call_kwargs.get("fallback_enabled") is False