generated from coulomb/repo-seed
feat: WP-0003 — RoutingPolicy (FR-2) and HTTP serve mode (FR-1)
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>
This commit is contained in:
91
tests/test_routing.py
Normal file
91
tests/test_routing.py
Normal file
@@ -0,0 +1,91 @@
|
||||
"""
|
||||
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
|
||||
Reference in New Issue
Block a user