62 lines
1.8 KiB
Python
62 lines
1.8 KiB
Python
"""契约测试共享 fixture: 后端参数化(M1 仅 memory,M2 增 redis 零改测试)。
|
|
|
|
FakeClock 仅对支持时钟注入的后端有效(memory);M2 接入 Redis 后端时,
|
|
依赖时钟推进的用例按后端能力跳过或改用真实等待。
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from polygateway.backends.memory.breaker import InMemoryGate
|
|
from polygateway.backends.memory.limiter import InMemoryLimiter
|
|
from polygateway.types import BreakerConfig, GlobalLimits, SourceConfig
|
|
|
|
|
|
class FakeClock:
|
|
"""确定性单调时钟;契约测试推进时间验证租约/冷却语义。"""
|
|
|
|
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
|
|
|
|
|
|
def make_source(name: str = "s1", **overrides) -> SourceConfig:
|
|
base = dict(
|
|
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)
|
|
|
|
|
|
@pytest.fixture
|
|
def clock() -> FakeClock:
|
|
return FakeClock()
|
|
|
|
|
|
@pytest.fixture(params=["memory"])
|
|
def limiter_factory(request, clock):
|
|
"""返回 (sources, global_limits, lease_ttl_s) -> RateLimiter 的工厂。"""
|
|
|
|
def make(sources: list[SourceConfig], global_limits: GlobalLimits, lease_ttl_s: float = 100.0):
|
|
return InMemoryLimiter(
|
|
scope="llm", sources={s.name: s for s in sources},
|
|
global_limits=global_limits, lease_ttl_s=lease_ttl_s, now=clock,
|
|
)
|
|
|
|
return make
|
|
|
|
|
|
@pytest.fixture(params=["memory"])
|
|
def gate_factory(request, clock):
|
|
"""返回 (BreakerConfig) -> ProviderGate 的工厂。"""
|
|
|
|
def make(config: BreakerConfig):
|
|
return InMemoryGate(config=config, now=clock)
|
|
|
|
return make
|