"""Redis 治理后端时间语义 1:1 真实等待变体(M2 设计 §2.3,人类拍板不缩放)。

契约文件中依赖 `clock.advance` 的用例在 redis 参数下被哨兵时钟 skip,
每个都在本文件有同名行为的 `test_variant_<原名去 test_>` 变体:配置用
真实量级(cooldown 60s / probe_ttl 120s / lease 30s),等待是真实的
`asyncio.sleep`。整文件约 13 分钟,标 slow(pytest addopts 默认排除,
显式 `pytest -m slow` 运行)。meta 用例(非 slow)解析契约源码保证
skip 集合与变体集合恒等,防契约新增时间用例而变体漏配。
"""

from __future__ import annotations

import ast
import asyncio
from pathlib import Path
from uuid import uuid4

import pytest

from polygateway.backends.redis.breaker import RedisGate
from polygateway.backends.redis.limiter import RedisLimiter
from polygateway.ports import GateState
from polygateway.types import BreakerConfig, GlobalLimits
from tests.contracts.conftest import (
    await_window_headroom,
    make_source,
    redis_url_from_env,
)

# 与契约 _CFG 同值——真实量级,不缩放(2026-07-20 人类拍板)
_CFG = BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0)
_NO_GLOBAL = GlobalLimits(max_concurrency=0, rpm=0, tpm=0)
_CONTRACT_DIR = Path(__file__).resolve().parents[1] / "contracts"


def _advance_dependent_cases() -> set[str]:
    """解析契约源码,提取调用 clock.advance 的测试函数名(去 test_ 前缀)。"""
    found: set[str] = set()
    for path in sorted(_CONTRACT_DIR.glob("test_*_contract.py")):
        tree = ast.parse(path.read_text(encoding="utf-8"))
        for node in ast.walk(tree):
            if not isinstance(node, ast.AsyncFunctionDef | ast.FunctionDef):
                continue
            if not node.name.startswith("test_"):
                continue
            calls = [
                sub
                for sub in ast.walk(node)
                if isinstance(sub, ast.Attribute) and sub.attr == "advance"
            ]
            if calls:
                found.add(node.name.removeprefix("test_"))
    return found


def test_meta_variants_cover_all_time_cases():
    """完整性守卫: 契约的时间用例集合 == 本文件变体集合(1:1 映射表机械化)。"""
    variants = {
        name.removeprefix("test_variant_") for name in globals() if name.startswith("test_variant_")
    }
    assert variants == _advance_dependent_cases()


pytestmark_slow = pytest.mark.slow


@pytest.fixture
async def redis_client():
    url = redis_url_from_env()
    if url is None:
        pytest.skip("REDIS_URL 未配置")
    import redis.asyncio as aioredis

    client = aioredis.from_url(url)
    try:
        yield client
    finally:
        await client.aclose()


def _limiter(redis_client, sources, *, lease_ttl_s: float = 30.0) -> RedisLimiter:
    return RedisLimiter(
        scope=f"t{uuid4().hex[:8]}",
        sources={s.name: s for s in sources},
        global_limits=_NO_GLOBAL,
        redis=redis_client,
        lease_ttl_s=lease_ttl_s,
    )


def _gate(redis_client) -> RedisGate:
    return RedisGate(config=_CFG, redis=redis_client, scope=f"t{uuid4().hex[:8]}")


async def _open_gate(gate: RedisGate, source: str = "s1"):
    """连续失败到阈值打开熔断(契约 _open_gate 同款)。"""
    update = None
    for _ in range(_CFG.fail_threshold):
        entry = await gate.try_enter(source, "w")
        assert entry.allowed
        update = await gate.record_failure(entry, "network_error", False)
    assert update.state is GateState.OPEN
    return update


# —— 限流侧 2 例 ——


@pytestmark_slow
async def test_variant_lease_expiry_reclaims_slot(redis_client):
    """契约 test_lease_expiry_reclaims_slot: 泄漏 permit 30s 后槽位回收。"""
    limiter = _limiter(redis_client, [make_source(max_concurrency=1)])
    _leaked = await limiter.try_acquire("s1", 0)
    assert _leaked is not None
    assert await limiter.try_acquire("s1", 0) is None
    await asyncio.sleep(31.0)
    reclaimed = await limiter.try_acquire("s1", 0)
    assert reclaimed is not None
    await reclaimed.release()


@pytestmark_slow
async def test_variant_progress_marks_fresh(redis_client):
    """契约 test_progress_marks_fresh: inf → mark → age 随真实时间推进。"""
    limiter = _limiter(redis_client, [make_source()])
    assert await limiter.progress_age_s() == float("inf")
    await limiter.mark_progress()
    assert await limiter.progress_age_s() < 5.0
    await asyncio.sleep(42.0)  # 精确 42s: 断言含上界 43
    assert 41.0 < await limiter.progress_age_s() < 43.0


# —— 熔断侧 7 例 ——


@pytestmark_slow
async def test_variant_cooldown_grants_single_probe(redis_client):
    gate = _gate(redis_client)
    await _open_gate(gate)
    await asyncio.sleep(_CFG.cooldown_s + 1)
    probe = await gate.try_enter("s1", "w1")
    assert probe.allowed and probe.is_probe and probe.probe_owner == "w1"
    second = await gate.try_enter("s1", "w2")
    assert not second.allowed and second.state is GateState.HALF_OPEN


@pytestmark_slow
async def test_variant_probe_success_closes(redis_client):
    gate = _gate(redis_client)
    await _open_gate(gate)
    await asyncio.sleep(_CFG.cooldown_s + 1)
    probe = await gate.try_enter("s1", "w1")
    update = await gate.record_success(probe)
    assert update.applied and update.state is GateState.CLOSED
    assert (await gate.try_enter("s1", "w2")).allowed


@pytestmark_slow
async def test_variant_probe_failure_reopens(redis_client):
    gate = _gate(redis_client)
    await _open_gate(gate)
    await asyncio.sleep(_CFG.cooldown_s + 1)
    probe = await gate.try_enter("s1", "w1")
    update = await gate.record_failure(probe, "network_error", False)
    assert update.state is GateState.OPEN
    assert not (await gate.try_enter("s1", "w2")).allowed


@pytestmark_slow
async def test_variant_probe_lease_expiry_allows_takeover(redis_client):
    """探针持有者死亡 → 120s 租约过期后新 caller 接管,防死锁。"""
    gate = _gate(redis_client)
    await _open_gate(gate)
    await asyncio.sleep(_CFG.cooldown_s + 1)
    stale = await gate.try_enter("s1", "dead-worker")
    assert stale.is_probe
    await asyncio.sleep(_CFG.probe_ttl_s + 1)
    takeover = await gate.try_enter("s1", "w2")
    assert takeover.allowed and takeover.is_probe and takeover.probe_owner == "w2"


@pytestmark_slow
async def test_variant_release_probe_hands_back_and_idempotent(redis_client):
    gate = _gate(redis_client)
    await _open_gate(gate)
    await asyncio.sleep(_CFG.cooldown_s + 1)
    probe = await gate.try_enter("s1", "w1")
    update = await gate.release_probe(probe)
    assert update.applied
    assert not (await gate.release_probe(probe)).applied  # 幂等
    nxt = await gate.try_enter("s1", "w2")
    assert nxt.allowed and nxt.is_probe


@pytestmark_slow
async def test_variant_stale_probe_owner_rejected(redis_client):
    gate = _gate(redis_client)
    await _open_gate(gate)
    await asyncio.sleep(_CFG.cooldown_s + 1)
    stale_probe = await gate.try_enter("s1", "dead-worker")
    await asyncio.sleep(_CFG.probe_ttl_s + 1)
    takeover = await gate.try_enter("s1", "w2")
    assert takeover.is_probe
    assert not (await gate.record_success(stale_probe)).applied  # owner 已易主被 fence


@pytestmark_slow
async def test_variant_retry_after_semantics(redis_client):
    gate = _gate(redis_client)
    assert await gate.retry_after_s(("s1",)) == 0.0
    await _open_gate(gate)
    wait = await gate.retry_after_s(("s1",))
    assert 0 < wait <= _CFG.cooldown_s
    await asyncio.sleep(_CFG.cooldown_s + 1)
    assert await gate.retry_after_s(("s1",)) == 0.0
    with pytest.raises(ValueError):
        await gate.retry_after_s(())


# —— 窗口翻滚(契约外补充,CHS test_redis_limiter.py 集成防抖同款)——


@pytestmark_slow
async def test_rpm_window_rollover_resets_quota(redis_client):
    """RPM 固定分钟窗口: 打满被拒 → 服务器时钟翻分钟后配额刷新。"""
    await await_window_headroom(redis_client, min_headroom_s=15)
    limiter = _limiter(redis_client, [make_source(rpm=2)])
    for _ in range(2):
        permit = await limiter.try_acquire("s1", 0)
        assert permit is not None
        await permit.release()
    assert await limiter.try_acquire("s1", 0) is None
    sec, _ = await redis_client.time()
    await asyncio.sleep(60 - int(sec) % 60 + 1)  # 睡到下一分钟边界
    fresh = await limiter.try_acquire("s1", 0)
    assert fresh is not None
    await fresh.release()


# —— M2.5 双通道/退避变体(小 cooldown 配置控时长;语义与契约用例同源)——

_M25_CFG = BreakerConfig(
    fail_threshold=100,
    cooldown_s=4.0,
    probe_ttl_s=8.0,
    min_calls=4,
    fail_rate=0.9,
    window_s=8.0,
    max_cooldown_s=16.0,
)
_M25_CONSEC_CFG = BreakerConfig(fail_threshold=3, cooldown_s=4.0, probe_ttl_s=8.0)


def _m25_gate(redis_client, cfg=_M25_CFG) -> RedisGate:
    return RedisGate(config=cfg, redis=redis_client, scope=f"t{uuid4().hex[:8]}")


async def _fail_n(gate, n, reason="timeout"):
    update = None
    for _ in range(n):
        entry = await gate.try_enter("s1", "w")
        assert entry.allowed
        update = await gate.record_failure(entry, reason, False)
    return update


@pytestmark_slow
async def test_variant_backoff_doubles_and_caps(redis_client):
    gate = _m25_gate(redis_client)
    update = await _fail_n(gate, 4)  # 率通道首开(4/4 ≥ 0.9): cooldown_eff = 4
    assert update.state is GateState.OPEN
    assert 0 < await gate.retry_after_s(("s1",)) <= 4.0
    await asyncio.sleep(5)
    probe = await gate.try_enter("s1", "w1")
    assert probe.is_probe
    await gate.record_failure(probe, "timeout", False)  # streak 2 → 8
    wait = await gate.retry_after_s(("s1",))
    assert 4.0 < wait <= 8.0
    await asyncio.sleep(9)
    probe = await gate.try_enter("s1", "w1")
    await gate.record_failure(probe, "timeout", False)  # streak 3 → 16(封顶)
    wait = await gate.retry_after_s(("s1",))
    assert 8.0 < wait <= 16.0


@pytestmark_slow
async def test_variant_streak_survives_close_then_decays(redis_client):
    gate = _m25_gate(redis_client)
    await _fail_n(gate, 4)  # streak 1(4s)
    await asyncio.sleep(5)
    probe = await gate.try_enter("s1", "w1")
    await gate.record_failure(probe, "timeout", False)  # streak 2(8s)
    await asyncio.sleep(9)
    probe = await gate.try_enter("s1", "w1")
    await gate.record_success(probe)  # 转 CLOSED,streak 不清零
    await _fail_n(gate, 9)  # 窗口 1 成功 + 9 失败 = 0.9 率开: streak 3 → 16
    assert await gate.retry_after_s(("s1",)) > 8.0
    await asyncio.sleep(17)
    probe = await gate.try_enter("s1", "w1")
    await gate.record_success(probe)  # closed_since 落点
    await asyncio.sleep(2 * 16.0 + 1)
    entry = await gate.try_enter("s1", "w")
    await gate.record_success(entry)  # CLOSED 稳定期满 → streak 衰减归零
    await _fail_n(gate, 9)  # 再开回基础档
    assert 0 < await gate.retry_after_s(("s1",)) <= 4.0


@pytestmark_slow
async def test_variant_consecutive_open_does_not_bump_streak(redis_client):
    gate = _m25_gate(redis_client, _M25_CONSEC_CFG)
    await _fail_n(gate, 3)  # 连续通道开路(不递增 streak)
    await asyncio.sleep(5)
    probe = await gate.try_enter("s1", "w1")
    await gate.record_success(probe)
    await _fail_n(gate, 3)  # 再次连续开路: 仍是基础档
    assert 0 < await gate.retry_after_s(("s1",)) <= 4.0


@pytestmark_slow
async def test_variant_probe_rate_limited_releases_not_hangs(redis_client):
    gate = _m25_gate(redis_client)
    await _fail_n(gate, 4)
    await asyncio.sleep(5)
    probe = await gate.try_enter("s1", "w1")
    assert probe.is_probe
    update = await gate.record_failure(probe, "rate_limited", False)
    assert update.applied
    nxt = await gate.try_enter("s1", "w2")
    assert nxt.allowed and nxt.is_probe  # 立即可再探,不等 probe_ttl


# —— issue #14: retry_after_s = 距离**确定**可再试的时刻,HALF_OPEN 无确定时刻 ——


@pytestmark_slow
async def test_variant_half_open_rejection_reports_no_certain_wait(redis_client):
    gate = _gate(redis_client)
    await _open_gate(gate)
    await asyncio.sleep(_CFG.cooldown_s + 1)
    probe = await gate.try_enter("s1", "w1")
    assert probe.is_probe
    blocked = await gate.try_enter("s1", "w2")
    assert not blocked.allowed and blocked.state is GateState.HALF_OPEN
    assert blocked.retry_after_s == 0.0


@pytestmark_slow
async def test_variant_probe_grant_reports_no_certain_wait(redis_client):
    gate = _gate(redis_client)
    await _open_gate(gate)
    await asyncio.sleep(_CFG.cooldown_s + 1)
    probe = await gate.try_enter("s1", "w1")
    assert probe.allowed and probe.is_probe
    assert probe.retry_after_s == 0.0


@pytestmark_slow
async def test_variant_retry_after_zero_while_probe_in_flight(redis_client):
    gate = _gate(redis_client)
    await _open_gate(gate)
    await asyncio.sleep(_CFG.cooldown_s + 1)
    assert (await gate.try_enter("s1", "w1")).is_probe
    assert await gate.retry_after_s(("s1",)) == 0.0


@pytestmark_slow
async def test_variant_fenced_write_in_half_open_reports_no_certain_wait(redis_client):
    gate = _gate(redis_client)
    stale = await gate.try_enter("s1", "slow-worker")  # epoch 0 的旧 entry
    await _open_gate(gate)  # 他人开路,epoch 推进
    await asyncio.sleep(_CFG.cooldown_s + 1)
    assert (await gate.try_enter("s1", "w1")).is_probe  # 门此刻 HALF_OPEN
    update = await gate.record_success(stale)
    assert not update.applied and update.state is GateState.HALF_OPEN
    assert update.retry_after_s == 0.0
