iomgaa
7b9815f4bc
Includes config aggregation for multi-source env keys, from_env and from_settings factories with explicit shared-backend injection, gather_bounded, top-level exports, tightened import-linter layers with the gate removed from the Makefile, and the finalized .env.example.
63 lines
2.2 KiB
Python
63 lines
2.2 KiB
Python
"""sources.py 选源策略与冷却备忘测试(蓝本 CHS app/providers/selector.py:20,36)。"""
|
|
|
|
from polygateway.sources import LeastInflightSelector, RoundRobinSelector, SourceCooldownMemo
|
|
from polygateway.types import SourceConfig, SourceStats
|
|
|
|
|
|
def _src(name):
|
|
return SourceConfig(
|
|
name=name,
|
|
provider="openai",
|
|
base_url="https://gw.example/v1",
|
|
api_key="sk",
|
|
model="m",
|
|
timeout_s=10.0,
|
|
)
|
|
|
|
|
|
class TestRoundRobin:
|
|
def test_rotation_advances_each_call(self):
|
|
sources = [_src("a"), _src("b"), _src("c")]
|
|
sel = RoundRobinSelector()
|
|
orders = [tuple(s.name for s in sel.order(sources, {})) for _ in range(4)]
|
|
assert orders == [("a", "b", "c"), ("b", "c", "a"), ("c", "a", "b"), ("a", "b", "c")]
|
|
|
|
def test_single_source_stable(self):
|
|
sources = [_src("only")]
|
|
sel = RoundRobinSelector()
|
|
assert [s.name for s in sel.order(sources, {})] == ["only"]
|
|
assert [s.name for s in sel.order(sources, {})] == ["only"]
|
|
|
|
|
|
class TestLeastInflight:
|
|
def test_orders_by_inflight_ascending(self):
|
|
sources = [_src("a"), _src("b"), _src("c")]
|
|
stats = {"a": SourceStats(5, 0, 0), "b": SourceStats(1, 0, 0), "c": SourceStats(3, 0, 0)}
|
|
sel = LeastInflightSelector()
|
|
assert [s.name for s in sel.order(sources, stats)] == ["b", "c", "a"]
|
|
|
|
def test_stable_on_ties(self):
|
|
sources = [_src("a"), _src("b")]
|
|
stats = {"a": SourceStats(2, 0, 0), "b": SourceStats(2, 0, 0)}
|
|
assert [s.name for s in LeastInflightSelector().order(sources, stats)] == ["a", "b"]
|
|
|
|
|
|
class TestCooldownMemo:
|
|
def test_active_until_expiry(self):
|
|
t = {"now": 100.0}
|
|
memo = SourceCooldownMemo(now=lambda: t["now"])
|
|
assert not memo.active("a")
|
|
memo.set_until("a", 130.0)
|
|
assert memo.active("a")
|
|
assert memo.remaining("a") == 30.0
|
|
t["now"] = 131.0
|
|
assert not memo.active("a")
|
|
assert memo.remaining("a") == 0.0
|
|
|
|
def test_extend_after_wait_hint(self):
|
|
t = {"now": 0.0}
|
|
memo = SourceCooldownMemo(now=lambda: t["now"])
|
|
memo.set_until("a", 10.0)
|
|
memo.set_until("a", 5.0) # 更早的截止不回退已有备忘
|
|
assert memo.remaining("a") == 10.0
|