#!/usr/bin/env python3 """Behavior tests for role routing, spending boundaries, and deterministic ties.""" import os import json import sys import tempfile import unittest from datetime import datetime, timezone from decimal import Decimal ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0, os.path.join(ROOT, "scripts")) import model_catalog import model_router NOW = datetime(2026, 9, 25, 12, 0, tzinfo=timezone.utc) NOW_TEXT = "2026-09-25T12:00:00Z" def candidate(model_key, *, provider="opencode", score="45", native_input="0.0000001", billing="zen_balance", **extra): result = { "tool": "opencode", "model_key": model_key, "provider": provider, "model_id": model_key.split("/", 1)[-1], "variant": None, "billing_mode": billing, "availability": {"local_listed": True, "authenticated": True, "entitled": True, "verified_at": NOW_TEXT, "local_verified_at": NOW_TEXT, "account_verified_at": NOW_TEXT, "balance_confirmed": True, "balance_verified_at": NOW_TEXT}, "score": {"metric_id": model_catalog.AA_METRIC, "version": "1.5", "score": score, "agent": "Codex", "harness": "OpenCode", "variant": None, "settings": {"effort": "max"}, "match_grade": "exact", "source_url": model_catalog.AA_URL, "observed_at": NOW_TEXT}, "openrouter": {"captured_at": NOW_TEXT, "price_per_token_usd": {"input": "0.0000001", "output": "0.0000005"}}, "native_tariff": {"billing_mode": billing, "account_confirmed": True, "source_url": "https://opencode.ai/docs/zen", "captured_at": NOW_TEXT, "available_balance_usd": "2", "price_per_token_usd": {"input": native_input, "output": "0.0000002", "cache_read": "0.00000001", "cache_write": "0.00000002"}}, } if billing == "go_quota": result["quota_verified_at"] = NOW_TEXT result["quota_source_url"] = "https://opencode.ai/docs/go" result["use_balance_fallback"] = False result.update(extra) return result class RoutePolicyTests(unittest.TestCase): def test_missing_cache_is_explicit_and_no_model_is_invented(self): with tempfile.TemporaryDirectory() as cache: result = model_router.route_from_catalog("opencode", "implementer", ["opencode"], cache) self.assertIsNone(result["selected"]) self.assertIn("snapshot_de_catalogo_ausente", result["reason"]) empty = model_router.route([], "explorer", allowed_providers=["opencode"], now=NOW) self.assertIsNone(empty["selected"]) self.assertEqual(empty["reason"], "no_scored_candidate") high_only = model_router.route([candidate("opencode/high-only", score="54")], "implementer", allowed_providers=["opencode"], budget_usd="1", now=NOW) self.assertIsNone(high_only["selected"]) self.assertEqual(high_only["reason"], "faixa_vazia:media") def test_benchmark_with_wrong_index_is_never_ranked(self): row = candidate("opencode/model-a", score="54") row["score"]["metric_id"] = "artificial_analysis_intelligence_index" result = model_router.route([row], "planner", allowed_providers=["opencode"], now=NOW) self.assertIsNone(result["selected"]) self.assertEqual(result["ranking"][0]["eligible"], False) self.assertIn("indice_incompativel_ou_ausente", result["ranking"][0]["exclusion_reasons"]) def test_role_band_precedes_cost_and_plateau_uses_native_cost_not_openrouter(self): cheap_high = candidate("opencode/high", score="54", native_input="0.000000001") media_best = candidate("opencode/media", score="49", native_input="0.0000001") media_cheaper = candidate("opencode/media-cheap", score="47", native_input="0.00000005") result = model_router.route([cheap_high, media_best, media_cheaper], "implementer", allowed_providers=["opencode"], tokens={"input": 1_000_000}, budget_usd="0.08", now=NOW) self.assertEqual(result["selected"]["model_key"], "opencode/media-cheap") self.assertEqual(result["selected"]["band"], "media") self.assertEqual(Decimal(result["selected"]["reference_cost_usd"]), Decimal("0.1")) self.assertEqual(Decimal(result["selected"]["estimated_native_usd"]), Decimal("0.05")) self.assertIn("platô", result["reason"]) def test_zen_balance_must_cover_native_estimate_and_be_confirmed(self): row = candidate("opencode/model-a") row["native_tariff"]["available_balance_usd"] = "0.05" result = model_router.route([row], "implementer", allowed_providers=["opencode"], tokens={"input": 1_000_000}, budget_usd="1", now=NOW) self.assertIsNone(result["selected"]) self.assertIn("saldo_insuficiente", result["ranking"][0]["exclusion_reasons"]) row["native_tariff"]["available_balance_usd"] = "1" row["availability"]["balance_confirmed"] = False result = model_router.route([row], "implementer", allowed_providers=["opencode"], tokens={"input": 1_000_000}, budget_usd="1", now=NOW) self.assertIsNone(result["selected"]) self.assertIn("saldo_nao_confirmado_ou_obsoleto", result["ranking"][0]["exclusion_reasons"]) def test_zen_balance_requires_explicit_budget(self): row = candidate("opencode/model-a") result = model_router.route([row], "implementer", allowed_providers=["opencode"], now=NOW) self.assertIsNone(result["selected"]) self.assertIn("orcamento_zen_nao_configurado", result["ranking"][0]["exclusion_reasons"]) def test_exact_benchmark_harness_must_match_tool_but_proxy_requires_approval(self): row = candidate("opencode/model-a") row["score"]["harness"] = "Codex" result = model_router.route([row], "implementer", allowed_providers=["opencode"], tokens={"input": 100_000}, budget_usd="1", now=NOW) self.assertIsNone(result["selected"]) self.assertIn("harness_incompativel", result["ranking"][0]["exclusion_reasons"]) row["score"]["match_grade"] = "proxy" result = model_router.route([row], "implementer", allowed_providers=["opencode"], tokens={"input": 100_000}, budget_usd="1", now=NOW) self.assertIsNone(result["selected"]) self.assertIn("match_proxy_nao_autorizado", result["ranking"][0]["exclusion_reasons"]) approved = model_router.route([row], "implementer", allowed_providers=["opencode"], tokens={"input": 100_000}, budget_usd="1", allow_proxy=True, now=NOW) self.assertEqual(approved["selected"]["model_key"], "opencode/model-a") def test_budget_is_applied_before_plateau_and_proxy_requires_explicit_permission(self): top = candidate("opencode/top", score="49", native_input="0.000001") lower = candidate("opencode/lower", score="46", native_input="0.00000005") row = candidate("opencode/proxy", score="45", match="proxy") row["score"]["match_grade"] = "proxy" result = model_router.route([top, lower, row], "implementer", allowed_providers=["opencode"], tokens={"input": 1_000_000}, budget_usd="0.1", now=NOW) self.assertEqual(result["selected"]["model_key"], "opencode/lower") exclusions = {entry["model_key"]: entry["exclusion_reasons"] for entry in result["ranking"]} self.assertIn("match_proxy_nao_autorizado", exclusions["opencode/proxy"]) proxy = model_router.route([row], "implementer", allowed_providers=["opencode"], tokens={"input": 1_000_000}, budget_usd="1", allow_proxy=True, now=NOW) self.assertEqual(proxy["selected"]["model_key"], "opencode/proxy") def test_unknown_entitlement_and_provider_are_not_eligible(self): row = candidate("opencode/model-a", score="54") row["availability"]["entitled"] = None result = model_router.route([row], "planner", allowed_providers=["opencode"], now=NOW) self.assertIsNone(result["selected"]) self.assertIn("entitlement_nao_confirmado", result["ranking"][0]["exclusion_reasons"]) unauthorized = model_router.route([candidate("opencode/model-a", score="54")], "planner", now=NOW) self.assertIn("provider_nao_autorizado", unauthorized["reason"]) def test_go_quota_checks_every_window_and_never_falls_back_to_zen(self): for window in model_router.WINDOWS: row = candidate("opencode-go/model-a", provider="opencode-go", score="54", billing="go_quota") row["quota_window_remaining"] = {"5h": "1", "weekly": "1", "monthly": "1"} row["quota_window_debit"] = {"5h": "0.1", "weekly": "0.1", "monthly": "0.1"} row["quota_window_remaining"][window] = "0.05" result = model_router.route([row], "planner", allowed_providers=["opencode-go"], tokens={"input": 1_000_000}, now=NOW) self.assertIsNone(result["selected"], window) self.assertIn("quota_go_insuficiente", result["reason"]) row = candidate("opencode-go/model-a", provider="opencode-go", score="54", billing="go_quota") row["quota_window_remaining"] = {"5h": "1", "weekly": "1", "monthly": "1"} row["quota_window_debit"] = {"5h": "0.1", "weekly": "0.1", "monthly": "0.1"} row["use_balance_fallback"] = True result = model_router.route([row], "planner", allowed_providers=["opencode-go"], tokens={"input": 1_000_000}, now=NOW) self.assertIsNone(result["selected"]) self.assertEqual(result["reason"], "fallback_go_para_saldo_zen_nao_autorizado") def test_go_success_exposes_quota_and_billing_mode_separately(self): row = candidate("opencode-go/model-a", provider="opencode-go", score="54", billing="go_quota") row["quota_window_remaining"] = {"5h": "1", "weekly": "2", "monthly": "3"} row["quota_window_debit"] = {"5h": "0.1", "weekly": "0.1", "monthly": "0.2"} result = model_router.route([row], "planner", allowed_providers=["opencode-go"], tokens={"input": 1_000_000}, now=NOW) selected = result["selected"] self.assertEqual(selected["billing_mode"], "go") self.assertEqual(selected["billing_mode_detail"], "go_quota") self.assertEqual(selected["quota_window_debit"]["monthly"], "0.2") self.assertIsNone(selected["estimated_native_usd"]) self.assertEqual(selected["native_cost_or_quota_debit_usd"], "0.2") def test_missing_cache_rate_disables_comparative_cost_not_native_selection(self): row = candidate("opencode/model-a", score="45") row["openrouter"]["price_per_token_usd"] = {"input": "0.0000001"} result = model_router.route([row], "implementer", allowed_providers=["opencode"], tokens={"input": 1_000_000, "cache_read": 1000}, budget_usd="1", now=NOW) self.assertEqual(result["selected"]["model_key"], "opencode/model-a") self.assertIsNone(result["selected"]["reference_cost_usd"]) self.assertEqual(Decimal(result["selected"]["estimated_native_usd"]), Decimal("0.10001")) def test_expired_paid_metadata_blocks_automatic_selection(self): row = candidate("opencode/model-a", score="54") row["openrouter"]["captured_at"] = "2026-09-23T00:00:00Z" result = model_router.route([row], "planner", allowed_providers=["opencode"], budget_usd="1", now=NOW) self.assertIsNone(result["selected"]) self.assertIn("preco_referencial_obsoleto_ou_ausente", result["ranking"][0]["exclusion_reasons"]) def test_ties_are_deterministic_independent_of_input_order(self): a = candidate("opencode/a", score="48", native_input="0.00000005") b = candidate("opencode/b", score="48", native_input="0.00000005") first = model_router.route([a, b], "implementer", allowed_providers=["opencode"], tokens={"input": 1_000_000}, budget_usd="1", now=NOW) second = model_router.route([b, a], "implementer", allowed_providers=["opencode"], tokens={"input": 1_000_000}, budget_usd="1", now=NOW) self.assertEqual(first, second) self.assertEqual(first["selected"]["model_key"], "opencode/a") def test_fixed_bands_metric_and_role_override_contract(self): self.assertEqual(model_router.validate_bands({"leve": [10, 20], "media": [20, 50], "alta": [50, 100]}), model_router.FIXED_BANDS) with self.assertRaises(ValueError): model_router.validate_bands({"leve": [0, 20], "media": [20, 50], "alta": [50, 100]}) row = candidate("opencode/model-a", score="54") result = model_router.route([row], "implementer", allowed_providers=["opencode"], band_override="alta", tokens={"input": 1_000_000}, budget_usd="1", now=NOW) self.assertEqual(result["selected"]["band"], "alta") with self.assertRaises(ValueError): model_router.route([row], "planner", allowed_providers=["opencode"], benchmark_metric="aa_coding_agent_index_v1.4", now=NOW) def test_omp_payg_uses_only_native_estimate_and_exact_dispatch_variant(self): row = candidate("openai/model-a#high", provider="openai", score="54", billing="omp_payg") row.update({"tool": "omp", "model_id": "model-a", "variant": "high"}) row["score"]["harness"] = "OMP" row["score"]["variant"] = "high" row["openrouter"] = None row["native_tariff"] = { "billing_mode": "omp_payg", "account_confirmed": True, "source_url": "https://provider.example/pricing", "captured_at": NOW_TEXT, "price_per_token_usd": {"input": "0.000001"}, "available_balance_usd": "0.2", } result = model_router.route([row], "planner", allowed_providers=["openai"], tokens={"input": 100_000}, budget_usd="0.2", now=NOW) selected = result["selected"] self.assertEqual(selected["model_key"], "openai/model-a#high") self.assertEqual(selected["dispatch_model_key"], "openai/model-a:high") self.assertEqual(selected["billing_mode"], "payg") self.assertEqual(selected["estimated_native_usd"], "0.100000") self.assertEqual(selected["available_balance_usd"], "0.2") self.assertIsNone(selected["reference_cost_usd"]) row["availability"]["balance_confirmed"] = False blocked = model_router.route([row], "planner", allowed_providers=["openai"], tokens={"input": 100_000}, budget_usd="0.2", now=NOW) self.assertIsNone(blocked["selected"]) self.assertIn("saldo_nao_confirmado_ou_obsoleto", blocked["ranking"][0]["exclusion_reasons"]) def test_omp_catalog_route_revalidates_local_list_and_keeps_manual_entitlement(self): row = candidate("openai/model-a#high", provider="openai", score="54", billing="omp_payg") row.update({"tool": "omp", "model_id": "model-a", "variant": "high"}) row["score"]["harness"] = "OMP" row["score"]["variant"] = "high" row["native_tariff"] = { "billing_mode": "omp_payg", "account_confirmed": True, "source_url": "https://provider.example/pricing", "captured_at": NOW_TEXT, "price_per_token_usd": {"input": "0.000001"}, "available_balance_usd": "0.2", } with tempfile.TemporaryDirectory() as cache: with open(os.path.join(cache, "catalog-omp.json"), "w", encoding="utf-8") as stream: json.dump({"tool": "omp", "models": [row]}, stream) result = model_router.route_from_catalog( "omp", "planner", ["openai"], cache, tokens={"input": 100_000}, budget_usd="0.2", omp_discoverer=lambda provider: {"ok": True, "models": ["openai/model-a#high"], "verified_at": NOW_TEXT}, now=NOW, ) unsupported = model_router.route_from_catalog( "omp", "planner", ["openai"], cache, tokens={"input": 100_000}, budget_usd="0.2", omp_discoverer=lambda provider: {"ok": True, "models": ["openai/model-a"], "verified_at": NOW_TEXT}, now=NOW, ) self.assertIsNone(unsupported["selected"]) self.assertIn("nao_listado_localmente", unsupported["ranking"][0]["exclusion_reasons"]) self.assertEqual(result["selected"]["dispatch_model_key"], "openai/model-a:high") if __name__ == "__main__": unittest.main(verbosity=1)