tzt
a043493548
- routes._fallback_chain：主条目之外的启用真实后端按（档位升序，单价和升序） 构成有序候选链（≤3），排除 mock/停用/同模型，仅作 D-P4 首 token 前 failover - _stream_response/_json_response 接入链式候选（原 [entry,entry] 同条目重试升级为真降级链） - upstream.stream：failover 发生即记 sink[upstream_fallback] 与模块级统计； JSON 路径经 X-Upstream-Fallback/Original/Used/Reason 四头如实上报 （流式路径头已发出不可追溯，由统计聚合暴露） - /proxy/admin/stats 新增 upstream_failover 块（total + recent 20 条） pytest 447 passed（T-X1 后 443 + 4）
131 lines
5.1 KiB
Python
131 lines
5.1 KiB
Python
"""上游有序降级链测试（T-X2）：链构造排序 + 首 token 前 failover 三元组上报。"""
|
|
import asyncio
|
|
import json
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from gateway.model_pool import PoolStore
|
|
from gateway.proxy.routes import _fallback_chain
|
|
from gateway.proxy.upstream import UpstreamError, failover_stats, reset_failover_stats
|
|
|
|
pytest.importorskip("fastapi")
|
|
|
|
|
|
def _make_pool(tmp_path) -> PoolStore:
|
|
store = PoolStore(path=tmp_path / "pool.json")
|
|
store.upsert({"id": "p-main", "name": "主", "tier": "premium", "backend": "openai",
|
|
"base_url": "http://bad", "model": "big-x", "price_in": 1.0,
|
|
"price_out": 2.0, "enabled": True})
|
|
store.upsert({"id": "p-cheap", "name": "便宜云", "tier": "budget", "backend": "openai",
|
|
"base_url": "http://good", "model": "mid-y", "price_in": 0.1,
|
|
"price_out": 0.2, "enabled": True})
|
|
store.upsert({"id": "p-local", "name": "本地", "tier": "local",
|
|
"backend": "llama_server", "base_url": "http://local:8901",
|
|
"model": "qwen-local", "price_in": 0.0, "price_out": 0.0,
|
|
"enabled": True})
|
|
store.upsert({"id": "p-off", "name": "停用", "tier": "local",
|
|
"backend": "openai", "base_url": "http://off", "model": "off-z",
|
|
"enabled": False})
|
|
store.upsert({"id": "p-mock", "name": "假", "tier": "local", "backend": "mock",
|
|
"model": "mock", "enabled": True})
|
|
return store
|
|
|
|
|
|
def test_fallback_chain_order_and_exclusions(tmp_path):
|
|
"""链序：档位升序 -> 单价和升序；排除 mock/停用/同模型；截断到 max_total。"""
|
|
pool = _make_pool(tmp_path)
|
|
main = pool.find_by_model("big-x")
|
|
chain = _fallback_chain(pool, main)
|
|
assert [e["model"] for e in chain] == ["qwen-local", "mid-y"] # local(0) < budget(1)
|
|
# 预算降档后的条目作主条目：不回排同档更贵者之外的高价条目在前
|
|
short = _fallback_chain(pool, main, max_total=2)
|
|
assert [e["model"] for e in short] == ["qwen-local"]
|
|
|
|
|
|
def test_fallback_chain_excludes_same_model(tmp_path):
|
|
"""同模型不同条目不进链（避免重复打同一上游）。"""
|
|
pool = _make_pool(tmp_path)
|
|
pool.upsert({"id": "p-dup", "name": "重", "tier": "budget", "backend": "openai",
|
|
"base_url": "http://dup", "model": "mid-y", "enabled": True})
|
|
mid = pool.find_by_model("mid-y")
|
|
assert all(e["model"] != "mid-y" for e in _fallback_chain(pool, mid))
|
|
|
|
|
|
def test_stream_failover_reports_triplet():
|
|
"""主上游首 token 前 500 -> 切换候选成功；sink 与统计如实上报三元组。"""
|
|
reset_failover_stats()
|
|
calls: list = []
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
calls.append(str(request.url))
|
|
if request.url.host == "bad":
|
|
return httpx.Response(500, text="boom")
|
|
body = ("\n\n".join([
|
|
'data: {"choices":[{"delta":{"content":"OK"}}]}',
|
|
"data: [DONE]",
|
|
]) + "\n\n").encode("utf-8")
|
|
return httpx.Response(200, content=body)
|
|
|
|
import gateway.proxy.upstream as up
|
|
client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
|
|
orig = up._client
|
|
up._client = client
|
|
primary = {"base_url": "http://bad", "model": "big-x", "provider": "openai"}
|
|
fallback = {"base_url": "http://good", "model": "mid-y", "provider": "openai"}
|
|
|
|
async def run():
|
|
sink: dict = {}
|
|
chunks = []
|
|
async for b in up.stream({"model": "big-x", "messages": []}, primary, sink,
|
|
[fallback]):
|
|
chunks.append(b)
|
|
return sink, chunks
|
|
|
|
try:
|
|
sink, chunks = asyncio.run(run())
|
|
finally:
|
|
up._client = orig
|
|
|
|
assert b"OK" in b"".join(chunks)
|
|
fb = sink["upstream_fallback"]
|
|
assert fb["used"] is True
|
|
assert fb["original"] == "big-x"
|
|
assert fb["used_model"] == "mid-y"
|
|
assert "500" in fb["reason"]
|
|
stats = failover_stats()
|
|
assert stats["total"] == 1
|
|
assert stats["recent"][0]["original"] == "big-x"
|
|
reset_failover_stats()
|
|
assert failover_stats()["total"] == 0
|
|
|
|
|
|
def test_stream_all_fail_raises_and_counts():
|
|
"""全部候选失败 -> UpstreamError；每次首 token 前失败均计入统计。"""
|
|
reset_failover_stats()
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(503, text="down")
|
|
|
|
import gateway.proxy.upstream as up
|
|
client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
|
|
orig = up._client
|
|
up._client = client
|
|
primary = {"base_url": "http://bad", "model": "big-x", "provider": "openai"}
|
|
fallback = {"base_url": "http://good", "model": "mid-y", "provider": "openai"}
|
|
|
|
async def run():
|
|
sink: dict = {}
|
|
async for _b in up.stream({"model": "big-x", "messages": []}, primary, sink,
|
|
[fallback]):
|
|
pass
|
|
|
|
try:
|
|
with pytest.raises(UpstreamError):
|
|
asyncio.run(run())
|
|
finally:
|
|
up._client = orig
|
|
|
|
assert failover_stats()["total"] == 2 # 两个候选各记一次
|
|
reset_failover_stats()
|