diff --git a/api/providers.py b/api/providers.py index 86a31155..b3ddb4d1 100644 --- a/api/providers.py +++ b/api/providers.py @@ -1441,6 +1441,255 @@ def _provider_is_oauth(provider_id: str) -> bool: return provider_id in _OAUTH_PROVIDERS +# ── OpenRouter cost-history snapshot helpers (#692) ────────────────────────── + +_COST_SNAPSHOTS_DIR_NAME = "cost-snapshots" +_COST_SNAPSHOT_MAX_DAYS = 365 # hard cap to prevent unbounded growth + + +def _cost_snapshots_dir() -> Path: + """Return the directory for cost-snapshot JSON files. + + Uses the Hermes home directory (profile-aware) so snapshots are + isolated per profile, matching the existing STATE_DIR convention. + """ + return _get_hermes_home() / _COST_SNAPSHOTS_DIR_NAME + + +def _fetch_openrouter_key_usage(api_key: str) -> dict[str, Any] | None: + """Fetch current usage/limit from the OpenRouter ``/auth/key`` endpoint. + + Returns a dict with ``usage``, ``limit``, ``label`` on success, or + ``None`` on any failure. Never raises; callers handle the None case. + """ + req = urllib.request.Request( + _OPENROUTER_KEY_URL, + headers={ + "Authorization": f"Bearer {api_key}", + "Accept": "application/json", + }, + ) + try: + with urllib.request.urlopen(req, timeout=_PROVIDER_QUOTA_TIMEOUT_SECONDS) as resp: + raw = resp.read() + payload = json.loads(raw.decode("utf-8") if isinstance(raw, (bytes, bytearray)) else raw) + sanitized = _sanitize_openrouter_quota(payload) + label = None + if isinstance(payload, dict): + data = payload.get("data", payload) + if isinstance(data, dict): + label = str(data.get("label") or "").strip() or None + return { + "usage": sanitized.get("usage"), + "limit": sanitized.get("limit"), + "label": label, + } + except Exception: + logger.debug("OpenRouter key usage fetch failed for cost-history", exc_info=True) + return None + + +def _read_cost_snapshots(provider: str) -> list[dict[str, Any]]: + """Read persisted daily snapshots for *provider* from disk. + + Returns a list of ``{date, used, limit}`` dicts sorted by date + ascending. Returns an empty list if the file does not exist or is + corrupt. + """ + path = _cost_snapshots_dir() / f"{provider}.json" + if not path.exists(): + return [] + try: + raw = path.read_text(encoding="utf-8") + data = json.loads(raw) + except (OSError, json.JSONDecodeError, ValueError): + return [] + if not isinstance(data, dict): + return [] + snapshots = data.get("snapshots") + if not isinstance(snapshots, list): + return [] + # Validate and sort + valid = [] + for entry in snapshots: + if not isinstance(entry, dict): + continue + date = str(entry.get("date") or "").strip() + if not date: + continue + valid.append({ + "date": date, + "used": _quota_number(entry.get("used")), + "limit": _quota_number(entry.get("limit")), + }) + valid.sort(key=lambda e: e["date"]) + return valid + + +def _write_cost_snapshots(provider: str, snapshots: list[dict[str, Any]]) -> None: + """Persist daily snapshots for *provider* to disk atomically.""" + snap_dir = _cost_snapshots_dir() + snap_dir.mkdir(parents=True, exist_ok=True) + path = snap_dir / f"{provider}.json" + payload = {"provider": provider, "snapshots": snapshots} + body = json.dumps(payload, ensure_ascii=False, indent=2) + import tempfile as _tempfile + _tmp_fd, _tmp_path = _tempfile.mkstemp( + dir=str(snap_dir), prefix=f".{provider}_", suffix=".tmp" + ) + try: + with os.fdopen(_tmp_fd, "w", encoding="utf-8") as _f: + _f.write(body) + _f.flush() + os.fsync(_f.fileno()) + os.replace(_tmp_path, path) + except BaseException: + try: + os.unlink(_tmp_path) + except OSError: + pass + raise + + +def _append_cost_snapshot(provider: str, usage: int | float | None, limit: int | float | None) -> list[dict[str, Any]]: + """Append today's snapshot and return the updated list. + + If a snapshot for today already exists it is updated in-place so + repeated calls within the same day are idempotent. + """ + today = datetime.now(timezone.utc).strftime("%Y-%m-%d") + snapshots = _read_cost_snapshots(provider) + # Update or append today's entry + updated = False + for entry in snapshots: + if entry["date"] == today: + entry["used"] = usage + entry["limit"] = limit + updated = True + break + if not updated: + snapshots.append({"date": today, "used": usage, "limit": limit}) + snapshots.sort(key=lambda e: e["date"]) + # Cap to _COST_SNAPSHOT_MAX_DAYS entries (keep most recent) + if len(snapshots) > _COST_SNAPSHOT_MAX_DAYS: + snapshots = snapshots[-_COST_SNAPSHOT_MAX_DAYS:] + _write_cost_snapshots(provider, snapshots) + return snapshots + + +def _compute_deltas(snapshots: list[dict[str, Any]], window_days: int) -> list[dict[str, Any]]: + """Compute daily deltas from cumulative usage snapshots. + + Each snapshot carries cumulative ``used``; the delta for a day is + the difference between that day's cumulative value and the previous + day's. The oldest day in the window has ``delta=None`` (no + previous baseline). + """ + # Take only the last *window_days* entries + window = snapshots[-window_days:] if len(snapshots) > window_days else list(snapshots) + result: list[dict[str, Any]] = [] + for i, entry in enumerate(window): + delta = None + if i > 0 and entry.get("used") is not None and window[i - 1].get("used") is not None: + delta = float(entry["used"]) - float(window[i - 1]["used"]) + # Rounding: avoid -0.0 and tiny floating-point noise + if abs(delta) < 1e-9: + delta = 0.0 + else: + delta = round(delta, 6) + result.append({ + "date": entry["date"], + "used": entry.get("used"), + "delta": delta, + }) + return result + + +def get_provider_cost_history(provider_id: str | None = None, days: int = 7) -> dict[str, Any]: + """Return daily cost-history snapshots with deltas for a provider. + + Currently only ``openrouter`` is supported. On each call the + endpoint fetches the current cumulative usage from the OpenRouter + ``/auth/key`` endpoint, appends/updates today's snapshot, and + returns the last *days* snapshots with per-day deltas. + + Returns a dict matching the existing API style (``ok``, ``provider``, + ``status``, ``message``, …). + """ + provider = (provider_id or "").strip().lower() + if not provider: + return { + "ok": False, + "provider": None, + "status": "missing_provider", + "message": "Provider parameter is required. Use ?provider=openrouter", + } + + if provider != "openrouter": + display_name = _PROVIDER_DISPLAY.get(provider, provider.replace("-", " ").title()) + return { + "ok": False, + "provider": provider, + "display_name": display_name, + "supported": False, + "status": "unsupported", + "message": f"Cost history is not available for {display_name}. Only openrouter is supported in this release.", + } + + display_name = _PROVIDER_DISPLAY.get("openrouter", "OpenRouter") + api_key = _get_provider_api_key("openrouter") + if not api_key: + return { + "ok": False, + "provider": "openrouter", + "display_name": display_name, + "supported": True, + "status": "no_key", + "message": "OpenRouter cost history needs an OPENROUTER_API_KEY configured on the server.", + } + + # Fetch current cumulative usage from OpenRouter + key_info = _fetch_openrouter_key_usage(api_key) + if key_info is None: + # Upstream failure — still return any previously persisted snapshots + # so the chart degrades gracefully instead of going blank. + snapshots = _read_cost_snapshots("openrouter") + deltas = _compute_deltas(snapshots, days) + return { + "ok": False, + "provider": "openrouter", + "display_name": display_name, + "supported": True, + "status": "unavailable", + "window_days": days, + "snapshots": deltas, + "limit": None, + "label": None, + "message": "OpenRouter cost history is temporarily unavailable. Showing last known data.", + } + + # Persist today's snapshot + try: + snapshots = _append_cost_snapshot("openrouter", key_info["usage"], key_info["limit"]) + except Exception: + logger.debug("Failed to persist cost snapshot for openrouter", exc_info=True) + snapshots = _read_cost_snapshots("openrouter") + + deltas = _compute_deltas(snapshots, days) + return { + "ok": True, + "provider": "openrouter", + "display_name": display_name, + "supported": True, + "status": "available", + "window_days": days, + "snapshots": deltas, + "limit": key_info.get("limit"), + "label": key_info.get("label") or "OpenRouter credits", + "message": "OpenRouter cost history loaded.", + } + + # SECTION: Public API diff --git a/api/routes.py b/api/routes.py index 3ad444e9..dd25e76b 100644 --- a/api/routes.py +++ b/api/routes.py @@ -2033,7 +2033,7 @@ from api.run_journal import ( read_run_events, stale_interrupted_event, ) -from api.providers import get_providers, get_provider_quota, set_provider_key, remove_provider_key +from api.providers import get_providers, get_provider_quota, get_provider_cost_history, set_provider_key, remove_provider_key from api.onboarding import ( apply_onboarding_setup, get_onboarding_status, @@ -3323,6 +3323,16 @@ def handle_get(handler, parsed) -> bool: refresh = (query.get("refresh", [""])[0] or "").strip().lower() in {"1", "true", "yes", "on"} return j(handler, get_provider_quota(provider_id, refresh=refresh)) + if parsed.path == "/api/provider/cost-history": + query = parse_qs(parsed.query) + provider_id = (query.get("provider", [""])[0] or None) + days_raw = (query.get("days", ["7"])[0] or "7").strip() + try: + days = max(1, min(int(days_raw), 365)) + except (ValueError, TypeError): + days = 7 + return j(handler, get_provider_cost_history(provider_id, days)) + if parsed.path == "/api/settings": settings = load_settings() # Never expose the stored password hash to clients diff --git a/tests/test_provider_cost_history.py b/tests/test_provider_cost_history.py new file mode 100644 index 00000000..7ddf96c9 --- /dev/null +++ b/tests/test_provider_cost_history.py @@ -0,0 +1,484 @@ +"""Regression coverage for OpenRouter cost-history endpoint (#692). + +Tests cover: + - Happy-path snapshot append and delta computation + - Missing credentials (no_key) + - Unsupported provider (non-openrouter) + - Upstream failure (graceful degradation with stale data) + - Malformed / corrupt snapshot file on disk + - Idempotent same-day updates + - No real network calls or private credential leakage +""" + +from __future__ import annotations + +import json +import os +import sys +import types +import urllib.error +from datetime import datetime, timezone +from io import BytesIO +from pathlib import Path +from types import SimpleNamespace + +import api.config as config +import api.profiles as profiles + +ROOT = Path(__file__).resolve().parents[1] + + +class _FakeResponse: + """Minimal stand-in for urllib.request.urlopen context manager.""" + + def __init__(self, payload: bytes): + self._payload = payload + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + def read(self): + return self._payload + + +def _with_config(model=None, providers=None): + old_cfg = dict(config.cfg) + old_mtime = config._cfg_mtime + config.cfg.clear() + config.cfg["model"] = model or {} + if providers is not None: + config.cfg["providers"] = providers + try: + config._cfg_mtime = config.Path(config._get_config_path()).stat().st_mtime + except Exception: + config._cfg_mtime = 0.0 + return old_cfg, old_mtime + + +def _restore_config(old_cfg, old_mtime): + config.cfg.clear() + config.cfg.update(old_cfg) + config._cfg_mtime = old_mtime + + +# ── Happy path: snapshot append + delta response ────────────────────────────── + + +def test_openrouter_cost_history_happy_path(monkeypatch, tmp_path): + """On-demand snapshot append returns deltas from cumulative usage.""" + monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path) + monkeypatch.delenv("OPENROUTER_API_KEY", raising=False) + (tmp_path / ".env").write_text("OPENROUTER_API_KEY=test-or-key\n", encoding="utf-8") + old_cfg, old_mtime = _with_config(model={"provider": "openrouter"}) + + import api.providers as providers + + call_count = {"n": 0} + + def fake_urlopen(req, timeout): + call_count["n"] += 1 + # Simulate cumulative usage of 5.0 credits used out of 20 limit + payload = { + "data": { + "limit_remaining": 15.0, + "usage": 5.0, + "limit": 20, + "label": "Test Label", + "key": "must-not-leak", + } + } + return _FakeResponse(json.dumps(payload).encode("utf-8")) + + monkeypatch.setattr(providers.urllib.request, "urlopen", fake_urlopen) + + # Freeze "today" so the test is deterministic + fake_today = "2030-04-15" + monkeypatch.setattr(providers, "datetime", type("DT", (), { + "now": staticmethod(lambda tz=None: datetime(2030, 4, 15, 12, 0, 0, tzinfo=tz or timezone.utc)), + "strftime": datetime.strftime, + })) + + try: + result = providers.get_provider_cost_history("openrouter", days=7) + finally: + _restore_config(old_cfg, old_mtime) + + assert result["ok"] is True + assert result["provider"] == "openrouter" + assert result["supported"] is True + assert result["status"] == "available" + assert result["window_days"] == 7 + assert result["limit"] == 20 + assert result["label"] == "Test Label" + assert result["message"] == "OpenRouter cost history loaded." + # One snapshot for today + assert len(result["snapshots"]) == 1 + snap = result["snapshots"][0] + assert snap["date"] == fake_today + assert snap["used"] == 5.0 + assert snap["delta"] is None # first entry has no previous baseline + # Verify the snapshot file was persisted + snap_file = tmp_path / "cost-snapshots" / "openrouter.json" + assert snap_file.exists() + persisted = json.loads(snap_file.read_text(encoding="utf-8")) + assert len(persisted["snapshots"]) == 1 + # No credential leakage + assert "test-or-key" not in repr(result) + assert "must-not-leak" not in repr(result) + + +def test_openrouter_cost_history_deltas_from_cumulative(monkeypatch, tmp_path): + """Deltas are computed as differences between consecutive cumulative values.""" + monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path) + monkeypatch.delenv("OPENROUTER_API_KEY", raising=False) + (tmp_path / ".env").write_text("OPENROUTER_API_KEY=test-or-key\n", encoding="utf-8") + old_cfg, old_mtime = _with_config(model={"provider": "openrouter"}) + + import api.providers as providers + + # Pre-seed two historical snapshots + snap_dir = tmp_path / "cost-snapshots" + snap_dir.mkdir(parents=True, exist_ok=True) + historical = { + "provider": "openrouter", + "snapshots": [ + {"date": "2030-04-13", "used": 3.0, "limit": 20}, + {"date": "2030-04-14", "used": 4.5, "limit": 20}, + ], + } + (snap_dir / "openrouter.json").write_text(json.dumps(historical), encoding="utf-8") + + call_count = {"n": 0} + + def fake_urlopen(req, timeout): + call_count["n"] += 1 + # Current cumulative usage is 7.0 + payload = {"data": {"usage": 7.0, "limit": 20, "label": "Credits"}} + return _FakeResponse(json.dumps(payload).encode("utf-8")) + + monkeypatch.setattr(providers.urllib.request, "urlopen", fake_urlopen) + + # Freeze "today" + monkeypatch.setattr(providers, "datetime", type("DT", (), { + "now": staticmethod(lambda tz=None: datetime(2030, 4, 15, 12, 0, 0, tzinfo=tz or timezone.utc)), + "strftime": datetime.strftime, + })) + + try: + result = providers.get_provider_cost_history("openrouter", days=7) + finally: + _restore_config(old_cfg, old_mtime) + + assert result["ok"] is True + snaps = result["snapshots"] + assert len(snaps) == 3 + # Day 1: no delta (baseline) + assert snaps[0]["date"] == "2030-04-13" + assert snaps[0]["used"] == 3.0 + assert snaps[0]["delta"] is None + # Day 2: delta = 4.5 - 3.0 = 1.5 + assert snaps[1]["date"] == "2030-04-14" + assert snaps[1]["used"] == 4.5 + assert snaps[1]["delta"] == 1.5 + # Day 3 (today): delta = 7.0 - 4.5 = 2.5 + assert snaps[2]["date"] == "2030-04-15" + assert snaps[2]["used"] == 7.0 + assert snaps[2]["delta"] == 2.5 + + +# ── Missing credentials ─────────────────────────────────────────────────────── + + +def test_openrouter_cost_history_no_key(monkeypatch, tmp_path): + """No API key → safe no_key response without network call.""" + monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path) + monkeypatch.delenv("OPENROUTER_API_KEY", raising=False) + old_cfg, old_mtime = _with_config(model={"provider": "openrouter"}) + + import api.providers as providers + + def explode(*_a, **_kw): + raise AssertionError("should not call network without a key") + + monkeypatch.setattr(providers.urllib.request, "urlopen", explode) + + try: + result = providers.get_provider_cost_history("openrouter", days=7) + finally: + _restore_config(old_cfg, old_mtime) + + assert result["ok"] is False + assert result["provider"] == "openrouter" + assert result["supported"] is True + assert result["status"] == "no_key" + assert "OPENROUTER_API_KEY" in result["message"] + + +# ── Unsupported provider ────────────────────────────────────────────────────── + + +def test_cost_history_unsupported_provider(monkeypatch, tmp_path): + """Non-openrouter providers return a clear unsupported response.""" + monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path) + old_cfg, old_mtime = _with_config(model={"provider": "anthropic"}) + + import api.providers as providers + + try: + result = providers.get_provider_cost_history("anthropic", days=7) + finally: + _restore_config(old_cfg, old_mtime) + + assert result["ok"] is False + assert result["provider"] == "anthropic" + assert result["supported"] is False + assert result["status"] == "unsupported" + assert "openrouter" in result["message"].lower() + + +def test_cost_history_missing_provider_param(monkeypatch, tmp_path): + """Empty provider parameter returns a clear error.""" + monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path) + + import api.providers as providers + + result = providers.get_provider_cost_history("", days=7) + assert result["ok"] is False + assert result["status"] == "missing_provider" + + result2 = providers.get_provider_cost_history(None, days=7) + assert result2["ok"] is False + assert result2["status"] == "missing_provider" + + +# ── Upstream failure / graceful degradation ──────────────────────────────────── + + +def test_openrouter_cost_history_upstream_failure_degrades_gracefully(monkeypatch, tmp_path): + """When OpenRouter API fails, previously persisted snapshots are still returned.""" + monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path) + monkeypatch.delenv("OPENROUTER_API_KEY", raising=False) + (tmp_path / ".env").write_text("OPENROUTER_API_KEY=test-or-key\n", encoding="utf-8") + old_cfg, old_mtime = _with_config(model={"provider": "openrouter"}) + + import api.providers as providers + + # Pre-seed a snapshot + snap_dir = tmp_path / "cost-snapshots" + snap_dir.mkdir(parents=True, exist_ok=True) + historical = { + "provider": "openrouter", + "snapshots": [ + {"date": "2030-04-13", "used": 3.0, "limit": 20}, + ], + } + (snap_dir / "openrouter.json").write_text(json.dumps(historical), encoding="utf-8") + + req = providers.urllib.request.Request("https://openrouter.ai/api/v1/key") + def fake_urlopen(_req, timeout=None): + raise urllib.error.HTTPError(req.full_url, 500, "Server Error", {}, BytesIO(b"error")) + + monkeypatch.setattr(providers.urllib.request, "urlopen", fake_urlopen) + + try: + result = providers.get_provider_cost_history("openrouter", days=7) + finally: + _restore_config(old_cfg, old_mtime) + + assert result["ok"] is False + assert result["status"] == "unavailable" + # Still returns previously persisted data + assert len(result["snapshots"]) == 1 + assert result["snapshots"][0]["date"] == "2030-04-13" + assert "temporarily unavailable" in result["message"].lower() + + +def test_openrouter_cost_history_timeout_is_safe(monkeypatch, tmp_path): + """Timeout from OpenRouter does not produce a traceback or leak secrets.""" + monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path) + monkeypatch.delenv("OPENROUTER_API_KEY", raising=False) + (tmp_path / ".env").write_text("OPENROUTER_API_KEY=test-or-key\n", encoding="utf-8") + old_cfg, old_mtime = _with_config(model={"provider": "openrouter"}) + + import api.providers as providers + + def fake_urlopen(_req, timeout=None): + raise TimeoutError("slow secret") + + monkeypatch.setattr(providers.urllib.request, "urlopen", fake_urlopen) + + try: + result = providers.get_provider_cost_history("openrouter", days=7) + finally: + _restore_config(old_cfg, old_mtime) + + assert result["ok"] is False + assert result["status"] == "unavailable" + assert "test-or-key" not in repr(result) + assert "secret" not in repr(result).lower() + + +# ── Malformed / corrupt snapshot file ───────────────────────────────────────── + + +def test_openrouter_cost_history_corrupt_snapshot_file(monkeypatch, tmp_path): + """A corrupt snapshot file on disk is handled gracefully (treated as empty).""" + monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path) + monkeypatch.delenv("OPENROUTER_API_KEY", raising=False) + (tmp_path / ".env").write_text("OPENROUTER_API_KEY=test-or-key\n", encoding="utf-8") + old_cfg, old_mtime = _with_config(model={"provider": "openrouter"}) + + import api.providers as providers + + # Write a corrupt file + snap_dir = tmp_path / "cost-snapshots" + snap_dir.mkdir(parents=True, exist_ok=True) + (snap_dir / "openrouter.json").write_text("NOT VALID JSON{{{{", encoding="utf-8") + + def fake_urlopen(req, timeout): + payload = {"data": {"usage": 2.0, "limit": 10, "label": "Credits"}} + return _FakeResponse(json.dumps(payload).encode("utf-8")) + + monkeypatch.setattr(providers.urllib.request, "urlopen", fake_urlopen) + + # Freeze "today" + monkeypatch.setattr(providers, "datetime", type("DT", (), { + "now": staticmethod(lambda tz=None: datetime(2030, 4, 15, 12, 0, 0, tzinfo=tz or timezone.utc)), + "strftime": datetime.strftime, + })) + + try: + result = providers.get_provider_cost_history("openrouter", days=7) + finally: + _restore_config(old_cfg, old_mtime) + + # Corrupt file is ignored; fresh snapshot is created + assert result["ok"] is True + assert len(result["snapshots"]) == 1 + assert result["snapshots"][0]["date"] == "2030-04-15" + + +# ── Idempotent same-day updates ─────────────────────────────────────────────── + + +def test_openrouter_cost_history_same_day_idempotent(monkeypatch, tmp_path): + """Repeated calls on the same day update the snapshot in-place.""" + monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path) + monkeypatch.delenv("OPENROUTER_API_KEY", raising=False) + (tmp_path / ".env").write_text("OPENROUTER_API_KEY=test-or-key\n", encoding="utf-8") + old_cfg, old_mtime = _with_config(model={"provider": "openrouter"}) + + import api.providers as providers + + call_count = {"n": 0} + + def fake_urlopen(req, timeout): + call_count["n"] += 1 + usage = 5.0 + call_count["n"] # usage grows each call + payload = {"data": {"usage": usage, "limit": 20, "label": "Credits"}} + return _FakeResponse(json.dumps(payload).encode("utf-8")) + + monkeypatch.setattr(providers.urllib.request, "urlopen", fake_urlopen) + + # Freeze "today" + monkeypatch.setattr(providers, "datetime", type("DT", (), { + "now": staticmethod(lambda tz=None: datetime(2030, 4, 15, 12, 0, 0, tzinfo=tz or timezone.utc)), + "strftime": datetime.strftime, + })) + + try: + r1 = providers.get_provider_cost_history("openrouter", days=7) + r2 = providers.get_provider_cost_history("openrouter", days=7) + finally: + _restore_config(old_cfg, old_mtime) + + # Both calls succeed; only one snapshot date (today) + assert r1["ok"] is True + assert r2["ok"] is True + assert len(r1["snapshots"]) == 1 + assert len(r2["snapshots"]) == 1 + # Second call updated the same day's used value + assert r2["snapshots"][0]["used"] == 7.0 # 5.0 + 2 (second call) + # Verify persisted file has only one entry for today + snap_file = tmp_path / "cost-snapshots" / "openrouter.json" + persisted = json.loads(snap_file.read_text(encoding="utf-8")) + assert len(persisted["snapshots"]) == 1 + + +# ── Window days parameter ───────────────────────────────────────────────────── + + +def test_openrouter_cost_history_window_days_truncation(monkeypatch, tmp_path): + """The window_days parameter limits how many snapshots are returned.""" + monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path) + monkeypatch.delenv("OPENROUTER_API_KEY", raising=False) + (tmp_path / ".env").write_text("OPENROUTER_API_KEY=test-or-key\n", encoding="utf-8") + old_cfg, old_mtime = _with_config(model={"provider": "openrouter"}) + + import api.providers as providers + + # Pre-seed 5 historical snapshots + snap_dir = tmp_path / "cost-snapshots" + snap_dir.mkdir(parents=True, exist_ok=True) + historical = { + "provider": "openrouter", + "snapshots": [ + {"date": f"2030-04-{d:02d}", "used": float(d), "limit": 20} + for d in range(10, 15) + ], + } + (snap_dir / "openrouter.json").write_text(json.dumps(historical), encoding="utf-8") + + def fake_urlopen(req, timeout): + payload = {"data": {"usage": 15.0, "limit": 20, "label": "Credits"}} + return _FakeResponse(json.dumps(payload).encode("utf-8")) + + monkeypatch.setattr(providers.urllib.request, "urlopen", fake_urlopen) + + # Freeze "today" + monkeypatch.setattr(providers, "datetime", type("DT", (), { + "now": staticmethod(lambda tz=None: datetime(2030, 4, 15, 12, 0, 0, tzinfo=tz or timezone.utc)), + "strftime": datetime.strftime, + })) + + try: + result = providers.get_provider_cost_history("openrouter", days=3) + finally: + _restore_config(old_cfg, old_mtime) + + # 5 historical + 1 today = 6 total, but window_days=3 returns last 3 + assert result["ok"] is True + assert result["window_days"] == 3 + assert len(result["snapshots"]) == 3 + # The returned snapshots are the most recent 3 + assert result["snapshots"][0]["date"] == "2030-04-13" + assert result["snapshots"][1]["date"] == "2030-04-14" + assert result["snapshots"][2]["date"] == "2030-04-15" + + +# ── No real network calls ───────────────────────────────────────────────────── + + +def test_cost_history_uses_no_real_network(monkeypatch, tmp_path): + """Every test path must monkeypatch urlopen; verify no real calls escape.""" + monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path) + monkeypatch.delenv("OPENROUTER_API_KEY", raising=False) + old_cfg, old_mtime = _with_config(model={"provider": "openrouter"}) + + import api.providers as providers + + # Without a key, no network call is made at all + def explode(*_a, **_kw): + raise AssertionError("real network call detected") + + monkeypatch.setattr(providers.urllib.request, "urlopen", explode) + + try: + result = providers.get_provider_cost_history("openrouter", days=7) + finally: + _restore_config(old_cfg, old_mtime) + + assert result["status"] == "no_key" \ No newline at end of file