mirror of
https://github.com/rennf93/roboco.git
synced 2026-08-03 07:23:24 +02:00
* [fastapi-guard] Phase 1a: gated config flags for the HTTP security layer Adds the ROBOCO_GUARD_* settings (all default-off / secure-default) for the upcoming fastapi-guard 7.2.0 hardening — guard_enabled (master switch), guard_fail_secure (fail-closed default; NAS overrides to false), guard_telemetry_enabled + guard_agent_api_key + guard_project_id (guard-agent telemetry, opt-in), guard_emergency + guard_emergency_whitelist (lockdown kill switch). Inert until consumed: nothing reads them yet, so the request path is unchanged. Foundation for v0.16.0. * [fastapi-guard] Phase 1b: security foundation module + gated wiring Add fastapi-guard 7.2.0 + guard-core 3.3.0 (bare, unpinned) and roboco/security.py: - build_security_config() from settings — behind-nginx real-IP (trusted_proxies + trust_x_forwarded_proto), HSTS/CSP headers, threat-ban + 404-sweep rules, redis-backed state, exclude_paths (/ws + health + docs), env-driven enforce_https, fail_secure (secure default), emergency lockdown, guard-agent telemetry (opt-in), passive-mode calibration switch. - guard_deco singleton (SecurityDecorator) for per-route decorators (Phase 2+). - Three custom content validators guard's WAF can't cover: prompt-injection / role-override, secret-exfil / credential-in-body, internal-SSRF. - apply_guard(app) + guarded_lifespan() wired into create_app AFTER settings. guard_passive_mode config flag added. Entirely gated by ROBOCO_GUARD_ENABLED (default off): create_app mounts nothing and returns the unchanged app when off (verified). make quality GREEN (cov 95.32%, pip-audit clean, import-linter 2/0). 12 new unit tests. * [fastapi-guard] Phase 2: critical-path decorators Apply guard decorators to the highest-value endpoints (metadata-only; enforced only when the middleware is mounted, so no-op while ROBOCO_GUARD_ENABLED is off): - provider keys (ollama/grok/self-hosted writes): strict rate_limit + max_request_size + block_clouds (no datacenter IP should touch secret writes). - settings write + release approve/reject (CEO-gated): strict rate_limit. - intake chat (prompter start/messages/events): rate_limit + max_request_size + custom_validation(prompt_injection_validator) — the prompt-facing free-text ingress gets the injection/role-override/secret-exfil content scan. make quality GREEN (cov 95.32%, contracts 2/0). App builds with guard off, decorators inert (verified). * [fastapi-guard] Phase 3: wide decorator coverage across ingress + sensitive routes Targeted-wide application (metadata-only; no-op until ROBOCO_GUARD_ENABLED). The global SecurityMiddleware already rate-limits + WAF-scans every request, so this adds the custom content validators on free-text ingress + tight limits on sensitive ops (not blanket per-route rate_limit on reads): - agent gateway do verbs (note/say/commit/dm/pitch/progress/draft_playbook/...): rate_limit + max_request_size + custom_validation(secret_exfil or prompt_injection). - a2a message/send + chat writes: rate_limit + size + prompt_injection. - optimal/RAG (kb/search, rag/query, mentor/ask, errors/decisions/standards/ learnings): prompt_injection on searches, secret_exfil on record writes; docs index → internal_ssrf. - tasks: create/update → prompt_injection; QA/doc/PM transitions → secret_exfil; CEO-gated verbs → tight rate_limit. - secretary chat → prompt_injection; research → internal_ssrf; orchestrator spawn/mutations → rate_limit; git ops + flow verbs → tight rate_limit. Pure GET/reads left to the global middleware. Applied via a Sonnet workflow, then verified: app builds with guard off (decorators inert), make quality GREEN (cov 95.37%, contracts 2/0). Decoy/honeypot-path surface deferred (needs verified guard ban-API integration — not rushed). * [fastapi-guard] Phase 5: arm the NAS composes in passive/log-only mode Arm ROBOCO_GUARD_ENABLED=true + ROBOCO_GUARD_PASSIVE_MODE=true + ROBOCO_GUARD_FAIL_SECURE=false on the two NAS composes (docker-compose.yaml + .yml). Passive = guard mounts and logs what it WOULD block but blocks nothing, so the next NAS deploy calibrates against real traffic; flip PASSIVE_MODE off after the false-positive review to enforce. fail_secure=false keeps a guard-internal error from 500ing the personal deploy. The registry (user-facing) compose is deliberately left unarmed so its published default stays conservative. Phase 4 (passive calibration) is the operational step this enables. * feat(security): Phase 3b — full-arsenal per-route guard enrichment Stack the applicable guard decorators per surface instead of the minimal rate_limit/max_request_size/custom_validation triad: content_type_filter on every JSON-body write, honeypot_detection form-traps on human-facing POSTs, block_clouds on key-writes + CEO release ops, behavior_analysis runaway-rate rules on the agent flow/do verbs, suspicious_detection + usage_monitor on the sensitive surfaces. Nine distinct decorators now applied thoughtfully per endpoint. All metadata-only — no-op while ROBOCO_GUARD_ENABLED is off. * fix(a2a): permit PR reviewer to deliver gate verdicts to the owning PM can_a2a_direct had no pr_reviewer rule, so a reviewer (team=board, or a cell team) fell through to the cell-member path and was cross-cell-denied when the in-path gate delivered a pr_fail change-request to main-pm (or a cross-cell cell-pm): "Cannot A2A main-pm ... Ask None to coordinate with None". The delivery is best-effort, so pr_fail still transitioned but the verdict never reached the owning PM — the blind-re-submit signal-gap the pr_fail fix closes. Add an explicit pr_reviewer handler: it may A2A only cell_pm / main_pm (its sole comms surface — everything else it posts on the PR itself), with a matching route hint. The cell reviewers kept same-team access by coincidence; this scopes every reviewer to PM-only, the correct model, with no other A2A caller affected. Refresh uv.lock to the current resolution. * feat(models): adopt Claude Sonnet 5 as the sonnet tier Point the 'sonnet' alias at claude-sonnet-5 (MODEL_MAP) and give pr_reviewer its own opus tier in ROLE_MODEL_MAP — it was falling through to the sonnet default, and the role gates untrusted external/fork PRs plus root→master, which warrants opus. Price claude-sonnet-5 at the promotional 33% off Sonnet 4.6 ($2.01 / $10.05, cache 0.201 / 0.5025) through 2026-08-31 via a dedicated pricing fragment that beats the bare 'sonnet' alias; revert to full rate when the promo ends. Bare 'sonnet' stays full-rate as a conservative fallback (prod prices the resolved claude-sonnet-5 id from the transcript). Update the model docs and the billing / usage / manifest / spawn tests. * feat(security): calibrate the guard WAF for RoboCo traffic + document the layer The first end-to-end run of the fastapi-guard layer showed active enforcement would block ~50% of legitimate agent traffic — RoboCo request bodies are code, SQL, diffs, file paths, HTML, and URLs, which the stock signature WAF reads as attacks. build_security_config now excludes RoboCo's free-text top-level body fields (derived from the real request models, including the free-form container fields whose nested prose is stringified and scanned) from WAF scanning, dropping the active-mode false-positive rate to zero while keeping the WAF on every non-excluded (id/enum/slug/branch) field and leaving the prompt-injection / secret-exfil / internal-SSRF validators — which run independently of the exclusion — fully in force. enable_penetration_detection is made explicit. Only excluded_detection_body_fields is reliable on guard 7.2.1: the per-route categories knob is bypassed for JSON bodies, and the body scanner excludes top-level keys only (scanning str(value) of every non-excluded field), so free-form container fields must be excluded wholesale. Adds tests/unit/test_security_middleware.py — the first end-to-end exercise of the middleware (mounts it, drives guard's lifespan, fires real requests): proves passive mode is log-only, active mode does not false-positive on realistic agent payloads, threats are still blocked inside excluded fields, and the WAF still fires on non-excluded fields. Docs: CHANGELOG (Unreleased); a user-facing Optional-subsystems page + nav + env reference for the HTTP security layer; the agent-facing RAG corpus (what it is + why a request could be blocked); and the roboco mapping (api-core-websocket / deployment-tooling / _complete_map). * feat(security): Surface N — scanner honeytrap (guard /api auto-ban + nginx edge-drop) Turns scanner probes against the scanner, in two layers matched to where traffic lands. Behind nginx only /api, /ws, /health, /ready reach the orchestrator, so guard can only see (and ban) scanner probes on those paths; the classic root probes (/.env, /wp-login.php, /phpmyadmin, /.git/config) hit the panel. So: - build_security_config's threat_ban_config gains recon / sensitive_file / cms_probing categories. A scanner probing those fingerprints on an /api path is detected on the URL-path scan; repeated probes from one IP trip an adaptive per-IP auto-ban (redis-backed, 24h). Only bans in active mode (passive logs the recon hit) and needs redis (the 24h ban exceeds the in-memory cap). The spec's decoy-route file is redundant — the WAF url-path scan bans regardless of a registered route — so it is intentionally omitted. - docker/nginx.conf drops the classic root scanner paths at the edge with 444 (connection closed, no response) before they reach the panel, anchored to known scanner fingerprints so /.well-known and every real panel/API route are untouched. Always on, independent of ROBOCO_GUARD_ENABLED. Tests: 2 unit (the exclusion set + the scanner-ban categories are present) and 2 integration (a decoy path is blocked in active mode, passes in passive). The nginx regex was validated against 15 scanner + 19 legit paths (0 false positives). Docs: CHANGELOG, the HTTP-security page, the roboco mapping, and the agent-facing RAG corpus. * Token optimization — per-role observability, compute policy, spawn preflight (#291) * test(models): lock the sonnet→claude-sonnet-5 MODEL_MAP invariant * feat(usage): surface cache tokens + cache_hit_rate in usage breakdowns * feat(usage): add per-role usage breakdown endpoint * feat(usage): add spawn-waste signal (per-role unproductive rate + respawn strikes) * feat(panel): surface per-role cost/cache + spawn-waste on the metrics page * feat(routing): Phase 2 per-role compute policy — qa→haiku, main_pm→sonnet, per-role effort env mechanism (default-inert) * feat(orchestrator): Phase 3 flag-gated spawn preflight — refuse non-gateway delivery roles (respawn-forever guard) * chore(compose): arm ROBOCO_SPAWN_PREFLIGHT_ENABLED on the NAS composes * docs: per-role usage observability, per-role compute policy, and spawn preflight --------- Co-authored-by: Renn F <rennf93@users.noreply.github.com> * fix(panel): pin outputFileTracingRoot so the standalone build isn't broken by stray lockfiles * feat(routing): populate ROLE_EFFORT_MAP + wire the verified --effort flag (cell_pm/board/auditor to medium) * feat(gateway): omit empty context_briefing sections (Phase 4 payload compaction) * refactor(orchestrator): extract spawn chokepoint guards to restore xenon rank B on spawn_agent --------- Co-authored-by: Renn F <rennf93@users.noreply.github.com>
500 lines
18 KiB
Python
500 lines
18 KiB
Python
"""
|
|
Unit tests for roboco.billing.pricing — calculate_cost().
|
|
|
|
Covers:
|
|
- Each model tier (opus, sonnet, haiku) with all 4 token types.
|
|
- Unknown model name returns 0.0 without raising.
|
|
- Empty model string returns 0.0 without raising.
|
|
- Substring match correctness: longer fragment wins
|
|
(e.g. 'claude-sonnet-4-6' matches 'claude-sonnet-4' not bare 'sonnet';
|
|
'claude-sonnet-5' matches its own promo entry).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from roboco.billing.pricing import (
|
|
CostResult,
|
|
_is_anthropic_model,
|
|
calculate_cost,
|
|
calculate_cost_result,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Named constants (ruff PLR2004: magic values in comparisons must be named).
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# Token counts
|
|
_M = 1_000_000 # 1 million tokens
|
|
|
|
# Pricing — per-1M USD, matches the _PRICING table in pricing.py
|
|
_OPUS_INPUT = 5.00
|
|
_OPUS_OUTPUT = 25.00
|
|
_OPUS_CACHE_READ = 0.50
|
|
_OPUS_CACHE_WRITE = 6.25
|
|
|
|
_SONNET_INPUT = 3.00
|
|
_SONNET_OUTPUT = 15.00
|
|
_SONNET_CACHE_READ = 0.30
|
|
_SONNET_CACHE_WRITE = 0.75
|
|
|
|
# Sonnet 5 — promotional pricing (33% off Sonnet 4.6, pay 67%) through 2026-08-31
|
|
_SONNET5_INPUT = 2.01
|
|
_SONNET5_OUTPUT = 10.05
|
|
_SONNET5_CACHE_READ = 0.201
|
|
_SONNET5_CACHE_WRITE = 0.5025
|
|
|
|
_HAIKU_INPUT = 1.00
|
|
_HAIKU_OUTPUT = 5.00
|
|
_HAIKU_CACHE_READ = 0.10
|
|
_HAIKU_CACHE_WRITE = 1.25
|
|
|
|
_HAIKU3_INPUT = 0.25 # claude-haiku-3 is cheaper than haiku-3-5 / haiku-4
|
|
|
|
# xAI Grok — priced non-Anthropic (per the xAI API)
|
|
_GROK_INPUT = 1.00
|
|
_GROK_OUTPUT = 2.00
|
|
_GROK_CACHE_READ = 0.20
|
|
_GROK_CACHE_WRITE = 1.00
|
|
|
|
# Tolerance for floating-point comparisons
|
|
_TOL = 1e-4
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Opus tier
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestOpusTier:
|
|
"""claude-opus-4 family pricing."""
|
|
|
|
def test_input_only(self) -> None:
|
|
cost = calculate_cost("claude-opus-4-5", tokens_input=_M, tokens_output=0)
|
|
assert abs(cost - _OPUS_INPUT) < _TOL
|
|
|
|
def test_output_only(self) -> None:
|
|
cost = calculate_cost("claude-opus-4-5", tokens_input=0, tokens_output=_M)
|
|
assert abs(cost - _OPUS_OUTPUT) < _TOL
|
|
|
|
def test_cache_read_only(self) -> None:
|
|
cost = calculate_cost(
|
|
"claude-opus-4-5",
|
|
tokens_input=0,
|
|
tokens_output=0,
|
|
tokens_cache_read=_M,
|
|
)
|
|
assert abs(cost - _OPUS_CACHE_READ) < _TOL
|
|
|
|
def test_cache_write_only(self) -> None:
|
|
cost = calculate_cost(
|
|
"claude-opus-4-5",
|
|
tokens_input=0,
|
|
tokens_output=0,
|
|
tokens_cache_write=_M,
|
|
)
|
|
assert abs(cost - _OPUS_CACHE_WRITE) < _TOL
|
|
|
|
def test_all_token_types(self) -> None:
|
|
cost = calculate_cost(
|
|
"claude-opus-4-5",
|
|
tokens_input=_M,
|
|
tokens_output=_M,
|
|
tokens_cache_read=_M,
|
|
tokens_cache_write=_M,
|
|
)
|
|
expected = _OPUS_INPUT + _OPUS_OUTPUT + _OPUS_CACHE_READ + _OPUS_CACHE_WRITE
|
|
assert abs(cost - expected) < _TOL
|
|
|
|
def test_short_alias(self) -> None:
|
|
"""Bare 'opus' alias resolves to the opus tier."""
|
|
cost = calculate_cost("opus", tokens_input=_M, tokens_output=0)
|
|
assert abs(cost - _OPUS_INPUT) < _TOL
|
|
|
|
def test_returns_float(self) -> None:
|
|
cost = calculate_cost("claude-opus-4", tokens_input=100, tokens_output=50)
|
|
assert isinstance(cost, float)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Sonnet tier
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSonnetTier:
|
|
"""claude-sonnet-4 family pricing (full rate — pre-promo / historical)."""
|
|
|
|
def test_input_only(self) -> None:
|
|
cost = calculate_cost("claude-sonnet-4-6", tokens_input=_M, tokens_output=0)
|
|
assert abs(cost - _SONNET_INPUT) < _TOL
|
|
|
|
def test_output_only(self) -> None:
|
|
cost = calculate_cost("claude-sonnet-4-6", tokens_input=0, tokens_output=_M)
|
|
assert abs(cost - _SONNET_OUTPUT) < _TOL
|
|
|
|
def test_cache_read_only(self) -> None:
|
|
cost = calculate_cost(
|
|
"claude-sonnet-4-6",
|
|
tokens_input=0,
|
|
tokens_output=0,
|
|
tokens_cache_read=_M,
|
|
)
|
|
assert abs(cost - _SONNET_CACHE_READ) < _TOL
|
|
|
|
def test_cache_write_only(self) -> None:
|
|
cost = calculate_cost(
|
|
"claude-sonnet-4-6",
|
|
tokens_input=0,
|
|
tokens_output=0,
|
|
tokens_cache_write=_M,
|
|
)
|
|
assert abs(cost - _SONNET_CACHE_WRITE) < _TOL
|
|
|
|
def test_all_token_types(self) -> None:
|
|
cost = calculate_cost(
|
|
"claude-sonnet-4-6",
|
|
tokens_input=_M,
|
|
tokens_output=_M,
|
|
tokens_cache_read=_M,
|
|
tokens_cache_write=_M,
|
|
)
|
|
expected = (
|
|
_SONNET_INPUT + _SONNET_OUTPUT + _SONNET_CACHE_READ + _SONNET_CACHE_WRITE
|
|
)
|
|
assert abs(cost - expected) < _TOL
|
|
|
|
def test_short_alias(self) -> None:
|
|
"""Bare 'sonnet' alias resolves to the sonnet tier."""
|
|
cost = calculate_cost("sonnet", tokens_input=_M, tokens_output=0)
|
|
assert abs(cost - _SONNET_INPUT) < _TOL
|
|
|
|
def test_35_variant(self) -> None:
|
|
"""claude-3-5-sonnet resolves to sonnet tier."""
|
|
cost = calculate_cost(
|
|
"claude-3-5-sonnet-20241022", tokens_input=_M, tokens_output=0
|
|
)
|
|
assert abs(cost - _SONNET_INPUT) < _TOL
|
|
|
|
|
|
class TestSonnet5PromoTier:
|
|
"""claude-sonnet-5 promotional pricing — 33% off Sonnet 4.6 (through
|
|
2026-08-31). A dedicated table entry wins over the bare 'sonnet' fragment."""
|
|
|
|
def test_input_only(self) -> None:
|
|
cost = calculate_cost("claude-sonnet-5", tokens_input=_M, tokens_output=0)
|
|
assert abs(cost - _SONNET5_INPUT) < _TOL
|
|
|
|
def test_output_only(self) -> None:
|
|
cost = calculate_cost("claude-sonnet-5", tokens_input=0, tokens_output=_M)
|
|
assert abs(cost - _SONNET5_OUTPUT) < _TOL
|
|
|
|
def test_cache_read_only(self) -> None:
|
|
cost = calculate_cost(
|
|
"claude-sonnet-5", tokens_input=0, tokens_output=0, tokens_cache_read=_M
|
|
)
|
|
assert abs(cost - _SONNET5_CACHE_READ) < _TOL
|
|
|
|
def test_cache_write_only(self) -> None:
|
|
cost = calculate_cost(
|
|
"claude-sonnet-5", tokens_input=0, tokens_output=0, tokens_cache_write=_M
|
|
)
|
|
assert abs(cost - _SONNET5_CACHE_WRITE) < _TOL
|
|
|
|
def test_all_token_types(self) -> None:
|
|
cost = calculate_cost(
|
|
"claude-sonnet-5",
|
|
tokens_input=_M,
|
|
tokens_output=_M,
|
|
tokens_cache_read=_M,
|
|
tokens_cache_write=_M,
|
|
)
|
|
expected = (
|
|
_SONNET5_INPUT
|
|
+ _SONNET5_OUTPUT
|
|
+ _SONNET5_CACHE_READ
|
|
+ _SONNET5_CACHE_WRITE
|
|
)
|
|
assert abs(cost - expected) < _TOL
|
|
|
|
def test_cheaper_than_sonnet4(self) -> None:
|
|
"""The promo must actually be cheaper than full Sonnet 4.6."""
|
|
five = calculate_cost("claude-sonnet-5", tokens_input=_M, tokens_output=_M)
|
|
four = calculate_cost("claude-sonnet-4-6", tokens_input=_M, tokens_output=_M)
|
|
assert five < four
|
|
|
|
def test_dated_variant_matches_promo(self) -> None:
|
|
"""A dated 'claude-sonnet-5-*' id still resolves to the promo entry."""
|
|
cost = calculate_cost(
|
|
"claude-sonnet-5-20260930", tokens_input=_M, tokens_output=0
|
|
)
|
|
assert abs(cost - _SONNET5_INPUT) < _TOL
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Haiku tier
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestHaikuTier:
|
|
"""claude-haiku family pricing."""
|
|
|
|
def test_input_only(self) -> None:
|
|
cost = calculate_cost("claude-haiku-4-5", tokens_input=_M, tokens_output=0)
|
|
assert abs(cost - _HAIKU_INPUT) < _TOL
|
|
|
|
def test_output_only(self) -> None:
|
|
cost = calculate_cost("claude-haiku-4-5", tokens_input=0, tokens_output=_M)
|
|
assert abs(cost - _HAIKU_OUTPUT) < _TOL
|
|
|
|
def test_cache_read_only(self) -> None:
|
|
cost = calculate_cost(
|
|
"claude-haiku-4-5",
|
|
tokens_input=0,
|
|
tokens_output=0,
|
|
tokens_cache_read=_M,
|
|
)
|
|
assert abs(cost - _HAIKU_CACHE_READ) < _TOL
|
|
|
|
def test_cache_write_only(self) -> None:
|
|
cost = calculate_cost(
|
|
"claude-haiku-4-5",
|
|
tokens_input=0,
|
|
tokens_output=0,
|
|
tokens_cache_write=_M,
|
|
)
|
|
assert abs(cost - _HAIKU_CACHE_WRITE) < _TOL
|
|
|
|
def test_all_token_types(self) -> None:
|
|
cost = calculate_cost(
|
|
"claude-haiku-4-5",
|
|
tokens_input=_M,
|
|
tokens_output=_M,
|
|
tokens_cache_read=_M,
|
|
tokens_cache_write=_M,
|
|
)
|
|
expected = _HAIKU_INPUT + _HAIKU_OUTPUT + _HAIKU_CACHE_READ + _HAIKU_CACHE_WRITE
|
|
assert abs(cost - expected) < _TOL
|
|
|
|
def test_short_alias(self) -> None:
|
|
"""Bare 'haiku' alias resolves to the haiku tier."""
|
|
cost = calculate_cost("haiku", tokens_input=_M, tokens_output=0)
|
|
assert abs(cost - _HAIKU_INPUT) < _TOL
|
|
|
|
def test_haiku3_variant(self) -> None:
|
|
"""claude-haiku-3 has lower pricing than haiku-3-5."""
|
|
cost = calculate_cost("claude-haiku-3", tokens_input=_M, tokens_output=0)
|
|
assert abs(cost - _HAIKU3_INPUT) < _TOL
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Grok tier (xAI — priced non-Anthropic)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestGrokTier:
|
|
"""grok-build-0.1 pricing — a non-Anthropic model that IS billed per token."""
|
|
|
|
def test_input_only(self) -> None:
|
|
cost = calculate_cost("grok-build-0.1", tokens_input=_M, tokens_output=0)
|
|
assert abs(cost - _GROK_INPUT) < _TOL
|
|
|
|
def test_output_only(self) -> None:
|
|
cost = calculate_cost("grok-build-0.1", tokens_input=0, tokens_output=_M)
|
|
assert abs(cost - _GROK_OUTPUT) < _TOL
|
|
|
|
def test_cached_input(self) -> None:
|
|
cost = calculate_cost(
|
|
"grok-build-0.1", tokens_input=0, tokens_output=0, tokens_cache_read=_M
|
|
)
|
|
assert abs(cost - _GROK_CACHE_READ) < _TOL
|
|
|
|
def test_all_token_types(self) -> None:
|
|
cost = calculate_cost(
|
|
"grok-build-0.1",
|
|
tokens_input=_M,
|
|
tokens_output=_M,
|
|
tokens_cache_read=_M,
|
|
tokens_cache_write=_M,
|
|
)
|
|
expected = _GROK_INPUT + _GROK_OUTPUT + _GROK_CACHE_READ + _GROK_CACHE_WRITE
|
|
assert abs(cost - expected) < _TOL
|
|
|
|
def test_grok_is_not_treated_as_anthropic(self) -> None:
|
|
"""Priced, but not an Anthropic model (no warn-on-unpriced path)."""
|
|
assert _is_anthropic_model("grok-build-0.1") is False
|
|
# Still resolves to a real (non-zero) per-token cost.
|
|
assert calculate_cost("grok-build-0.1", tokens_input=_M, tokens_output=0) > 0.0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Unknown / edge cases — must return 0.0 without raising
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestUnknownModels:
|
|
def test_unknown_model_name_returns_zero(self) -> None:
|
|
cost = calculate_cost("gpt-4o", tokens_input=_M, tokens_output=_M)
|
|
assert cost == 0.0
|
|
|
|
def test_empty_string_returns_zero(self) -> None:
|
|
cost = calculate_cost("", tokens_input=_M, tokens_output=_M)
|
|
assert cost == 0.0
|
|
|
|
def test_gibberish_returns_zero(self) -> None:
|
|
cost = calculate_cost(
|
|
"totally-unknown-model-xyz", tokens_input=100, tokens_output=100
|
|
)
|
|
assert cost == 0.0
|
|
|
|
def test_zero_tokens_with_unknown_model_returns_zero(self) -> None:
|
|
cost = calculate_cost("unknown", tokens_input=0, tokens_output=0)
|
|
assert cost == 0.0
|
|
|
|
def test_does_not_raise_on_unknown_model(self) -> None:
|
|
"""Must not raise regardless of token counts."""
|
|
try:
|
|
calculate_cost(
|
|
"not-a-claude-model",
|
|
tokens_input=999_999,
|
|
tokens_output=999_999,
|
|
)
|
|
except Exception as exc:
|
|
pytest.fail(f"calculate_cost raised unexpectedly: {exc}")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Substring match correctness
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# Named constants for the comparison floor/ceiling used in these tests.
|
|
_ZERO_COST = 0.0
|
|
_SONNET_CHEAPER_THAN_OPUS = True # structural assertion in the test below
|
|
|
|
|
|
class TestSubstringMatchPriority:
|
|
def test_claude_sonnet_4_resolves_non_zero(self) -> None:
|
|
"""'claude-sonnet-4-6' must find a match (non-zero cost)."""
|
|
cost = calculate_cost("claude-sonnet-4-6", tokens_input=_M, tokens_output=0)
|
|
assert cost > _ZERO_COST
|
|
|
|
def test_haiku3_cheaper_than_haiku4(self) -> None:
|
|
"""claude-haiku-3 is cheaper than claude-haiku-4 — longest-match wins."""
|
|
haiku3_cost = calculate_cost("claude-haiku-3", tokens_input=_M, tokens_output=0)
|
|
haiku4_cost = calculate_cost("claude-haiku-4", tokens_input=_M, tokens_output=0)
|
|
# haiku-3 ($0.25/1M) < haiku-4 ($1.00/1M)
|
|
assert haiku3_cost < haiku4_cost
|
|
|
|
def test_non_claude_model_returns_zero(self) -> None:
|
|
"""A random non-Claude model must not match any Claude pricing entry."""
|
|
non_opus_cost = calculate_cost("llama-3-70b", tokens_input=_M, tokens_output=0)
|
|
assert non_opus_cost == _ZERO_COST
|
|
|
|
def test_opus_model_non_zero(self) -> None:
|
|
"""Claude opus model resolves to non-zero cost."""
|
|
opus_cost = calculate_cost("claude-opus-4", tokens_input=_M, tokens_output=0)
|
|
assert opus_cost > _ZERO_COST
|
|
|
|
def test_zero_tokens_returns_zero_for_known_model(self) -> None:
|
|
"""Known model with 0 tokens has 0 cost."""
|
|
cost = calculate_cost("claude-opus-4", tokens_input=0, tokens_output=0)
|
|
assert cost == _ZERO_COST
|
|
|
|
def test_case_insensitive_matching(self) -> None:
|
|
"""Model name matching is case-insensitive."""
|
|
lower_cost = calculate_cost(
|
|
"claude-sonnet-5", tokens_input=1000, tokens_output=1000
|
|
)
|
|
upper_cost = calculate_cost(
|
|
"CLAUDE-SONNET-5", tokens_input=1000, tokens_output=1000
|
|
)
|
|
assert lower_cost == upper_cost
|
|
assert lower_cost > _ZERO_COST
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Provider awareness — non-Anthropic models have no per-token cost
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestProviderAwareness:
|
|
"""Non-Anthropic models (local Ollama / Ollama Cloud) cost 0.0 per token."""
|
|
|
|
def test_ollama_prefixed_model_returns_zero(self) -> None:
|
|
"""Self-hosted Ollama models (``ollama/`` prefix) have no API cost."""
|
|
cost = calculate_cost("ollama/llama3", tokens_input=_M, tokens_output=_M)
|
|
assert cost == _ZERO_COST
|
|
|
|
def test_ollama_cloud_model_returns_zero(self) -> None:
|
|
"""Ollama Cloud (``:cloud`` tag) is subscription-billed, not per token."""
|
|
cost = calculate_cost("glm-5.2:cloud", tokens_input=_M, tokens_output=_M)
|
|
assert cost == _ZERO_COST
|
|
|
|
def test_bare_local_model_returns_zero(self) -> None:
|
|
"""A bare local embedding model has no per-token cost."""
|
|
cost = calculate_cost("qwen3-embedding:0.6b", tokens_input=_M, tokens_output=0)
|
|
assert cost == _ZERO_COST
|
|
|
|
def test_is_anthropic_model_true_for_claude_names(self) -> None:
|
|
for name in ("claude-opus-4-6", "claude-fable-5", "opus", "sonnet", "haiku"):
|
|
assert _is_anthropic_model(name) is True, name
|
|
|
|
def test_is_anthropic_model_false_for_non_claude_names(self) -> None:
|
|
for name in ("ollama/llama3", "glm-5.2:cloud", "qwen3-embedding", "gpt-4o"):
|
|
assert _is_anthropic_model(name) is False, name
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Structured cost result — distinguish unpriced Anthropic from genuinely-free
|
|
# (#65). ``calculate_cost`` keeps returning a plain float for existing callers;
|
|
# ``calculate_cost_result`` returns a ``CostResult`` so a caller can tell real
|
|
# spend we failed to price ($0, unpriced=True) apart from local inference
|
|
# ($0, unpriced=False).
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestCostResult:
|
|
def test_unpriced_anthropic_model_is_flagged(self) -> None:
|
|
result = calculate_cost_result(
|
|
"claude-brand-new-unpriced", tokens_input=_M, tokens_output=0
|
|
)
|
|
assert isinstance(result, CostResult)
|
|
assert result.cost_usd == 0.0
|
|
assert result.unpriced is True
|
|
assert result.is_anthropic is True
|
|
|
|
def test_free_non_anthropic_model_is_not_unpriced(self) -> None:
|
|
result = calculate_cost_result(
|
|
"ollama/qwen3-embedding", tokens_input=_M, tokens_output=0
|
|
)
|
|
assert result.cost_usd == 0.0
|
|
assert result.unpriced is False
|
|
assert result.is_anthropic is False
|
|
|
|
def test_priced_anthropic_model_is_not_unpriced(self) -> None:
|
|
result = calculate_cost_result(
|
|
"claude-sonnet-5", tokens_input=_M, tokens_output=0
|
|
)
|
|
assert result.cost_usd > 0.0
|
|
assert result.unpriced is False
|
|
|
|
def test_priced_non_anthropic_grok_is_not_unpriced(self) -> None:
|
|
result = calculate_cost_result(
|
|
"grok-build-0.1", tokens_input=_M, tokens_output=0
|
|
)
|
|
assert result.cost_usd > 0.0
|
|
assert result.unpriced is False
|
|
assert result.is_anthropic is False
|
|
|
|
def test_calculate_cost_matches_structured_cost_usd(self) -> None:
|
|
model = "claude-opus-4-6"
|
|
assert (
|
|
calculate_cost(model, tokens_input=_M, tokens_output=_M)
|
|
== calculate_cost_result(model, tokens_input=_M, tokens_output=_M).cost_usd
|
|
)
|
|
|
|
def test_empty_model_is_not_unpriced(self) -> None:
|
|
# An empty model name is a caller bug, not an unpriced-Anthropic miss.
|
|
result = calculate_cost_result("", tokens_input=_M, tokens_output=0)
|
|
assert result.cost_usd == 0.0
|
|
assert result.unpriced is False
|