xsl
93b88db2f5
Covers routing, failover, the circuit breaker, structured-output retries, the cache-prefix validator, cost computation including cached-token pricing, the daily budget breaker, the quota protocol, the task state machine and log redaction. HTTP is mocked -- no unit test spends money. The integration test runs the real slice against PostgreSQL and Redis with a faked provider.
78 lines
2.8 KiB
Python
Executable File
78 lines
2.8 KiB
Python
Executable File
"""Scene routing configuration (架构文档 §8.1: 改模型不改代码)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from packages.llm.errors import SceneNotRoutedError
|
|
from packages.llm.routing import SceneRouter, build_scene_configs
|
|
from packages.llm.types import Scene
|
|
|
|
|
|
def test_defaults_put_the_primary_first_and_the_backup_second() -> None:
|
|
configs = build_scene_configs(primary_provider="deepseek", backup_provider="zhipu")
|
|
assert configs[Scene.AGENT_CHAT].providers == ("deepseek", "zhipu")
|
|
|
|
|
|
def test_a_missing_backup_yields_a_single_provider_chain() -> None:
|
|
configs = build_scene_configs(primary_provider="deepseek", backup_provider=None)
|
|
assert configs[Scene.AGENT_CHAT].providers == ("deepseek",)
|
|
|
|
|
|
def test_every_scene_is_routed_and_bound_to_a_prompt_version() -> None:
|
|
configs = build_scene_configs(primary_provider="deepseek")
|
|
assert set(configs) == set(Scene)
|
|
for config in configs.values():
|
|
assert config.prompt_id
|
|
assert config.prompt_version
|
|
|
|
|
|
def test_intent_classification_is_critical_so_budget_pressure_cannot_disable_it() -> None:
|
|
configs = build_scene_configs(primary_provider="deepseek")
|
|
assert configs[Scene.INTENT_CLASSIFY].critical is True
|
|
assert configs[Scene.AGENT_CHAT].critical is False
|
|
|
|
|
|
def test_agent_chat_output_budget_matches_the_token_plan() -> None:
|
|
# 架构文档 §7.2: 输出 ≤ 800
|
|
assert build_scene_configs(primary_provider="x")[Scene.AGENT_CHAT].max_tokens <= 800
|
|
|
|
|
|
def test_overrides_change_routing_without_a_code_change() -> None:
|
|
configs = build_scene_configs(
|
|
primary_provider="deepseek",
|
|
backup_provider="zhipu",
|
|
overrides={
|
|
"agent_chat": {
|
|
"max_tokens": 1200,
|
|
"prompt_version": "v2",
|
|
"models": {"deepseek": "deepseek-v4-pro"},
|
|
"providers": ["zhipu", "deepseek"],
|
|
}
|
|
},
|
|
)
|
|
config = configs[Scene.AGENT_CHAT]
|
|
assert config.max_tokens == 1200
|
|
assert config.prompt_version == "v2"
|
|
assert config.providers == ("zhipu", "deepseek")
|
|
assert config.model_for("deepseek", "fallback") == "deepseek-v4-pro"
|
|
assert config.model_for("zhipu", "fallback") == "fallback"
|
|
# Untouched scenes keep their defaults.
|
|
assert configs[Scene.REPORT_COPY].max_tokens == 3000
|
|
|
|
|
|
def test_structured_retry_budget_is_configurable() -> None:
|
|
configs = build_scene_configs(primary_provider="x", structured_max_retries=5)
|
|
assert configs[Scene.AGENT_CHAT].structured_max_retries == 5
|
|
|
|
|
|
def test_router_raises_for_an_unconfigured_scene() -> None:
|
|
router = SceneRouter({})
|
|
with pytest.raises(SceneNotRoutedError):
|
|
router.resolve(Scene.AGENT_CHAT)
|
|
|
|
|
|
def test_router_lists_configured_scenes() -> None:
|
|
router = SceneRouter(build_scene_configs(primary_provider="x"))
|
|
assert set(router.scenes()) == set(Scene)
|