"""Regression coverage for active-provider quota status (#706).""" from __future__ import annotations import base64 import json import inspect import os import re import sys import threading 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: 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 old_path = getattr(config, "_cfg_path", None) 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 config._cfg_path = config._get_config_path() return old_cfg, (old_mtime, old_path) def _restore_config(old_cfg, old_mtime): if isinstance(old_mtime, tuple): old_mtime, old_path = old_mtime else: old_path = getattr(config, "_cfg_path", None) config.cfg.clear() config.cfg.update(old_cfg) config._cfg_mtime = old_mtime config._cfg_path = old_path def test_openrouter_quota_fetches_key_endpoint_and_sanitizes_response(monkeypatch, tmp_path): """OpenRouter's documented key endpoint should be called server-side only.""" 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-openrouter-key-private\n", encoding="utf-8") old_cfg, old_mtime = _with_config(model={"provider": "openrouter"}) import api.providers as providers seen = {} def fake_urlopen(req, timeout): seen["url"] = req.full_url seen["timeout"] = timeout seen["authorization"] = req.headers.get("Authorization") payload = {"data": {"limit_remaining": "12.5", "usage": 3, "limit": 20, "key": "must-not-leak"}} return _FakeResponse(json.dumps(payload).encode("utf-8")) monkeypatch.setattr(providers.urllib.request, "urlopen", fake_urlopen) try: result = providers.get_provider_quota() finally: _restore_config(old_cfg, old_mtime) assert seen == { "url": "https://openrouter.ai/api/v1/key", "timeout": 3.0, "authorization": "Bearer test-openrouter-key-private", } assert result == { "ok": True, "provider": "openrouter", "display_name": "OpenRouter", "supported": True, "status": "available", "label": "OpenRouter credits", "quota": {"limit_remaining": 12.5, "usage": 3, "limit": 20}, "message": "OpenRouter quota status loaded.", } assert "test-openrouter-key-private" not in repr(result) assert "must-not-leak" not in repr(result) def test_openrouter_quota_no_key_returns_safe_no_key_without_network(monkeypatch, tmp_path): """No-key state must not call OpenRouter or leak environment details.""" 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(*_args, **_kwargs): raise AssertionError("quota lookup should not call the network without a key") monkeypatch.setattr(providers.urllib.request, "urlopen", explode) try: result = providers.get_provider_quota() 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 result["quota"] is None assert "OPENROUTER_API_KEY" in result["message"] def test_openrouter_quota_invalid_key_and_timeout_are_sanitized(monkeypatch, tmp_path): """Invalid-key and timeout/error paths should expose statuses, not 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-openrouter-key-private\n", encoding="utf-8") old_cfg, old_mtime = _with_config(model={"provider": "openrouter"}) import api.providers as providers req = providers.urllib.request.Request("https://openrouter.ai/api/v1/key") invalid = urllib.error.HTTPError(req.full_url, 401, "Unauthorized", {}, BytesIO(b"secret body")) errors = [invalid, TimeoutError("slow secret")] try: for expected in ("invalid_key", "unavailable"): def fake_urlopen(_req, timeout=None, *, _err=errors.pop(0)): raise _err monkeypatch.setattr(providers.urllib.request, "urlopen", fake_urlopen) result = providers.get_provider_quota("openrouter") assert result["ok"] is False assert result["status"] == expected assert result["quota"] is None assert "test-openrouter-key-private" not in repr(result) assert "secret" not in repr(result).lower() finally: _restore_config(old_cfg, old_mtime) def test_unsupported_provider_reports_followup_state(monkeypatch, tmp_path): """Providers without safe quota APIs should return a clear unsupported state.""" monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path) old_cfg, old_mtime = _with_config(model={"provider": "openai"}) import api.providers as providers try: result = providers.get_provider_quota() finally: _restore_config(old_cfg, old_mtime) assert result["ok"] is False assert result["provider"] == "openai" assert result["supported"] is False assert result["status"] == "unsupported" assert result["quota"] is None assert "follow-up" in result["message"] def test_codex_account_usage_is_fetched_under_active_profile_home(monkeypatch, tmp_path): """Codex account limits must use the selected WebUI profile's HERMES_HOME.""" monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path) old_cfg, old_mtime = _with_config(model={"provider": "openai-codex"}) import api.providers as providers seen = {} previous_home = os.environ.get("HERMES_HOME") def fake_fetch(provider, home, api_key=None): seen["provider"] = provider seen["home"] = str(home) seen["api_key"] = api_key return SimpleNamespace( provider="openai-codex", source="usage_api", title="Account limits", plan="Pro", fetched_at=datetime(2030, 3, 17, 12, 30, tzinfo=timezone.utc), available=True, windows=( SimpleNamespace( label="Session", used_percent=15.0, reset_at=datetime(2030, 3, 17, 17, 30, tzinfo=timezone.utc), detail=None, ), SimpleNamespace( label="Weekly", used_percent=40.0, reset_at=datetime(2030, 3, 24, 12, 30, tzinfo=timezone.utc), detail=None, ), ), details=("Credits balance: $12.50",), unavailable_reason=None, ) monkeypatch.setattr(providers, "_agent_fetch_account_usage_for_home", fake_fetch) try: result = providers.get_provider_quota() finally: _restore_config(old_cfg, old_mtime) assert seen == { "provider": "openai-codex", "home": str(tmp_path), "api_key": None, } assert os.environ.get("HERMES_HOME") == previous_home assert result["ok"] is True assert result["provider"] == "openai-codex" assert result["supported"] is True assert result["status"] == "available" assert result["quota"] is None assert result["account_limits"] == { "provider": "openai-codex", "source": "usage_api", "title": "Account limits", "plan": "Pro", "windows": [ { "label": "Session", "used_percent": 15.0, "remaining_percent": 85.0, "reset_at": "2030-03-17T17:30:00Z", "detail": None, }, { "label": "Weekly", "used_percent": 40.0, "remaining_percent": 60.0, "reset_at": "2030-03-24T12:30:00Z", "detail": None, }, ], "details": ["Credits balance: $12.50"], "available": True, "unavailable_reason": None, "fetched_at": "2030-03-17T12:30:00Z", } def test_codex_account_usage_unavailable_is_sanitized(monkeypatch, tmp_path): """Auth/network failures should not leak raw token or exception details.""" monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path) old_cfg, old_mtime = _with_config(model={"provider": "openai-codex"}) import api.providers as providers def fake_fetch(*_args, **_kwargs): raise RuntimeError("secret access token should not leak") monkeypatch.setattr(providers, "_agent_fetch_account_usage_for_home", fake_fetch) try: result = providers.get_provider_quota() finally: _restore_config(old_cfg, old_mtime) assert result["ok"] is False assert result["provider"] == "openai-codex" assert result["supported"] is True assert result["status"] == "unavailable" assert result["account_limits"] is None assert "Confirm provider authentication" in result["message"] assert "secret" not in repr(result).lower() def test_codex_account_usage_subprocess_reports_read_only_credential_pool(monkeypatch, capsys): """Codex quota probes should inspect pool entries without mutating selection order.""" import api.providers as providers def b64url(payload: bytes) -> str: return base64.urlsafe_b64encode(payload).rstrip(b"=").decode("ascii") def token_for(account_id: str) -> str: return ".".join(( b64url(b'{"alg":"none","typ":"JWT"}'), b64url(json.dumps({ "https://api.openai.com/auth": { "chatgpt_account_id": account_id, }, }).encode("utf-8")), b64url(b"signature"), )) primary_token = token_for("acct-primary") exhausted_token = token_for("acct-exhausted") fetch_calls = [] load_pool_calls = [] entries_called = [] seen = [] agent_mod = types.ModuleType("agent") agent_mod.__path__ = [] account_usage_mod = types.ModuleType("agent.account_usage") credential_pool_mod = types.ModuleType("agent.credential_pool") def fake_fetch_account_usage(provider, *, base_url=None, api_key=None): fetch_calls.append((provider, base_url, api_key)) return None class FakePool: def entries(self): entries_called.append(True) return [ SimpleNamespace( label="Team primary", runtime_api_key=primary_token, runtime_base_url="https://chatgpt.com/backend-api/codex", last_status=None, ), SimpleNamespace( label="Plus backup", runtime_api_key=exhausted_token, runtime_base_url="https://chatgpt.com/backend-api/codex", last_status="exhausted", last_status_at=1_900_000_000, ), ] def select(self): raise AssertionError("quota display must not rotate credential_pool selection") def fake_load_pool(provider): load_pool_calls.append(provider) return FakePool() def fake_urlopen(req, timeout): headers = {key.lower(): value for key, value in req.header_items()} seen.append({ "url": req.full_url, "timeout": timeout, "headers": headers, }) payload = { "plan_type": "pro" if headers.get("chatgpt-account-id") == "acct-primary" else "plus", "rate_limit": { "primary_window": {"used_percent": 15 if headers.get("chatgpt-account-id") == "acct-primary" else 95, "reset_at": 1_900_000_000}, "secondary_window": {"used_percent": 40, "reset_at": "2030-03-24T12:30:00Z"}, }, "credits": {"has_credits": True, "balance": 12.5}, } return _FakeResponse(json.dumps(payload).encode("utf-8")) account_usage_mod.fetch_account_usage = fake_fetch_account_usage credential_pool_mod.load_pool = fake_load_pool monkeypatch.setitem(sys.modules, "agent", agent_mod) monkeypatch.setitem(sys.modules, "agent.account_usage", account_usage_mod) monkeypatch.setitem(sys.modules, "agent.credential_pool", credential_pool_mod) monkeypatch.setattr(providers.urllib.request, "urlopen", fake_urlopen) monkeypatch.setattr(sys, "argv", ["quota-probe", "openai-codex", ""]) exec(providers._ACCOUNT_USAGE_SUBPROCESS_CODE, {"__name__": "__main__"}) output = capsys.readouterr().out.strip() snapshot = json.loads(output) assert fetch_calls == [("openai-codex", None, None)] assert load_pool_calls == ["openai-codex"] assert entries_called == [True] assert [call["url"] for call in seen] == [ "https://chatgpt.com/backend-api/wham/usage", ] assert [call["timeout"] for call in seen] == [4.0] assert seen[0]["headers"]["authorization"] == f"Bearer {primary_token}" assert seen[0]["headers"]["chatgpt-account-id"] == "acct-primary" assert snapshot["provider"] == "openai-codex" assert snapshot["source"] == "usage_api_pool" assert snapshot["windows"][0]["label"] == "Session" assert snapshot["windows"][0]["used_percent"] == 15 assert snapshot["details"] == ["1/2 credentials available", "1 exhausted", "Plans: Pro"] assert snapshot["available"] is True assert snapshot["pool"] == { "total_credentials": 2, "queried_credentials": 1, "available_credentials": 1, "exhausted_credentials": 1, "failed_credentials": 0, "plans": ["Pro"], "next_reset_at": "2030-03-17T17:46:40Z", "best_remaining_by_window": [ { "label": "Session", "remaining_percent": 85.0, "used_percent": 15.0, "reset_at": "2030-03-17T17:46:40Z", "detail": None, "credential_label": "Team primary", }, { "label": "Weekly", "remaining_percent": 60.0, "used_percent": 40.0, "reset_at": "2030-03-24T12:30:00Z", "detail": None, "credential_label": "Team primary", }, ], "credentials": [ { "label": "Team primary", "status": "available", "plan": "Pro", "windows": [ { "label": "Session", "used_percent": 15.0, "remaining_percent": 85.0, "reset_at": "2030-03-17T17:46:40Z", "detail": None, }, { "label": "Weekly", "used_percent": 40.0, "remaining_percent": 60.0, "reset_at": "2030-03-24T12:30:00Z", "detail": None, }, ], "details": ["Credits balance: $12.50"], "unavailable_reason": None, "fetched_at": snapshot["pool"]["credentials"][0]["fetched_at"], }, { "label": "Plus backup", "status": "exhausted", "plan": None, "windows": [], "details": [], "unavailable_reason": "Credential pool marked this credential exhausted; retry after 2030-03-17T18:46:40Z.", "retry_after": "2030-03-17T18:46:40Z", "fetched_at": None, }, ], } assert primary_token not in output assert exhausted_token not in output def test_codex_account_usage_subprocess_retries_expired_pool_exhaustion(monkeypatch, capsys): """Expired pool cooldowns should be probed instead of shown as still exhausted.""" import api.providers as providers def b64url(payload: bytes) -> str: return base64.urlsafe_b64encode(payload).rstrip(b"=").decode("ascii") token = ".".join(( b64url(b'{"alg":"none","typ":"JWT"}'), b64url(json.dumps({ "https://api.openai.com/auth": { "chatgpt_account_id": "acct-expired", }, }).encode("utf-8")), b64url(b"signature"), )) seen = [] agent_mod = types.ModuleType("agent") agent_mod.__path__ = [] account_usage_mod = types.ModuleType("agent.account_usage") credential_pool_mod = types.ModuleType("agent.credential_pool") def fake_fetch_account_usage(provider, *, base_url=None, api_key=None): return None class FakePool: def entries(self): return [ SimpleNamespace( label="Expired cooldown", runtime_api_key=token, runtime_base_url="https://chatgpt.com/backend-api/codex", last_status="exhausted", last_status_at=1, last_error_code=None, last_error_reset_at=None, ), ] def select(self): raise AssertionError("quota display must not rotate credential_pool selection") def fake_load_pool(provider): return FakePool() def fake_urlopen(req, timeout): seen.append(req.full_url) payload = { "plan_type": "team", "rate_limit": { "primary_window": {"used_percent": 10, "reset_at": "2030-03-17T17:30:00Z"}, }, } return _FakeResponse(json.dumps(payload).encode("utf-8")) account_usage_mod.fetch_account_usage = fake_fetch_account_usage credential_pool_mod.load_pool = fake_load_pool monkeypatch.setitem(sys.modules, "agent", agent_mod) monkeypatch.setitem(sys.modules, "agent.account_usage", account_usage_mod) monkeypatch.setitem(sys.modules, "agent.credential_pool", credential_pool_mod) monkeypatch.setattr(providers.urllib.request, "urlopen", fake_urlopen) monkeypatch.setattr(sys, "argv", ["quota-probe", "openai-codex", ""]) exec(providers._ACCOUNT_USAGE_SUBPROCESS_CODE, {"__name__": "__main__"}) output = capsys.readouterr().out.strip() snapshot = json.loads(output) assert seen == ["https://chatgpt.com/backend-api/wham/usage"] assert snapshot["pool"]["queried_credentials"] == 1 assert snapshot["pool"]["exhausted_credentials"] == 0 assert snapshot["pool"]["credentials"][0]["status"] == "available" assert snapshot["pool"]["credentials"][0]["unavailable_reason"] is None assert token not in output def test_codex_account_usage_subprocess_probes_pool_entries_concurrently(monkeypatch, capsys): """Eligible pool credentials should be probed concurrently and reported in pool order.""" import api.providers as providers def b64url(payload: bytes) -> str: return base64.urlsafe_b64encode(payload).rstrip(b"=").decode("ascii") def token_for(account_id: str) -> str: return ".".join(( b64url(b'{"alg":"none","typ":"JWT"}'), b64url(json.dumps({ "https://api.openai.com/auth": { "chatgpt_account_id": account_id, }, }).encode("utf-8")), b64url(b"signature"), )) token_a = token_for("acct-a") token_b = token_for("acct-b") events = [] events_lock = threading.Lock() barrier = threading.Barrier(2, timeout=2) agent_mod = types.ModuleType("agent") agent_mod.__path__ = [] account_usage_mod = types.ModuleType("agent.account_usage") credential_pool_mod = types.ModuleType("agent.credential_pool") def fake_fetch_account_usage(provider, *, base_url=None, api_key=None): return None class FakePool: def entries(self): return [ SimpleNamespace( label="Slow A", runtime_api_key=token_a, runtime_base_url="https://chatgpt.com/backend-api/codex", last_status=None, ), SimpleNamespace( label="Slow B", runtime_api_key=token_b, runtime_base_url="https://chatgpt.com/backend-api/codex", last_status=None, ), ] def select(self): raise AssertionError("quota display must not rotate credential_pool selection") def fake_load_pool(provider): return FakePool() def fake_urlopen(req, timeout): headers = {key.lower(): value for key, value in req.header_items()} account_id = headers.get("chatgpt-account-id") with events_lock: events.append(("enter", account_id)) barrier.wait() with events_lock: events.append(("exit", account_id)) used = 80 if account_id == "acct-a" else 10 payload = { "plan_type": "team", "rate_limit": { "primary_window": {"used_percent": used, "reset_at": "2030-03-17T17:30:00Z"}, }, } return _FakeResponse(json.dumps(payload).encode("utf-8")) account_usage_mod.fetch_account_usage = fake_fetch_account_usage credential_pool_mod.load_pool = fake_load_pool monkeypatch.setitem(sys.modules, "agent", agent_mod) monkeypatch.setitem(sys.modules, "agent.account_usage", account_usage_mod) monkeypatch.setitem(sys.modules, "agent.credential_pool", credential_pool_mod) monkeypatch.setattr(providers.urllib.request, "urlopen", fake_urlopen) monkeypatch.setattr(sys, "argv", ["quota-probe", "openai-codex", ""]) exec(providers._ACCOUNT_USAGE_SUBPROCESS_CODE, {"__name__": "__main__"}) output = capsys.readouterr().out.strip() snapshot = json.loads(output) first_exit = next(index for index, event in enumerate(events) if event[0] == "exit") assert [event[0] for event in events[:first_exit]] == ["enter", "enter"] assert snapshot["pool"]["queried_credentials"] == 2 assert [row["label"] for row in snapshot["pool"]["credentials"]] == ["Slow A", "Slow B"] assert snapshot["pool"]["best_remaining_by_window"][0]["credential_label"] == "Slow B" assert snapshot["pool"]["best_remaining_by_window"][0]["remaining_percent"] == 90.0 assert token_a not in output assert token_b not in output def test_codex_account_usage_subprocess_sanitizes_pool_entry_errors(monkeypatch, capsys): """Pool per-entry failures must not leak bearer/JWT-like exception text.""" import api.providers as providers fetch_calls = [] agent_mod = types.ModuleType("agent") agent_mod.__path__ = [] account_usage_mod = types.ModuleType("agent.account_usage") credential_pool_mod = types.ModuleType("agent.credential_pool") def fake_fetch_account_usage(provider, *, base_url=None, api_key=None): fetch_calls.append((provider, base_url, api_key)) return None class FakePool: def entries(self): return [ SimpleNamespace( label="Bad token", runtime_api_key="header.payload.signature", runtime_base_url="https://chatgpt.com/backend-api/codex", last_status=None, ), ] def select(self): raise AssertionError("quota display must not rotate credential_pool selection") def fake_load_pool(provider): return FakePool() def fake_urlopen(_req, timeout): raise RuntimeError("Bearer eyJsecret-token-like-value should not leak") account_usage_mod.fetch_account_usage = fake_fetch_account_usage credential_pool_mod.load_pool = fake_load_pool monkeypatch.setitem(sys.modules, "agent", agent_mod) monkeypatch.setitem(sys.modules, "agent.account_usage", account_usage_mod) monkeypatch.setitem(sys.modules, "agent.credential_pool", credential_pool_mod) monkeypatch.setattr(providers.urllib.request, "urlopen", fake_urlopen) monkeypatch.setattr(sys, "argv", ["quota-probe", "openai-codex", ""]) exec(providers._ACCOUNT_USAGE_SUBPROCESS_CODE, {"__name__": "__main__"}) output = capsys.readouterr().out.strip() snapshot = json.loads(output) assert fetch_calls == [("openai-codex", None, None)] assert snapshot["available"] is False assert snapshot["pool"]["failed_credentials"] == 1 assert snapshot["pool"]["credentials"][0]["unavailable_reason"] == "Usage unavailable for this credential." assert "eyJsecret" not in output assert "Bearer" not in output def test_codex_account_usage_subprocess_keeps_legacy_reason_when_pool_misses(monkeypatch, capsys): """A failed pool fallback should not discard the legacy unavailable reason.""" import api.providers as providers fetch_calls = [] load_pool_calls = [] agent_mod = types.ModuleType("agent") agent_mod.__path__ = [] account_usage_mod = types.ModuleType("agent.account_usage") credential_pool_mod = types.ModuleType("agent.credential_pool") def fake_fetch_account_usage(provider, *, base_url=None, api_key=None): fetch_calls.append((provider, base_url, api_key)) return SimpleNamespace( provider="openai-codex", source="usage_api", title="Account limits", plan=None, windows=(), details=(), available=False, unavailable_reason="Codex account limits are not available for this credential.", fetched_at=datetime(2030, 3, 17, 12, 30, tzinfo=timezone.utc), ) class EmptyPool: def select(self): return None def fake_load_pool(provider): load_pool_calls.append(provider) return EmptyPool() def explode_urlopen(*_args, **_kwargs): raise AssertionError("no network call should happen when the pool has no selected entry") account_usage_mod.fetch_account_usage = fake_fetch_account_usage credential_pool_mod.load_pool = fake_load_pool monkeypatch.setitem(sys.modules, "agent", agent_mod) monkeypatch.setitem(sys.modules, "agent.account_usage", account_usage_mod) monkeypatch.setitem(sys.modules, "agent.credential_pool", credential_pool_mod) monkeypatch.setattr(providers.urllib.request, "urlopen", explode_urlopen) monkeypatch.setattr(sys, "argv", ["quota-probe", "openai-codex", ""]) exec(providers._ACCOUNT_USAGE_SUBPROCESS_CODE, {"__name__": "__main__"}) snapshot = json.loads(capsys.readouterr().out.strip()) assert fetch_calls == [("openai-codex", None, None)] assert load_pool_calls == ["openai-codex"] assert snapshot["available"] is False assert snapshot["unavailable_reason"] == "Codex account limits are not available for this credential." assert snapshot["fetched_at"] == "2030-03-17T12:30:00Z" def test_account_usage_pool_payload_round_trips_to_provider_quota_status(): """Parent process serialization must preserve pooled credential summaries.""" import api.providers as providers payload = { "provider": "openai-codex", "source": "usage_api_pool", "title": "Account limits", "plan": None, "windows": [ {"label": "Session", "used_percent": 25, "reset_at": "2030-03-17T17:30:00Z", "detail": "Best of 2"}, ], "details": ["2/3 credentials available"], "available": True, "unavailable_reason": None, "fetched_at": "2030-03-17T12:30:00Z", "pool": { "total_credentials": 3, "available_credentials": 2, "exhausted_credentials": 1, "failed_credentials": 0, "credentials": [ {"label": "Credential 1", "status": "available", "windows": []}, { "label": "Credential 2", "status": "exhausted", "windows": [], "retry_after": "2030-03-17T18:46:40Z", }, ], }, } snapshot = providers._account_usage_payload_to_snapshot(payload) serialized = providers._serialize_account_usage_snapshot(snapshot) assert serialized["windows"][0]["remaining_percent"] == 75.0 assert serialized["pool"] == payload["pool"] def test_anthropic_oauth_usage_unavailable_reason_is_reported(monkeypatch, tmp_path): """Hermes Agent can report why account limits are not available.""" monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path) old_cfg, old_mtime = _with_config(model={"provider": "anthropic"}) import api.providers as providers monkeypatch.setattr( providers, "_agent_fetch_account_usage_for_home", lambda *_args, **_kwargs: SimpleNamespace( provider="anthropic", source="oauth_usage_api", title="Account limits", plan=None, fetched_at=datetime(2030, 3, 17, 12, 30, tzinfo=timezone.utc), available=False, windows=(), details=(), unavailable_reason="Anthropic account limits are only available for OAuth-backed Claude accounts.", ), ) try: result = providers.get_provider_quota() finally: _restore_config(old_cfg, old_mtime) assert result["ok"] is False assert result["provider"] == "anthropic" assert result["supported"] is True assert result["status"] == "unavailable" assert result["account_limits"]["unavailable_reason"].startswith("Anthropic account limits") assert "OAuth-backed Claude accounts" in result["message"] def test_account_usage_profile_fetch_does_not_enter_cron_env_context(): """Quota probes must not reuse cron's process-global env/module swapper.""" import api.providers as providers body = inspect.getsource(providers._fetch_account_usage_with_profile_context) assert "cron_profile_context_for_home" not in body assert "_agent_fetch_account_usage_for_home" in body def test_account_usage_profile_env_is_child_scoped(monkeypatch, tmp_path): """Profile .env values should be passed to the child probe only.""" import api.providers as providers home = tmp_path / "profile-a" home.mkdir() (home / ".env").write_text("ANTHROPIC_API_KEY=profile-key\n", encoding="utf-8") monkeypatch.setenv("ANTHROPIC_API_KEY", "process-key") env = providers._account_usage_subprocess_env(home, "anthropic", None) assert env["HERMES_HOME"] == str(home) assert env["ANTHROPIC_API_KEY"] == "profile-key" assert os.environ["ANTHROPIC_API_KEY"] == "process-key" def test_account_usage_profile_fetch_uses_short_lived_cache(monkeypatch, tmp_path): """Repeated Settings refreshes should not re-query pooled account usage immediately.""" import api.providers as providers monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path) old_cfg, old_mtime = _with_config(model={"provider": "openai-codex"}) providers._account_usage_status_cache.clear() calls = [] snapshots = [ SimpleNamespace( provider="openai-codex", source="usage_api_pool", title="Account limits", plan=None, windows=(), details=(), available=True, unavailable_reason=None, fetched_at=datetime(2030, 3, 17, 12, 30, tzinfo=timezone.utc), pool={"total_credentials": 1, "credentials": []}, ), SimpleNamespace( provider="openai-codex", source="usage_api_pool", title="Account limits", plan=None, windows=(), details=(), available=True, unavailable_reason=None, fetched_at=datetime(2030, 3, 17, 12, 31, tzinfo=timezone.utc), pool={"total_credentials": 1, "credentials": []}, ), ] def fake_fetch(provider, home, api_key=None): calls.append((provider, str(home), api_key)) return snapshots[len(calls) - 1] monkeypatch.setattr(providers, "_agent_fetch_account_usage_for_home", fake_fetch) try: first = providers._fetch_account_usage_with_profile_context("openai-codex") second = providers._fetch_account_usage_with_profile_context("openai-codex") refreshed_status = providers.get_provider_quota("openai-codex", refresh=True) after_refresh = providers._fetch_account_usage_with_profile_context("openai-codex") finally: providers._account_usage_status_cache.clear() _restore_config(old_cfg, old_mtime) assert first is snapshots[0] assert second is snapshots[0] assert refreshed_status["account_limits"]["fetched_at"] == "2030-03-17T12:31:00Z" assert after_refresh is snapshots[1] assert calls == [ ("openai-codex", str(tmp_path), None), ("openai-codex", str(tmp_path), None), ] def test_account_usage_forced_refresh_failure_preserves_warm_snapshot(monkeypatch, tmp_path): """A failed manual refresh should not discard the last usable account snapshot.""" import api.providers as providers monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path) old_cfg, old_mtime = _with_config(model={"provider": "openai-codex"}) providers._account_usage_status_cache.clear() calls = [] good_snapshot = SimpleNamespace( provider="openai-codex", source="usage_api_pool", title="Account limits", plan=None, windows=(), details=(), available=True, unavailable_reason=None, fetched_at=datetime(2030, 3, 17, 12, 30, tzinfo=timezone.utc), pool={"total_credentials": 1, "credentials": []}, ) def fake_fetch(provider, home, api_key=None): calls.append((provider, str(home), api_key)) return good_snapshot if len(calls) == 1 else None monkeypatch.setattr(providers, "_agent_fetch_account_usage_for_home", fake_fetch) try: first = providers._fetch_account_usage_with_profile_context("openai-codex") refreshed = providers._fetch_account_usage_with_profile_context( "openai-codex", refresh=True, ) after_refresh_failure = providers._fetch_account_usage_with_profile_context( "openai-codex", ) finally: providers._account_usage_status_cache.clear() _restore_config(old_cfg, old_mtime) assert first is good_snapshot assert refreshed is None assert after_refresh_failure is good_snapshot assert calls == [ ("openai-codex", str(tmp_path), None), ("openai-codex", str(tmp_path), None), ] def test_account_usage_profile_cache_invalidates_with_credential_pool_cache(monkeypatch, tmp_path): """Credential-pool invalidation should also clear pooled account usage.""" import api.providers as providers monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path) old_cfg, old_mtime = _with_config(model={"provider": "openai-codex"}) providers._account_usage_status_cache.clear() calls = [] snapshots = [ SimpleNamespace( provider="openai-codex", source="usage_api_pool", title="Account limits", plan=None, windows=(), details=(), available=True, unavailable_reason=None, fetched_at=datetime(2030, 3, 17, 12, 30, tzinfo=timezone.utc), pool={"total_credentials": 1, "credentials": []}, ), SimpleNamespace( provider="openai-codex", source="usage_api_pool", title="Account limits", plan=None, windows=(), details=(), available=True, unavailable_reason=None, fetched_at=datetime(2030, 3, 17, 12, 31, tzinfo=timezone.utc), pool={"total_credentials": 1, "credentials": []}, ), ] def fake_fetch(provider, home, api_key=None): calls.append((provider, str(home), api_key)) return snapshots[len(calls) - 1] monkeypatch.setattr(providers, "_agent_fetch_account_usage_for_home", fake_fetch) try: first = providers._fetch_account_usage_with_profile_context("openai-codex") config.invalidate_credential_pool_cache("openai-codex") second = providers._fetch_account_usage_with_profile_context("openai-codex") finally: providers._account_usage_status_cache.clear() _restore_config(old_cfg, old_mtime) assert first is snapshots[0] assert second is snapshots[1] assert calls == [ ("openai-codex", str(tmp_path), None), ("openai-codex", str(tmp_path), None), ] def test_account_usage_profile_fetch_caches_unavailable_snapshots(monkeypatch, tmp_path): """Known unavailable account snapshots should be cached like available ones.""" import api.providers as providers monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path) old_cfg, old_mtime = _with_config(model={"provider": "openai-codex"}) providers._account_usage_status_cache.clear() calls = [] unavailable_snapshot = SimpleNamespace( provider="openai-codex", source="usage_api_pool", title="Account limits", plan=None, windows=(), details=("0/1 credentials available", "1 exhausted"), available=False, unavailable_reason="Credential pool exhausted until 2030-03-17T18:46:40Z.", fetched_at=datetime(2030, 3, 17, 12, 30, tzinfo=timezone.utc), pool={"total_credentials": 1, "available_credentials": 0, "exhausted_credentials": 1, "credentials": []}, ) def fake_fetch(provider, home, api_key=None): calls.append((provider, str(home), api_key)) return unavailable_snapshot monkeypatch.setattr(providers, "_agent_fetch_account_usage_for_home", fake_fetch) try: first = providers._fetch_account_usage_with_profile_context("openai-codex") second = providers._fetch_account_usage_with_profile_context("openai-codex") finally: providers._account_usage_status_cache.clear() _restore_config(old_cfg, old_mtime) assert first is unavailable_snapshot assert second is unavailable_snapshot assert calls == [("openai-codex", str(tmp_path), None)] def test_account_usage_profile_fetch_does_not_cache_transient_none_results(monkeypatch, tmp_path): """Transient None probe results should not mask the next successful status check.""" import api.providers as providers monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path) old_cfg, old_mtime = _with_config(model={"provider": "openai-codex"}) providers._account_usage_status_cache.clear() calls = [] recovered_snapshot = SimpleNamespace( provider="openai-codex", source="usage_api_pool", title="Account limits", plan=None, windows=(), details=(), available=True, unavailable_reason=None, fetched_at=datetime(2030, 3, 17, 12, 31, tzinfo=timezone.utc), pool={"total_credentials": 1, "credentials": []}, ) def fake_fetch(provider, home, api_key=None): calls.append((provider, str(home), api_key)) return None if len(calls) == 1 else recovered_snapshot monkeypatch.setattr(providers, "_agent_fetch_account_usage_for_home", fake_fetch) try: first = providers._fetch_account_usage_with_profile_context("openai-codex") second = providers._fetch_account_usage_with_profile_context("openai-codex") third = providers._fetch_account_usage_with_profile_context("openai-codex") finally: providers._account_usage_status_cache.clear() _restore_config(old_cfg, old_mtime) assert first is None assert second is recovered_snapshot assert third is recovered_snapshot assert calls == [ ("openai-codex", str(tmp_path), None), ("openai-codex", str(tmp_path), None), ] def test_account_usage_profile_fetches_can_overlap_for_different_homes(monkeypatch, tmp_path): """Different profile quota fetches should not serialize on cron's global lock.""" import api.providers as providers homes = { "quota-a": tmp_path / "a", "quota-b": tmp_path / "b", } for home in homes.values(): home.mkdir() barrier = threading.Barrier(2, timeout=2) events = [] errors = [] def fake_home(): return homes[threading.current_thread().name] def fake_fetch(provider, home, api_key=None): events.append(("enter", str(home))) barrier.wait() events.append(("exit", str(home))) return None monkeypatch.setattr(providers, "_get_hermes_home", fake_home) monkeypatch.setattr(providers, "_agent_fetch_account_usage_for_home", fake_fetch) def worker(): try: providers._fetch_account_usage_with_profile_context("openai-codex") except Exception as exc: errors.append(exc) threads = [ threading.Thread(target=worker, name="quota-a"), threading.Thread(target=worker, name="quota-b"), ] for thread in threads: thread.start() for thread in threads: thread.join() assert not errors assert [kind for kind, _home in events[:2]] == ["enter", "enter"] def test_openai_api_key_detection_ignores_codex_oauth_jwt(monkeypatch, tmp_path): """A Codex OAuth JWT in OPENAI_API_KEY should not show a bare OpenAI card.""" monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path) monkeypatch.delenv("OPENAI_API_KEY", raising=False) def b64url(payload: bytes) -> str: return base64.urlsafe_b64encode(payload).rstrip(b"=").decode("ascii") codex_token = ".".join(( b64url(b'{"alg":"none","typ":"JWT"}'), b64url(json.dumps({ "https://api.openai.com/auth": {"chatgpt_account_id": "acct-codex"}, }).encode("utf-8")), b64url(b"signature"), )) (tmp_path / ".env").write_text(f"OPENAI_API_KEY={codex_token}\n", encoding="utf-8") old_cfg, old_mtime = _with_config(model={"provider": "openai"}) import api.providers as providers try: assert providers._provider_has_key("openai") is False assert providers._get_provider_api_key("openai") is None provider_ids = [p["id"] for p in providers.get_providers()["providers"]] finally: _restore_config(old_cfg, old_mtime) assert "openai" not in provider_ids assert "openai-codex" in provider_ids assert codex_token not in repr(provider_ids) def test_openai_api_key_detection_still_accepts_real_api_keys(monkeypatch, tmp_path): """Filtering Codex OAuth tokens must not hide real OpenAI API keys.""" monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path) monkeypatch.delenv("OPENAI_API_KEY", raising=False) (tmp_path / ".env").write_text("OPENAI_API_KEY=sk-test-real-openai-key\n", encoding="utf-8") old_cfg, old_mtime = _with_config(model={"provider": "openai"}) import api.providers as providers try: assert providers._provider_has_key("openai") is True assert providers._get_provider_api_key("openai") == "sk-test-real-openai-key" finally: _restore_config(old_cfg, old_mtime) def test_openai_api_key_detection_falls_through_after_codex_jwt_config_value(monkeypatch, tmp_path): """A filtered OpenAI config value should not mask a later real API key source.""" monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path) monkeypatch.delenv("OPENAI_API_KEY", raising=False) def b64url(payload: bytes) -> str: return base64.urlsafe_b64encode(payload).rstrip(b"=").decode("ascii") codex_token = ".".join(( b64url(b'{"alg":"none","typ":"JWT"}'), b64url(json.dumps({ "https://api.openai.com/auth": {"chatgpt_account_id": "acct-codex"}, }).encode("utf-8")), b64url(b"signature"), )) old_cfg, old_mtime = _with_config( model={"provider": "openai", "api_key": codex_token}, providers={"openai": {"api_key": "sk-config-openai-key"}}, ) import api.providers as providers try: assert providers._provider_has_key("openai") is True assert providers._get_provider_api_key("openai") == "sk-config-openai-key" finally: _restore_config(old_cfg, old_mtime) def test_provider_quota_route_is_registered(): """The backend must expose a route for the UI to poll quota status.""" routes = (ROOT / "api" / "routes.py").read_text(encoding="utf-8") assert 'parsed.path == "/api/provider/quota"' in routes assert 'query.get("refresh", [""])' in routes assert "get_provider_quota(provider_id, refresh=refresh)" in routes def test_provider_quota_card_is_rendered_in_providers_panel(): """The Providers panel should show active provider quota/status before cards.""" panels = (ROOT / "static" / "panels.js").read_text(encoding="utf-8") assert "_fetchProviderQuotaStatus(false)" in panels assert "'/api/provider/quota'" in panels assert "function _buildProviderQuotaCard" in panels assert "provider_quota_title" in panels assert "provider-quota-card" in panels assert "account_limits" in panels assert "remaining_percent" in panels assert "provider-quota-details" in panels assert "provider_quota_credential_pool" in panels assert "provider-quota-pool-row" in panels assert "_buildProviderQuotaPoolBreakdown" in panels assert "_providerQuotaPoolShouldDefaultOpen" in panels assert "hermes-provider-quota-pool-open" in panels assert "provider-quota-pool-chevron" in panels assert 'aria-hidden="true"' in panels assert "count>0&&count<=3" in panels assert "status.status==='available'||accountLimits.pool" in panels assert "provider-quota-window-detail" in panels assert "provider_quota_session_limit" in panels assert "provider_quota_weekly_limit" in panels assert "_providerQuotaUnavailableReason" in panels assert "provider_quota_retry_after" in panels assert "accountLimits.details)&&!accountLimits.pool" in panels def test_provider_quota_card_has_manual_refresh_control(): """The quota card should let users force an immediate fresh usage lookup.""" panels = (ROOT / "static" / "panels.js").read_text(encoding="utf-8") assert "function _refreshProviderQuota" in panels assert "function _fetchProviderQuotaStatus" in panels assert "refresh=1" in panels assert "cache:'no-store'" in panels assert "data-provider-quota-refresh" in panels assert "provider_quota_refresh_usage" in panels assert "provider_quota_refresh_succeeded" in panels assert "provider_quota_refresh_failed" in panels assert "card.isConnected&&button" in panels assert "provider_quota_last_checked" in panels def test_provider_quota_i18n_keys_exist_for_all_locales(): """Provider quota UI keys must be present in every locale block.""" i18n = (ROOT / "static" / "i18n.js").read_text(encoding="utf-8") locale_count = len( re.findall(r"^ (?:[A-Za-z_][A-Za-z0-9_]*|'[^']+'):\s*\{", i18n, re.MULTILINE) ) keys = sorted(set(re.findall(r"provider_quota_[a-z0-9_]+", (ROOT / "static" / "panels.js").read_text(encoding="utf-8")))) assert locale_count >= 1 assert "provider_quota_retry_after" in keys for key in keys: assert len(re.findall(rf"^\s+{re.escape(key)}:", i18n, re.MULTILINE)) == locale_count, key def test_settings_label_and_description_i18n_keys_exist_for_all_locales(): """Settings labels/descriptions referenced by the page need every locale.""" i18n = (ROOT / "static" / "i18n.js").read_text(encoding="utf-8") index_html = (ROOT / "static" / "index.html").read_text(encoding="utf-8") locale_count = len( re.findall(r"^ (?:[A-Za-z_][A-Za-z0-9_]*|'[^']+'):\s*\{", i18n, re.MULTILINE) ) keys = sorted( set(re.findall(r'data-i18n="(settings_(?:label|desc)_[a-z0-9_]+)"', index_html)) ) assert locale_count >= 1 assert "settings_label_fade_text_effect" in keys assert "settings_desc_fade_text_effect" in keys for key in keys: assert len(re.findall(rf"^\s+{re.escape(key)}:", i18n, re.MULTILINE)) == locale_count, key def test_provider_quota_styles_exist(): """Quota UI should have visible supported/unavailable/invalid states.""" css = (ROOT / "static" / "style.css").read_text(encoding="utf-8") for token in ( ".provider-quota-card", ".provider-quota-metric", ".provider-quota-card-available", ".provider-quota-card-no_key", ".provider-quota-card-invalid_key", ".provider-quota-details", ".provider-quota-window", ".provider-quota-actions", ".provider-quota-refresh", ".provider-quota-checked", ".provider-quota-pool", ".provider-quota-pool-chevron", ".provider-quota-pool[open] .provider-quota-pool-chevron", ".provider-quota-pool-row", ".provider-quota-pool-window", ".provider-quota-window-detail", ): assert token in css # ── Regression tests for #1912 ──────────────────────────────────────────────── class _FakeAccountUsageWorkerProcess: def __init__(self, payloads=None): self.payloads = list(payloads or []) self.requests = [] self.returncode = None self.terminated = False self.killed = False self.stdin = self._Stdin(self) self.stdout = self._Stdout(self) class _Stdin: def __init__(self, proc): self.proc = proc self.closed = False def write(self, value): self.proc.requests.append(json.loads(value)) def flush(self): pass def close(self): self.closed = True class _Stdout: def __init__(self, proc): self.proc = proc self.closed = False def readline(self): request = self.proc.requests[-1] if self.proc.payloads: payload = self.proc.payloads.pop(0) else: payload = { "provider": request["provider"], "source": "usage_api", "title": "Account limits", "windows": [], "details": [], "available": True, "unavailable_reason": None, "fetched_at": "2030-03-17T12:30:00Z", } return json.dumps(payload) + "\n" def close(self): self.closed = True def poll(self): return self.returncode def terminate(self): self.terminated = True self.returncode = -15 def wait(self, timeout=None): return self.returncode def kill(self): self.killed = True self.returncode = -9 def test_account_usage_worker_reuses_process_for_same_home(monkeypatch, tmp_path): import api.providers as providers import subprocess launched = [] def fake_popen(*args, **kwargs): proc = _FakeAccountUsageWorkerProcess() launched.append((args, kwargs, proc)) return proc monkeypatch.setattr(subprocess, "Popen", fake_popen) providers._close_account_usage_probe_workers() try: first = providers._agent_fetch_account_usage_for_home("openai-codex", tmp_path) second = providers._agent_fetch_account_usage_for_home("anthropic", tmp_path, api_key="sk-test") finally: providers._close_account_usage_probe_workers() assert len(launched) == 1 assert first.provider == "openai-codex" assert second.provider == "anthropic" assert launched[0][2].requests == [ {"provider": "openai-codex", "api_key": "", "env_var": None}, {"provider": "anthropic", "api_key": "sk-test", "env_var": "ANTHROPIC_API_KEY"}, ] def test_account_usage_worker_pool_is_keyed_by_home(monkeypatch, tmp_path): import api.providers as providers import subprocess launched = [] def fake_popen(*args, **kwargs): proc = _FakeAccountUsageWorkerProcess() launched.append((args, kwargs, proc)) return proc home_a = tmp_path / "a" home_b = tmp_path / "b" monkeypatch.setattr(subprocess, "Popen", fake_popen) providers._close_account_usage_probe_workers() try: providers._agent_fetch_account_usage_for_home("openai-codex", home_a) providers._agent_fetch_account_usage_for_home("openai-codex", home_b) finally: providers._close_account_usage_probe_workers() assert len(launched) == 2 def test_account_usage_worker_idle_cleanup_closes_stale_process(monkeypatch, tmp_path): import api.providers as providers import subprocess launched = [] def fake_popen(*args, **kwargs): proc = _FakeAccountUsageWorkerProcess() launched.append((args, kwargs, proc)) return proc monkeypatch.setattr(subprocess, "Popen", fake_popen) providers._close_account_usage_probe_workers() try: providers._agent_fetch_account_usage_for_home("openai-codex", tmp_path) workers = providers._account_usage_worker_pool[str(tmp_path)] # Pool is now a list of workers; pick the one that was last used worker = max(workers, key=lambda w: w.last_used) providers._cleanup_account_usage_probe_workers( now=worker.last_used + providers._ACCOUNT_USAGE_WORKER_IDLE_SECONDS + 1 ) providers._agent_fetch_account_usage_for_home("openai-codex", tmp_path) finally: providers._close_account_usage_probe_workers() assert len(launched) >= 2 assert launched[0][2].terminated is True def test_busy_account_usage_worker_uses_one_shot_fallback(monkeypatch, tmp_path): import api.providers as providers worker = providers._AccountUsageProbeWorker(tmp_path) calls = [] def fake_one_shot(provider, home, *, api_key=None): calls.append((provider, Path(home), api_key)) return SimpleNamespace(provider=provider, source="usage_api", windows=(), details=(), available=True) monkeypatch.setattr(providers, "_fetch_account_usage_once_for_home", fake_one_shot) locked = threading.Event() release = threading.Event() def hold_lock(): with worker._lock: locked.set() release.wait(timeout=5) holder = threading.Thread(target=hold_lock) holder.start() assert locked.wait(timeout=5) try: snapshot = worker.fetch("anthropic", api_key="sk-test") finally: release.set() holder.join(timeout=5) worker.close() assert snapshot.provider == "anthropic" assert calls == [("anthropic", tmp_path, "sk-test")] def test_account_usage_cleanup_removes_null_proc_worker(monkeypatch, tmp_path): import api.providers as providers import subprocess launched = [] def fake_popen(*args, **kwargs): proc = _FakeAccountUsageWorkerProcess() launched.append(proc) return proc monkeypatch.setattr(subprocess, "Popen", fake_popen) providers._close_account_usage_probe_workers() try: providers._agent_fetch_account_usage_for_home("openai-codex", tmp_path) workers = providers._account_usage_worker_pool[str(tmp_path)] # Pool is now a list of workers; close all of them to simulate null proc state for worker in workers: worker.close() providers._cleanup_account_usage_probe_workers() assert str(tmp_path) not in providers._account_usage_worker_pool finally: providers._close_account_usage_probe_workers() assert len(launched) >= 1 def test_provider_key_mutation_invalidates_warm_account_usage_workers(monkeypatch, tmp_path): import api.providers as providers invalidated = [] monkeypatch.setattr(providers, "_get_hermes_home", lambda: tmp_path) monkeypatch.setattr(providers, "invalidate_models_cache", lambda: None) monkeypatch.setattr( providers, "invalidate_account_usage_status_cache", lambda provider_id=None: invalidated.append(provider_id), ) updated = providers.set_provider_key("anthropic", "sk-test-quota-worker") removed = providers.set_provider_key("anthropic", None) assert updated["ok"] is True assert removed["ok"] is True assert invalidated == ["anthropic", "anthropic"] def test_account_usage_worker_uses_controlled_pipe_stdin(monkeypatch): """Account-usage probe workers must not inherit process stdin.""" import api.providers as providers import subprocess seen_stdin = None def capturing_popen(*args, **kwargs): nonlocal seen_stdin seen_stdin = kwargs.get('stdin') return _FakeAccountUsageWorkerProcess() monkeypatch.setattr(subprocess, 'Popen', capturing_popen) providers._close_account_usage_probe_workers() try: providers._agent_fetch_account_usage_for_home( 'openai-codex', Path('/nonexistent'), api_key=None ) finally: providers._close_account_usage_probe_workers() assert seen_stdin is subprocess.PIPE, ( f'expected stdin=subprocess.PIPE, got {seen_stdin!r}' ) def test_account_usage_probe_semaphore_has_correct_bound(monkeypatch, tmp_path): """The probe semaphore must enforce the declared concurrency cap. Verifying the bound directly ensures the cap actually prevents resource exhaustion when the UI polls multiple providers in rapid succession. """ import api.providers as providers monkeypatch.setattr(profiles, 'get_active_hermes_home', lambda: tmp_path) old_cfg, old_mtime = _with_config(model={'provider': 'openai-codex'}) sem = providers._get_account_usage_probe_semaphore() try: bound = sem._value assert bound == providers._MAX_CONCURRENT_ACCOUNT_USAGE_PROBES, ( f'semaphore bound is {bound}, expected ' f'{providers._MAX_CONCURRENT_ACCOUNT_USAGE_PROBES}' ) finally: _restore_config(old_cfg, old_mtime) def test_account_usage_preexec_fn_is_wired_on_posix(monkeypatch): """On POSIX systems the probe subprocess must receive a parent-death preexec_fn. The preexec_fn arranges prctl(PR_SET_PDEATHSIG, SIGTERM) so the child is terminated when the WebUI parent dies (OOM kill, systemctl restart, etc.). This test verifies the wiring and skips harmlessly on non-POSIX (Windows). """ import api.providers as providers assert callable(providers._account_usage_preexec_fn) try: providers._account_usage_preexec_fn() except Exception as exc: raise AssertionError( f'_account_usage_preexec_fn raised {exc!r}; it should be ' 'safe to call unconditionally' ) from exc if hasattr(os, 'fork'): import subprocess captured_kwargs = {} def capture_popen(*args, **kwargs): captured_kwargs.update(kwargs) return _FakeAccountUsageWorkerProcess() monkeypatch.setattr(subprocess, 'Popen', capture_popen) providers._close_account_usage_probe_workers() try: providers._agent_fetch_account_usage_for_home( 'openai-codex', Path('/nonexistent'), api_key=None ) finally: providers._close_account_usage_probe_workers() assert 'preexec_fn' in captured_kwargs, ( 'preexec_fn should be in subprocess.Popen kwargs on POSIX' ) assert captured_kwargs['preexec_fn'] is providers._account_usage_preexec_fn def test_account_usage_semaphore_caps_concurrency(monkeypatch, tmp_path): """The probe semaphore must actually serialise callers beyond its bound. Verifies the bounded semaphore is used in the call path and genuinely prevents more than _MAX_CONCURRENT_ACCOUNT_USAGE_PROBES probes running. """ import api.providers as providers import threading monkeypatch.setattr(profiles, 'get_active_hermes_home', lambda: tmp_path) old_cfg, old_mtime = _with_config(model={'provider': 'openai-codex'}) active = 0 max_active = 0 entered = 0 first_two_entered = threading.Event() third_entered = threading.Event() lock = threading.Lock() unblock = threading.Event() def slow_fetch(provider, home, api_key=None): nonlocal active, max_active, entered with lock: active += 1 entered += 1 max_active = max(max_active, active) if entered == providers._MAX_CONCURRENT_ACCOUNT_USAGE_PROBES: first_two_entered.set() if entered > providers._MAX_CONCURRENT_ACCOUNT_USAGE_PROBES: third_entered.set() unblock.wait(timeout=5) with lock: active -= 1 return None monkeypatch.setattr(providers, '_agent_fetch_account_usage_for_home', slow_fetch) results = [] errors = [] def worker(): try: results.append( providers._fetch_account_usage_with_profile_context('openai-codex') ) except Exception as exc: errors.append(exc) threads = [ threading.Thread(target=worker) for _ in range(providers._MAX_CONCURRENT_ACCOUNT_USAGE_PROBES + 1) ] for t in threads: t.start() try: assert first_two_entered.wait(timeout=2), 'first probe batch did not start' assert not third_entered.wait(timeout=0.2), 'third probe bypassed semaphore' unblock.set() for t in threads: t.join(timeout=10) assert not errors, f'workers raised: {errors}' assert len(results) == len(threads), f'expected {len(threads)} results, got {len(results)}' assert max_active <= providers._MAX_CONCURRENT_ACCOUNT_USAGE_PROBES finally: unblock.set() _restore_config(old_cfg, old_mtime)