"""熔断后端契约测试(状态机 + 半开单探针租约 + 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


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