1462 lines
59 KiB
Python
1462 lines
59 KiB
Python
"""Tests for routing.py — hard filters and weighted ranking.
|
|
|
|
The filters are the part that disqualifies outright, so each one gets a case
|
|
proving it rejects and a case proving it does not over-reject.
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from config import FlexPreference
|
|
from routing import (
|
|
apply_flex_preference,
|
|
capability_gate_reason,
|
|
credit_attenuation_multiplier,
|
|
estimated_cost,
|
|
get_flex_sibling,
|
|
is_eligible,
|
|
rank_candidates,
|
|
rejection_reason,
|
|
select_candidates,
|
|
)
|
|
|
|
|
|
|
|
def _row(**overrides) -> dict:
|
|
"""A routable model row; override one field per test."""
|
|
row = {
|
|
"model_id": "m",
|
|
"provider": "neuralwatt",
|
|
"tier": 2,
|
|
"cost": 1.0,
|
|
"energy": 1.0e-5,
|
|
"effective_context_window": 100_000,
|
|
"availability": "active",
|
|
"deprecated": 0,
|
|
"access_level": "public",
|
|
"latency_class": "standard",
|
|
"reasoning_mode": "default",
|
|
"context_variant": "full",
|
|
"supports_vision": 1,
|
|
"supports_json_mode": 1,
|
|
}
|
|
row.update(overrides)
|
|
return row
|
|
|
|
|
|
def _eligible(row, **overrides) -> bool:
|
|
kwargs = {
|
|
"required_context_tokens": 10_000,
|
|
"required_tier": 2,
|
|
"latency_tolerance": "interactive",
|
|
"allowed_access_levels": ["public"],
|
|
"exclude_stale": True,
|
|
"exclude_deprecated": True,
|
|
}
|
|
kwargs.update(overrides)
|
|
return is_eligible(row, **kwargs)
|
|
|
|
|
|
# --- context window -------------------------------------------------------
|
|
|
|
def test_context_window_too_small_is_rejected():
|
|
assert _eligible(_row(effective_context_window=5_000)) is False
|
|
|
|
|
|
def test_context_window_exactly_equal_is_accepted():
|
|
# The filter is >=, so a model that exactly fits is still a candidate
|
|
assert _eligible(_row(effective_context_window=10_000)) is True
|
|
|
|
|
|
def test_null_context_window_is_rejected():
|
|
# Given: a row whose window never got derived. An unknown window cannot be
|
|
# shown to fit, and truncating mid-task is worse than routing elsewhere.
|
|
assert _eligible(_row(effective_context_window=None)) is False
|
|
|
|
|
|
# --- tier floor -----------------------------------------------------------
|
|
|
|
def test_tier_below_required_is_rejected():
|
|
assert _eligible(_row(tier=1), required_tier=2) is False
|
|
|
|
|
|
def test_tier_above_required_is_accepted():
|
|
# tier is a floor, not an equality match
|
|
assert _eligible(_row(tier=3), required_tier=2) is True
|
|
|
|
|
|
def test_null_tier_is_rejected():
|
|
assert _eligible(_row(tier=None)) is False
|
|
|
|
|
|
# --- freshness ------------------------------------------------------------
|
|
|
|
def test_stale_row_is_rejected_when_configured():
|
|
assert _eligible(_row(availability="stale")) is False
|
|
|
|
|
|
def test_stale_row_is_kept_when_not_excluded():
|
|
assert _eligible(_row(availability="stale"), exclude_stale=False) is True
|
|
|
|
|
|
def test_deprecated_row_is_rejected():
|
|
assert _eligible(_row(availability="deprecated")) is False
|
|
assert _eligible(_row(deprecated=1)) is False
|
|
|
|
|
|
# --- circuit breaker (exclude_models) -------------------------------------
|
|
|
|
def test_circuit_open_row_is_rejected_by_exclude_models():
|
|
assert _eligible(_row(model_id="dead-model"), exclude_models={"dead-model"}) is False
|
|
|
|
|
|
def test_circuit_open_row_is_kept_when_not_excluded():
|
|
assert _eligible(_row(model_id="dead-model"), exclude_models=set()) is True
|
|
|
|
|
|
def test_exclude_models_reason_is_circuit_open():
|
|
row = _row(model_id="dead-model")
|
|
reason = rejection_reason(
|
|
row,
|
|
required_context_tokens=10_000,
|
|
required_tier=2,
|
|
latency_tolerance="interactive",
|
|
allowed_access_levels=["public"],
|
|
exclude_stale=True,
|
|
exclude_deprecated=True,
|
|
exclude_models={"dead-model"},
|
|
)
|
|
assert reason == "circuit_open"
|
|
|
|
|
|
|
|
# --- access gating --------------------------------------------------------
|
|
|
|
def test_grant_gated_rows_are_rejected_by_default():
|
|
# Given: the glm-5.2-short* rows, which are private preview. Routing to
|
|
# one earns a 403 at dispatch, so they are excluded before scoring.
|
|
assert _eligible(_row(access_level="preview")) is False
|
|
assert _eligible(_row(access_level="canary")) is False
|
|
|
|
|
|
def test_gated_rows_are_admitted_when_the_account_holds_the_grant():
|
|
assert (
|
|
_eligible(_row(access_level="preview"), allowed_access_levels=["public", "preview"])
|
|
is True
|
|
)
|
|
|
|
|
|
# --- latency tolerance ----------------------------------------------------
|
|
|
|
def test_flex_row_is_rejected_for_interactive_work():
|
|
# Given: a flex row, held server-side during peak until capacity frees up
|
|
assert _eligible(_row(latency_class="flex"), latency_tolerance="interactive") is False
|
|
|
|
|
|
def test_flex_row_is_admitted_for_batch_work():
|
|
assert _eligible(_row(latency_class="flex"), latency_tolerance="batch") is True
|
|
|
|
|
|
def test_standard_row_is_admitted_for_batch_work():
|
|
# Batch tolerates flex; it does not require it
|
|
assert _eligible(_row(latency_class="standard"), latency_tolerance="batch") is True
|
|
|
|
|
|
# --- select_candidates ----------------------------------------------------
|
|
|
|
def test_select_candidates_filters_and_preserves_order():
|
|
rows = [
|
|
_row(model_id="keep-1"),
|
|
_row(model_id="drop-flex", latency_class="flex"),
|
|
_row(model_id="keep-2", tier=3),
|
|
_row(model_id="drop-gated", access_level="preview"),
|
|
]
|
|
selected = select_candidates(
|
|
rows,
|
|
required_context_tokens=10_000,
|
|
required_tier=2,
|
|
latency_tolerance="interactive",
|
|
allowed_access_levels=["public"],
|
|
exclude_stale=True,
|
|
exclude_deprecated=True,
|
|
)
|
|
assert [r["model_id"] for r in selected] == ["keep-1", "keep-2"]
|
|
|
|
|
|
# --- measured cost --------------------------------------------------------
|
|
|
|
# --- ranking: quality first, cheapest among equals -------------------------
|
|
|
|
def test_a_real_quality_gap_decides_outright():
|
|
# tool_use_agentic spans 0.67 across the catalog; a gap that size must
|
|
# beat any cost saving
|
|
rows = [_row(model_id="cheap-bad", cost=1e-6, proficiency=0.33),
|
|
_row(model_id="dear-good", cost=1e-3, proficiency=1.00)]
|
|
assert rank_candidates(rows)[0]["model_id"] == "dear-good"
|
|
|
|
|
|
def test_within_tolerance_the_cheaper_model_wins():
|
|
# 0.02 apart on 2-3 samples is sampling noise, not a quality difference.
|
|
# Paying 100x for it would be buying noise.
|
|
rows = [_row(model_id="cheap", cost=1e-5, proficiency=0.98),
|
|
_row(model_id="dear", cost=1e-3, proficiency=1.00)]
|
|
assert rank_candidates(rows, quality_tolerance=0.1)[0]["model_id"] == "cheap"
|
|
|
|
|
|
def test_narrowing_the_tolerance_makes_small_gaps_count():
|
|
# As samples accumulate and confidence rises, the band should shrink
|
|
rows = [_row(model_id="cheap", cost=1e-5, proficiency=0.98),
|
|
_row(model_id="dear", cost=1e-3, proficiency=1.00)]
|
|
assert rank_candidates(rows, quality_tolerance=0.001)[0]["model_id"] == "dear"
|
|
|
|
|
|
def test_eco_no_longer_influences_the_decision():
|
|
# Carbon is still logged; it is simply not what this router optimizes
|
|
rows = [_row(model_id="clean", cost=1e-3, proficiency=0.5, eco=1e-6),
|
|
_row(model_id="dirty", cost=1e-5, proficiency=0.5, eco=1e9)]
|
|
assert rank_candidates(rows)[0]["model_id"] == "dirty"
|
|
|
|
|
|
def test_missing_proficiency_is_neutral_not_penalized():
|
|
ranked = rank_candidates([_row()])
|
|
assert ranked[0]["proficiency_score"] == 0.5
|
|
|
|
|
|
def test_unknown_cost_does_not_disqualify(db_free=None):
|
|
# A model never swept must stay pickable, or it can never acquire a
|
|
# measurement — the same trap the neutral-0.5 default avoids
|
|
ranked = rank_candidates([_row(model_id="unswept", cost=None)])
|
|
assert ranked[0]["model_id"] == "unswept"
|
|
|
|
|
|
def test_ties_break_deterministically():
|
|
rows = [_row(model_id="glm-b"), _row(model_id="glm-a"), _row(model_id="glm-c")]
|
|
first = [r["model_id"] for r in rank_candidates(rows)]
|
|
second = [r["model_id"] for r in rank_candidates(list(reversed(rows)))]
|
|
assert first == ["glm-a", "glm-b", "glm-c"] == second
|
|
|
|
|
|
def test_empty_candidate_set_ranks_to_empty():
|
|
assert rank_candidates([]) == []
|
|
|
|
|
|
# --- the budget ceiling: the quota mandate as a guarantee ------------------
|
|
|
|
def test_ceiling_excludes_models_over_budget():
|
|
# Denominated in kWh: the plan is a fixed quota, and a quota is a wall you
|
|
# hit mid-task rather than a bill that accrues
|
|
rows = [_row(model_id="affordable", energy=1e-6, proficiency=0.5),
|
|
_row(model_id="expensive", energy=1e-3, proficiency=1.0)]
|
|
ranked = rank_candidates(rows, max_energy_per_request=1e-4)
|
|
assert [r["model_id"] for r in ranked] == ["affordable"]
|
|
|
|
|
|
def test_ceiling_binds_even_against_the_best_model():
|
|
# This is the point of a constraint rather than a weight: no amount of
|
|
# quality buys past the ceiling
|
|
rows = [_row(model_id="expensive", energy=1e-3, proficiency=1.0)]
|
|
assert rank_candidates(rows, max_energy_per_request=1e-6) == []
|
|
|
|
|
|
def test_no_ceiling_admits_everything():
|
|
rows = [_row(model_id="expensive", energy=1e9, proficiency=1.0)]
|
|
assert len(rank_candidates(rows, max_energy_per_request=None)) == 1
|
|
|
|
|
|
def test_unmeasured_cost_is_admitted_under_a_ceiling():
|
|
# Excluding the unmeasured would mean a new model could never be picked
|
|
# and so could never acquire a measurement
|
|
rows = [_row(model_id="unswept", energy=None, proficiency=1.0)]
|
|
assert len(rank_candidates(rows, max_energy_per_request=1e-9)) == 1
|
|
|
|
|
|
def test_ceiling_at_exactly_the_cost_admits():
|
|
rows = [_row(model_id="borderline", energy=1e-6, proficiency=0.5)]
|
|
assert len(rank_candidates(rows, max_energy_per_request=1e-6)) == 1
|
|
|
|
|
|
# --- tool competence is read from the request, not guessed at -------------
|
|
|
|
def test_tools_in_the_request_exclude_a_model_that_overreaches_for_them():
|
|
# deepseek-v4-flash measures 0.33 here. The recorded failure is a
|
|
# NON-agentic prompt ("it is 1:20pm, my meeting is at 3pm, how many
|
|
# minutes?") where it called two tools instead of subtracting -- so the
|
|
# hazard is tools being available, not the task being agentic.
|
|
weak = _row(model_id="overreacher", proficiency=1.0, tool_proficiency=0.33)
|
|
strong = _row(model_id="reliable", proficiency=1.0, tool_proficiency=1.0)
|
|
assert _eligible(weak, min_tool_proficiency=0.5) is False
|
|
assert _eligible(strong, min_tool_proficiency=0.5) is True
|
|
|
|
|
|
def test_without_tools_the_same_model_is_fine():
|
|
# The filter applies only when the request carries tool definitions;
|
|
# otherwise a cheap tool-clumsy model is a perfectly good choice.
|
|
weak = _row(model_id="overreacher", tool_proficiency=0.33)
|
|
assert _eligible(weak, min_tool_proficiency=None) is True
|
|
|
|
|
|
def test_an_unmeasured_model_is_unproven_not_disqualified():
|
|
# Same principle as the tier-1 context gate: absent evidence must not
|
|
# decide anything. A model nobody has evaluated for tool use yet is
|
|
# unproven, not proven bad.
|
|
unknown = _row(model_id="unevaluated", tool_proficiency=None)
|
|
assert _eligible(unknown, min_tool_proficiency=0.5) is True
|
|
|
|
|
|
def test_the_threshold_is_exclusive_at_the_boundary():
|
|
assert _eligible(_row(tool_proficiency=0.5), min_tool_proficiency=0.5) is True
|
|
assert _eligible(_row(tool_proficiency=0.49), min_tool_proficiency=0.5) is False
|
|
|
|
|
|
def test_the_tool_filter_can_change_the_winner():
|
|
# The whole point: it is a hard filter, so it removes a candidate that
|
|
# would otherwise win on cost rather than merely penalizing it.
|
|
rows = [
|
|
_row(model_id="cheap-clumsy", proficiency=1.0, tool_proficiency=0.33,
|
|
cost_per_1m_prompt=0.1, cost_per_1m_completion=0.2),
|
|
_row(model_id="dearer-reliable", proficiency=1.0, tool_proficiency=1.0,
|
|
cost_per_1m_prompt=1.0, cost_per_1m_completion=2.0),
|
|
]
|
|
no_tools = select_candidates(
|
|
rows, required_context_tokens=1000, required_tier=1,
|
|
latency_tolerance="interactive", allowed_access_levels=["public"],
|
|
exclude_stale=True, exclude_deprecated=True, min_tool_proficiency=None)
|
|
with_tools = select_candidates(
|
|
rows, required_context_tokens=1000, required_tier=1,
|
|
latency_tolerance="interactive", allowed_access_levels=["public"],
|
|
exclude_stale=True, exclude_deprecated=True, min_tool_proficiency=0.5)
|
|
|
|
assert rank_candidates(no_tools, prompt_tokens=1000)[0]["model_id"] == "cheap-clumsy"
|
|
assert rank_candidates(with_tools, prompt_tokens=1000)[0]["model_id"] == "dearer-reliable"
|
|
|
|
|
|
# --- capability filters ----------------------------------------------------
|
|
|
|
def test_vision_required_excludes_a_model_without_vision():
|
|
blind = _row(model_id="blind", supports_vision=0)
|
|
sighted = _row(model_id="sighted", supports_vision=1)
|
|
assert _eligible(blind, require_vision=True) is False
|
|
assert _eligible(sighted, require_vision=True) is True
|
|
|
|
|
|
def test_vision_not_required_admits_everyone():
|
|
# The gate is off by default; not asking for vision must not exclude
|
|
blind = _row(model_id="blind", supports_vision=0)
|
|
assert _eligible(blind, require_vision=False) is True
|
|
|
|
|
|
def test_unknown_vision_capability_fails_closed():
|
|
# A capability FLAG that is absent means "cannot confirm". Unlike a
|
|
# proficiency MEASUREMENT, admit-on-None would route a vision request to a
|
|
# model that might lack it -- a guaranteed provider 400.
|
|
unknown = _row(model_id="unknown", supports_vision=None)
|
|
assert _eligible(unknown, require_vision=True) is False
|
|
|
|
|
|
def test_json_mode_required_excludes_without_support():
|
|
plain = _row(model_id="plain", supports_json_mode=0)
|
|
json = _row(model_id="json", supports_json_mode=1)
|
|
assert _eligible(plain, require_json_mode=True) is False
|
|
assert _eligible(json, require_json_mode=True) is True
|
|
|
|
|
|
def test_unknown_json_mode_fails_closed():
|
|
unknown = _row(model_id="unknown", supports_json_mode=None)
|
|
assert _eligible(unknown, require_json_mode=True) is False
|
|
|
|
|
|
def test_json_mode_gate_admits_supporting_model():
|
|
assert _eligible(_row(supports_json_mode=1), require_json_mode=True) is True
|
|
|
|
|
|
def test_the_vision_filter_can_change_the_winner():
|
|
# Same shape as the tool-filter winner test: the gate is a hard filter, so
|
|
# it removes a model that would otherwise win on cost rather than merely
|
|
# penalizing it.
|
|
rows = [
|
|
_row(model_id="cheap-blind", proficiency=1.0, supports_vision=0,
|
|
cost_per_1m_prompt=0.1, cost_per_1m_completion=0.2),
|
|
_row(model_id="dearer-sighted", proficiency=1.0, supports_vision=1,
|
|
cost_per_1m_prompt=1.0, cost_per_1m_completion=2.0),
|
|
]
|
|
no_images = select_candidates(
|
|
rows, required_context_tokens=1000, required_tier=1,
|
|
latency_tolerance="interactive", allowed_access_levels=["public"],
|
|
exclude_stale=True, exclude_deprecated=True, require_vision=False)
|
|
with_images = select_candidates(
|
|
rows, required_context_tokens=1000, required_tier=1,
|
|
latency_tolerance="interactive", allowed_access_levels=["public"],
|
|
exclude_stale=True, exclude_deprecated=True, require_vision=True)
|
|
|
|
assert rank_candidates(no_images, prompt_tokens=1000)[0]["model_id"] == "cheap-blind"
|
|
assert rank_candidates(with_images, prompt_tokens=1000)[0]["model_id"] == "dearer-sighted"
|
|
|
|
|
|
def test_rejection_reason_names_the_missing_capability():
|
|
assert _reason(_row(supports_vision=0), require_vision=True) == "vision(unsupported)"
|
|
assert _reason(_row(supports_vision=None), require_vision=True) == "vision(unknown)"
|
|
assert _reason(_row(supports_json_mode=0), require_json_mode=True) == "json_mode(unsupported)"
|
|
assert _reason(_row(supports_json_mode=None), require_json_mode=True) == "json_mode(unknown)"
|
|
|
|
|
|
def test_zero_tolerance_ranks_strictly_on_quality():
|
|
""""Never trade quality for cost" is a legitimate setting.
|
|
|
|
The validator has always accepted 0; `band()` divided by it, so every
|
|
request raised ZeroDivisionError on a config that had loaded cleanly.
|
|
"""
|
|
rows = [_row(model_id="cheap", cost=1e-9, proficiency=0.98),
|
|
_row(model_id="dear", cost=1e-3, proficiency=1.00)]
|
|
|
|
assert rank_candidates(rows, quality_tolerance=0)[0]["model_id"] == "dear"
|
|
|
|
|
|
def test_zero_tolerance_still_breaks_exact_ties_on_cost():
|
|
rows = [_row(model_id="dear", cost=1e-3, proficiency=1.00),
|
|
_row(model_id="cheap", cost=1e-9, proficiency=1.00)]
|
|
|
|
assert rank_candidates(rows, quality_tolerance=0)[0]["model_id"] == "cheap"
|
|
|
|
|
|
# --- rejection reasons ----------------------------------------------------
|
|
#
|
|
# is_eligible returns a bool, so a dropped model used to vanish without
|
|
# explanation and "no model satisfies the hard filters" was a dead end. The
|
|
# reason string is what the debug log prints, so it has to be exact rather than
|
|
# re-derived somewhere else.
|
|
|
|
def _reason(row, **overrides):
|
|
kwargs = {
|
|
"required_context_tokens": 10_000,
|
|
"required_tier": 2,
|
|
"latency_tolerance": "interactive",
|
|
"allowed_access_levels": ["public"],
|
|
"exclude_stale": True,
|
|
"exclude_deprecated": True,
|
|
}
|
|
kwargs.update(overrides)
|
|
return rejection_reason(row, **kwargs)
|
|
|
|
|
|
def test_an_eligible_row_has_no_reason():
|
|
assert _reason(_row()) is None
|
|
|
|
|
|
def test_the_reason_names_the_filter_and_its_numbers():
|
|
assert _reason(_row(effective_context_window=5_000)) == "context(5000<10000)"
|
|
assert _reason(_row(tier=1)) == "tier(1<2)"
|
|
assert _reason(_row(availability="stale")) == "stale"
|
|
assert _reason(_row(availability="deprecated")) == "deprecated"
|
|
assert _reason(_row(access_level="canary")) == "access_level(canary)"
|
|
assert _reason(_row(latency_class="flex")) == "latency_class(flex)"
|
|
|
|
|
|
def test_unknown_values_say_unknown_rather_than_comparing():
|
|
assert _reason(_row(effective_context_window=None)) == "context(unknown)"
|
|
assert _reason(_row(tier=None)) == "tier(unknown)"
|
|
|
|
|
|
def test_the_tool_filter_reports_the_measured_score():
|
|
reason = _reason(_row(tool_proficiency=0.33), min_tool_proficiency=0.5)
|
|
|
|
assert reason == "tool_proficiency(0.33<0.5)"
|
|
|
|
|
|
def test_reasons_are_single_tokens():
|
|
"""They go straight into a logfmt value; a space would force quoting."""
|
|
for row in (_row(tier=1), _row(availability="stale"),
|
|
_row(access_level="canary"), _row(latency_class="flex"),
|
|
_row(supports_vision=0), _row(supports_vision=None),
|
|
_row(supports_json_mode=0), _row(supports_json_mode=None)):
|
|
reason = _reason(row, require_vision=True, require_json_mode=True)
|
|
assert reason is not None
|
|
assert " " not in reason
|
|
|
|
|
|
# --- credit attenuation multiplier ----------------------------------------
|
|
|
|
|
|
def test_credit_attenuation_unknown_balance_is_healthy():
|
|
assert credit_attenuation_multiplier(
|
|
None, soft_floor_usd=5.0, zero_floor_usd=0.0, max_multiplier=5.0
|
|
) == 1.0
|
|
|
|
|
|
def test_credit_attenuation_above_soft_floor_is_healthy():
|
|
assert credit_attenuation_multiplier(
|
|
10.0, soft_floor_usd=5.0, zero_floor_usd=0.0, max_multiplier=5.0
|
|
) == 1.0
|
|
|
|
|
|
def test_credit_attenuation_at_soft_floor_is_healthy():
|
|
assert credit_attenuation_multiplier(
|
|
5.0, soft_floor_usd=5.0, zero_floor_usd=0.0, max_multiplier=5.0
|
|
) == 1.0
|
|
|
|
|
|
def test_credit_attenuation_at_zero_floor_is_max():
|
|
assert credit_attenuation_multiplier(
|
|
0.0, soft_floor_usd=5.0, zero_floor_usd=0.0, max_multiplier=5.0
|
|
) == 5.0
|
|
|
|
|
|
def test_credit_attenuation_below_zero_floor_is_max():
|
|
assert credit_attenuation_multiplier(
|
|
-1.0, soft_floor_usd=5.0, zero_floor_usd=0.0, max_multiplier=5.0
|
|
) == 5.0
|
|
|
|
|
|
def test_credit_attenuation_midpoint_linear():
|
|
# balance 2.5 is halfway between soft=5 and zero=0 with max=5.
|
|
# frac = (5 - 2.5) / (5 - 0) = 0.5
|
|
# result = 1.0 + 0.5 * (5 - 1.0) = 3.0
|
|
assert credit_attenuation_multiplier(
|
|
2.5, soft_floor_usd=5.0, zero_floor_usd=0.0, max_multiplier=5.0
|
|
) == pytest.approx(3.0)
|
|
|
|
|
|
def test_credit_attenuation_multiplier_range_independence():
|
|
# Same fractional position with max_multiplier=2.0 should yield 1.5.
|
|
assert credit_attenuation_multiplier(
|
|
2.5, soft_floor_usd=5.0, zero_floor_usd=0.0, max_multiplier=2.0
|
|
) == pytest.approx(1.5)
|
|
|
|
|
|
def test_credit_attenuation_equal_floors_is_safe():
|
|
# With soft == zero, the guard order (>= soft first, <= zero second)
|
|
# means the interpolation branch is unreachable regardless of balance.
|
|
# This construction-safety test hits the <= zero branch.
|
|
assert credit_attenuation_multiplier(
|
|
0.0, soft_floor_usd=5.0, zero_floor_usd=5.0, max_multiplier=5.0
|
|
) == 5.0
|
|
|
|
|
|
# --- capability_gate_reason (the extracted flag rule) -----------------------
|
|
#
|
|
# rejection_reason delegates its vision/json-mode arm to this function, and
|
|
# dispatcher._check_pinned_capabilities reuses it so a pinned-model check and
|
|
# routed traffic cannot drift apart. It is unit-tested directly because the
|
|
# pinned-model path has no other cheap way to exercise these branches without
|
|
# a full request round-trip.
|
|
|
|
def test_capability_gate_passes_when_not_required():
|
|
# With no capability required, a model that lacks both still passes.
|
|
assert capability_gate_reason(_row(supports_vision=0, supports_json_mode=0)) is None
|
|
|
|
|
|
def test_capability_gate_rejects_missing_vision_flag():
|
|
assert capability_gate_reason(
|
|
_row(supports_vision=None), require_vision=True
|
|
) == "vision(unknown)"
|
|
|
|
|
|
def test_capability_gate_rejects_vision_unsupported():
|
|
assert capability_gate_reason(
|
|
_row(supports_vision=0), require_vision=True
|
|
) == "vision(unsupported)"
|
|
|
|
|
|
def test_capability_gate_rejects_missing_json_mode_flag():
|
|
assert capability_gate_reason(
|
|
_row(supports_json_mode=None), require_json_mode=True
|
|
) == "json_mode(unknown)"
|
|
|
|
|
|
def test_capability_gate_rejects_json_mode_unsupported():
|
|
assert capability_gate_reason(
|
|
_row(supports_json_mode=0), require_json_mode=True
|
|
) == "json_mode(unsupported)"
|
|
|
|
|
|
def test_capability_gate_admits_a_row_that_supports_both():
|
|
assert capability_gate_reason(
|
|
_row(supports_vision=1, supports_json_mode=1),
|
|
require_vision=True, require_json_mode=True,
|
|
) is None
|
|
|
|
|
|
def test_capability_gate_unknown_reason_on_empty_row():
|
|
assert capability_gate_reason({}, require_vision=True) == "vision(unknown)"
|
|
assert capability_gate_reason({}, require_json_mode=True) == "json_mode(unknown)"
|
|
|
|
|
|
def test_is_eligible_still_agrees_with_the_reason():
|
|
"""One copy of the rules, two views of it."""
|
|
for row in (_row(), _row(tier=1), _row(latency_class="flex")):
|
|
assert is_eligible(row, **{
|
|
"required_context_tokens": 10_000, "required_tier": 2,
|
|
"latency_tolerance": "interactive", "allowed_access_levels": ["public"],
|
|
"exclude_stale": True, "exclude_deprecated": True,
|
|
}) == (_reason(row) is None)
|
|
|
|
|
|
# --- get_flex_sibling -------------------------------------------------------
|
|
#
|
|
# The flex-preference knob can route a request to a `-flex` row (deferred
|
|
# during peak). For an operator who prefers flex latency but not a hard
|
|
# failure, the standard winner has a flex sibling that carries identical
|
|
# weights/serving class except latency_class. This helper finds it; it is
|
|
# deliberately a pure function of the rows, so the dispatcher owns only the
|
|
# "which preference applies" decision and not the catalog lookup.
|
|
|
|
def test_flex_sibling_when_standard_row_has_one():
|
|
# Given: the standard kimi-k3 winner and its flex sibling, differing only
|
|
# by latency_class
|
|
selected = _row(model_id="kimi-k3", base_model_id="kimi-k3",
|
|
reasoning_mode="default", context_variant="full",
|
|
latency_class="standard")
|
|
flex = _row(model_id="kimi-k3-flex", base_model_id="kimi-k3",
|
|
reasoning_mode="default", context_variant="full",
|
|
latency_class="flex")
|
|
sibling = get_flex_sibling(selected, [selected, flex])
|
|
assert sibling is not None
|
|
assert sibling["model_id"] == "kimi-k3-flex"
|
|
assert sibling["latency_class"] == "flex"
|
|
|
|
|
|
def test_flex_sibling_must_match_all_three_dims():
|
|
# A flex row that shares base_model_id but not every serving dimension is
|
|
# NOT a sibling: -fast/-short change weights or context budget.
|
|
selected = _row(model_id="kimi-k3", base_model_id="kimi-k3",
|
|
reasoning_mode="default", context_variant="full",
|
|
latency_class="standard")
|
|
wrong_reasoning = _row(model_id="kimi-k3-fast-flex", base_model_id="kimi-k3",
|
|
reasoning_mode="reduced", context_variant="full",
|
|
latency_class="flex")
|
|
wrong_ctx = _row(model_id="kimi-k3-short-flex", base_model_id="kimi-k3",
|
|
reasoning_mode="default", context_variant="short",
|
|
latency_class="flex")
|
|
# A matching flex must still be found among non-matching rows.
|
|
right = _row(model_id="kimi-k3-flex", base_model_id="kimi-k3",
|
|
reasoning_mode="default", context_variant="full",
|
|
latency_class="flex")
|
|
assert get_flex_sibling(selected, [wrong_reasoning, wrong_ctx, right])["model_id"] == "kimi-k3-flex"
|
|
|
|
|
|
def test_flex_sibling_is_none_when_selected_is_already_flex():
|
|
selected = _row(model_id="kimi-k3-flex", base_model_id="kimi-k3",
|
|
reasoning_mode="default", context_variant="full",
|
|
latency_class="flex")
|
|
assert get_flex_sibling(selected, [selected]) is None
|
|
|
|
|
|
def test_flex_sibling_is_none_when_no_matching_row_exists():
|
|
# The flex row present belongs to a different base model entirely
|
|
selected = _row(model_id="kimi-k3", base_model_id="kimi-k3",
|
|
reasoning_mode="default", context_variant="full",
|
|
latency_class="standard")
|
|
other = _row(model_id="glm-5.2-flex", base_model_id="glm-5.2",
|
|
reasoning_mode="default", context_variant="full",
|
|
latency_class="flex")
|
|
assert get_flex_sibling(selected, [selected, other]) is None
|
|
|
|
|
|
def test_flex_sibling_ignores_stale_or_unroutable_flex_rows():
|
|
# The helper is purely structural: it matches by identity dims regardless
|
|
# of freshness/access. Routing-level gating is the caller's job.
|
|
selected = _row(model_id="kimi-k3", base_model_id="kimi-k3",
|
|
reasoning_mode="default", context_variant="full",
|
|
latency_class="standard")
|
|
flex = _row(model_id="kimi-k3-flex", base_model_id="kimi-k3",
|
|
reasoning_mode="default", context_variant="full",
|
|
latency_class="flex", availability="stale")
|
|
sibling = get_flex_sibling(selected, [selected, flex])
|
|
assert sibling is not None
|
|
assert sibling["model_id"] == "kimi-k3-flex"
|
|
|
|
|
|
# --- apply_flex_preference: the post-rank serving-class switch --------------
|
|
#
|
|
# rank_candidates already decides quality and cost; this runs AFTER it to
|
|
# override only the serving class when the operator's flex-preference stance
|
|
# asks for it. The mode matrix below pins the behavior of every preference
|
|
# under both latency tolerances, plus the no-sibling fallback.
|
|
|
|
def _std_flex_pair(**overrides):
|
|
"""A ranked standard winner plus its weight-equivalent flex twin.
|
|
|
|
The flex twin shares base_model_id/reasoning_mode/context_variant and
|
|
differs only by latency_class. ``base_model_id`` defaults to the model_id
|
|
(matching the get_flex_sibling tests which must set it explicitly).
|
|
"""
|
|
std = _row(model_id="kimi-k3", base_model_id="kimi-k3",
|
|
reasoning_mode="default", context_variant="full",
|
|
latency_class="standard", proficiency=0.5,
|
|
cost_per_1m_prompt=1.0, cost_per_1m_completion=2.0, **overrides)
|
|
flex = _row(model_id="kimi-k3-flex", base_model_id="kimi-k3",
|
|
reasoning_mode="default", context_variant="full",
|
|
latency_class="flex", proficiency=0.5,
|
|
cost_per_1m_prompt=1.0, cost_per_1m_completion=2.0, **overrides)
|
|
return std, flex
|
|
|
|
|
|
def test_apply_no_flex_never_swaps_under_interactive():
|
|
# no-flex: never route to a flex row, even when one exists and would help.
|
|
std, flex = _std_flex_pair()
|
|
row, swapped, forced, cost = apply_flex_preference(
|
|
std, [std, flex], FlexPreference.no_flex, "interactive",
|
|
prompt_tokens=1000, completion_tokens=500, cache_rate=0.5,
|
|
required_context_tokens=1000, required_tier=2,
|
|
allowed_access_levels=["public"], exclude_stale=True,
|
|
exclude_deprecated=True)
|
|
assert row["model_id"] == "kimi-k3"
|
|
assert swapped is False
|
|
assert forced is False
|
|
assert cost == std["cost"]
|
|
|
|
|
|
def test_apply_no_flex_leaves_an_already_flex_winner_alone():
|
|
# If the rank winner is already flex (batch admitted it), no-flex does not
|
|
# bounce it back to standard; it leaves the decision as ranked.
|
|
flex = _row(model_id="kimi-k3-flex", base_model_id="kimi-k3",
|
|
reasoning_mode="default", context_variant="full",
|
|
latency_class="flex", proficiency=0.5)
|
|
row, swapped, forced, cost = apply_flex_preference(
|
|
flex, [flex], FlexPreference.no_flex, "interactive",
|
|
prompt_tokens=1000, completion_tokens=500, cache_rate=0.5,
|
|
required_context_tokens=1000, required_tier=2,
|
|
allowed_access_levels=["public"], exclude_stale=True,
|
|
exclude_deprecated=True)
|
|
assert row["model_id"] == "kimi-k3-flex"
|
|
assert swapped is False
|
|
assert forced is False
|
|
assert cost == flex["cost"]
|
|
|
|
|
|
def test_apply_auto_makes_no_swap_even_when_a_flex_sibling_exists():
|
|
# auto defers to the latency filter that already ran; a flex sibling
|
|
# existing does not change the ranked winner.
|
|
std, flex = _std_flex_pair()
|
|
row, swapped, forced, cost = apply_flex_preference(
|
|
std, [std, flex], FlexPreference.auto, "batch",
|
|
prompt_tokens=1000, completion_tokens=500, cache_rate=0.5,
|
|
required_context_tokens=1000, required_tier=2,
|
|
allowed_access_levels=["public"], exclude_stale=True,
|
|
exclude_deprecated=True)
|
|
assert row["model_id"] == "kimi-k3"
|
|
assert swapped is False
|
|
assert forced is False
|
|
assert cost == std["cost"]
|
|
|
|
|
|
def test_apply_prefer_flex_holds_back_under_interactive():
|
|
# prefer-flex swaps only when the flex sibling is eligible under the
|
|
# current hard filters; interactive's latency filter dismisses it.
|
|
std, flex = _std_flex_pair()
|
|
row, swapped, forced, cost = apply_flex_preference(
|
|
std, [std, flex], FlexPreference.prefer_flex, "interactive",
|
|
prompt_tokens=1000, completion_tokens=500, cache_rate=0.5,
|
|
required_context_tokens=1000, required_tier=2,
|
|
allowed_access_levels=["public"], exclude_stale=True,
|
|
exclude_deprecated=True)
|
|
assert row["model_id"] == "kimi-k3"
|
|
assert swapped is False
|
|
assert forced is False
|
|
assert cost == std["cost"]
|
|
|
|
|
|
def test_apply_prefer_flex_swaps_under_batch():
|
|
# Under batch the latency filter admits flex, so the sibling is eligible
|
|
# and the swap happens without bypassing anything.
|
|
std, flex = _std_flex_pair()
|
|
row, swapped, forced, cost = apply_flex_preference(
|
|
std, [std, flex], FlexPreference.prefer_flex, "batch",
|
|
prompt_tokens=1000, completion_tokens=500, cache_rate=0.5,
|
|
required_context_tokens=1000, required_tier=2,
|
|
allowed_access_levels=["public"], exclude_stale=True,
|
|
exclude_deprecated=True)
|
|
assert row["model_id"] == "kimi-k3-flex"
|
|
assert swapped is True
|
|
assert forced is False
|
|
assert cost is not None
|
|
|
|
|
|
def test_apply_force_flex_swaps_under_interactive_and_marks_forced():
|
|
# force-flex bypasses the interactive latency filter: swap happens and the
|
|
# bypass is recorded as flex_forced=True.
|
|
std, flex = _std_flex_pair()
|
|
row, swapped, forced, cost = apply_flex_preference(
|
|
std, [std, flex], FlexPreference.force_flex, "interactive",
|
|
prompt_tokens=1000, completion_tokens=500, cache_rate=0.5,
|
|
required_context_tokens=1000, required_tier=2,
|
|
allowed_access_levels=["public"], exclude_stale=True,
|
|
exclude_deprecated=True)
|
|
assert row["model_id"] == "kimi-k3-flex"
|
|
assert swapped is True
|
|
assert forced is True
|
|
assert cost is not None
|
|
|
|
|
|
def test_apply_force_flex_swaps_under_batch_without_forced_flag():
|
|
# Under batch there is no filter to bypass, so force-flex swaps but the
|
|
# forced flag stays False.
|
|
std, flex = _std_flex_pair()
|
|
row, swapped, forced, cost = apply_flex_preference(
|
|
std, [std, flex], FlexPreference.force_flex, "batch",
|
|
prompt_tokens=1000, completion_tokens=500, cache_rate=0.5,
|
|
required_context_tokens=1000, required_tier=2,
|
|
allowed_access_levels=["public"], exclude_stale=True,
|
|
exclude_deprecated=True)
|
|
assert row["model_id"] == "kimi-k3-flex"
|
|
assert swapped is True
|
|
assert forced is False
|
|
assert cost is not None
|
|
|
|
|
|
def test_apply_falls_back_when_no_flex_sibling_exists():
|
|
# No flex twin in the catalog: every preference falls back to the ranked
|
|
# winner and its cost.
|
|
std = _row(model_id="kimi-k3", base_model_id="kimi-k3",
|
|
reasoning_mode="default", context_variant="full",
|
|
latency_class="standard", proficiency=0.5)
|
|
for pref in (FlexPreference.no_flex, FlexPreference.auto,
|
|
FlexPreference.prefer_flex, FlexPreference.force_flex):
|
|
row, swapped, forced, cost = apply_flex_preference(
|
|
std, [std], pref, "batch",
|
|
prompt_tokens=1000, completion_tokens=500, cache_rate=0.5,
|
|
required_context_tokens=1000, required_tier=2,
|
|
allowed_access_levels=["public"], exclude_stale=True,
|
|
exclude_deprecated=True)
|
|
assert row["model_id"] == "kimi-k3"
|
|
assert swapped is False
|
|
assert forced is False
|
|
assert cost == std["cost"]
|
|
|
|
|
|
def test_apply_reestimates_cost_for_the_swapped_flex_row():
|
|
# The flex twin carries the same list price as its standard sibling, so a
|
|
# swap recomputes the estimate from the flex row's own pricing fields. Give
|
|
# the flex row a distinct price and confirm the returned cost tracks it.
|
|
std = _row(model_id="kimi-k3", base_model_id="kimi-k3",
|
|
reasoning_mode="default", context_variant="full",
|
|
latency_class="standard", proficiency=0.5,
|
|
cost_per_1m_prompt=1.0, cost_per_1m_completion=2.0,
|
|
cost_per_1m_prompt_cached=0.5)
|
|
flex = _row(model_id="kimi-k3-flex", base_model_id="kimi-k3",
|
|
reasoning_mode="default", context_variant="full",
|
|
latency_class="flex", proficiency=0.5,
|
|
cost_per_1m_prompt=2.0, cost_per_1m_completion=4.0,
|
|
cost_per_1m_prompt_cached=1.0)
|
|
row, swapped, _, _ = apply_flex_preference(
|
|
std, [std, flex], FlexPreference.prefer_flex, "batch",
|
|
prompt_tokens=10_000, completion_tokens=500, cache_rate=0.5,
|
|
required_context_tokens=1000, required_tier=2,
|
|
allowed_access_levels=["public"], exclude_stale=True,
|
|
exclude_deprecated=True)
|
|
assert swapped is True
|
|
# estimated_cost scales prompt/completion prices to this request shape, so
|
|
# the flex row's higher price must drive a higher estimate than standard's.
|
|
std_cost = rank_candidates(
|
|
[std], prompt_tokens=10_000, completion_tokens=500, cache_rate=0.5
|
|
)[0]["cost"]
|
|
flex_cost = rank_candidates(
|
|
[flex], prompt_tokens=10_000, completion_tokens=500, cache_rate=0.5
|
|
)[0]["cost"]
|
|
assert row["cost"] == flex_cost
|
|
assert row["cost"] > std_cost
|
|
|
|
|
|
def test_apply_prefer_flex_refuses_stale_sibling_under_batch():
|
|
# A stale flex sibling fails the freshness filter; prefer-flex under batch
|
|
# must keep the ranked standard winner rather than dispatch to staleness.
|
|
std, flex = _std_flex_pair(availability="stale")
|
|
row, swapped, forced, cost = apply_flex_preference(
|
|
std, [std, flex], FlexPreference.prefer_flex, "batch",
|
|
prompt_tokens=1000, completion_tokens=500, cache_rate=0.5,
|
|
required_context_tokens=1000, required_tier=2,
|
|
allowed_access_levels=["public"], exclude_stale=True,
|
|
exclude_deprecated=True)
|
|
assert row["model_id"] == "kimi-k3"
|
|
assert swapped is False
|
|
assert forced is False
|
|
assert cost == std["cost"]
|
|
|
|
|
|
def test_apply_force_flex_refuses_stale_sibling_under_interactive():
|
|
# force-flex bypasses the latency filter, but it still must not dispatch to
|
|
# a stale sibling; the standard winner stands.
|
|
std, flex = _std_flex_pair(availability="stale")
|
|
row, swapped, forced, cost = apply_flex_preference(
|
|
std, [std, flex], FlexPreference.force_flex, "interactive",
|
|
prompt_tokens=1000, completion_tokens=500, cache_rate=0.5,
|
|
required_context_tokens=1000, required_tier=2,
|
|
allowed_access_levels=["public"], exclude_stale=True,
|
|
exclude_deprecated=True)
|
|
assert row["model_id"] == "kimi-k3"
|
|
assert swapped is False
|
|
assert forced is False
|
|
assert cost == std["cost"]
|
|
|
|
|
|
def test_apply_prefer_flex_refuses_canary_sibling_under_batch():
|
|
# An access-restricted flex sibling must not be swapped to when the caller
|
|
# is restricted to public rows, even under batch where latency would admit it.
|
|
std, flex = _std_flex_pair(access_level="canary")
|
|
row, swapped, forced, cost = apply_flex_preference(
|
|
std, [std, flex], FlexPreference.prefer_flex, "batch",
|
|
prompt_tokens=1000, completion_tokens=500, cache_rate=0.5,
|
|
required_context_tokens=1000, required_tier=2,
|
|
allowed_access_levels=["public"], exclude_stale=True,
|
|
exclude_deprecated=True)
|
|
assert row["model_id"] == "kimi-k3"
|
|
assert swapped is False
|
|
assert forced is False
|
|
assert cost == std["cost"]
|
|
|
|
|
|
def test_apply_force_flex_refuses_canary_sibling_under_interactive():
|
|
# force-flex bypasses only the latency filter — access restriction still
|
|
# gates the sibling, so the standard winner stands.
|
|
std, flex = _std_flex_pair(access_level="canary")
|
|
row, swapped, forced, cost = apply_flex_preference(
|
|
std, [std, flex], FlexPreference.force_flex, "interactive",
|
|
prompt_tokens=1000, completion_tokens=500, cache_rate=0.5,
|
|
required_context_tokens=1000, required_tier=2,
|
|
allowed_access_levels=["public"], exclude_stale=True,
|
|
exclude_deprecated=True)
|
|
assert row["model_id"] == "kimi-k3"
|
|
assert swapped is False
|
|
assert forced is False
|
|
assert cost == std["cost"]
|
|
|
|
|
|
def test_apply_force_flex_prices_swapped_row_at_challenger_cache_rate():
|
|
# When challenger_cache_rate is set, the flex twin is priced at that rate
|
|
# rather than the caller's cache_rate — the twin is billed as a cold
|
|
# challenger because the session's incumbent cache does not carry across
|
|
# the swap, keeping parity with the ranking dial at every setting.
|
|
std = _row(model_id="kimi-k3", base_model_id="kimi-k3",
|
|
reasoning_mode="default", context_variant="full",
|
|
latency_class="standard", proficiency=0.5,
|
|
cost_per_1m_prompt=1.0, cost_per_1m_completion=2.0,
|
|
cost_per_1m_prompt_cached=0.5)
|
|
flex = _row(model_id="kimi-k3-flex", base_model_id="kimi-k3",
|
|
reasoning_mode="default", context_variant="full",
|
|
latency_class="flex", proficiency=0.5,
|
|
cost_per_1m_prompt=1.0, cost_per_1m_completion=2.0,
|
|
cost_per_1m_prompt_cached=0.5)
|
|
cache_rate = 0.8
|
|
challenger_rate = 0.1
|
|
row, swapped, forced, cost = apply_flex_preference(
|
|
std, [std, flex], FlexPreference.force_flex, "interactive",
|
|
prompt_tokens=10_000, completion_tokens=500, cache_rate=cache_rate,
|
|
required_context_tokens=1000, required_tier=2,
|
|
allowed_access_levels=["public"], exclude_stale=True,
|
|
exclude_deprecated=True,
|
|
challenger_cache_rate=challenger_rate)
|
|
assert row["model_id"] == "kimi-k3-flex"
|
|
assert swapped is True
|
|
assert forced is True
|
|
# The cost must track the challenger rate, not the standard cached rate.
|
|
assert cost == estimated_cost(flex, 10_000, 500, challenger_rate)
|
|
assert cost != estimated_cost(flex, 10_000, 500, cache_rate)
|
|
|
|
|
|
def test_apply_challenger_cache_rate_none_falls_back_to_cache_rate():
|
|
# Without challenger_cache_rate (the common caller case), swap behavior is
|
|
# unchanged — the flex twin prices at the same cache_rate as before.
|
|
std, flex = _std_flex_pair()
|
|
cache_rate = 0.5
|
|
row, swapped, forced, cost = apply_flex_preference(
|
|
std, [std, flex], FlexPreference.force_flex, "interactive",
|
|
prompt_tokens=10_000, completion_tokens=500, cache_rate=cache_rate,
|
|
required_context_tokens=1000, required_tier=2,
|
|
allowed_access_levels=["public"], exclude_stale=True,
|
|
exclude_deprecated=True)
|
|
assert swapped is True
|
|
# Default None → use cache_rate, so the cost must match what we get when
|
|
# we explicitly pass cache_rate.
|
|
assert cost == estimated_cost(flex, 10_000, 500, cache_rate)
|
|
|
|
|
|
# --- eligible_categories hard filter ----------------------------------------
|
|
|
|
def test_category_restricted_row_rejected_when_task_outside_list():
|
|
# A row with eligible_categories=["coding"] must be ineligible for a
|
|
# summarization task.
|
|
row = _row(eligible_categories=["coding", "refactoring"])
|
|
assert _eligible(row, task_category="summarization") is False
|
|
|
|
|
|
def test_category_restricted_row_eligible_when_task_in_list():
|
|
# The same row is eligible when the task category is in the list.
|
|
row = _row(eligible_categories=["coding", "refactoring"])
|
|
assert _eligible(row, task_category="refactoring") is True
|
|
|
|
|
|
def test_null_eligible_categories_unaffected_by_task():
|
|
# NULL eligible_categories NEVER rejects — not even when task_category is
|
|
# provided, and not when it is None.
|
|
row = _row() # missing eligible_categories, behaves as None
|
|
assert _eligible(row, task_category="coding") is True
|
|
assert _eligible(row, task_category=None) is True
|
|
|
|
|
|
def test_raw_comma_string_is_parsed_not_substring_matched():
|
|
"""SQLite stores this column as a comma-joined string, and `in` on a
|
|
string is a SUBSTRING test — so an unparsed row admitted `summarization`
|
|
to a model eligible only for `file_summarization`. dispatcher parsed the
|
|
column; admin.py's profile probe passed rows straight out of SELECT *.
|
|
"""
|
|
row = _row(eligible_categories="file_summarization,diff_checking")
|
|
assert _eligible(row, task_category="summarization") is False
|
|
assert _eligible(row, task_category="file_summarization") is True
|
|
assert _eligible(row, task_category="diff_checking") is True
|
|
|
|
|
|
def test_parse_eligible_categories_is_idempotent():
|
|
"""Called on rows that may or may not have been normalized already."""
|
|
from routing import parse_eligible_categories
|
|
|
|
assert parse_eligible_categories(None) is None
|
|
once = parse_eligible_categories(" a , b ,, ")
|
|
assert once == ["a", "b"]
|
|
assert parse_eligible_categories(once) == ["a", "b"]
|
|
|
|
|
|
def test_category_ineligible_reason_is_single_token():
|
|
# Reason strings go into logfmt values — no spaces allowed.
|
|
row = _row(eligible_categories=["coding"])
|
|
reason = _reason(row, task_category="summarization")
|
|
assert reason == "category_ineligible"
|
|
assert " " not in reason
|
|
|
|
|
|
# --- restrict_to profile allowlist ----------------------------------------
|
|
|
|
|
|
def test_restrict_to_none_is_unrestricted():
|
|
row = _row(model_id="m1")
|
|
assert _eligible(row, restrict_to=None) is True
|
|
|
|
|
|
def test_restrict_to_allows_model_in_set():
|
|
row = _row(model_id="m1")
|
|
assert _eligible(row, restrict_to={"m1"}) is True
|
|
|
|
|
|
def test_restrict_to_excludes_model_not_in_set():
|
|
row = _row(model_id="m2")
|
|
assert _eligible(row, restrict_to={"m1"}) is False
|
|
|
|
|
|
def test_restrict_to_empty_set_excludes_every_model():
|
|
row = _row(model_id="m1")
|
|
assert _eligible(row, restrict_to=set()) is False
|
|
|
|
|
|
def test_restrict_to_reason_is_profile_excluded():
|
|
row = _row(model_id="m2")
|
|
reason = _reason(row, restrict_to={"m1"})
|
|
assert reason == "profile_excluded"
|
|
|
|
|
|
def test_restrict_to_is_single_token_reason():
|
|
row = _row(model_id="m2")
|
|
reason = _reason(row, restrict_to={"m1"})
|
|
assert " " not in reason
|
|
|
|
|
|
def test_select_candidates_applies_restrict_to_allowlist():
|
|
rows = [_row(model_id="m1"), _row(model_id="m2"), _row(model_id="m3")]
|
|
selected = select_candidates(
|
|
rows, required_context_tokens=10_000, required_tier=2,
|
|
latency_tolerance="interactive", allowed_access_levels=["public"],
|
|
exclude_stale=True, exclude_deprecated=True, restrict_to={"m1"})
|
|
assert [r["model_id"] for r in selected] == ["m1"]
|
|
|
|
|
|
def test_select_candidates_empty_restrict_to_yields_no_candidates():
|
|
rows = [_row(model_id="m1"), _row(model_id="m2")]
|
|
selected = select_candidates(
|
|
rows, required_context_tokens=10_000, required_tier=2,
|
|
latency_tolerance="interactive", allowed_access_levels=["public"],
|
|
exclude_stale=True, exclude_deprecated=True, restrict_to=set())
|
|
assert selected == []
|
|
|
|
|
|
def test_restrict_to_and_exclude_models_compose_as_and():
|
|
# An allowlist of {m1, m2} plus a circuit-open exclusion of m1 should
|
|
# leave only m2.
|
|
rows = [_row(model_id="m1"), _row(model_id="m2"), _row(model_id="m3")]
|
|
selected = select_candidates(
|
|
rows, required_context_tokens=10_000, required_tier=2,
|
|
latency_tolerance="interactive", allowed_access_levels=["public"],
|
|
exclude_stale=True, exclude_deprecated=True,
|
|
exclude_models={"m1"}, restrict_to={"m1", "m2"})
|
|
assert [r["model_id"] for r in selected] == ["m2"]
|
|
|
|
|
|
def test_select_candidates_drops_outside_category():
|
|
# Even a perfect row (tier 1, zero cost) is dropped for the wrong category.
|
|
rows = [
|
|
_row(
|
|
model_id="restricted",
|
|
tier=1,
|
|
cost=0.01,
|
|
proficiency=1.0,
|
|
eligible_categories=["coding"],
|
|
),
|
|
_row(
|
|
model_id="open",
|
|
tier=2,
|
|
cost=1.0,
|
|
proficiency=0.5,
|
|
),
|
|
]
|
|
filtered = select_candidates(
|
|
rows, required_context_tokens=1000, required_tier=1,
|
|
latency_tolerance="interactive", allowed_access_levels=["public"],
|
|
exclude_stale=True, exclude_deprecated=True, task_category="summarization")
|
|
ids = {r["model_id"] for r in filtered}
|
|
assert "restricted" not in ids
|
|
assert "open" in ids
|
|
|
|
|
|
def test_through_is_eligible_via_filters():
|
|
# task_category must reach rejection_reason through the **filters passthrough
|
|
# in is_eligible, not only via direct parameter passing.
|
|
row = _row(eligible_categories=["coding"])
|
|
assert is_eligible(
|
|
row,
|
|
required_context_tokens=10_000,
|
|
required_tier=2,
|
|
latency_tolerance="interactive",
|
|
allowed_access_levels=["public"],
|
|
exclude_stale=True,
|
|
exclude_deprecated=True,
|
|
task_category="summarization",
|
|
) is False
|
|
|
|
|
|
# QA cases (a)-(g) for provider_cost_multipliers effective_cost tiebreak.
|
|
|
|
def _ranked_ids(rows, **kwargs):
|
|
return [r["model_id"] for r in rank_candidates(rows, **kwargs)]
|
|
|
|
|
|
def test_provider_cost_multiplier_flips_tiebreak_toward_healthy_provider():
|
|
rows = [
|
|
_row(model_id="A", provider="provA", cost=0.10, proficiency=0.80),
|
|
_row(model_id="B", provider="provB", cost=0.10, proficiency=0.80),
|
|
]
|
|
assert _ranked_ids(rows) == ["A", "B"]
|
|
assert _ranked_ids(
|
|
rows, provider_cost_multipliers={"provA": 5.0}
|
|
) == ["B", "A"]
|
|
|
|
|
|
def test_provider_cost_multiplier_cannot_override_quality_band_gap():
|
|
rows = [
|
|
_row(model_id="A", provider="provA", cost=0.10, proficiency=0.95),
|
|
_row(model_id="B", provider="provB", cost=0.01, proficiency=0.80),
|
|
]
|
|
assert _ranked_ids(
|
|
rows, provider_cost_multipliers={"provA": 5.0}, quality_tolerance=0.1
|
|
) == ["A", "B"]
|
|
|
|
|
|
def test_ranked_rows_carry_raw_cost_and_effective_cost():
|
|
rows = [
|
|
_row(model_id="A", provider="provA", cost=0.10, proficiency=0.80),
|
|
]
|
|
ranked = rank_candidates(rows, provider_cost_multipliers={"provA": 5.0})
|
|
assert len(ranked) == 1
|
|
assert ranked[0]["cost"] == pytest.approx(0.10)
|
|
assert ranked[0]["cost_score"] == pytest.approx(1.0)
|
|
assert ranked[0]["effective_cost"] == pytest.approx(0.50)
|
|
|
|
|
|
def test_provider_multiplier_applies_to_fallback_cost_when_estimate_missing():
|
|
rows = [
|
|
_row(
|
|
model_id="A",
|
|
provider="provA",
|
|
cost=0.20,
|
|
proficiency=0.80,
|
|
prompt_cost_1m=None,
|
|
completion_cost_1m=None,
|
|
),
|
|
]
|
|
ranked = rank_candidates(
|
|
rows,
|
|
prompt_tokens=0,
|
|
completion_tokens=0,
|
|
provider_cost_multipliers={"provA": 3.0},
|
|
)
|
|
assert ranked[0]["cost"] == pytest.approx(0.20)
|
|
assert ranked[0]["effective_cost"] == pytest.approx(0.60)
|
|
|
|
|
|
def test_provider_cost_multiplier_unknown_provider_ignored():
|
|
rows = [
|
|
_row(model_id="A", provider="provA", cost=0.10, proficiency=0.80),
|
|
]
|
|
ranked = rank_candidates(
|
|
rows, provider_cost_multipliers={"provUnknown": 99.0}
|
|
)
|
|
assert ranked[0]["effective_cost"] == pytest.approx(0.10)
|
|
|
|
|
|
def test_provider_cost_multiplier_defaults_to_one_for_unmapped_provider():
|
|
rows = [
|
|
_row(model_id="A", provider="provA", cost=0.10, proficiency=0.80),
|
|
_row(model_id="B", provider="provB", cost=0.10, proficiency=0.80),
|
|
]
|
|
assert _ranked_ids(
|
|
rows, provider_cost_multipliers={"provA": 5.0}
|
|
) == ["B", "A"]
|
|
|
|
|
|
def test_provider_cost_multiplier_empty_dict_matches_no_argument():
|
|
rows = [
|
|
_row(model_id="A", provider="provA", cost=0.10, proficiency=0.80),
|
|
_row(model_id="B", provider="provB", cost=0.10, proficiency=0.80),
|
|
]
|
|
assert _ranked_ids(rows) == _ranked_ids(
|
|
rows, provider_cost_multipliers={}
|
|
)
|
|
|
|
|
|
# --- incumbency pricing: per-candidate cache pricing in rank_candidates -----
|
|
|
|
class TestIncumbentPricing:
|
|
"""8 tests (a)-(h) for per-candidate cache pricing inside rank_candidates.
|
|
|
|
Uses cached-price fields (cost_per_1m_prompt_cached != cost_per_1m_prompt)
|
|
so that varying cache_rate produces differentiable costs.
|
|
"""
|
|
|
|
INCUMBENT = ("provA", "incumbent-model")
|
|
INCUMBENT_PROVIDER, INCUMBENT_MODEL = INCUMBENT
|
|
|
|
# Consistent across all tests that need estimated_cost:
|
|
_PTOKENS = 1000
|
|
_CTOKENS = 500
|
|
_PROMPT_PRICE = 10.0
|
|
_CACHED_PRICE = 1.0
|
|
_COMPLETION_PRICE = 2.0
|
|
|
|
@classmethod
|
|
def _pricing_rows(cls):
|
|
"""Two rows with cached pricing and same-band proficiency."""
|
|
return [
|
|
_row(
|
|
model_id=cls.INCUMBENT_MODEL,
|
|
provider=cls.INCUMBENT_PROVIDER,
|
|
proficiency=0.90,
|
|
cost_per_1m_prompt=cls._PROMPT_PRICE,
|
|
cost_per_1m_prompt_cached=cls._CACHED_PRICE,
|
|
cost_per_1m_completion=cls._COMPLETION_PRICE,
|
|
),
|
|
_row(
|
|
model_id="challenger",
|
|
provider="provB",
|
|
proficiency=0.90,
|
|
cost_per_1m_prompt=cls._PROMPT_PRICE,
|
|
cost_per_1m_prompt_cached=cls._CACHED_PRICE,
|
|
cost_per_1m_completion=cls._COMPLETION_PRICE,
|
|
),
|
|
]
|
|
|
|
# (a) EQUIVALENCE: incumbent=None produces pre-change ranking
|
|
def test_equivalence_incumbent_none_matches_default(self):
|
|
rows = [
|
|
_row(model_id="A", proficiency=0.95, cost=0.10),
|
|
_row(model_id="B", proficiency=0.80, cost=0.01),
|
|
]
|
|
default = rank_candidates(rows)
|
|
explicit = rank_candidates(rows, incumbent=None)
|
|
assert [r["model_id"] for r in default] == [r["model_id"] for r in explicit]
|
|
for d, e in zip(default, explicit):
|
|
assert d["cost"] == e["cost"]
|
|
assert d["effective_cost"] == e["effective_cost"]
|
|
assert d["proficiency_score"] == pytest.approx(e["proficiency_score"])
|
|
|
|
# (b) Incumbent protection: same band + dial OFF neutral → incumbent wins
|
|
def test_incumbent_wins_on_cost_when_measured_rate_is_lower(self):
|
|
rows = self._pricing_rows()
|
|
# cache_rate=0.5 → cheaper (more tokens at $1 cached, fewer at $10 fresh)
|
|
# challenger_cache_rate=0.1 → clamp=min(0.1, 0.5)=0.1 → more expensive
|
|
ranked = rank_candidates(
|
|
rows,
|
|
incumbent=self.INCUMBENT,
|
|
measured_cache_rates={self.INCUMBENT: 0.5},
|
|
challenger_cache_rate=0.1,
|
|
prompt_tokens=self._PTOKENS,
|
|
completion_tokens=self._CTOKENS,
|
|
)
|
|
assert ranked[0]["model_id"] == self.INCUMBENT_MODEL
|
|
assert ranked[0]["cost"] < ranked[1]["cost"]
|
|
|
|
# (c) Quality wins outright: challenger above tolerance band beats incumbent
|
|
def test_quality_gap_still_decides_outright(self):
|
|
rows = [
|
|
_row(
|
|
model_id=self.INCUMBENT_MODEL,
|
|
provider=self.INCUMBENT_PROVIDER,
|
|
proficiency=0.80,
|
|
cost_per_1m_prompt=self._PROMPT_PRICE,
|
|
cost_per_1m_prompt_cached=self._CACHED_PRICE,
|
|
cost_per_1m_completion=self._COMPLETION_PRICE,
|
|
),
|
|
_row(
|
|
model_id="superior-challenger",
|
|
provider="provB",
|
|
proficiency=1.00,
|
|
cost_per_1m_prompt=100.0,
|
|
cost_per_1m_prompt_cached=10.0,
|
|
cost_per_1m_completion=20.0,
|
|
),
|
|
]
|
|
ranked = rank_candidates(
|
|
rows,
|
|
incumbent=self.INCUMBENT,
|
|
measured_cache_rates={self.INCUMBENT: 0.5},
|
|
challenger_cache_rate=0.1,
|
|
prompt_tokens=self._PTOKENS,
|
|
completion_tokens=self._CTOKENS,
|
|
quality_tolerance=0.1,
|
|
)
|
|
assert ranked[0]["model_id"] == "superior-challenger"
|
|
|
|
# (d) Challenger cost matches estimated_cost with min(dial, inc_rate) at 3 dials
|
|
def test_challenger_cost_matches_clamped_rate(self):
|
|
rows = self._pricing_rows()
|
|
inc_rate = 0.7
|
|
cache_rate = 0.84
|
|
for challenger_cache_rate, expected_rate in [
|
|
(None, cache_rate), # neutral → unchanged
|
|
(cache_rate, cache_rate), # neutral → unchanged
|
|
(0.5, min(0.5, inc_rate)), # mid → clamped
|
|
(0.0, min(0.0, inc_rate)), # 0.0 → clamped
|
|
]:
|
|
ranked = rank_candidates(
|
|
rows,
|
|
incumbent=self.INCUMBENT,
|
|
measured_cache_rates={self.INCUMBENT: inc_rate},
|
|
challenger_cache_rate=challenger_cache_rate,
|
|
prompt_tokens=self._PTOKENS,
|
|
completion_tokens=self._CTOKENS,
|
|
cache_rate=cache_rate,
|
|
)
|
|
# Challenger is always index 1
|
|
challenger = ranked[1]
|
|
expected = estimated_cost(
|
|
rows[1], self._PTOKENS, self._CTOKENS, expected_rate
|
|
)
|
|
assert challenger["cost"] == pytest.approx(expected)
|
|
|
|
# (e) Rate ladder: dict hit → source="measured"; dict miss → source="assumed"
|
|
def test_rate_ladder_source_metadata(self):
|
|
rows = self._pricing_rows()
|
|
# Hit: incumbent in measured_cache_rates
|
|
ranked_hit = rank_candidates(
|
|
rows,
|
|
incumbent=self.INCUMBENT,
|
|
measured_cache_rates={self.INCUMBENT: 0.5},
|
|
challenger_cache_rate=0.3,
|
|
prompt_tokens=self._PTOKENS,
|
|
completion_tokens=self._CTOKENS,
|
|
)
|
|
assert ranked_hit[0]["incumbent_pricing"]["source"] == "measured"
|
|
|
|
# Miss: measured_cache_rates doesn't include incumbent
|
|
rows2 = self._pricing_rows()
|
|
ranked_miss = rank_candidates(
|
|
rows2,
|
|
incumbent=self.INCUMBENT,
|
|
measured_cache_rates={}, # empty → no entry
|
|
challenger_cache_rate=0.3,
|
|
prompt_tokens=self._PTOKENS,
|
|
completion_tokens=self._CTOKENS,
|
|
)
|
|
assert ranked_miss[0]["incumbent_pricing"]["source"] == "assumed"
|
|
|
|
# (f) Composition: effective_cost == cost * provider_cost_multipliers[provider]
|
|
def test_effective_cost_applies_multiplier_for_incumbent(self):
|
|
rows = self._pricing_rows()
|
|
ranked = rank_candidates(
|
|
rows,
|
|
incumbent=self.INCUMBENT,
|
|
measured_cache_rates={self.INCUMBENT: 0.5},
|
|
challenger_cache_rate=0.3,
|
|
prompt_tokens=self._PTOKENS,
|
|
completion_tokens=self._CTOKENS,
|
|
provider_cost_multipliers={self.INCUMBENT_PROVIDER: 5.0},
|
|
)
|
|
# Find the incumbent row by model_id (it may not be ranked[0])
|
|
inc_row = [r for r in ranked if r["model_id"] == self.INCUMBENT_MODEL][0]
|
|
assert inc_row["effective_cost"] == pytest.approx(
|
|
inc_row["cost"] * 5.0
|
|
)
|
|
|
|
# (g) NEUTRAL-EQUIVALENCE BLOCKING TEST: both forms byte-identical to reference
|
|
def test_neutral_forms_are_byte_identical_to_reference(self):
|
|
rows = self._pricing_rows()
|
|
# Pre-condition: incumbent IS in measured_cache_rates (non-vacuous)
|
|
rates = {self.INCUMBENT: 0.3}
|
|
assert self.INCUMBENT in rates
|
|
|
|
# Reference: no incumbent pricing
|
|
ref = rank_candidates(rows, prompt_tokens=self._PTOKENS,
|
|
completion_tokens=self._CTOKENS)
|
|
|
|
# Form (i): challenger_cache_rate=None
|
|
form1 = rank_candidates(rows, prompt_tokens=self._PTOKENS,
|
|
completion_tokens=self._CTOKENS,
|
|
incumbent=self.INCUMBENT,
|
|
measured_cache_rates=rates,
|
|
challenger_cache_rate=None)
|
|
|
|
# Form (ii): challenger_cache_rate == cache_rate
|
|
form2 = rank_candidates(rows, prompt_tokens=self._PTOKENS,
|
|
completion_tokens=self._CTOKENS,
|
|
incumbent=self.INCUMBENT,
|
|
measured_cache_rates=rates,
|
|
challenger_cache_rate=0.84)
|
|
|
|
# Explicit field-by-field comparison for byte identity
|
|
for f in form1, form2:
|
|
assert len(f) == len(ref)
|
|
for i in range(len(ref)):
|
|
for key in ref[i]:
|
|
assert ref[i][key] == f[i][key], \
|
|
f"Mismatch at row {i}, key {key}: {ref[i][key]} != {f[i][key]}"
|
|
|
|
# (h) INVARIANT PROPERTY TEST: incumbent_rate >= every challenger_rate
|
|
def test_incumbent_rate_always_ge_challenger_rate(self):
|
|
rows = self._pricing_rows()
|
|
for inc_rate in (0.731, 0.953): # straddle assumed cache_rate (0.84)
|
|
for dial in (0.0, 0.1, 0.3, 0.5, 0.7, 0.9, 1.0):
|
|
ranked = rank_candidates(
|
|
rows,
|
|
incumbent=self.INCUMBENT,
|
|
measured_cache_rates={self.INCUMBENT: inc_rate},
|
|
challenger_cache_rate=dial,
|
|
prompt_tokens=self._PTOKENS,
|
|
completion_tokens=self._CTOKENS,
|
|
)
|
|
inc_row = next(
|
|
r for r in ranked if r["model_id"] == self.INCUMBENT_MODEL
|
|
)
|
|
chal_row = next(
|
|
r for r in ranked if r["model_id"] != self.INCUMBENT_MODEL
|
|
)
|
|
# THE CLAMP pinned: challenger priced at min(dial, inc_rate)
|
|
# for EVERY dial — including every dial ABOVE the incumbent's
|
|
# measured rate (the inversion zone). Without the clamp
|
|
# (pricing challengers at the raw dial) this fails.
|
|
assert chal_row["cost"] == pytest.approx(estimated_cost(
|
|
next(r for r in rows if r["model_id"] == chal_row["model_id"]),
|
|
self._PTOKENS, self._CTOKENS, min(dial, inc_rate),
|
|
)), f"inc_rate={inc_rate} dial={dial}: clamp violated"
|
|
assert inc_row["cost"] == pytest.approx(estimated_cost(
|
|
next(r for r in rows if r["model_id"] == inc_row["model_id"]),
|
|
self._PTOKENS, self._CTOKENS, inc_rate,
|
|
))
|
|
# Invariant in cost terms: cached is cheaper than fresh, so
|
|
# the challenger's rate (min(dial, inc_rate)) never exceeds
|
|
# the incumbent's measured rate — it can never cost less.
|
|
assert chal_row["cost"] >= inc_row["cost"]
|
|
|