evandro
1aa2c68818
Adds qa/test_branch_paths.py pinning the main path and one inverted path per decision for validate_aa_snapshot, validate_availability_snapshot, validate_native_tariff_overrides, model_router._staleness and _native_cost_and_headroom, then splits each into small helpers. Messages, check order and results are unchanged.
456 lines
26 KiB
Python
456 lines
26 KiB
Python
#!/usr/bin/env python3
|
|
"""Characterization tests: one main path plus one path per decision for the validators and
|
|
the staleness/cost helpers. They pin current behavior so the functions can be split safely.
|
|
|
|
python3 qa/test_branch_paths.py
|
|
"""
|
|
import copy
|
|
import os
|
|
import sys
|
|
import unittest
|
|
from datetime import datetime, timedelta, 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"))
|
|
sys.path.insert(0, os.path.join(ROOT, "qa"))
|
|
import model_catalog as catalog
|
|
import model_router
|
|
from test_model_router import NOW, NOW_TEXT, candidate
|
|
|
|
PAST = "2026-01-01T00:00:00Z"
|
|
FUTURE = "2099-01-01T00:00:00Z"
|
|
OLD = "2026-08-01T00:00:00Z"
|
|
|
|
|
|
def mutated(base, *edits):
|
|
"""Deep copy of base with edits [(path, value)]; value DELETE removes the key."""
|
|
result = copy.deepcopy(base)
|
|
for path, value in edits:
|
|
node = result
|
|
for key in path[:-1]:
|
|
node = node[key]
|
|
if value is DELETE:
|
|
node.pop(path[-1], None)
|
|
else:
|
|
node[path[-1]] = value
|
|
return result
|
|
|
|
|
|
DELETE = object()
|
|
|
|
|
|
class ValidationCases(unittest.TestCase):
|
|
def assert_rejected(self, function, base, cases):
|
|
for name, edits, message in cases:
|
|
with self.subTest(name):
|
|
with self.assertRaisesRegex(ValueError, message):
|
|
function(mutated(base, *edits))
|
|
|
|
|
|
class NativeTariffTests(ValidationCases):
|
|
ZEN = {"opencode/m": {
|
|
"billing_mode": "zen_balance", "account_confirmed": True, "source_url": "https://opencode.ai/docs/zen",
|
|
"captured_at": PAST, "price_per_token_usd": {"input": "0.1", "output": "0.2"},
|
|
"available_balance_usd": "2"}}
|
|
GO = {"opencode/m": {
|
|
"billing_mode": "go_quota", "account_confirmed": True, "source_url": "https://opencode.ai/docs/go",
|
|
"captured_at": PAST,
|
|
"quota_window_remaining": {"5h": "1", "weekly": "2", "monthly": "3"},
|
|
"quota_window_debit": {"5h": "0.1", "weekly": "0.2", "monthly": "0.3"}}}
|
|
|
|
def test_main_paths_normalize_records(self):
|
|
zen = catalog.validate_native_tariff_overrides(self.ZEN)["opencode/m"]
|
|
self.assertEqual(zen["price_per_token_usd"], {"input": "0.1", "output": "0.2"})
|
|
self.assertEqual(zen["available_balance_usd"], "2")
|
|
self.assertNotIn("quota_window_remaining", zen)
|
|
go = catalog.validate_native_tariff_overrides(self.GO)["opencode/m"]
|
|
self.assertEqual(go["quota_window_remaining"], {"5h": "1", "weekly": "2", "monthly": "3"})
|
|
self.assertEqual(go["quota_window_debit"]["monthly"], "0.3")
|
|
self.assertIs(go["use_balance_fallback"], False)
|
|
self.assertNotIn("available_balance_usd", go)
|
|
self.assertEqual(catalog.validate_native_tariff_overrides({}), {})
|
|
|
|
def test_omp_payg_follows_the_zen_branch_and_fallback_bool_is_kept(self):
|
|
omp = mutated(self.ZEN, (("opencode/m", "billing_mode"), "omp_payg"))
|
|
self.assertEqual(catalog.validate_native_tariff_overrides(omp)["opencode/m"]["billing_mode"], "omp_payg")
|
|
go = mutated(self.GO, (("opencode/m", "use_balance_fallback"), True))
|
|
self.assertIs(catalog.validate_native_tariff_overrides(go)["opencode/m"]["use_balance_fallback"], True)
|
|
|
|
def test_shared_decisions_reject(self):
|
|
z = "opencode/m"
|
|
self.assert_rejected(catalog.validate_native_tariff_overrides, self.ZEN, [
|
|
("unknown field", [((z, "extra"), 1)], "estrutura/campo inválido"),
|
|
("value not a dict", [((z,), "x")], "estrutura/campo inválido"),
|
|
("not confirmed", [((z, "account_confirmed"), False)], "confirmação explícita"),
|
|
("bad mode", [((z, "billing_mode"), "other")], "billing_mode native inválido"),
|
|
("http url", [((z, "source_url"), "http://opencode.ai/docs")], "source_url HTTPS"),
|
|
("captured_at missing", [((z, "captured_at"), DELETE)], "captured_at de tarifa/cota native inválido"),
|
|
("captured_at garbage", [((z, "captured_at"), "x")], "captured_at de tarifa/cota native inválido"),
|
|
("captured_at naive", [((z, "captured_at"), "2026-01-01T00:00:00")], "aware e não futuro"),
|
|
("captured_at future", [((z, "captured_at"), FUTURE)], "aware e não futuro"),
|
|
])
|
|
with self.assertRaisesRegex(ValueError, "deve ser objeto JSON"):
|
|
catalog.validate_native_tariff_overrides([])
|
|
with self.assertRaisesRegex(ValueError, "estrutura/campo inválido"):
|
|
catalog.validate_native_tariff_overrides({1: self.ZEN["opencode/m"]})
|
|
|
|
def test_balance_branch_rejects(self):
|
|
z = "opencode/m"
|
|
self.assert_rejected(catalog.validate_native_tariff_overrides, self.ZEN, [
|
|
("rates missing", [((z, "price_per_token_usd"), DELETE)], "precisa de price_per_token_usd"),
|
|
("rates empty", [((z, "price_per_token_usd"), {})], "precisa de price_per_token_usd"),
|
|
("unknown component", [((z, "price_per_token_usd"), {"input": "1", "foo": "1"})], "componentes de tarifa"),
|
|
("negative component", [((z, "price_per_token_usd"), {"input": "-1"})], "componentes de tarifa"),
|
|
("empty component", [((z, "price_per_token_usd"), {"input": ""})], "componentes de tarifa"),
|
|
("balance missing", [((z, "available_balance_usd"), DELETE)], "saldo disponível"),
|
|
("balance negative", [((z, "available_balance_usd"), "-1")], "saldo disponível"),
|
|
])
|
|
|
|
def test_go_branch_rejects(self):
|
|
z = "opencode/m"
|
|
self.assert_rejected(catalog.validate_native_tariff_overrides, self.GO, [
|
|
("remaining missing", [((z, "quota_window_remaining"), DELETE)], "5h/weekly/monthly"),
|
|
("debit not dict", [((z, "quota_window_debit"), "1")], "5h/weekly/monthly"),
|
|
("window missing in remaining", [((z, "quota_window_remaining"), {"5h": "1", "weekly": "2"})], "5h/weekly/monthly"),
|
|
("window missing in debit", [((z, "quota_window_debit"), {"5h": "1", "monthly": "2"})], "5h/weekly/monthly"),
|
|
("negative remaining", [((z, "quota_window_remaining", "5h"), "-1")], "não negativos"),
|
|
("empty debit", [((z, "quota_window_debit", "weekly"), "")], "não negativos"),
|
|
("fallback not bool", [((z, "use_balance_fallback"), "yes")], "use_balance_fallback precisa ser bool"),
|
|
])
|
|
|
|
|
|
class AvailabilityTests(ValidationCases):
|
|
BASE = {
|
|
"tool": "opencode", "ingestion": "manual_audited", "user_confirmed": True, "captured_at": PAST,
|
|
"records": [{"provider": "opencode", "model_id": "m", "variant": None, "local_listed": True,
|
|
"authenticated": True, "entitled": None, "verified_at": PAST, "source": "manual"}]}
|
|
|
|
def rec(self, **edits):
|
|
return [(("records", 0, key), value) for key, value in edits.items()]
|
|
|
|
def test_main_path_returns_audited_record_keyed_by_identity(self):
|
|
result = catalog.validate_availability_snapshot(self.BASE)
|
|
record = result[("opencode", "m", None)]
|
|
self.assertEqual((record["local_listed"], record["authenticated"], record["entitled"]), (True, True, None))
|
|
self.assertEqual(record["source"], "manual_audited")
|
|
self.assertIsNone(record["balance_confirmed"])
|
|
self.assertEqual(len(record["source_hash"]), 64)
|
|
|
|
def test_balance_confirmed_with_valid_time_is_kept(self):
|
|
payload = mutated(self.BASE, *self.rec(balance_confirmed=True, balance_verified_at=PAST))
|
|
record = catalog.validate_availability_snapshot(payload)[("opencode", "m", None)]
|
|
self.assertEqual((record["balance_confirmed"], record["balance_verified_at"]), (True, PAST))
|
|
|
|
def test_snapshot_level_decisions_reject(self):
|
|
self.assert_rejected(catalog.validate_availability_snapshot, self.BASE, [
|
|
("tool", [(("tool",), "claude")], "manual auditado"),
|
|
("ingestion", [(("ingestion",), "auto")], "manual auditado"),
|
|
("not confirmed", [(("user_confirmed",), False)], "user_confirmed=true"),
|
|
("captured_at not str", [(("captured_at",), 5)], "captured_at de disponibilidade obrigatório"),
|
|
("captured_at garbage", [(("captured_at",), "x")], "captured_at de disponibilidade inválido"),
|
|
("captured_at naive", [(("captured_at",), "2026-01-01T00:00:00")], "UTC/offset-aware"),
|
|
("captured_at future", [(("captured_at",), FUTURE)], "UTC/offset-aware"),
|
|
("records not list", [(("records",), {})], "deve ser lista"),
|
|
])
|
|
with self.assertRaisesRegex(ValueError, "manual auditado"):
|
|
catalog.validate_availability_snapshot([])
|
|
|
|
def test_record_level_decisions_reject(self):
|
|
self.assert_rejected(catalog.validate_availability_snapshot, self.BASE, [
|
|
("record not dict", [(("records",), ["x"])], "campos não permitidos"),
|
|
("extra field", self.rec(extra=1), "campos não permitidos"),
|
|
("provider not str", self.rec(provider=1), "exatos são obrigatórios"),
|
|
("model empty", self.rec(model_id=""), "exatos são obrigatórios"),
|
|
("model with slash", self.rec(model_id="a/b"), "exatos são obrigatórios"),
|
|
("model with hash", self.rec(model_id="a#b"), "exatos são obrigatórios"),
|
|
("variant not str", self.rec(variant=1), "exatos são obrigatórios"),
|
|
("field absent", self.rec(entitled=DELETE), "todos os campos"),
|
|
("flag not bool", self.rec(authenticated="yes"), "authenticated deve ser bool ou null"),
|
|
("balance flag not bool", self.rec(balance_confirmed="yes"), "balance_confirmed deve ser bool ou null"),
|
|
("balance time without confirmation", self.rec(balance_verified_at=PAST), "requer confirmação explícita"),
|
|
("balance time future", self.rec(balance_confirmed=True, balance_verified_at=FUTURE), "requer confirmação explícita"),
|
|
("confirmed without time", self.rec(balance_confirmed=True), "exige balance_verified_at"),
|
|
("verified_at invalid", self.rec(verified_at="x"), "source e verified_at"),
|
|
("verified_at future", self.rec(verified_at=FUTURE), "source e verified_at"),
|
|
("source empty", self.rec(source=""), "source e verified_at"),
|
|
])
|
|
|
|
def test_duplicate_identity_rejects_but_variant_distinguishes(self):
|
|
first = self.BASE["records"][0]
|
|
with self.assertRaisesRegex(ValueError, "duplicado"):
|
|
catalog.validate_availability_snapshot(mutated(self.BASE, (("records",), [first, dict(first)])))
|
|
other = dict(first, variant="high")
|
|
result = catalog.validate_availability_snapshot(mutated(self.BASE, (("records",), [first, other])))
|
|
self.assertEqual(set(result), {("opencode", "m", None), ("opencode", "m", "high")})
|
|
|
|
|
|
class AaSnapshotTests(ValidationCases):
|
|
URL = "https://artificialanalysis.ai/agents/coding-agents"
|
|
BASE = {
|
|
"metric_id": catalog.AA_METRIC, "version": "1.5", "ingestion": "manual_audited", "terms_accepted": True,
|
|
"source_url": URL, "attribution": "Artificial Analysis", "captured_at": PAST,
|
|
"records": [{"provider": "p", "model_id": "m", "score": "50", "match_grade": "exact", "agent": "a",
|
|
"harness": "h", "settings": {"effort": "max"}, "evidence_url": URL}]}
|
|
|
|
def rec(self, **edits):
|
|
return [(("records", 0, key), value) for key, value in edits.items()]
|
|
|
|
def test_main_path_adds_provenance_and_normalizes_score(self):
|
|
row = catalog.validate_aa_snapshot(self.BASE)[0]
|
|
self.assertEqual((row["score"], row["metric_id"], row["version"]), ("50", catalog.AA_METRIC, "1.5"))
|
|
self.assertEqual((row["observed_at"], row["source_url"], row["attribution"]), (PAST, self.URL, "Artificial Analysis"))
|
|
self.assertEqual(len(row["snapshot_hash"]), 64)
|
|
self.assertEqual(catalog.validate_aa_snapshot(mutated(self.BASE, (("records",), []))), [])
|
|
|
|
def test_snapshot_level_decisions_reject(self):
|
|
self.assert_rejected(catalog.validate_aa_snapshot, self.BASE, [
|
|
("extra root field", [(("extra",), 1)], "campo não permitido"),
|
|
("metric", [(("metric_id",), "other")], "somente aa_coding_agent_index v1.5"),
|
|
("version", [(("version",), "1.4")], "somente aa_coding_agent_index v1.5"),
|
|
("ingestion", [(("ingestion",), "auto")], "importação manual consentida"),
|
|
("terms", [(("terms_accepted",), False)], "importação manual consentida"),
|
|
("source url", [(("source_url",), "https://example.com/x")], "source_url oficial"),
|
|
("attribution missing", [(("attribution",), DELETE)], "atribuição AA obrigatória"),
|
|
("attribution empty", [(("attribution",), "")], "atribuição AA obrigatória"),
|
|
("captured_at invalid", [(("captured_at",), "x")], "captured_at obrigatório"),
|
|
("captured_at future", [(("captured_at",), FUTURE)], "captured_at obrigatório"),
|
|
("records not list", [(("records",), {})], "records deve ser lista"),
|
|
])
|
|
with self.assertRaisesRegex(ValueError, "deve ser objeto"):
|
|
catalog.validate_aa_snapshot([])
|
|
|
|
def test_record_level_decisions_reject(self):
|
|
self.assert_rejected(catalog.validate_aa_snapshot, self.BASE, [
|
|
("record not dict", [(("records",), ["x"])], "registro AA inválido"),
|
|
("record extra field", self.rec(extra=1), "registro AA inválido"),
|
|
("record metric", self.rec(metric_id="other"), "não pode ser misturado"),
|
|
("record version", self.rec(version="1.4"), "não pode ser misturado"),
|
|
("provider not str", self.rec(provider=1), "mapeamento provider/model_id"),
|
|
("model not str", self.rec(model_id=1), "mapeamento provider/model_id"),
|
|
("model empty", self.rec(model_id=""), "mapeamento provider/model_id"),
|
|
("model slash", self.rec(model_id="a/b"), "mapeamento provider/model_id"),
|
|
("model hash", self.rec(model_id="a#b"), "mapeamento provider/model_id"),
|
|
("variant not str", self.rec(variant=1), "variante AA inválida"),
|
|
("variant empty", self.rec(variant=""), "variante AA inválida"),
|
|
("score missing", self.rec(score=DELETE), "intervalo 0..100"),
|
|
("score negative", self.rec(score="-1"), "intervalo 0..100"),
|
|
("score above 100", self.rec(score="101"), "intervalo 0..100"),
|
|
("grade", self.rec(match_grade="fuzzy"), "exact ou proxy"),
|
|
("agent missing", self.rec(agent=DELETE), "proveniência AA obrigatório: agent"),
|
|
("harness empty", self.rec(harness=""), "proveniência AA obrigatório: harness"),
|
|
("settings none", self.rec(settings=None), "proveniência AA obrigatório: settings"),
|
|
("evidence missing", self.rec(evidence_url=DELETE), "proveniência AA obrigatório: evidence_url"),
|
|
("evidence url", self.rec(evidence_url="http://artificialanalysis.ai/x"), "evidence_url AA inválida"),
|
|
])
|
|
|
|
def test_boundaries_duplicates_and_variant(self):
|
|
for score in ("0", "100"):
|
|
row = catalog.validate_aa_snapshot(mutated(self.BASE, *self.rec(score=score)))[0]
|
|
self.assertEqual(row["score"], score)
|
|
proxy = catalog.validate_aa_snapshot(mutated(self.BASE, *self.rec(match_grade="proxy", variant="high")))[0]
|
|
self.assertEqual((proxy["match_grade"], proxy["variant"]), ("proxy", "high"))
|
|
first = self.BASE["records"][0]
|
|
with self.assertRaisesRegex(ValueError, "duplicado"):
|
|
catalog.validate_aa_snapshot(mutated(self.BASE, (("records",), [first, dict(first)])))
|
|
both = catalog.validate_aa_snapshot(mutated(self.BASE, (("records",), [first, dict(first, variant="high")])))
|
|
self.assertEqual(len(both), 2)
|
|
|
|
|
|
TOKENS = {"input": 1000, "output": 100}
|
|
HOUR = timedelta(hours=1)
|
|
|
|
|
|
def stale(row, **kwargs):
|
|
return model_router._staleness(row, kwargs.get("now", NOW), Decimal(24), Decimal(30), kwargs.get("tokens", TOKENS))
|
|
|
|
|
|
def ago(delta):
|
|
return (NOW - delta).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
|
|
|
|
class StalenessTests(unittest.TestCase):
|
|
def go_row(self):
|
|
return candidate("opencode/g", billing="go_quota")
|
|
|
|
def check(self, row, expected, **kwargs):
|
|
self.assertEqual(stale(row, **kwargs), expected)
|
|
|
|
def test_fresh_paid_rows_have_no_reasons(self):
|
|
self.check(candidate("opencode/m"), [])
|
|
self.check(self.go_row(), [])
|
|
|
|
def test_unpaid_mode_is_never_stale(self):
|
|
row = mutated(candidate("opencode/m", billing="free", availability={}), (("score",), {}))
|
|
self.check(row, [])
|
|
|
|
def test_reference_price(self):
|
|
no_price = candidate("opencode/m", openrouter={"captured_at": OLD, "price_per_token_usd": {}})
|
|
self.check(no_price, [])
|
|
for name, value in (("old", OLD), ("future", FUTURE), ("missing", None)):
|
|
with self.subTest(name):
|
|
row = candidate("opencode/m", openrouter={"captured_at": value, "price_per_token_usd": {"input": "1"}})
|
|
self.check(row, ["preco_referencial_obsoleto_ou_ausente"])
|
|
|
|
def test_benchmark_score(self):
|
|
for name, value in (("old", OLD), ("future", FUTURE), ("missing", None)):
|
|
with self.subTest(name):
|
|
row = mutated(candidate("opencode/m"), (("score", "observed_at"), value))
|
|
self.check(row, ["benchmark_obsoleto_ou_ausente"])
|
|
self.check(mutated(candidate("opencode/m"), (("score",), None)), ["benchmark_obsoleto_ou_ausente"])
|
|
|
|
def test_native_tariff_for_balance_modes(self):
|
|
base = candidate("opencode/m")
|
|
cases = [
|
|
("not confirmed", (("native_tariff", "account_confirmed"), False)),
|
|
("http", (("native_tariff", "source_url"), "http://x")),
|
|
("missing url", (("native_tariff", "source_url"), DELETE)),
|
|
("old", (("native_tariff", "captured_at"), OLD)),
|
|
("future", (("native_tariff", "captured_at"), FUTURE)),
|
|
("missing time", (("native_tariff", "captured_at"), DELETE)),
|
|
]
|
|
for name, edit in cases:
|
|
with self.subTest(name):
|
|
self.check(mutated(base, edit), ["tarifa_nativa_obsoleta_ou_nao_confirmada"])
|
|
omp = candidate("opencode/m", billing="omp_payg")
|
|
self.check(mutated(omp, (("native_tariff", "account_confirmed"), False)), ["tarifa_nativa_obsoleta_ou_nao_confirmada"])
|
|
|
|
def test_go_quota_freshness_window_is_sixty_seconds(self):
|
|
base = self.go_row()
|
|
self.check(mutated(base, (("quota_verified_at",), ago(timedelta(seconds=59)))), [])
|
|
for name, edit in (("old", ago(timedelta(seconds=61))), ("missing", None), ("future", FUTURE)):
|
|
with self.subTest(name):
|
|
self.check(mutated(base, (("quota_verified_at",), edit)), ["quota_go_obsoleta_ou_nao_confirmada"])
|
|
self.check(mutated(base, (("quota_source_url",), "http://x")), ["quota_go_obsoleta_ou_nao_confirmada"])
|
|
self.check(mutated(base, (("quota_source_url",), DELETE)), ["quota_go_obsoleta_ou_nao_confirmada"])
|
|
|
|
def test_local_availability(self):
|
|
base = candidate("opencode/m")
|
|
for name, edit in (("old", ago(25 * HOUR)), ("future", FUTURE), ("missing", None)):
|
|
with self.subTest(name):
|
|
row = mutated(base, (("availability", "local_verified_at"), edit), (("availability", "verified_at"), edit))
|
|
self.check(row, ["disponibilidade_local_precisa_revalidacao"])
|
|
fallback = mutated(base, (("availability", "local_verified_at"), DELETE))
|
|
self.check(fallback, [])
|
|
self.check(mutated(base, (("availability", "local_verified_at"), ago(23 * HOUR))), [])
|
|
|
|
def test_entitlement_is_checked_only_when_authenticated_and_entitled(self):
|
|
base = candidate("opencode/m")
|
|
for name, edit in (("old", ago(25 * HOUR)), ("future", FUTURE), ("missing", DELETE)):
|
|
with self.subTest(name):
|
|
self.check(mutated(base, (("availability", "account_verified_at"), edit)), ["entitlement_precisa_revalidacao"])
|
|
stale_account = (("availability", "account_verified_at"), DELETE)
|
|
self.check(mutated(base, stale_account, (("availability", "entitled"), False)), [])
|
|
self.check(mutated(base, stale_account, (("availability", "authenticated"), None)), [])
|
|
|
|
def test_balance_confirmation_for_balance_modes_only(self):
|
|
base = candidate("opencode/m")
|
|
cases = [
|
|
("unconfirmed", (("availability", "balance_confirmed"), False)),
|
|
("old", (("availability", "balance_verified_at"), ago(timedelta(seconds=61)))),
|
|
("future", (("availability", "balance_verified_at"), FUTURE)),
|
|
("missing time", (("availability", "balance_verified_at"), DELETE)),
|
|
]
|
|
for name, edit in cases:
|
|
with self.subTest(name):
|
|
self.check(mutated(base, edit), ["saldo_nao_confirmado_ou_obsoleto"])
|
|
go = self.go_row()
|
|
self.check(mutated(go, (("availability", "balance_confirmed"), False)), [])
|
|
|
|
def test_reasons_accumulate_in_order(self):
|
|
row = mutated(candidate("opencode/m", openrouter={"captured_at": OLD, "price_per_token_usd": {"input": "1"}},
|
|
availability={}), (("score",), {}))
|
|
self.check(row, ["preco_referencial_obsoleto_ou_ausente", "benchmark_obsoleto_ou_ausente",
|
|
"disponibilidade_local_precisa_revalidacao", "saldo_nao_confirmado_ou_obsoleto"])
|
|
|
|
def test_legacy_or_price_key_and_configurable_limits(self):
|
|
row = candidate("opencode/m", openrouter=None, or_price_per_token={"input": "1"})
|
|
self.check(row, ["preco_referencial_obsoleto_ou_ausente"])
|
|
row = mutated(candidate("opencode/m"), (("openrouter", "captured_at"), ago(2 * HOUR)))
|
|
self.assertEqual(model_router._staleness(row, NOW, Decimal(1), Decimal(30), TOKENS),
|
|
["preco_referencial_obsoleto_ou_ausente"])
|
|
self.assertEqual(model_router._staleness(row, NOW, Decimal(3), Decimal(30), TOKENS), [])
|
|
|
|
|
|
def cost(row, tokens=TOKENS):
|
|
return model_router._native_cost_and_headroom(row, tokens)
|
|
|
|
|
|
class NativeCostTests(unittest.TestCase):
|
|
def zen(self, *edits, billing="zen_balance"):
|
|
return mutated(candidate("opencode/m", billing=billing), *edits)
|
|
|
|
def go(self, *edits, **extra):
|
|
remaining = {"5h": "10", "weekly": "20", "monthly": "30"}
|
|
debit = {"5h": "1", "weekly": "2", "monthly": "3"}
|
|
row = candidate("opencode/g", billing="go_quota", quota_window_remaining=remaining, quota_window_debit=debit, **extra)
|
|
return mutated(row, *edits)
|
|
|
|
def test_balance_main_path_and_exact_balance(self):
|
|
amount, headroom, reason = cost(self.zen())
|
|
self.assertEqual((amount, reason), (Decimal("0.0001") + Decimal("0.00002"), None))
|
|
self.assertEqual(headroom, Decimal(2) - amount)
|
|
exact = self.zen((("native_tariff", "available_balance_usd"), str(amount)))
|
|
self.assertEqual(cost(exact), (amount, Decimal(0), None))
|
|
self.assertEqual(cost(self.zen(billing="omp_payg"))[2], None)
|
|
|
|
def test_balance_decisions(self):
|
|
cases = [
|
|
("no token profile", self.zen(), {}, (None, None, "perfil_tokens_native_nao_configurado")),
|
|
("tariff missing", self.zen((("native_tariff",), None)), TOKENS, (None, None, "tarifa_native_nao_confirmada")),
|
|
("rates not dict", self.zen((("native_tariff", "price_per_token_usd"), None)), TOKENS,
|
|
(None, None, "tarifa_native_nao_confirmada")),
|
|
("not confirmed", self.zen((("native_tariff", "account_confirmed"), False)), TOKENS,
|
|
(None, None, "tarifa_native_nao_confirmada")),
|
|
("rates incomplete", self.zen((("native_tariff", "price_per_token_usd"), {"input": "1"})), TOKENS,
|
|
(None, None, "tarifa_native_incompleta_para_perfil")),
|
|
("balance not confirmed", self.zen((("availability", "balance_confirmed"), None)), TOKENS,
|
|
(None, None, "saldo_nao_confirmado")),
|
|
("balance unknown", self.zen((("native_tariff", "available_balance_usd"), None)), TOKENS,
|
|
(None, None, "saldo_desconhecido")),
|
|
("balance negative", self.zen((("native_tariff", "available_balance_usd"), "-1")), TOKENS,
|
|
(None, None, "saldo_desconhecido")),
|
|
("balance short", self.zen((("native_tariff", "available_balance_usd"), "0.00001")), TOKENS,
|
|
(None, Decimal("0.00001"), "saldo_insuficiente")),
|
|
]
|
|
for name, row, tokens, expected in cases:
|
|
with self.subTest(name):
|
|
self.assertEqual(cost(row, tokens), expected)
|
|
|
|
def test_go_main_path_uses_monthly_debit_and_smallest_headroom(self):
|
|
self.assertEqual(cost(self.go()), (Decimal(3), Decimal(9), None))
|
|
|
|
def test_go_scalar_debit_applies_to_every_window(self):
|
|
row = self.go((("quota_window_debit",), "5"))
|
|
self.assertEqual(cost(row), (Decimal(5), Decimal(5), None))
|
|
|
|
def test_go_decisions(self):
|
|
cases = [
|
|
("no token profile", self.go(), {}, (None, None, "perfil_tokens_go_nao_configurado")),
|
|
("balance fallback", self.go((("use_balance_fallback",), True)), TOKENS,
|
|
(None, None, "fallback_go_para_saldo_zen_nao_autorizado")),
|
|
("remaining not dict", self.go((("quota_window_remaining",), None)), TOKENS,
|
|
(None, None, "quota_go_nao_confirmada")),
|
|
("debit missing", self.go((("quota_window_debit",), None)), TOKENS, (None, None, "quota_go_nao_confirmada")),
|
|
("window unknown", self.go((("quota_window_remaining", "weekly"), DELETE)), TOKENS,
|
|
(None, None, "janela_go_weekly_desconhecida")),
|
|
("debit negative", self.go((("quota_window_debit", "5h"), "-1")), TOKENS,
|
|
(None, None, "janela_go_5h_desconhecida")),
|
|
("insufficient", self.go((("quota_window_debit", "monthly"), "31")), TOKENS,
|
|
(None, None, "quota_go_monthly_insuficiente")),
|
|
]
|
|
for name, row, tokens, expected in cases:
|
|
with self.subTest(name):
|
|
self.assertEqual(cost(row, tokens), expected)
|
|
|
|
def test_claude_subscription_and_unknown_modes(self):
|
|
sub = candidate("claude/s", billing="claude_subscription", subscription_consumption_usd="1.5")
|
|
self.assertEqual(cost(sub), (Decimal("1.5"), None, None))
|
|
unmeasured = candidate("claude/s", billing="claude_subscription")
|
|
self.assertEqual(cost(unmeasured), (None, None, "consumo_assinatura_nao_mensuravel"))
|
|
self.assertEqual(cost(candidate("x/y", billing="mystery")), (None, None, "modo_de_cobranca_desconhecido"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|