generated from coulomb/repo-seed
Bernd Worsch
d51d6303e2
FR-2 RoutingPolicy: - RoutingPolicy + RoutingRule dataclasses in llm_connect/routing.py - resolve(task_type, estimated_cost_per_1k=None) with cost-cap fallback - Exported from llm_connect.__init__; contract doc at contracts/functional/routing-policy.md - 11 tests covering rule match, cost-cap, fallback, unknown type, no-match FR-1 HTTP serve mode: - LLMServer in llm_connect/server.py (stdlib http.server, zero extra deps) - POST /execute + GET /health; CLI via python -m llm_connect.server - [server] optional-dep group added to pyproject.toml - Contract doc at contracts/functional/server.md - 9 tests: health, round-trip, 400/404/500 errors, config forwarding - Added "mock" provider to factory for CLI default All 101 tests pass. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
92 lines
3.3 KiB
Python
92 lines
3.3 KiB
Python
"""
|
|
Tests for RoutingPolicy (FR-2).
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from llm_connect.routing import RoutingPolicy, RoutingRule
|
|
from llm_connect.adapter import MockLLMAdapter
|
|
|
|
|
|
class TestRoutingPolicy:
|
|
def _adapters(self, n: int = 3):
|
|
return [MockLLMAdapter(mock_response=f"resp-{i}") for i in range(n)]
|
|
|
|
def test_rule_match_returns_prefer(self):
|
|
prefer, *_ = self._adapters()
|
|
policy = RoutingPolicy(rules=[RoutingRule("triage", prefer=prefer)])
|
|
assert policy.resolve("triage") is prefer
|
|
|
|
def test_first_matching_rule_wins(self):
|
|
a, b = self._adapters(2)
|
|
policy = RoutingPolicy(rules=[
|
|
RoutingRule("triage", prefer=a),
|
|
RoutingRule("triage", prefer=b),
|
|
])
|
|
assert policy.resolve("triage") is a
|
|
|
|
def test_cost_cap_within_limit_returns_prefer(self):
|
|
prefer, fallback = self._adapters(2)
|
|
policy = RoutingPolicy(rules=[
|
|
RoutingRule("triage", prefer=prefer, max_cost_per_1k=1.0, fallback=fallback)
|
|
])
|
|
assert policy.resolve("triage", estimated_cost_per_1k=0.5) is prefer
|
|
|
|
def test_cost_cap_exceeded_returns_fallback(self):
|
|
prefer, fallback = self._adapters(2)
|
|
policy = RoutingPolicy(rules=[
|
|
RoutingRule("triage", prefer=prefer, max_cost_per_1k=1.0, fallback=fallback)
|
|
])
|
|
assert policy.resolve("triage", estimated_cost_per_1k=2.0) is fallback
|
|
|
|
def test_cost_cap_exceeded_no_fallback_returns_prefer(self):
|
|
"""When cost exceeds cap but no fallback is set, still return prefer."""
|
|
prefer, *_ = self._adapters()
|
|
policy = RoutingPolicy(rules=[
|
|
RoutingRule("triage", prefer=prefer, max_cost_per_1k=0.1)
|
|
])
|
|
assert policy.resolve("triage", estimated_cost_per_1k=5.0) is prefer
|
|
|
|
def test_no_estimated_cost_ignores_cap(self):
|
|
prefer, fallback = self._adapters(2)
|
|
policy = RoutingPolicy(rules=[
|
|
RoutingRule("triage", prefer=prefer, max_cost_per_1k=0.01, fallback=fallback)
|
|
])
|
|
# No cost estimate → cap not applied
|
|
assert policy.resolve("triage") is prefer
|
|
|
|
def test_unknown_task_type_returns_default(self):
|
|
prefer, default = self._adapters(2)
|
|
policy = RoutingPolicy(
|
|
rules=[RoutingRule("triage", prefer=prefer)],
|
|
default=default,
|
|
)
|
|
assert policy.resolve("unknown") is default
|
|
|
|
def test_no_match_no_default_raises_lookup_error(self):
|
|
prefer, *_ = self._adapters()
|
|
policy = RoutingPolicy(rules=[RoutingRule("triage", prefer=prefer)])
|
|
with pytest.raises(LookupError, match="unknown"):
|
|
policy.resolve("unknown")
|
|
|
|
def test_empty_rules_with_default_returns_default(self):
|
|
default, *_ = self._adapters()
|
|
policy = RoutingPolicy(default=default)
|
|
assert policy.resolve("anything") is default
|
|
|
|
def test_empty_policy_raises(self):
|
|
policy = RoutingPolicy()
|
|
with pytest.raises(LookupError):
|
|
policy.resolve("triage")
|
|
|
|
def test_multiple_task_types(self):
|
|
a, b, c = self._adapters(3)
|
|
policy = RoutingPolicy(rules=[
|
|
RoutingRule("fast", prefer=a),
|
|
RoutingRule("smart", prefer=b),
|
|
RoutingRule("cheap", prefer=c),
|
|
])
|
|
assert policy.resolve("fast") is a
|
|
assert policy.resolve("smart") is b
|
|
assert policy.resolve("cheap") is c
|