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