"""自适应线程池测试。

覆盖决策函数（增/减/保持/边界）、map 顺序返回、worker 异常隔离，
以及"10s 窗口内平均响应 < 0.3s 加线程 / > 1.0s 减线程"的弹性行为
（通过注入假时钟做确定性验证）。
"""

import threading
import time

from nodes.adaptive_pool import AdaptiveThreadPool, decide


class FakeClock:
    """可手动拨动的假时钟，用于确定性验证弹性窗口逻辑。"""

    def __init__(self, now: float = 0.0) -> None:
        self.now = now

    def __call__(self) -> float:
        return self.now

    def advance(self, seconds: float) -> None:
        self.now += seconds


def test_decide_increase_when_fast() -> None:
    """平均响应低于 fast_threshold 且未达上限：线程数 +1。"""
    assert decide(1, 0.1, 1, 16, 0.3, 1.0) == 2


def test_decide_decrease_when_slow() -> None:
    """平均响应高于 slow_threshold 且高于下限：线程数 -1。"""
    assert decide(3, 2.0, 1, 16, 0.3, 1.0) == 2


def test_decide_keep_when_mid() -> None:
    """平均响应介于两阈值之间：保持不变。"""
    assert decide(2, 0.5, 1, 16, 0.3, 1.0) == 2


def test_decide_bounds() -> None:
    """已达上限不再增、已达下限不再减。"""
    assert decide(16, 0.1, 1, 16, 0.3, 1.0) == 16
    assert decide(1, 2.0, 1, 16, 0.3, 1.0) == 1


def test_pool_map_ordered_results() -> None:
    """map 按输入顺序返回结果，worker 简单映射。"""
    pool = AdaptiveThreadPool(worker=lambda item: item * 2)
    assert pool.map([1, 2, 3, 4]) == [2, 4, 6, 8]


def test_pool_on_progress_callback() -> None:
    """进度回调：每次完成触发一次，携带已完成数/总数/速度/平均耗时/线程数。"""
    progress: list[tuple[int, int, float, float, int]] = []
    pool = AdaptiveThreadPool(
        worker=lambda item: item,
        on_progress=lambda done, total, rate, avg_time, workers: progress.append(
            (done, total, rate, avg_time, workers)
        ),
    )
    pool.map([10, 20, 30])
    assert [item[0] for item in progress] == [1, 2, 3]  # 已完成数递增。
    assert all(item[1] == 3 for item in progress)  # 总数固定。
    assert all(item[2] > 0 for item in progress)  # 速度为正值。
    # 窗口未满时平均耗时回退为累计平均（>0）；线程数 ∈ [1, 上限]。
    assert all(item[3] > 0 for item in progress)
    assert all(1 <= item[4] <= pool.max_workers for item in progress)


def test_pool_progress_reports_window_avg_after_first_window() -> None:
    """窗口评估后：回调携带最近窗口平均耗时（扩缩容依据）与扩容后的线程数。

    覆盖 `_current_avg_time` 两个分支：窗口评估前回退累计平均，评估后使用
    最近窗口平均（0.0s，响应远快于 fast_threshold 0.3s → 线程 +1）。
    """
    clock = FakeClock()
    seen: list[tuple[int, int, float, float, int]] = []
    def progress(done, total, rate, avg_time, workers):
        # 推进窗口时钟但不增加 worker 耗时，确定性触发快响应扩容。
        seen.append((done, total, rate, avg_time, workers))
        if done == 1:
            clock.advance(11)

    pool = AdaptiveThreadPool(
        worker=lambda item: item,
        min_workers=1, max_workers=16,
        window_seconds=10.0, fast_threshold=0.3,
        clock=clock, on_progress=progress,
    )
    pool.map(list(range(5)))
    # 第一个完成的任务回调在窗口评估前：真实 worker 耗时为 0，均值也为 0。
    assert seen[0][3] == 0.0
    # 窗口评估后：回调携带最近窗口平均（≈0.0），且线程已扩容到 2。
    assert any(item[3] == 0.0 for item in seen)
    assert any(item[4] == 2 for item in seen)
    assert pool.max_concurrency == 2
def test_pool_cancel_suppresses_progress() -> None:
    """worker 触发 cancel（如检测到暂停信号）后：剩余任务不再触发进度回调。

    暂停场景：队列中剩余的大量帧会逐帧快速失败退出，若每完成一项都打印
    进度日志，会在数秒内打出上万行日志；cancel 后抑制后续进度回调。
    """
    progress: list[tuple[int, int, float, float, int]] = []

    def worker(item):
        if item == 1:
            pool.cancel()  # 模拟某帧检测到暂停信号。
        return item

    pool = AdaptiveThreadPool(
        worker=worker,
        on_progress=lambda done, total, rate, avg_time, workers: progress.append(
            (done, total, rate, avg_time, workers)
        ),
    )
    pool.map([0, 1, 2, 3])
    # 只有 cancel 之前的任务（item=0）触发了进度回调。
    assert len(progress) == 1
    assert progress[0][1] == 4  # 总数仍是 4。


def test_pool_cancel_resets_between_maps() -> None:
    """取消状态按批（map）重置：下一批任务进度回调恢复正常。"""
    progress: list[tuple[int, int, float, float, int]] = []

    def worker(item):
        if item == "stop":
            pool.cancel()
        return item

    pool = AdaptiveThreadPool(
        worker=worker,
        on_progress=lambda done, total, rate, avg_time, workers: progress.append(
            (done, total, rate, avg_time, workers)
        ),
    )
    pool.map(["a", "stop", "b"])
    assert len(progress) == 1  # 只有 cancel 前的 a 触发回调。
    pool.map(["c", "d"])
    # 新一批恢复回调（done 从 1 重新计数）：共 3 次回调（1 + 2）。
    assert [item[0] for item in progress] == [1, 1, 2]

def test_pool_map_empty() -> None:

    """空输入：不启动任务，直接返回空列表。"""
    pool = AdaptiveThreadPool(worker=lambda item: item)
    assert pool.map([]) == []


def test_pool_worker_exception_isolated() -> None:
    """worker 抛异常时以异常对象作为结果，不拖垮整体。"""
    def boom(item):
        raise RuntimeError("boom")

    pool = AdaptiveThreadPool(worker=boom)
    results = pool.map([1, 2])
    assert len(results) == 2
    assert all(isinstance(result, RuntimeError) for result in results)


def test_pool_grows_when_fast() -> None:
    """10s 窗口内平均响应 < 0.3s：线程数从 1 增至 2（弹性扩容）。"""
    clock = FakeClock()
    pool = AdaptiveThreadPool(
        worker=lambda item: item,
        min_workers=1, max_workers=16,
        window_seconds=10.0, fast_threshold=0.3,
        clock=clock,
        on_progress=lambda done, *_: clock.advance(11) if done == 1 else None,
    )
    # 首个任务完成后越过窗口 → 平均响应≈0 < 0.3 → +1 个在途额度。
    pool.map(list(range(4)))
    assert pool.max_concurrency == 2


def test_pool_shrink_when_slow() -> None:
    """窗口平均响应 > 1.0s：线程数从 2 减至 1（弹性退避）。"""
    clock = FakeClock()
    pool = AdaptiveThreadPool(
        worker=lambda item: item,
        min_workers=1, max_workers=16,
        window_seconds=10.0, fast_threshold=0.3, slow_threshold=1.0,
        clock=clock,
    )
    pool._resize(2)  # 先扩到 2 个线程。
    clock.advance(11)
    pool._tick(2.0)  # 窗口内平均 2.0 > 1.0 → 缩回 1。
    # 缩容只调整提交额度；实际积压任务行为由下方并发回归测试验证。
    assert pool._target_workers == 1


def test_resize_shrink_idempotent() -> None:
    """回归：重复缩容到同一目标不会重复放哨兵（曾因并发缩容毒死全部线程而死锁）。"""
    pool = AdaptiveThreadPool(worker=lambda item: item, min_workers=1, max_workers=8)
    pool._resize(3)
    assert pool._target_workers == 3
    pool._resize(2)
    pool._resize(2)  # 目标已是 2：幂等，不再放哨兵。
    assert pool._target_workers == 2
    # 重复缩容不会遗留哨兵影响下一批，真实 map 必须返回全部输入。
    assert pool.map(list(range(20))) == list(range(20))


def test_pool_survives_mixed_grow_shrink() -> None:
    """回归：扩容+缩容混合场景 map 必须完成且保序（修复前会死锁挂起）。"""
    clock = FakeClock()
    state = {"count": 0}

    def worker(item):
        state["count"] += 1
        clock.advance(0.06 if state["count"] <= 20 else 0.6)
        return item

    pool = AdaptiveThreadPool(
        worker=worker, min_workers=1, max_workers=4,
        window_seconds=0.5, fast_threshold=0.2, slow_threshold=0.4,
        clock=clock,
    )
    out = pool.map(list(range(60)))
    assert out == list(range(60))


def test_pool_report_failure_lowers_effective_max() -> None:
    """消费错误（如 API 限流）临时降低有效最大线程数，下限为 min_workers。

    自适应：并发打到配额线触发 429 时，report_failure 收紧有效上限，
    后续请求减少从而避开持续限流。
    """
    pool = AdaptiveThreadPool(worker=lambda item: item, min_workers=1, max_workers=16)
    assert pool._effective_max_workers == 16
    pool.report_failure()
    assert pool._effective_max_workers == 15
    for _ in range(30):
        pool.report_failure()
    assert pool._effective_max_workers == 1  # 下限 min_workers。


def test_pool_effective_max_recovers_after_clean_window() -> None:
    """连续无错误窗口后有效上限逐步回升到 max_workers。"""
    clock = FakeClock()
    pool = AdaptiveThreadPool(
        worker=lambda item: item, min_workers=1, max_workers=16,
        window_seconds=10.0, fast_threshold=0.3, clock=clock,
    )
    pool.report_failure()  # 有效上限 16 -> 15。
    clock.advance(11)
    pool._tick(0.01)  # 错误所在窗口：上限不恢复。
    assert pool._effective_max_workers == 15
    clock.advance(11)
    pool._tick(0.01)  # 下一个干净窗口：恢复 +1。
    assert pool._effective_max_workers == 16
    pool._tick(0.01)  # 窗口未满早退，上限不变。
    assert pool._effective_max_workers == 16


def test_failure_at_single_worker_never_expands() -> None:
    """真实 map 内报告限流：1 个在途任务不能因上限 20 变 19 而突然扩容。"""
    targets = []

    def worker(item):
        if item == 0:
            pool.report_failure()
            targets.append(pool._target_workers)
        return item

    pool = AdaptiveThreadPool(worker=worker, max_workers=20, window_seconds=1000)
    assert pool.map(list(range(30))) == list(range(30))
    assert targets == [1]


def test_failure_shrink_limits_backlogged_work() -> None:
    """积压任务中从 4 降到 1：已开始任务可完成，后续实际并发必须为 1。"""
    barrier = threading.Barrier(4, timeout=3)
    reduced = threading.Event()
    lock = threading.Lock()
    active = 0
    subsequent_peaks = []

    def worker(item):
        nonlocal active
        with lock:
            active += 1
            if item >= 5:
                subsequent_peaks.append(active)
        try:
            if 1 <= item <= 4:
                barrier.wait()
                if item == 1:
                    # 连续限流将有效上限压到下限，存量请求不强制中断。
                    for _ in range(3):
                        pool.report_failure()
                    reduced.set()
                assert reduced.wait(3)
            elif item >= 5:
                # 模拟 I/O 等待，给其他工作线程实际进入任务的机会。
                time.sleep(0.005)
            return item
        finally:
            with lock:
                active -= 1

    def progress(done, *_):
        if done == 1:
            pool._resize(4)

    pool = AdaptiveThreadPool(worker=worker, max_workers=4, window_seconds=1000,
                              on_progress=progress)
    items = list(range(40))
    assert pool.map(items) == items
    assert subsequent_peaks and max(subsequent_peaks) == 1


def test_error_window_does_not_regrow() -> None:
    """报告限流的同一窗口即使响应很快，也不能重新加并发。"""
    clock = FakeClock()
    pool = AdaptiveThreadPool(worker=lambda item: item, max_workers=4, clock=clock)
    pool.report_failure()
    clock.advance(11)
    pool._tick(0.01)
    assert pool._target_workers <= 1


def test_retry_map_preserves_reduced_limit() -> None:
    """两轮真实 map：首轮连续限流后，重试期间实际并发及进度都遵守新上限。"""
    lock = threading.Lock()
    active = 0
    peak = 0
    progress_counts = []

    def worker(item):
        nonlocal active, peak
        with lock:
            active += 1
            peak = max(peak, active)
        try:
            if item == "limited":
                for _ in range(3):
                    pool.report_failure()
                raise RuntimeError("rate limited")
            time.sleep(0.002)
            return item
        finally:
            with lock:
                active -= 1

    pool = AdaptiveThreadPool(
        worker=worker, max_workers=4, window_seconds=1000,
        on_progress=lambda done, *_: progress_counts.append(done),
    )
    assert isinstance(pool.map(["limited"])[0], RuntimeError)
    assert pool._effective_max_workers == 1
    # 主动申请扩容也受已收紧额度约束，下一轮不能重置有效上限。
    pool._resize(4)
    assert pool._target_workers == 1
    assert pool.map(list(range(12))) == list(range(12))
    assert peak == 1
    assert progress_counts == [1] + list(range(1, 13))


def test_pool_decide_uses_effective_max() -> None:
    """扩容上限按有效最大线程数：错误窗口内即使响应快也不超过收紧后的上限。"""
    clock = FakeClock()
    pool = AdaptiveThreadPool(
        worker=lambda item: item, min_workers=1, max_workers=16,
        window_seconds=10.0, fast_threshold=0.3, clock=clock,
    )
    pool.report_failure()  # 有效上限 16 -> 15。
    pool._resize(15)
    pool._window_failures = 1  # 本窗口内仍有错误 → 不恢复上限。
    clock.advance(11)
    pool._tick(0.01)  # 响应快，但 15 已是有效上限 → 不扩。
    assert pool._target_workers == 15
    assert pool._effective_max_workers == 15