From 7d77877358dd09003846dda154e14fd475800eeb Mon Sep 17 00:00:00 2001 From: Hermes Team Date: Sat, 22 Aug 2026 23:45:37 +0700 Subject: [PATCH] fix: show live quotas in fixed account cards --- .../router/quota_collector.py | 141 +++++++++++++----- .../router/ui/components.py | 116 +++++++++----- .../router/ui/views/accounts_view.py | 28 +--- .../router/ui/views/dashboard_view.py | 17 ++- tests/test_accounts_tariffs_quotas.py | 71 +++++++++ tests/test_ui_contract_v11.py | 23 +++ 6 files changed, 288 insertions(+), 108 deletions(-) diff --git a/src/antigravity_provider/router/quota_collector.py b/src/antigravity_provider/router/quota_collector.py index bb0a314..5b70dc5 100644 --- a/src/antigravity_provider/router/quota_collector.py +++ b/src/antigravity_provider/router/quota_collector.py @@ -150,6 +150,11 @@ class AccountQuotaService: except Exception as e: logger.warning("Error fetching quota for %s/%s: %s", provider, profile_id, e) snap = self._generate_baseline_snapshot(provider, profile_id) + status = getattr(e, "status", None) + if status == 401 or "401" in str(e) or "unauthenticated" in str(e).lower(): + snap.unavailable_reason = "Авторизация истекла — обновите подключение" + else: + snap.unavailable_reason = "Провайдер не вернул данные лимитов" with self._cache_lock: self._snapshots[key] = snap @@ -330,9 +335,32 @@ class AccountQuotaService: if not access_token: raise RuntimeError("В профиле Antigravity отсутствует access token") + refresh_token = token_data.get("refresh_token") or token_data.get("refresh") + + def _refresh_and_save() -> str: + if not refresh_token: + raise RuntimeError("OAuth-сессия истекла, refresh token отсутствует") + refreshed = refresh_access_token(str(refresh_token)) + token_data.update(refreshed) + auth_data["token"] = token_data + ProfileAuthManager.save_profile_auth("antigravity", profile_id, auth_data) + return str(refreshed["access_token"]) + + expiry = _parse_datetime(token_data.get("expires_at") or token_data.get("expiry")) + if expiry and expiry <= now + timedelta(seconds=60): + access_token = _refresh_and_save() + project_id = auth_data.get("project_id") or auth_data.get("projectId") if not project_id: - project_id = load_or_onboard_project(str(access_token)) + try: + project_id = load_or_onboard_project(str(access_token)) + except Exception as exc: + if (getattr(exc, "status", None) != 401 and "401" not in str(exc)) or not refresh_token: + raise + access_token = _refresh_and_save() + project_id = load_or_onboard_project(str(access_token)) + auth_data["project_id"] = project_id + ProfileAuthManager.save_profile_auth("antigravity", profile_id, auth_data) def _fetch(token: str) -> dict[str, Any]: request = urllib.request.Request( @@ -351,15 +379,9 @@ class AccountQuotaService: try: payload = _fetch(str(access_token)) except urllib.error.HTTPError as exc: - refresh_token = token_data.get("refresh_token") or token_data.get("refresh") if exc.code != 401 or not refresh_token: raise - refreshed = refresh_access_token(str(refresh_token)) - token_data.update(refreshed) - auth_data["token"] = token_data - auth_data["project_id"] = project_id - ProfileAuthManager.save_profile_auth("antigravity", profile_id, auth_data) - payload = _fetch(str(refreshed["access_token"])) + payload = _fetch(_refresh_and_save()) models = payload.get("models") or {} if not isinstance(models, dict): @@ -440,44 +462,85 @@ class AccountQuotaService: ) def _collect_opencode_quota(self, profile_id: str, auth_data: dict) -> QuotaSnapshot: - """Collect Sliding, Weekly, and Monthly usage for OpenCode Go.""" + """Validate OpenCode Go entitlement and read usage when its API exposes it.""" now = _utc_now() - b_sliding = QuotaBucket( - id="opencode.sliding", - display_name="Скользящее", - model_family="opencode", - used_percent=None, - remaining_percent=None, - period="sliding", - status="unknown", - ) - b_weekly = QuotaBucket( - id="opencode.weekly", - display_name="Недельное", - model_family="opencode", - used_percent=None, - remaining_percent=None, - period="7d", - reset_at=now + timedelta(days=7), - status="unknown", - ) - b_monthly = QuotaBucket( - id="opencode.monthly", - display_name="Ежемесячное", - model_family="opencode", - used_percent=None, - remaining_percent=None, - period="30d", - reset_at=now + timedelta(days=30), - status="unknown", + api_key = auth_data.get("api_key") + if not api_key: + raise RuntimeError("Ключ OpenCode Go не сохранён") + + base_url = "https://opencode.ai/zen/go/v1" + + def _get(path: str) -> dict[str, Any]: + request = urllib.request.Request( + f"{base_url}{path}", + headers={ + "Authorization": f"Bearer {api_key}", + "Accept": "application/json", + "User-Agent": "hermes-hub/1.0", + }, + ) + with urllib.request.urlopen(request, timeout=20) as response: + return json.loads(response.read().decode("utf-8") or "{}") + + # /models is the documented read-only endpoint and confirms that the + # key is accepted without spending a request from the user's limit. + _get("/models") + usage: dict[str, Any] = {} + unavailable_reason: Optional[str] = None + try: + usage = _get("/usage") + except urllib.error.HTTPError as exc: + raw = exc.read().decode("utf-8", "replace") + if exc.code == 403 and "subscription required" in raw.lower(): + unavailable_reason = "Для этого ключа не активна подписка OpenCode Go" + elif exc.code in (403, 404): + unavailable_reason = "OpenCode Go не предоставляет остаток через публичный API" + else: + raise + + def _metric(*names: str) -> dict[str, Any]: + for name in names: + value = usage.get(name) + if isinstance(value, dict): + return value + return {} + + buckets: list[QuotaBucket] = [] + specs = ( + ("5h", "Лимит 5 часов", 12, _metric("five_hour", "fiveHour", "sliding")), + ("7d", "Недельный лимит", 30, _metric("weekly", "seven_day", "sevenDay")), + ("30d", "Месячный лимит", 60, _metric("monthly", "thirty_day", "thirtyDay")), ) + for period, label, limit_value, metric in specs: + remaining_percent = metric.get("remaining_percent", metric.get("remainingPercentage")) + remaining_absolute = metric.get("remaining", metric.get("remaining_amount")) + used_absolute = metric.get("used", metric.get("used_amount")) + reset_at = _parse_datetime(metric.get("reset_at") or metric.get("resetTime")) + buckets.append( + QuotaBucket( + id=f"opencode.{period}", + display_name=label, + model_family="opencode", + remaining_percent=float(remaining_percent) if isinstance(remaining_percent, (int, float)) else None, + used_absolute=int(used_absolute) if isinstance(used_absolute, (int, float)) else None, + remaining_absolute=( + int(remaining_absolute) if isinstance(remaining_absolute, (int, float)) else None + ), + limit_absolute=limit_value, + reset_at=reset_at, + period=period, + unit="USD", + scope="account", + ) + ) return QuotaSnapshot( account_id=profile_id, provider="opencode-go", - buckets=[b_sliding, b_weekly, b_monthly], + buckets=buckets, fetched_at=now, - source="baseline", + source="provider_api", + unavailable_reason=unavailable_reason, ) def _collect_claude_quota(self, profile_id: str, auth_data: dict) -> QuotaSnapshot: diff --git a/src/antigravity_provider/router/ui/components.py b/src/antigravity_provider/router/ui/components.py index 29e6472..e2284c8 100644 --- a/src/antigravity_provider/router/ui/components.py +++ b/src/antigravity_provider/router/ui/components.py @@ -868,7 +868,7 @@ class AccountCardWidget(HubCard): self.profile_id = profile_id self.profile_model: Any = None self.on_action = on_action - self.compact = compact + self.compact = True self._quota_widgets: Dict[str, QuotaBucketWidget] = {} self.widgets_created = 0 self.widgets_destroyed = 0 @@ -877,8 +877,8 @@ class AccountCardWidget(HubCard): top.pack(fill="x", padx=Theme.CARD_PAD_X, pady=(Theme.CARD_PAD_Y, Theme.SPACE_XS)) self.provider = ctk.CTkLabel(top, text=provider, font=Theme.font_micro(), text_color=Theme.TEXT_MUTED) self.provider.pack(side="left") - self.toggle = IconButton(top, text="▾", command=self.toggle_compact) - self.toggle.pack(side="right") + self.profile_mark = ctk.CTkLabel(top, text="◇", font=Theme.font_caption(), text_color=Theme.TEXT_ACCENT) + self.profile_mark.pack(side="right") self.identity = EllipsizedLabel(self, text=identity, font=Theme.font_body_bold(), text_color=Theme.TEXT_PRIMARY) self.identity.pack(anchor="w", padx=Theme.CARD_PAD_X) @@ -889,21 +889,37 @@ class AccountCardWidget(HubCard): self.status.pack(anchor="w", padx=Theme.CARD_PAD_X, pady=Theme.SPACE_SM) self.compact_quota = ctk.CTkFrame(self, fg_color="transparent") - self.compact_quota_labels = [ - ctk.CTkLabel( - self.compact_quota, - text="", - font=Theme.font_micro(), - text_color=Theme.TEXT_SECONDARY, - anchor="w", + self.compact_quota_cells: list[dict[str, Any]] = [] + for index in range(4): + cell = ctk.CTkFrame(self.compact_quota, fg_color="transparent") + cell.grid(row=index // 2, column=index % 2, sticky="nsew", padx=(0, 10), pady=(2, 6)) + title = ctk.CTkLabel(cell, text="", font=Theme.font_micro(), text_color=Theme.TEXT_PRIMARY, anchor="w") + title.pack(fill="x") + value = ctk.CTkLabel(cell, text="", font=Theme.font_caption(), text_color=Theme.TEXT_SECONDARY, anchor="e") + value.pack(fill="x") + progress = ctk.CTkProgressBar( + cell, + height=5, + corner_radius=3, + fg_color=Theme.SURFACE_MUTED, + progress_color=Theme.STATUS_HEALTHY, + ) + progress.pack(fill="x", pady=(2, 1)) + progress.set(0) + reset = ctk.CTkLabel(cell, text="", font=Theme.font_micro(), text_color=Theme.TEXT_MUTED, anchor="w") + reset.pack(fill="x") + self.compact_quota_cells.append( + {"frame": cell, "title": title, "value": value, "progress": progress, "reset": reset} ) - for _ in range(4) - ] - for index, label in enumerate(self.compact_quota_labels): - label.grid(row=index // 2, column=index % 2, sticky="ew", padx=(0, 8), pady=2) self.compact_quota.grid_columnconfigure(index % 2, weight=1) self.compact_actions = ctk.CTkFrame(self, fg_color="transparent") - for text, action in (("⚡", "test"), ("↻", "refresh_account"), ("Роль", "assign_role")): + for text, action in ( + ("⚡", "test"), + ("★", "set_main"), + ("♛", "set_orchestrator"), + ("Роль", "assign_role"), + ("↻", "refresh_account"), + ): ActionButton( self.compact_actions, text=text, @@ -961,7 +977,7 @@ class AccountCardWidget(HubCard): width=72, command=lambda: self._trigger("delete_credentials"), ).pack(side="right") - self.set_compact(compact) + self.set_compact(True) @staticmethod def resolve_identity(profile: Any) -> str: @@ -992,19 +1008,14 @@ class AccountCardWidget(HubCard): self.action_feedback.pack(fill="x", pady=(0, Theme.SPACE_XS), before=self.actions.winfo_children()[1]) def toggle_compact(self) -> None: - self.set_compact(not self.compact) + # Account cards intentionally have one fixed Cockpit-style layout. + self.set_compact(True) def set_compact(self, compact: bool) -> None: - self.compact = compact - self.toggle.configure(text="▸" if compact else "▾") - if compact: - self.details.pack_forget() - self.compact_quota.pack(fill="x", padx=Theme.CARD_PAD_X, pady=(0, Theme.SPACE_XS)) - self.compact_actions.pack(fill="x", padx=Theme.CARD_PAD_X, pady=(0, Theme.CARD_PAD_Y)) - else: - self.compact_quota.pack_forget() - self.compact_actions.pack_forget() - self.details.pack(fill="x") + self.compact = True + self.details.pack_forget() + self.compact_quota.pack(fill="x", padx=Theme.CARD_PAD_X, pady=(0, Theme.SPACE_XS)) + self.compact_actions.pack(fill="x", padx=Theme.CARD_PAD_X, pady=(0, Theme.CARD_PAD_Y)) def update_account(self, profile: Any, quota_snapshot: Optional[Any] = None) -> None: self.profile_model = profile @@ -1024,28 +1035,53 @@ class AccountCardWidget(HubCard): ) else: self.plan_badge.pack_forget() - self.status.set_status(profile.health_state, getattr(profile, "health_label_ru", None)) self.configure(border_color=Theme.BORDER_ACCENT if profile.is_main_account else Theme.BORDER) snapshot = quota_snapshot or getattr(profile, "quota_snapshot", None) buckets = list(getattr(snapshot, "buckets", None) or []) estimated = bool(getattr(snapshot, "is_estimated", True)) if snapshot else True - for index, label in enumerate(self.compact_quota_labels): + unavailable_reason = getattr(snapshot, "unavailable_reason", None) if snapshot else None + measured_remaining = [ + float(bucket.remaining_percent) + for bucket in buckets + if getattr(bucket, "remaining_percent", None) is not None + ] + if measured_remaining and not estimated: + if all(remaining <= 0 for remaining in measured_remaining): + card_health, card_label = "quota_exhausted", "Квота исчерпана" + elif any(remaining <= 0 for remaining in measured_remaining): + card_health, card_label = "warning", "Часть квот исчерпана" + else: + card_health, card_label = "healthy", "Работает" + else: + card_health = profile.health_state + card_label = getattr(profile, "health_label_ru", None) + self.status.set_status(card_health, card_label) + for index, cell in enumerate(self.compact_quota_cells): if index < len(buckets): bucket = buckets[index] - label.configure( - text=f"{bucket.display_name}: {bucket.formatted_remaining()}", - text_color=( - Theme.STATUS_HEALTHY - if getattr(bucket, "status", "unknown") == "healthy" - else Theme.STATUS_WARNING - if getattr(bucket, "status", "unknown") == "warning" - else Theme.TEXT_MUTED - ), + remaining = getattr(bucket, "remaining_percent", None) + color = ( + Theme.STATUS_HEALTHY + if getattr(bucket, "status", "unknown") == "healthy" + else Theme.STATUS_WARNING + if getattr(bucket, "status", "unknown") == "warning" + else Theme.TEXT_MUTED ) - label.grid() + detail = bucket.formatted_remaining() + if detail == "Н/Д" and unavailable_reason: + limit = getattr(bucket, "limit_absolute", None) + unit = getattr(bucket, "unit", None) + prefix = f"Лимит ${limit} • " if limit is not None and unit == "USD" else "" + detail = f"{prefix}{unavailable_reason}" + cell["title"].configure(text=bucket.display_name) + cell["value"].configure(text=detail, text_color=color) + cell["progress"].configure(progress_color=color) + cell["progress"].set(float(remaining) / 100.0 if remaining is not None else 0) + cell["reset"].configure(text=bucket.formatted_reset() or "Период указан провайдером") + cell["frame"].grid() else: - label.grid_remove() + cell["frame"].grid_remove() seen: set[str] = set() for bucket in buckets: key = str(getattr(bucket, "id", "") or getattr(bucket, "display_name", "bucket")) diff --git a/src/antigravity_provider/router/ui/views/accounts_view.py b/src/antigravity_provider/router/ui/views/accounts_view.py index bf5ac17..13d0e34 100644 --- a/src/antigravity_provider/router/ui/views/accounts_view.py +++ b/src/antigravity_provider/router/ui/views/accounts_view.py @@ -29,7 +29,7 @@ PROVIDER_LABELS = { class ProviderGroup(ctk.CTkFrame): - """Collapsible provider section that retains its account cards while hidden.""" + """Fixed provider section with always-visible Cockpit-style cards.""" def __init__(self, master: Any, provider: str): super().__init__(master, fg_color="transparent") @@ -37,21 +37,13 @@ class ProviderGroup(ctk.CTkFrame): self.collapsed = False self.header = ctk.CTkFrame(self, fg_color=Theme.BG_HEADER, corner_radius=Theme.RADIUS_SM) self.header.pack(fill="x", pady=(Theme.SPACE_SM, Theme.SPACE_XS)) - self.toggle = ActionButton( - self.header, - text="▾", - variant="ghost", - width=34, - command=self.toggle_collapsed, - ) - self.toggle.pack(side="left", padx=(Theme.SPACE_SM, 0)) self.title = ctk.CTkLabel( self.header, text=PROVIDER_LABELS.get(provider, provider), font=Theme.font_heading(), text_color=Theme.TEXT_PRIMARY, ) - self.title.pack(side="left", padx=Theme.SPACE_SM, pady=Theme.SPACE_SM) + self.title.pack(side="left", padx=Theme.SPACE_MD, pady=Theme.SPACE_SM) self.count = ctk.CTkLabel(self.header, text="0", font=Theme.font_caption(), text_color=Theme.TEXT_MUTED) self.count.pack(side="right", padx=Theme.SPACE_MD) self.body = ctk.CTkFrame(self, fg_color="transparent") @@ -60,12 +52,8 @@ class ProviderGroup(ctk.CTkFrame): self.body.grid_columnconfigure(column, weight=1) def toggle_collapsed(self) -> None: - self.collapsed = not self.collapsed - self.toggle.configure(text="▸" if self.collapsed else "▾") - if self.collapsed: - self.body.pack_forget() - else: - self.body.pack(fill="x") + self.collapsed = False + self.body.pack(fill="x") class AccountsView(ctk.CTkFrame): @@ -91,7 +79,7 @@ class AccountsView(ctk.CTkFrame): header = SectionHeader( self, title="Аккаунты и квоты", - subtitle="Реальные идентичности, независимые лимитные корзины и резервные роли", + subtitle="Компактные карточки аккаунтов, реальные остатки и быстрые действия", action_text="+ Добавить аккаунт", action_cmd=lambda: self._emit("add_account", {}), ) @@ -203,7 +191,6 @@ class AccountsView(ctk.CTkFrame): return self._snapshot = snapshot live_ids = {profile.profile_id for profile in self._profiles()} - collapse_new_collection = len(self._cards) <= 1 and len(live_ids) > 1 for profile_id in list(self._cards): if profile_id not in live_ids: self._cards.pop(profile_id).destroy() @@ -221,15 +208,12 @@ class AccountsView(ctk.CTkFrame): profile.profile_id, AccountCardWidget.resolve_identity(profile), profile.provider_display_name, - compact=len(live_ids) > 1, + compact=True, on_action=self._emit, ) self._cards[profile.profile_id] = card self.cards_created += 1 card.update_account(profile, snapshot.quotas.get(profile.profile_id)) - if collapse_new_collection: - for card in self._cards.values(): - card.set_compact(True) self._render_visibility() def _render_visibility(self) -> None: diff --git a/src/antigravity_provider/router/ui/views/dashboard_view.py b/src/antigravity_provider/router/ui/views/dashboard_view.py index 418237e..9e81904 100644 --- a/src/antigravity_provider/router/ui/views/dashboard_view.py +++ b/src/antigravity_provider/router/ui/views/dashboard_view.py @@ -83,10 +83,10 @@ class _EndpointCard(HubCard): self.subtitle.pack(anchor="w", pady=(0, 1)) self.status = ctk.CTkLabel(text, text="", font=Theme.font_micro(), text_color=Theme.STATUS_HEALTHY) self.status.pack(anchor="w") - quota = ctk.CTkFrame(self, fg_color="transparent", width=58) + quota = ctk.CTkFrame(self, fg_color="transparent", width=92) quota.pack(side="right", fill="y", padx=(3, 8), pady=8) quota.pack_propagate(False) - self.quota_label = ctk.CTkLabel(quota, text="Н/Д", font=Theme.font_micro(), text_color=Theme.TEXT_SECONDARY) + self.quota_label = ctk.CTkLabel(quota, text="—", font=Theme.font_micro(), text_color=Theme.TEXT_SECONDARY) self.quota_label.pack(anchor="e") self.progress = ctk.CTkProgressBar( quota, height=4, corner_radius=2, progress_color=Theme.STATUS_HEALTHY, fg_color=Theme.SURFACE_MUTED @@ -195,7 +195,10 @@ class _RouteDiagram(ctk.CTkFrame): self.context, text="▤ Хранилище контекста", font=Theme.font_caption(), text_color=Theme.TEXT_PRIMARY ).pack(pady=(6, 0)) self.context_status = ctk.CTkLabel( - self.context, text="● Состояние: Н/Д", font=Theme.font_micro(), text_color=Theme.TEXT_MUTED + self.context, + text="● Нет телеметрии хранилища", + font=Theme.font_micro(), + text_color=Theme.TEXT_MUTED, ) self.context_status.pack() self._left_labels = ["", "", ""] @@ -513,7 +516,7 @@ class DashboardView(ctk.CTkFrame): for bucket in quota.buckets: if bucket.remaining_percent is not None: return f"{bucket.remaining_percent:.0f}%", float(bucket.remaining_percent) - return "Н/Д", None + return "Нет данных API", None @staticmethod def _sync_endpoint_cards( @@ -633,7 +636,7 @@ class DashboardView(ctk.CTkFrame): agent.role_name_ru, f"{agent.provider_display_name} • {agent.model}", "Здорово" if agent.is_active else agent.status_label_ru, - agent.active_quota_label or "Н/Д", + agent.active_quota_label or "Нет данных API", None, "healthy" if agent.is_active else "warning", ) @@ -658,12 +661,12 @@ class DashboardView(ctk.CTkFrame): left_labels: list[str] = [] for provider in providers: share = dict(provider_telemetry.get(provider.provider_id) or {}).get("call_share") - left_labels.append(f"{share:.0%}" if share is not None else "Н/Д") + left_labels.append(f"{share:.0%}" if share is not None else "") right_labels: list[str] = [] for agent in agents: measured = dict(role_telemetry.get(agent.role_id) or {}) calls = measured.get("total_calls") if measured.get("has_data") else None - right_labels.append(f"{calls} выз." if calls is not None else "Н/Д") + right_labels.append(f"{calls} выз." if calls is not None else "") self.route_diagram.set_labels(left_labels, right_labels) live_provider_ids = {provider.provider_id for provider in providers} diff --git a/tests/test_accounts_tariffs_quotas.py b/tests/test_accounts_tariffs_quotas.py index 9b47450..f642477 100644 --- a/tests/test_accounts_tariffs_quotas.py +++ b/tests/test_accounts_tariffs_quotas.py @@ -2,9 +2,11 @@ from __future__ import annotations import json +import io import os import tempfile import time +import urllib.error from datetime import datetime, timedelta, timezone from pathlib import Path from unittest.mock import MagicMock, patch @@ -222,6 +224,75 @@ def test_antigravity_separate_claude_and_gemini_buckets(): assert b_g.remaining_percent == 87.0 +def test_antigravity_refreshes_expired_token_before_project_discovery(): + service = AccountQuotaService() + response = MagicMock() + response.__enter__.return_value.read.return_value = json.dumps( + { + "models": { + "claude-sonnet-4-6": { + "quotaInfo": {"remainingFraction": 0.64, "resetTime": "2026-08-23T00:00:00Z"} + }, + "gemini-3.7-flash": { + "quotaInfo": {"remainingFraction": 0.91, "resetTime": "2026-08-23T00:00:00Z"} + }, + } + } + ).encode() + auth_data = { + "token": { + "access_token": "expired-token", + "refresh_token": "refresh-token", + "expires_at": "2026-08-22T00:00:00Z", + } + } + + with patch( + "antigravity_provider.oauth.refresh_access_token", + return_value={"access_token": "fresh-token", "expires_at": "2026-08-23T00:00:00Z"}, + ) as refresh, patch( + "antigravity_provider.cloudcode.load_or_onboard_project", return_value="project-1" + ) as discover, patch( + "antigravity_provider.router.quota_collector.ProfileAuthManager.save_profile_auth" + ) as save, patch( + "antigravity_provider.router.quota_collector.urllib.request.urlopen", return_value=response + ) as urlopen: + snapshot = service._collect_antigravity_quota("ag-w1", auth_data) + + refresh.assert_called_once_with("refresh-token") + discover.assert_called_once_with("fresh-token") + assert "Bearer fresh-token" == urlopen.call_args.args[0].headers["Authorization"] + assert save.called + assert auth_data["project_id"] == "project-1" + assert snapshot.source == "provider_api" + assert snapshot.get_bucket_for_model("claude-sonnet-4-6").remaining_percent == 64.0 + + +def test_opencode_shows_published_limits_and_subscription_error(): + service = AccountQuotaService() + models_response = MagicMock() + models_response.__enter__.return_value.read.return_value = b'{"data": []}' + entitlement_error = urllib.error.HTTPError( + "https://opencode.ai/zen/go/v1/usage", + 403, + "Forbidden", + {}, + io.BytesIO(b'{"message":"OpenCode Go subscription required"}'), + ) + + with patch( + "antigravity_provider.router.quota_collector.urllib.request.urlopen", + side_effect=[models_response, entitlement_error], + ): + snapshot = service._collect_opencode_quota("opengo-1", {"api_key": "test-key"}) + + assert snapshot.source == "provider_api" + assert snapshot.unavailable_reason == "Для этого ключа не активна подписка OpenCode Go" + assert [bucket.limit_absolute for bucket in snapshot.buckets] == [12, 30, 60] + assert [bucket.period for bucket in snapshot.buckets] == ["5h", "7d", "30d"] + assert all(bucket.remaining_percent is None for bucket in snapshot.buckets) + + def test_health_tracker_antigravity_claude_exhaustion_does_not_block_gemini(tmp_path): state_file = tmp_path / "router_state.json" tracker = HealthTracker(state_file=state_file) diff --git a/tests/test_ui_contract_v11.py b/tests/test_ui_contract_v11.py index 3605bf1..8fad135 100644 --- a/tests/test_ui_contract_v11.py +++ b/tests/test_ui_contract_v11.py @@ -97,6 +97,29 @@ def test_plan_badge_distinguishes_trusted_inferred_and_unknown(ui_root) -> None: card.destroy() +@pytest.mark.ui +def test_live_quota_overrides_stale_exhausted_card_status(ui_root) -> None: + card = AccountCardWidget(ui_root, "account-1", "user@example.test", "Antigravity") + profile = replace(_profile(), health_state="quota_exhausted", health_label_ru="Квота исчерпана") + snapshot = QuotaSnapshot( + account_id="account-1", + provider="antigravity", + source="provider_api", + buckets=[ + QuotaBucket(id="claude", display_name="Claude", remaining_percent=100.0), + QuotaBucket(id="gemini", display_name="Gemini", remaining_percent=100.0), + ], + ) + try: + card.pack() + card.update_account(profile, snapshot) + ui_root.update_idletasks() + assert card.status.label.cget("text") == "Работает" + assert card.status.dot.cget("text_color") == Theme.STATUS_HEALTHY + finally: + card.destroy() + + @pytest.mark.ui def test_quota_missing_is_not_rendered_as_zero_and_reason_is_visible(ui_root) -> None: widget = QuotaBucketWidget(ui_root, "bucket", "Claude 5h")