143 lines
4.5 KiB
Python
143 lines
4.5 KiB
Python
"""契约测试共享 fixture: 后端参数化(memory + redis,D3 双后端同一契约)。
|
|
|
|
结构(M2 设计 §2.3 定案): 单一 `backend` fixture 承载参数化;`clock` 与两工厂
|
|
都依赖它——memory 用 FakeClock(时钟注入,时间语义可快进);redis 用真实实验室
|
|
Redis(db3),FakeClock 对服务器时钟不可注入,故 `clock.advance()` 是哨兵:
|
|
触发 skip,对应用例由 tests/integration/test_redis_governance_time.py 的
|
|
1:1 真实等待变体覆盖(人类拍板: 不缩放时长)。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from dotenv import dotenv_values
|
|
|
|
from polygateway.backends.memory.breaker import InMemoryGate
|
|
from polygateway.backends.memory.limiter import InMemoryLimiter
|
|
from polygateway.types import BreakerConfig, GlobalLimits, SourceConfig
|
|
|
|
_HEADROOM_S = 10 # 分钟窗口剩余不足则等翻滚,防 RPM/TPM 用例跨窗 flake(CHS 同款)
|
|
|
|
|
|
class FakeClock:
|
|
"""确定性单调时钟;memory 后端的契约测试推进时间验证租约/冷却语义。"""
|
|
|
|
def __init__(self, start: float = 1000.0) -> None:
|
|
self.t = start
|
|
|
|
def __call__(self) -> float:
|
|
return self.t
|
|
|
|
def advance(self, seconds: float) -> None:
|
|
self.t += seconds
|
|
|
|
|
|
class SkipClock:
|
|
"""redis 后端的哨兵时钟: 服务器时钟不可注入,依赖快进的用例整例跳过。"""
|
|
|
|
def __call__(self) -> float: # pragma: no cover - 不应被消费
|
|
raise AssertionError("redis 后端不消费注入时钟")
|
|
|
|
def advance(self, seconds: float) -> None:
|
|
pytest.skip("redis 时间语义由 tests/integration/test_redis_governance_time.py 变体覆盖")
|
|
|
|
|
|
def make_source(name: str = "s1", **overrides) -> SourceConfig:
|
|
base = {
|
|
"name": name,
|
|
"provider": "openai",
|
|
"base_url": "https://gw.example/v1",
|
|
"api_key": "sk-test",
|
|
"model": "m",
|
|
"timeout_s": 10.0,
|
|
}
|
|
base.update(overrides)
|
|
return SourceConfig(**base)
|
|
|
|
|
|
def redis_url_from_env() -> str | None:
|
|
"""读 REDIS_URL(.env 与进程环境合并,后者优先);供契约与集成测试共用。"""
|
|
merged = {**dotenv_values(".env"), **os.environ}
|
|
return merged.get("REDIS_URL") or None
|
|
|
|
|
|
async def await_window_headroom(client, min_headroom_s: int = _HEADROOM_S) -> None:
|
|
"""按 Redis 服务器时钟等待分钟窗口翻滚防抖(CHS test_redis_limiter.py:17-30)。"""
|
|
sec, _ = await client.time()
|
|
remaining = 60 - int(sec) % 60
|
|
if remaining < min_headroom_s:
|
|
await asyncio.sleep(remaining + 0.5)
|
|
|
|
|
|
class _Backend:
|
|
def __init__(self, name: str, redis=None) -> None:
|
|
self.name = name
|
|
self.redis = redis
|
|
|
|
|
|
@pytest.fixture(params=["memory", "redis"])
|
|
async def backend(request):
|
|
if request.param == "memory":
|
|
yield _Backend("memory")
|
|
return
|
|
url = redis_url_from_env()
|
|
if url is None:
|
|
pytest.skip("REDIS_URL 未配置,跳过 redis 后端契约")
|
|
import redis.asyncio as aioredis
|
|
|
|
client = aioredis.from_url(url)
|
|
try:
|
|
await await_window_headroom(client)
|
|
yield _Backend("redis", client)
|
|
finally:
|
|
await client.aclose()
|
|
|
|
|
|
@pytest.fixture
|
|
def clock(backend):
|
|
return FakeClock() if backend.name == "memory" else SkipClock()
|
|
|
|
|
|
@pytest.fixture
|
|
def limiter_factory(backend, clock):
|
|
"""返回 (sources, global_limits, lease_ttl_s) -> RateLimiter 的工厂。"""
|
|
|
|
def make(sources: list[SourceConfig], global_limits: GlobalLimits, lease_ttl_s: float = 100.0):
|
|
if backend.name == "memory":
|
|
return InMemoryLimiter(
|
|
scope="llm",
|
|
sources={s.name: s for s in sources},
|
|
global_limits=global_limits,
|
|
lease_ttl_s=lease_ttl_s,
|
|
now=clock,
|
|
)
|
|
from polygateway.backends.redis.limiter import RedisLimiter
|
|
|
|
return RedisLimiter(
|
|
scope=f"t{uuid4().hex[:8]}",
|
|
sources={s.name: s for s in sources},
|
|
global_limits=global_limits,
|
|
redis=backend.redis,
|
|
lease_ttl_s=lease_ttl_s,
|
|
)
|
|
|
|
return make
|
|
|
|
|
|
@pytest.fixture
|
|
def gate_factory(backend, clock):
|
|
"""返回 (BreakerConfig) -> ProviderGate 的工厂。"""
|
|
|
|
def make(config: BreakerConfig):
|
|
if backend.name == "memory":
|
|
return InMemoryGate(config=config, now=clock)
|
|
from polygateway.backends.redis.breaker import RedisGate
|
|
|
|
return RedisGate(config=config, redis=backend.redis, scope=f"t{uuid4().hex[:8]}")
|
|
|
|
return make
|