"""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 routing import ( capability_gate_reason, 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 # --- 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 # --- 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)