test: add real-wait time-semantics variants for redis backends

2026-07-21 00:43:39 -04:00
parent 733f413b08
commit 7ceeab3366
2 changed files with 242 additions and 0 deletions
@@ -38,6 +38,9 @@ where = ["src"]
 pythonpath = ["src"]
 testpaths = ["tests"]
 asyncio_mode = "auto"
+# slow(真实等待的时间语义变体,单文件 12-15 分钟)默认排除;
+# 显式 `pytest -m slow` 覆盖(CLI 的 -m 后到优先)。M2 T12 全量跑。
+addopts = "-m 'not slow'"
 markers = [
    "requires_redis: 需要可达的 Redis（无则 skip）",
    "requires_llm: 需要真实 LLM 网关（无则 skip，输出落 tests/outputs/）",
@@ -0,0 +1,239 @@
+"""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()