b84b4839dc
Post a task to POST /route and print the full decision tree (classification, candidates, selected model, est cost, rejections) without dispatching a provider call. Supports --category/--tier/--context overrides and --json. Route-only and no-spend.
226 lines
6.6 KiB
Python
226 lines
6.6 KiB
Python
"""Tests for router_cli.py -- the one-shot routing-decision printer.
|
|
|
|
The CLI is route-only: it POSTs to the FastAPI ``POST /route`` endpoint and
|
|
prints the decision tree without ever dispatching to a provider. These tests
|
|
exercise the decision-rendering / posting logic through ``run()`` with an
|
|
injected ``post`` callback so no live server or provider is involved.
|
|
"""
|
|
|
|
import json
|
|
|
|
import router_cli
|
|
|
|
|
|
def _candidate(model_id, cost=0.0005, proficiency=0.95):
|
|
return {
|
|
"model_id": model_id,
|
|
"provider": "neuralwatt",
|
|
"tier": 2,
|
|
"latency_class": "standard",
|
|
"reasoning_mode": "default",
|
|
"context_variant": "full",
|
|
"effective_context_window": 262128,
|
|
"cost": cost,
|
|
"energy": 1.23e-05,
|
|
"eco": 0.5,
|
|
"list_price_per_1m": 0.42,
|
|
"composite": 0.8,
|
|
"cost_score": 0.9,
|
|
"proficiency_score": proficiency,
|
|
}
|
|
|
|
|
|
def _route_payload(selected=True, rejected_reason=None):
|
|
payload = {
|
|
"classification": {
|
|
"task_category": "coding_general",
|
|
"task_tier": 2,
|
|
"required_context_tokens": 5000,
|
|
"confidence": 0.9,
|
|
"escalated": False,
|
|
"source": "classifier",
|
|
},
|
|
"latency_tolerance": "interactive",
|
|
"candidates_considered": 4,
|
|
"selected": (
|
|
_candidate("deepseek-v4-flash", cost=1.6e-05, proficiency=0.97)
|
|
if selected
|
|
else None
|
|
),
|
|
"runners_up": [
|
|
_candidate("gemma-4-31b", cost=5.02e-05, proficiency=0.91),
|
|
_candidate("kimi-k3", cost=2.2e-05, proficiency=0.9),
|
|
],
|
|
}
|
|
# The 422 no-candidate body carries the limits/rejected reason in `detail`.
|
|
if rejected_reason is not None:
|
|
payload["detail"] = rejected_reason
|
|
return payload
|
|
|
|
|
|
class _FakeResponse:
|
|
"""minimal requests.Response stand-in with the fields router_cli reads."""
|
|
|
|
def __init__(self, status_code, payload):
|
|
self.status_code = status_code
|
|
self._payload = payload
|
|
|
|
def json(self):
|
|
return self._payload
|
|
|
|
@property
|
|
def text(self):
|
|
return json.dumps(self._payload)
|
|
|
|
|
|
def _fake_post(route_payload, status_code=200):
|
|
captured = {}
|
|
|
|
def post(url, json=None, **kwargs):
|
|
captured["url"] = url
|
|
captured["body"] = json
|
|
return _FakeResponse(status_code, route_payload)
|
|
|
|
return post, captured
|
|
|
|
|
|
# --- happy path -----------------------------------------------------------
|
|
|
|
|
|
def test_happy_path_renders_decision_tree(capsys):
|
|
"""The rendered text names the selected model, the category, and a cost."""
|
|
post, _ = _fake_post(_route_payload())
|
|
code = router_cli.run(
|
|
"Refactor this Django view", post=post
|
|
)
|
|
out = capsys.readouterr().out
|
|
|
|
assert code == 0
|
|
assert "selected" in out.lower()
|
|
assert "coding_general" in out
|
|
# A US$ cost figure for the selected model.
|
|
assert "1.6e-05" in out or "US$" in out
|
|
|
|
|
|
def test_happy_path_includes_runners_up_and_candidates(capsys):
|
|
post, _ = _fake_post(_route_payload())
|
|
router_cli.run("Refactor this Django view", post=post)
|
|
out = capsys.readouterr().out
|
|
|
|
assert "candidates_considered" in out.lower() or "considered" in out.lower()
|
|
assert "gemma-4-31b" in out # a runner-up
|
|
assert "kimi-k3" in out
|
|
|
|
|
|
def test_json_flag_emits_raw_route_json(capsys):
|
|
payload = _route_payload()
|
|
post, _ = _fake_post(payload)
|
|
code = router_cli.run("hello", as_json=True, post=post)
|
|
out = capsys.readouterr().out.strip()
|
|
|
|
assert code == 0
|
|
parsed = json.loads(out)
|
|
assert parsed["classification"]["task_category"] == "coding_general"
|
|
assert parsed["selected"]["model_id"] == "deepseek-v4-flash"
|
|
assert parsed["candidates_considered"] == 4
|
|
assert len(parsed["runners_up"]) == 2
|
|
|
|
|
|
def test_payload_includes_overrides_only_when_given(capsys):
|
|
post, captured = _fake_post(_route_payload())
|
|
router_cli.run(
|
|
"task",
|
|
category="coding_refactor",
|
|
tier=3,
|
|
context=12000,
|
|
post=post,
|
|
)
|
|
capsys.readouterr()
|
|
|
|
body = captured["body"]
|
|
assert body["task"] == "task"
|
|
assert body["task_category"] == "coding_refactor"
|
|
assert body["task_tier"] == 3
|
|
assert body["required_context_tokens"] == 12000
|
|
|
|
|
|
def test_payload_omits_overrides_when_not_given(capsys):
|
|
post, captured = _fake_post(_route_payload())
|
|
router_cli.run("task", post=post)
|
|
capsys.readouterr()
|
|
|
|
body = captured["body"]
|
|
assert body["task"] == "task"
|
|
assert "task_category" not in body
|
|
assert "task_tier" not in body
|
|
assert "required_context_tokens" not in body
|
|
|
|
|
|
# --- failure: 422 no candidate -------------------------------------------
|
|
|
|
def test_422_no_candidate_prints_rejection_and_exits_nonzero(capsys):
|
|
"""A no-candidate /route returns 422 with the rejected reason; the CLI
|
|
prints it and exits non-zero without a traceback."""
|
|
post, _ = _fake_post(
|
|
_route_payload(selected=False, rejected_reason="no model fits"),
|
|
status_code=422,
|
|
)
|
|
code = router_cli.run("task", post=post)
|
|
captured = capsys.readouterr()
|
|
out, err = captured.out, captured.err
|
|
|
|
assert code != 0
|
|
assert "no model fits" in out or "no model fits" in err
|
|
# No traceback leaked.
|
|
assert "Traceback" not in out
|
|
assert "Traceback" not in err
|
|
|
|
|
|
# --- failure: connection error -------------------------------------------
|
|
|
|
def test_connection_error_prints_clean_error_and_exits_nonzero(capsys):
|
|
def post(url, json=None, **kwargs):
|
|
raise ConnectionError("refused")
|
|
|
|
code = router_cli.run("task", post=post)
|
|
captured = capsys.readouterr()
|
|
out, err = captured.out, captured.err
|
|
|
|
assert code != 0
|
|
assert "refused" in out or "refused" in err
|
|
assert "Traceback" not in out
|
|
assert "Traceback" not in err
|
|
|
|
|
|
# --- malformed input ------------------------------------------------------
|
|
|
|
def test_tier_flag_rejects_out_of_range(capsys):
|
|
"""A --tier outside 1..3 must be rejected cleanly, non-zero, no traceback."""
|
|
code = router_cli.run("task", tier=9)
|
|
out = capsys.readouterr().out
|
|
err = capsys.readouterr().err
|
|
|
|
assert code != 0
|
|
assert "Traceback" not in out
|
|
assert "Traceback" not in err
|
|
|
|
|
|
def test_tier_flag_rejects_non_int(capsys):
|
|
code = router_cli.run("task", tier="abc")
|
|
out = capsys.readouterr().out
|
|
err = capsys.readouterr().err
|
|
|
|
assert code != 0
|
|
assert "Traceback" not in out
|
|
assert "Traceback" not in err
|
|
|
|
|
|
# --- arg parsing / usage --------------------------------------------------
|
|
|
|
def test_help_exits_zero():
|
|
try:
|
|
code = router_cli.main(["--help"])
|
|
except SystemExit as e:
|
|
code = e.code
|
|
assert code == 0
|