qihu-bot
7e002ac2b5
新增 test_capability_discovery.py (30) + test_llm_router.py (21) 共 51 个哨兵单测,覆盖 #874 双 Claw 能力发现 / #875 LLM 路由熔断 / #876 OTel 桥接关键路径. 套件: 116 -> 165 passed.
268 lines
7.9 KiB
Python
268 lines
7.9 KiB
Python
"""P1-1: 单测 — llm_router 路由/熔断/上下文(双 Claw 共享 #875)。
|
||
|
||
为什么这个文件存在:
|
||
LLM Router 是 #875 加速带的关键能力,但 #937 副密钥 failover 设计依托
|
||
同一熔断 / 决策路径。锁死关键纯函数 + mock 路径,防止回归。
|
||
|
||
验证范围:
|
||
1. 上下文注入: set_route_ctx → get_route_ctx 含 tenant/claw/trace
|
||
2. OTel 桥接: set_route_ctx 同步注入 OTel 双 Claw 标签
|
||
3. _is_text_only: 排除 media-only 模型(关键坑:failover 误用图像模型给文本)
|
||
4. _tenant_authorized: 全局 / 私有模型归属校验
|
||
5. _fallbacks: 排除 exclude_ids + 仅纯文本 + is_default desc
|
||
6. 熔断: report_failure ×3 → _fail_count = 3(内存降级)
|
||
7. 熔断: report_success 清零
|
||
8. 熔断: TTL 过期自动清零
|
||
9. RouteDecision 数据类可序列化 / 字段完整
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import sys
|
||
import time
|
||
|
||
import pytest
|
||
|
||
_HERE = os.path.dirname(os.path.abspath(__file__))
|
||
sys.path.insert(0, os.path.join(_HERE, ".."))
|
||
|
||
from app.core import llm_router as lr # noqa: E402
|
||
from app.core.llm_router import ( # noqa: E402
|
||
SHADOW,
|
||
_is_text_only,
|
||
_tenant_authorized,
|
||
_fallbacks,
|
||
_fail_count,
|
||
_BREAKER_LOCAL,
|
||
report_failure,
|
||
report_success,
|
||
set_route_ctx,
|
||
get_route_ctx,
|
||
RouteDecision,
|
||
)
|
||
|
||
|
||
# =====================================================================
|
||
# 1. 上下文注入
|
||
# =====================================================================
|
||
|
||
def test_set_route_ctx_default():
|
||
set_route_ctx()
|
||
ctx = get_route_ctx()
|
||
assert ctx["claw_type"] == "web"
|
||
assert ctx["tenant_id"] == ""
|
||
assert "trace_id" in ctx
|
||
|
||
|
||
def test_set_route_ctx_im():
|
||
set_route_ctx(tenant_id="t-1", claw_type="im")
|
||
ctx = get_route_ctx()
|
||
assert ctx["tenant_id"] == "t-1"
|
||
assert ctx["claw_type"] == "im"
|
||
|
||
|
||
def test_set_route_ctx_otel_bridge(monkeypatch):
|
||
"""set_route_ctx 应同步注入 OTel 双 Claw 标签"""
|
||
captured = {}
|
||
|
||
def fake_set_otel_ctx(*, tenant_id, claw_type):
|
||
captured["tenant_id"] = tenant_id
|
||
captured["claw_type"] = claw_type
|
||
|
||
# patch the lazy import target
|
||
import app.core.otel as otel_mod
|
||
monkeypatch.setattr(otel_mod, "set_otel_ctx", fake_set_otel_ctx)
|
||
set_route_ctx(tenant_id="t-9", claw_type="im")
|
||
assert captured == {"tenant_id": "t-9", "claw_type": "im"}
|
||
|
||
|
||
def test_set_route_ctx_otel_silently_swallows_failure(monkeypatch):
|
||
"""otel 不可用时静默跳过,不冒泡"""
|
||
import app.core.otel as otel_mod
|
||
def boom(**kw):
|
||
raise RuntimeError("otel down")
|
||
monkeypatch.setattr(otel_mod, "set_otel_ctx", boom)
|
||
# 应不抛
|
||
set_route_ctx(tenant_id="t-x", claw_type="web")
|
||
|
||
|
||
# =====================================================================
|
||
# 2. _is_text_only — failover 备选关键判断
|
||
# =====================================================================
|
||
|
||
class FakeModel:
|
||
def __init__(self, capabilities):
|
||
self.capabilities = capabilities
|
||
|
||
|
||
def test_is_text_only_pure_text():
|
||
assert _is_text_only(FakeModel(["text"])) is True
|
||
|
||
|
||
def test_is_text_only_excludes_image():
|
||
"""纯图像模型不是文本 failover 备选"""
|
||
assert _is_text_only(FakeModel(["image"])) is False
|
||
|
||
|
||
def test_is_text_only_excludes_mixed_with_media():
|
||
"""含 text 但同时有 image → 视为媒体"""
|
||
assert _is_text_only(FakeModel(["text", "image"])) is False
|
||
|
||
|
||
def test_is_text_only_handles_empty():
|
||
assert _is_text_only(FakeModel([])) is False
|
||
assert _is_text_only(FakeModel(None)) is False
|
||
|
||
|
||
# =====================================================================
|
||
# 3. _tenant_authorized
|
||
# =====================================================================
|
||
|
||
class FakeModelWithTenant:
|
||
def __init__(self, tenant_id):
|
||
self.tenant_id = tenant_id
|
||
|
||
|
||
def test_tenant_authorized_global_open():
|
||
"""全局模型(无 tenant_id)任何人可用"""
|
||
m = FakeModelWithTenant(None)
|
||
assert _tenant_authorized(m, "t-1") is True
|
||
assert _tenant_authorized(m, "") is True
|
||
|
||
|
||
def test_tenant_authorized_private_match():
|
||
m = FakeModelWithTenant("t-1")
|
||
assert _tenant_authorized(m, "t-1") is True
|
||
|
||
|
||
def test_tenant_authorized_private_mismatch():
|
||
m = FakeModelWithTenant("t-1")
|
||
assert _tenant_authorized(m, "t-2") is False
|
||
|
||
|
||
def test_tenant_authorized_no_tenant_blocked():
|
||
"""私有模型 + caller 没传 tenant → 拒绝"""
|
||
m = FakeModelWithTenant("t-1")
|
||
assert _tenant_authorized(m, "") is False
|
||
|
||
|
||
# =====================================================================
|
||
# 4. _fallbacks: 排除 exclude_ids + 纯文本 + is_default desc
|
||
# =====================================================================
|
||
|
||
class FakeAimModel:
|
||
def __init__(self, id, capabilities, is_default=False, is_enabled=True):
|
||
self.id = id
|
||
self.capabilities = capabilities
|
||
self.is_default = is_default
|
||
self.is_enabled = is_enabled
|
||
|
||
|
||
class FakeDb:
|
||
def __init__(self, rows):
|
||
self._rows = rows
|
||
def query(self, cls):
|
||
rows = self._rows
|
||
class Q:
|
||
def filter(self_inner, *a, **kw):
|
||
return self_inner
|
||
def order_by(self_inner, *a, **kw):
|
||
return self_inner
|
||
def all(self_inner):
|
||
return rows
|
||
return Q()
|
||
|
||
|
||
def test_fallbacks_excludes_exclude_ids():
|
||
db = FakeDb([
|
||
FakeAimModel("a", ["text"], is_default=True),
|
||
FakeAimModel("b", ["text"], is_default=False),
|
||
])
|
||
out = _fallbacks(db, exclude_ids={"a"}, limit=4)
|
||
assert [m.id for m in out] == ["b"]
|
||
|
||
|
||
def test_fallbacks_excludes_media():
|
||
db = FakeDb([
|
||
FakeAimModel("a", ["text"], is_default=True),
|
||
FakeAimModel("b", ["image"], is_default=True), # 媒体不算
|
||
])
|
||
out = _fallbacks(db, exclude_ids=set(), limit=4)
|
||
assert [m.id for m in out] == ["a"]
|
||
|
||
|
||
def test_fallbacks_limit_caps():
|
||
db = FakeDb([
|
||
FakeAimModel(f"m{i}", ["text"], is_default=False) for i in range(10)
|
||
])
|
||
out = _fallbacks(db, exclude_ids=set(), limit=3)
|
||
assert len(out) == 3
|
||
|
||
|
||
# =====================================================================
|
||
# 5. 熔断: 内存降级(无 Redis)
|
||
# =====================================================================
|
||
|
||
def setup_function(_):
|
||
"""每个测试前清空内存 breaker 状态"""
|
||
_BREAKER_LOCAL.clear()
|
||
|
||
|
||
def test_breaker_increments_and_reads():
|
||
report_failure("m-1")
|
||
report_failure("m-1")
|
||
report_failure("m-1")
|
||
assert _fail_count("m-1") == 3
|
||
|
||
|
||
def test_breaker_success_clears():
|
||
report_failure("m-2")
|
||
report_failure("m-2")
|
||
assert _fail_count("m-2") == 2
|
||
report_success("m-2")
|
||
assert _fail_count("m-2") == 0
|
||
|
||
|
||
def test_breaker_ttl_expires(monkeypatch):
|
||
"""TTL 过期后 _fail_count 返回 0,本地条目清理"""
|
||
report_failure("m-3")
|
||
# 假装时间穿越到未来(after TTL)
|
||
fake_now = time.time() + 1000
|
||
monkeypatch.setattr(lr.time, "time", lambda: fake_now)
|
||
assert _fail_count("m-3") == 0
|
||
assert "m-3" not in _BREAKER_LOCAL
|
||
|
||
|
||
def test_breaker_empty_model_id():
|
||
report_failure("")
|
||
assert _fail_count("") == 0
|
||
|
||
|
||
# =====================================================================
|
||
# 6. RouteDecision 数据类
|
||
# =====================================================================
|
||
|
||
def test_route_decision_defaults():
|
||
rd = RouteDecision(capability="text", claw_type="web", tenant_id="t-1")
|
||
assert rd.capability == "text"
|
||
assert rd.claw_type == "web"
|
||
assert rd.tenant_id == "t-1"
|
||
assert rd.models == []
|
||
assert rd.primary is None
|
||
assert rd.shadow is True
|
||
assert rd.matched_legacy is None
|
||
assert rd.reason == ""
|
||
|
||
|
||
def test_route_decision_fields_assignable():
|
||
rd = RouteDecision(capability="text", claw_type="im", tenant_id="t-1")
|
||
rd.models = ["a", "b"]
|
||
rd.primary = "a"
|
||
rd.reason = "explicit"
|
||
rd.matched_legacy = True
|
||
assert rd.models == ["a", "b"]
|
||
assert rd.primary == "a"
|
||
assert rd.reason == "explicit"
|
||
assert rd.matched_legacy is True
|
||
|