"""熔断后端契约测试(状态机 + 半开单探针租约 + epoch fencing)。

蓝本: VT adapters/breaker.py 状态机语义 + CHS provider_gate.py 契约;
M2 的 Redis 实现复用本套件。
"""

import pytest

from polygateway.ports import GateState
from polygateway.types import BreakerConfig

_CFG = BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0)


async def _open_gate(gate, source="s1"):
    """连续失败到阈值,打开熔断,返回最后一次 Update。"""
    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


class TestStateMachine:
    async def test_closed_admits_and_success_resets(self, gate_factory):
        gate = gate_factory(_CFG)
        entry = await gate.try_enter("s1", "w")
        assert entry.allowed and entry.state is GateState.CLOSED and not entry.is_probe
        update = await gate.record_success(entry)
        assert update.applied and update.failure_count == 0

    async def test_threshold_opens_and_rejects(self, gate_factory):
        gate = gate_factory(_CFG)
        await _open_gate(gate)
        entry = await gate.try_enter("s1", "w")
        assert not entry.allowed and entry.state is GateState.OPEN
        assert entry.retry_after_s > 0

    async def test_success_before_threshold_resets_count(self, gate_factory):
        gate = gate_factory(_CFG)
        for _ in range(_CFG.fail_threshold - 1):
            entry = await gate.try_enter("s1", "w")
            await gate.record_failure(entry, "timeout", False)
        entry = await gate.try_enter("s1", "w")
        update = await gate.record_success(entry)
        assert update.applied and update.failure_count == 0
        assert (await gate.try_enter("s1", "w")).allowed

    async def test_force_open_single_strike(self, gate_factory):
        gate = gate_factory(_CFG)
        entry = await gate.try_enter("s1", "w")
        update = await gate.record_failure(entry, "source_dead", True)
        assert update.state is GateState.OPEN
        assert not (await gate.try_enter("s1", "w")).allowed


class TestHalfOpenProbe:
    async def test_cooldown_grants_single_probe(self, gate_factory, clock):
        gate = gate_factory(_CFG)
        await _open_gate(gate)
        clock.advance(_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

    async def test_probe_success_closes(self, gate_factory, clock):
        gate = gate_factory(_CFG)
        await _open_gate(gate)
        clock.advance(_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

    async def test_probe_failure_reopens(self, gate_factory, clock):
        gate = gate_factory(_CFG)
        await _open_gate(gate)
        clock.advance(_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

    async def test_probe_lease_expiry_allows_takeover(self, gate_factory, clock):
        """探针持有者死亡 → 租约过期后新 caller 接管探针,防死锁。"""
        gate = gate_factory(_CFG)
        await _open_gate(gate)
        clock.advance(_CFG.cooldown_s + 1)
        stale = await gate.try_enter("s1", "dead-worker")
        assert stale.is_probe
        clock.advance(_CFG.probe_ttl_s + 1)
        takeover = await gate.try_enter("s1", "w2")
        assert takeover.allowed and takeover.is_probe and takeover.probe_owner == "w2"

    async def test_release_probe_hands_back_and_idempotent(self, gate_factory, clock):
        gate = gate_factory(_CFG)
        await _open_gate(gate)
        clock.advance(_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
        # 源保持可接管状态: 下一 caller 拿到探针
        nxt = await gate.try_enter("s1", "w2")
        assert nxt.allowed and nxt.is_probe


class TestEpochFencing:
    async def test_stale_epoch_write_rejected(self, gate_factory, clock):
        """entry 取得后世代已推进(他人触发开路)→ 迟到写回被 fencing 拒绝。"""
        gate = gate_factory(_CFG)
        stale_entry = await gate.try_enter("s1", "slow-worker")
        await _open_gate(gate)  # 他人连续失败 → 开路,epoch 推进
        late = await gate.record_success(stale_entry)
        assert not late.applied
        # 门仍是 OPEN,未被迟到的成功污染
        assert not (await gate.try_enter("s1", "w")).allowed

    async def test_stale_probe_owner_rejected(self, gate_factory, clock):
        gate = gate_factory(_CFG)
        await _open_gate(gate)
        clock.advance(_CFG.cooldown_s + 1)
        stale_probe = await gate.try_enter("s1", "dead-worker")
        clock.advance(_CFG.probe_ttl_s + 1)
        takeover = await gate.try_enter("s1", "w2")
        assert takeover.is_probe
        # 死亡探针的迟到写回被拒(owner 已易主)
        assert not (await gate.record_success(stale_probe)).applied


_RATE_CFG = BreakerConfig(
    fail_threshold=100,  # 连续通道抬高失声,单测率通道
    cooldown_s=60.0,
    probe_ttl_s=120.0,
    min_calls=4,
    fail_rate=0.9,
    window_s=60.0,
    max_cooldown_s=240.0,
)


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


class TestRateChannel:
    """M2.5 失败率通道(设计 §3.1;全部确定性序列)。"""

    async def test_rate_opens_at_min_calls(self, gate_factory):
        # 病灶 1 回归: 高失败率源在 min_calls 样本处开路(连续通道静默)
        gate = gate_factory(_RATE_CFG)
        await _fail(gate, n=3)  # attempts 3 < min_calls 4 → 仍 CLOSED
        assert (await gate.try_enter("s1", "w")).allowed
        update = await _fail(gate, n=1)  # attempts 4, 失败率 1.0 ≥ 0.9
        assert update.state is GateState.OPEN
        assert not (await gate.try_enter("s1", "w")).allowed

    async def test_below_min_calls_never_rate_opens(self, gate_factory):
        gate = gate_factory(_RATE_CFG)
        await _fail(gate, n=3)
        assert (await gate.try_enter("s1", "w")).allowed

    async def test_real_success_dilutes_window(self, gate_factory):
        # 真实成功计入 attempts: 3 失败 + 1 成功 + 1 失败 = 4/5 = 0.8 < 0.9 不开
        gate = gate_factory(_RATE_CFG)
        await _fail(gate, n=3)
        entry = await gate.try_enter("s1", "w")
        await gate.record_success(entry)
        update = await _fail(gate, n=1)
        assert update.state is GateState.CLOSED
        assert (await gate.try_enter("s1", "w")).allowed

    async def test_result_invalid_not_counted(self, gate_factory):
        # 坏结果 ≠ 坏服务: count_attempt=False 完全不动窗口 →
        # 3 失败 + 1 不计成功 + 1 失败 = attempts 4, 失败率 1.0 → 开路
        gate = gate_factory(_RATE_CFG)
        await _fail(gate, n=3)
        entry = await gate.try_enter("s1", "w")
        await gate.record_success(entry, count_attempt=False)
        update = await _fail(gate, n=1)
        assert update.state is GateState.OPEN

    async def test_rate_limited_bypasses_both_channels(self, gate_factory):
        # C1 对抗: 429 是背压不是故障——连续通道(阈值 3)与率通道都不吃
        cfg = BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0, min_calls=4)
        gate = gate_factory(cfg)
        await _fail(gate, reason="rate_limited", n=10)
        assert (await gate.try_enter("s1", "w")).allowed
        # timeout 脉冲照常走连续通道开路
        update = await _fail(gate, reason="timeout", n=3)
        assert update.state is GateState.OPEN

    async def test_probe_rate_limited_releases_not_hangs(self, gate_factory, clock):
        # 探针撞 429: 非故障证据也非成功——按无果归还语义放下家接管,
        # 不得把探针租约挂到 TTL(否则源被锁死 probe_ttl_s)
        gate = gate_factory(_RATE_CFG)
        await _fail(gate, n=4)
        clock.advance(61)
        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

    async def test_rate_open_failure_count_capped(self, gate_factory):
        gate = gate_factory(_RATE_CFG)
        await _fail(gate, n=3)
        update = await _fail(gate, n=1)
        assert update.state is GateState.OPEN
        assert update.failure_count == _RATE_CFG.fail_threshold  # 顶格语义沿用


class TestReopenBackoff:
    """开路时长指数递增与衰减(设计 §3.1;memory 假时钟,redis 走真实等待变体)。"""

    async def test_backoff_doubles_and_caps(self, gate_factory, clock):
        gate = gate_factory(_RATE_CFG)
        await _fail(gate, n=4)  # 率通道首开: cooldown_eff = 60
        assert 0 < await gate.retry_after_s(("s1",)) <= 60.0
        clock.advance(61)
        probe = await gate.try_enter("s1", "w1")
        assert probe.is_probe
        await gate.record_failure(probe, "timeout", False)  # 探针失败重开: streak 2 → 120
        wait = await gate.retry_after_s(("s1",))
        assert 60.0 < wait <= 120.0
        clock.advance(121)
        probe = await gate.try_enter("s1", "w1")
        await gate.record_failure(probe, "timeout", False)  # streak 3 → 240(封顶)
        wait = await gate.retry_after_s(("s1",))
        assert 120.0 < wait <= 240.0

    async def test_streak_survives_close_then_decays(self, gate_factory, clock):
        gate = gate_factory(_RATE_CFG)
        await _fail(gate, n=4)
        clock.advance(61)
        probe = await gate.try_enter("s1", "w1")
        await gate.record_failure(probe, "timeout", False)  # streak 2
        clock.advance(121)
        probe = await gate.try_enter("s1", "w1")
        await gate.record_success(probe)  # 转 CLOSED,streak 不清零
        # 窗口含探针成功 1 次,9 失败 → 9/10 = 0.9 率开;streak 递增至封顶
        await _fail(gate, n=9)
        assert await gate.retry_after_s(("s1",)) > 60.0
        # 衰减: CLOSED 稳定 2×cooldown_eff 后首次 record_success 归零
        clock.advance(241)
        probe = await gate.try_enter("s1", "w1")
        await gate.record_success(probe)
        clock.advance(2 * 240.0 + 1)
        entry = await gate.try_enter("s1", "w")
        await gate.record_success(entry)  # 触发衰减
        await _fail(gate, n=9)  # 再开(9/10)回到基础档
        assert 0 < await gate.retry_after_s(("s1",)) <= 60.0

    async def test_consecutive_open_does_not_bump_streak(self, gate_factory, clock):
        # C1 对抗: 连续通道误熔健康源的代价封顶为单次 cooldown_s
        cfg = BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0)
        gate = gate_factory(cfg)
        await _fail(gate, reason="timeout", n=3)  # 连续通道开路
        clock.advance(61)
        probe = await gate.try_enter("s1", "w1")
        await gate.record_success(probe)  # 恢复
        await _fail(gate, reason="timeout", n=3)  # 再次连续开路
        assert 0 < await gate.retry_after_s(("s1",)) <= 60.0  # 无翻倍


class TestRetryAfter:
    async def test_retry_after_semantics(self, gate_factory, clock):
        gate = gate_factory(_CFG)
        assert await gate.retry_after_s(("s1",)) == 0.0  # 健康 → 0
        await _open_gate(gate)
        wait = await gate.retry_after_s(("s1",))
        assert 0 < wait <= _CFG.cooldown_s
        clock.advance(_CFG.cooldown_s + 1)
        assert await gate.retry_after_s(("s1",)) == 0.0  # 冷却到期 → 0
        with pytest.raises(ValueError):
            await gate.retry_after_s(())

    async def test_retry_after_takes_min_across_sources(self, gate_factory, clock):
        gate = gate_factory(_CFG)
        await _open_gate(gate, "s1")  # s1 开路;s2 健康
        assert await gate.retry_after_s(("s1", "s2")) == 0.0

    async def test_half_open_rejection_reports_no_certain_wait(self, gate_factory, clock):
        """探针在途时被拒 → 0.0(issue #14): 探针随时可能出结果,不存在确定时刻。

        旧行为返回探针租约剩余,而租约长度是**死锁保护参数**(派生自
        `2 × 最慢源 timeout`),与"这个源多久能恢复"没有因果关系。现场
        `TIMEOUT_S=300` 时它是 600s,而冷却期只有 60s。
        """
        gate = gate_factory(_CFG)
        await _open_gate(gate)
        clock.advance(_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

    async def test_probe_grant_reports_no_certain_wait(self, gate_factory, clock):
        """准入被允许 → 恒 0.0(现在就能试);此前 redis 侧返回探针 TTL。"""
        gate = gate_factory(_CFG)
        await _open_gate(gate)
        clock.advance(_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

    async def test_retry_after_zero_while_probe_in_flight(self, gate_factory, clock):
        """集合查询同口径: 探针在途的源不贡献等待时间。"""
        gate = gate_factory(_CFG)
        await _open_gate(gate)
        clock.advance(_CFG.cooldown_s + 1)
        assert (await gate.try_enter("s1", "w1")).is_probe
        assert await gate.retry_after_s(("s1",)) == 0.0

    async def test_fenced_write_in_half_open_reports_no_certain_wait(self, gate_factory, clock):
        """写回被 fencing 拒时的快照同口径;此前 redis 侧返回探针租约剩余。"""
        gate = gate_factory(_CFG)
        stale = await gate.try_enter("s1", "slow-worker")  # epoch 0 的旧 entry
        await _open_gate(gate)  # 他人开路,epoch 推进
        clock.advance(_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


class TestConsecutiveSuppression:
    """迭代 6: 窗口证据充足且健康时,连败是噪声,不开路(设计 §3.39)。"""

    async def test_streak_suppressed_on_evidently_healthy_source(self, gate_factory):
        # 20 成功垫底(窗口样本充足、失败率低)后 3 连败(阈值 3)→ 不开路
        cfg = BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0, min_calls=10)
        gate = gate_factory(cfg)
        for _ in range(20):
            entry = await gate.try_enter("s1", "w")
            await gate.record_success(entry)
        await _fail(gate, reason="timeout", n=3)
        assert (await gate.try_enter("s1", "w")).allowed

    async def test_streak_fires_when_window_insufficient(self, gate_factory):
        # 冷启动(窗口样本不足)3 连败照常开路——连续通道本职保留
        cfg = BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0, min_calls=10)
        gate = gate_factory(cfg)
        update = await _fail(gate, reason="timeout", n=3)
        assert update.state is GateState.OPEN

    async def test_warm_source_going_dead_caught_by_rate(self, gate_factory):
        # 温热源猝死: 连败被抑制,但失败率窗口随失败累积必然接管
        cfg = BreakerConfig(
            fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0, min_calls=4, fail_rate=0.6
        )
        gate = gate_factory(cfg)
        for _ in range(3):
            entry = await gate.try_enter("s1", "w")
            await gate.record_success(entry)
        update = await _fail(gate, reason="timeout", n=5)  # 5/8 = 0.625 ≥ 0.6
        assert update.state is GateState.OPEN