austnv/remnawave-minishop
1
0
Fork 0
Code Issues 1 Pull Requests Activity

feat(admin): surface configuration problems in the admin panel

Browse Source 
Add GET /api/admin/health powered by a config health service that
detects common deployment mistakes: missing or read-only data volume,
broken tariffs/locale-override/guides JSON files, payment providers
enabled without credentials, webhook providers without
WEBHOOK_BASE_URL, no enabled payment methods, missing or non-https
mini app URL, missing Redis, partially configured SMTP, untrusted
reverse proxy, invalid bot token, missing/mismatched/failing Telegram
webhook and unreachable Remnawave panel. Network checks (Telegram,
panel) are cached for two minutes; ?refresh=1 forces a re-check.

The admin UI shows the alerts as a banner on the dashboard with
per-section navigation chips and a manual re-check button, and as a
filtered banner inside each affected section. Alerts are localized
via admin_health_* keys with built-in Russian fallbacks.
This commit is contained in:
3252a8
2026-06-10 12:31:41 +03:00
parent ce6273a652
commit 1b2290ea66
12 changed files with 1208 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/bot/app/web/admin_api.py
+2
View File
@@ -9,6 +9,7 @@ from bot.app.web.admin_api_impl import (
backups as _backups,
broadcast as _broadcast,
common as _common,
health as _health,
logs as _logs,
panel as _panel,
payments as _payments,
@@ -28,6 +29,7 @@ _MODULES = (
_runtime,
_auth,
_common,
_health,
_stats,
_users,
_payments,
backend/bot/app/web/admin_api_impl/health.py
+19
View File
@@ -0,0 +1,19 @@
# ruff: noqa: F401,F403,F405,I001
from datetime import datetime, timezone
from ._runtime import * # noqa: F403,F405
from .auth import _require_admin_user_id
from .common import _ok
from bot.services.config_health_service import collect_config_alerts
async def admin_health_route(request: web.Request) -> web.Response:
_require_admin_user_id(request)
refresh = str(request.query.get("refresh", "")).strip().lower() in {"1", "true", "yes"}
alerts = await collect_config_alerts(request, refresh=refresh)
return _ok(
{
"alerts": alerts,
"checked_at": datetime.now(timezone.utc).isoformat(),
}
)
backend/bot/app/web/admin_api_impl/routes.py
+1
View File
@@ -6,6 +6,7 @@ def setup_admin_routes(app: web.Application) -> None:
router = app.router
router.add_get("/api/admin/me", admin_me_route)
router.add_get("/api/admin/stats", admin_stats_route)
router.add_get("/api/admin/health", admin_health_route)
router.add_get("/api/admin/users", admin_users_list_route)
router.add_get("/api/admin/users/{user_id:-?\\d+}", admin_user_detail_route)
backend/bot/services/config_health_service.py
+507
View File
@@ -0,0 +1,507 @@
"""Detect common deployment misconfigurations for the admin panel.
Each check returns :class:`ConfigAlert` items the admin UI renders as
banners on the dashboard and inside the affected sections. Local checks
(filesystem, settings flags) run on every request; network checks
(Telegram webhook, Remnawave panel) are cached for a couple of minutes so
the dashboard stays fast and external APIs are not hammered.
"""
from __future__ import annotations
import asyncio
import json
import logging
import time
import uuid
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
from bot.utils.request_security import ip_in_allowlist
logger = logging.getLogger(__name__)
APP_ROOT = Path(__file__).resolve().parents[3]
NETWORK_CHECKS_TTL_SECONDS = 120.0
NETWORK_CHECK_TIMEOUT_SECONDS = 8.0
_WEBHOOK_ERROR_RECENT_SECONDS = 3600
_WEBHOOK_PENDING_THRESHOLD = 50
SEVERITY_ERROR = "error"
SEVERITY_WARNING = "warning"
# Admin section ids the frontend routes alerts to.
SECTION_SETTINGS = "settings"
SECTION_PAYMENTS = "payments"
SECTION_BACKUPS = "backups"
SECTION_TARIFFS = "tariffs"
SECTION_APPEARANCE = "appearance"
SECTION_TRANSLATIONS = "translations"
SECTION_USERS = "users"
_DATA_DIR_SECTIONS = (
SECTION_BACKUPS,
SECTION_TARIFFS,
SECTION_APPEARANCE,
SECTION_TRANSLATIONS,
SECTION_SETTINGS,
)
# Every message key an alert can carry. Tests assert each has
# ``admin_health_<key>`` entries in both locale files.
ALL_MESSAGE_KEYS = (
"data_dir_missing",
"data_dir_not_writable",
"backups_dir_not_writable",
"tariffs_config_invalid",
"locale_overrides_invalid",
"subscription_page_config_invalid",
"provider_not_configured",
"provider_webhook_needs_base_url",
"no_payment_methods",
"mini_app_url_missing",
"mini_app_url_not_https",
"redis_not_configured",
"smtp_incomplete",
"proxy_not_trusted",
"bot_token_invalid",
"telegram_api_error",
"telegram_webhook_missing",
"telegram_webhook_mismatch",
"telegram_webhook_error",
"telegram_webhook_pending",
"panel_api_not_configured",
"panel_api_unreachable",
)
@dataclass(frozen=True)
class ConfigAlert:
id: str
severity: str
sections: Tuple[str, ...]
params: Dict[str, Any] = field(default_factory=dict)
# Locale key suffix; defaults to ``id``. Per-provider alerts carry ids
# like ``provider_not_configured:wata`` but share one message key.
message_key: Optional[str] = None
def as_payload(self) -> Dict[str, Any]:
return {
"id": self.id,
"severity": self.severity,
"sections": list(self.sections),
"message_key": self.message_key or self.id,
"params": dict(self.params),
}
# ─── Filesystem checks ─────────────────────────────────────────────
def _dir_is_writable(path: Path) -> bool:
probe = path / f".health-probe-{uuid.uuid4().hex}.tmp"
try:
probe.write_text("ok", encoding="utf-8")
probe.unlink()
return True
except OSError:
try:
probe.unlink()
except OSError:
pass
return False
def _resolve_data_path(value: str) -> Path:
path = Path(value)
return path if path.is_absolute() else APP_ROOT / path
def data_dir_alerts(settings: Any, app_root: Path = APP_ROOT) -> List[ConfigAlert]:
alerts: List[ConfigAlert] = []
data_dir = app_root / "data"
if not data_dir.is_dir():
return [
ConfigAlert(
id="data_dir_missing",
severity=SEVERITY_ERROR,
sections=_DATA_DIR_SECTIONS,
params={"path": str(data_dir)},
)
]
if not _dir_is_writable(data_dir):
alerts.append(
ConfigAlert(
id="data_dir_not_writable",
severity=SEVERITY_ERROR,
sections=_DATA_DIR_SECTIONS,
params={"path": str(data_dir)},
)
)
backup_dir = _resolve_data_path(str(getattr(settings, "BACKUP_DIR", "") or "data/backups"))
if backup_dir.is_dir() and not _dir_is_writable(backup_dir):
alerts.append(
ConfigAlert(
id="backups_dir_not_writable",
severity=SEVERITY_WARNING,
sections=(SECTION_BACKUPS,),
params={"path": str(backup_dir)},
)
)
return alerts
def config_file_alerts(settings: Any) -> List[ConfigAlert]:
alerts: List[ConfigAlert] = []
tariffs_path = _resolve_data_path(
str(getattr(settings, "TARIFFS_CONFIG_PATH", "") or "data/tariffs.json")
)
if tariffs_path.is_file():
try:
from config.tariffs_config import load_tariffs_config
load_tariffs_config(tariffs_path)
except Exception as exc:
alerts.append(
ConfigAlert(
id="tariffs_config_invalid",
severity=SEVERITY_ERROR,
sections=(SECTION_TARIFFS,),
params={"path": str(tariffs_path), "error": str(exc)[:300]},
)
)
locale_overrides_path = APP_ROOT / "data" / "locales-overrides.json"
if locale_overrides_path.is_file():
try:
json.loads(locale_overrides_path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
alerts.append(
ConfigAlert(
id="locale_overrides_invalid",
severity=SEVERITY_WARNING,
sections=(SECTION_TRANSLATIONS,),
params={"path": str(locale_overrides_path), "error": str(exc)[:300]},
)
)
try:
from config.subscription_guides_config import (
SubscriptionGuidesConfigError,
subscription_guides_admin_config_json,
)
try:
subscription_guides_admin_config_json(settings)
except SubscriptionGuidesConfigError as exc:
alerts.append(
ConfigAlert(
id="subscription_page_config_invalid",
severity=SEVERITY_WARNING,
sections=(SECTION_SETTINGS,),
params={"error": str(exc)[:300]},
)
)
except Exception: # pragma: no cover - defensive import guard
logger.exception("Subscription guides config check failed unexpectedly")
return alerts
# ─── Settings checks ───────────────────────────────────────────────
def payment_provider_alerts(settings: Any, app: Any) -> List[ConfigAlert]:
from bot.payment_providers import iter_provider_specs
alerts: List[ConfigAlert] = []
any_enabled = False
seen_services: set = set()
for spec in iter_provider_specs():
try:
enabled = spec.is_effectively_enabled(settings)
except Exception: # pragma: no cover - provider config errors
logger.exception("Provider %s enabled check failed", spec.id)
continue
if not enabled:
continue
any_enabled = True
if spec.service_key in seen_services:
continue
if spec.service_key:
seen_services.add(spec.service_key)
if not spec.is_service_configured(app):
alerts.append(
ConfigAlert(
id=f"provider_not_configured:{spec.id}",
severity=SEVERITY_ERROR,
sections=(SECTION_SETTINGS,),
params={"provider": spec.label},
message_key="provider_not_configured",
)
)
if spec.webhook_requires_base_url and not getattr(settings, "WEBHOOK_BASE_URL", None):
alerts.append(
ConfigAlert(
id=f"provider_webhook_needs_base_url:{spec.id}",
severity=SEVERITY_ERROR,
sections=(SECTION_SETTINGS,),
params={"provider": spec.label},
message_key="provider_webhook_needs_base_url",
)
)
if not any_enabled:
alerts.append(
ConfigAlert(
id="no_payment_methods",
severity=SEVERITY_WARNING,
sections=(SECTION_SETTINGS, SECTION_PAYMENTS),
)
)
return alerts
def settings_alerts(settings: Any) -> List[ConfigAlert]:
alerts: List[ConfigAlert] = []
mini_app_url = str(getattr(settings, "SUBSCRIPTION_MINI_APP_URL", "") or "").strip()
if not mini_app_url:
alerts.append(
ConfigAlert(
id="mini_app_url_missing",
severity=SEVERITY_WARNING,
sections=(SECTION_SETTINGS,),
)
)
elif not mini_app_url.lower().startswith("https://"):
alerts.append(
ConfigAlert(
id="mini_app_url_not_https",
severity=SEVERITY_ERROR,
sections=(SECTION_SETTINGS,),
params={"url": mini_app_url},
)
)
if not getattr(settings, "REDIS_URL", None):
alerts.append(
ConfigAlert(
id="redis_not_configured",
severity=SEVERITY_WARNING,
sections=(SECTION_SETTINGS,),
)
)
smtp_partial = any(
getattr(settings, key, None)
for key in ("SMTP_USERNAME", "SMTP_PASSWORD", "SMTP_FROM_EMAIL")
)
if smtp_partial and not getattr(settings, "email_auth_configured", False):
alerts.append(
ConfigAlert(
id="smtp_incomplete",
severity=SEVERITY_WARNING,
sections=(SECTION_SETTINGS,),
)
)
return alerts
def proxy_alerts(request: Any, settings: Any) -> List[ConfigAlert]:
"""Warn when the admin request itself came through an untrusted proxy.
In that case provider webhooks with IP allowlists will see the proxy
address instead of the real sender and may reject valid callbacks.
"""
headers = getattr(request, "headers", None) or {}
forwarded = headers.get("X-Forwarded-For")
remote = getattr(request, "remote", None)
if not forwarded or not remote:
return []
if ip_in_allowlist(remote, getattr(settings, "trusted_proxies", None)):
return []
return [
ConfigAlert(
id="proxy_not_trusted",
severity=SEVERITY_WARNING,
sections=(SECTION_SETTINGS,),
params={"remote": str(remote)},
)
]
# ─── Network checks (cached) ───────────────────────────────────────
async def telegram_alerts(bot: Any, settings: Any) -> List[ConfigAlert]:
if bot is None:
return []
try:
info = await asyncio.wait_for(bot.get_webhook_info(), timeout=NETWORK_CHECK_TIMEOUT_SECONDS)
except Exception as exc:
if exc.__class__.__name__ in {"TelegramUnauthorizedError", "TelegramNotFound"}:
return [
ConfigAlert(
id="bot_token_invalid",
severity=SEVERITY_ERROR,
sections=(SECTION_SETTINGS,),
)
]
return [
ConfigAlert(
id="telegram_api_error",
severity=SEVERITY_WARNING,
sections=(SECTION_SETTINGS,),
params={"error": str(exc)[:300]},
)
]
alerts: List[ConfigAlert] = []
actual_url = str(getattr(info, "url", "") or "")
base_url = str(getattr(settings, "WEBHOOK_BASE_URL", "") or "").rstrip("/")
expected_url = (
f"{base_url}{getattr(settings, 'telegram_webhook_path', '/tg/webhook')}" if base_url else ""
)
if not actual_url:
alerts.append(
ConfigAlert(
id="telegram_webhook_missing",
severity=SEVERITY_ERROR,
sections=(SECTION_SETTINGS,),
)
)
elif expected_url and actual_url != expected_url:
alerts.append(
ConfigAlert(
id="telegram_webhook_mismatch",
severity=SEVERITY_WARNING,
sections=(SECTION_SETTINGS,),
params={"actual": actual_url, "expected": expected_url},
)
)
last_error_date = getattr(info, "last_error_date", None)
last_error_ts: Optional[float] = None
if last_error_date is not None:
last_error_ts = (
last_error_date.timestamp()
if hasattr(last_error_date, "timestamp")
else float(last_error_date)
)
if last_error_ts and (time.time() - last_error_ts) < _WEBHOOK_ERROR_RECENT_SECONDS:
alerts.append(
ConfigAlert(
id="telegram_webhook_error",
severity=SEVERITY_WARNING,
sections=(SECTION_SETTINGS,),
params={"error": str(getattr(info, "last_error_message", "") or "")[:300]},
)
)
pending = int(getattr(info, "pending_update_count", 0) or 0)
if pending > _WEBHOOK_PENDING_THRESHOLD:
alerts.append(
ConfigAlert(
id="telegram_webhook_pending",
severity=SEVERITY_WARNING,
sections=(SECTION_SETTINGS,),
params={"count": pending},
)
)
return alerts
async def panel_alerts(panel_service: Any, settings: Any) -> List[ConfigAlert]:
if not getattr(settings, "PANEL_API_URL", None) or not getattr(settings, "PANEL_API_KEY", None):
return [
ConfigAlert(
id="panel_api_not_configured",
severity=SEVERITY_ERROR,
sections=(SECTION_SETTINGS, SECTION_USERS, SECTION_TARIFFS),
)
]
if panel_service is None:
return []
try:
stats = await asyncio.wait_for(
panel_service.get_system_stats(), timeout=NETWORK_CHECK_TIMEOUT_SECONDS
)
except Exception as exc:
logger.debug("Panel health check failed: %s", exc)
stats = None
if stats is None:
return [
ConfigAlert(
id="panel_api_unreachable",
severity=SEVERITY_ERROR,
sections=(SECTION_SETTINGS, SECTION_USERS),
params={"url": str(getattr(settings, "PANEL_API_URL", "") or "")},
)
]
return []
# ─── Aggregation ───────────────────────────────────────────────────
_network_cache: Dict[int, Tuple[float, List[ConfigAlert]]] = {}
_network_cache_lock = asyncio.Lock()
def local_alerts(request: Any, settings: Any, app: Any) -> List[ConfigAlert]:
alerts: List[ConfigAlert] = []
for collect in (
lambda: data_dir_alerts(settings),
lambda: config_file_alerts(settings),
lambda: payment_provider_alerts(settings, app),
lambda: settings_alerts(settings),
lambda: proxy_alerts(request, settings),
):
try:
alerts.extend(collect())
except Exception: # pragma: no cover - one broken check must not hide others
logger.exception("Config health check failed")
return alerts
async def network_alerts(app: Any, settings: Any, *, refresh: bool = False) -> List[ConfigAlert]:
cache_key = id(settings)
now = time.monotonic()
if not refresh:
cached = _network_cache.get(cache_key)
if cached and (now - cached[0]) < NETWORK_CHECKS_TTL_SECONDS:
return cached[1]
async with _network_cache_lock:
if not refresh:
cached = _network_cache.get(cache_key)
if cached and (time.monotonic() - cached[0]) < NETWORK_CHECKS_TTL_SECONDS:
return cached[1]
results = await asyncio.gather(
telegram_alerts(app.get("bot"), settings),
panel_alerts(app.get("panel_service"), settings),
return_exceptions=True,
)
alerts: List[ConfigAlert] = []
for result in results:
if isinstance(result, BaseException):
logger.exception("Network config health check failed", exc_info=result)
continue
alerts.extend(result)
_network_cache[cache_key] = (time.monotonic(), alerts)
return alerts
async def collect_config_alerts(request: Any, *, refresh: bool = False) -> List[Dict[str, Any]]:
app = request.app
settings = app["settings"]
alerts = local_alerts(request, settings, app)
alerts.extend(await network_alerts(app, settings, refresh=refresh))
order = {SEVERITY_ERROR: 0, SEVERITY_WARNING: 1}
alerts.sort(key=lambda alert: (order.get(alert.severity, 2), alert.id))
return [alert.as_payload() for alert in alerts]
frontend/src/admin/AdminPanel.svelte
+11
View File
@@ -45,9 +45,11 @@
import AppearanceSection from "./sections/AppearanceSection.svelte";
import UserDetailModal from "./sections/UserDetailModal.svelte";
import UsersSection from "./sections/UsersSection.svelte";
import ConfigAlertsBanner from "./ConfigAlertsBanner.svelte";
import { createAdsStore } from "../lib/admin/stores/adsStore.js";
import { createBackupsStore } from "../lib/admin/stores/backupsStore.js";
import { createBroadcastStore } from "../lib/admin/stores/broadcastStore.js";
import { createHealthStore } from "../lib/admin/stores/healthStore.js";
import { createLogsStore } from "../lib/admin/stores/logsStore.js";
import { createPaymentsStore } from "../lib/admin/stores/paymentsStore.js";
import { createPromosStore } from "../lib/admin/stores/promosStore.js";
@@ -247,6 +249,7 @@
const adsStore = createAdsStore({ api, onToast: flash, at });
const backupsStore = createBackupsStore({ api, onToast: flash, at });
const broadcastStore = createBroadcastStore({ api, onToast: flash, at });
const healthStore = createHealthStore({ api });
const logsStore = createLogsStore({ api, at });
const paymentsStore = createPaymentsStore({ api, onToast: flash, at, routePrefix });
const promosStore = createPromosStore({ api, onToast: flash, at });
@@ -260,6 +263,7 @@
setContext("promosStore", promosStore);
setContext("adsStore", adsStore);
setContext("healthStore", healthStore);
setContext("backupsStore", backupsStore);
setContext("broadcastStore", broadcastStore);
setContext("logsStore", logsStore);
@@ -531,6 +535,11 @@
window.addEventListener("popstate", onPopState);
}
void broadcastStore.loadCounts();
void healthStore.loadHealth();
const healthTimer =
typeof window !== "undefined"
? window.setInterval(() => void healthStore.loadHealth(), 5 * 60 * 1000)
: null;
return () => {
if (motionMql) motionMql.removeEventListener("change", onMotionChange);
if (compactMql) {
@@ -539,6 +548,7 @@
else if (compactMql.removeListener) compactMql.removeListener(onCompactChange);
}
if (typeof window !== "undefined") window.removeEventListener("popstate", onPopState);
if (healthTimer !== null) window.clearInterval(healthTimer);
clearAdminLanguageClickGuard();
};
});
@@ -802,6 +812,7 @@
</header>
<main class="admin-main">
<ConfigAlertsBanner {at} section={active} onNavigate={setActive} />
{#key active}
<div class="admin-section-stage" in:fade={sectionFade} out:fade={sectionFade}>
{#if active === "stats"}
frontend/src/admin/ConfigAlertsBanner.svelte
+122
View File
@@ -0,0 +1,122 @@
<script>
import { getContext } from "svelte";
import { RefreshCw, TriangleAlert } from "$components/ui/icons.js";
import { AdminButton } from "$components/patterns/admin/index.js";
export let at = (key, _params = {}, fallback = "") => fallback || key;
export let section = "stats";
export let onNavigate = () => {};
const healthStore = getContext("healthStore");
const MESSAGE_FALLBACKS = {
data_dir_missing:
"Каталог data ({path}) не найден. Проверьте, что том data смонтирован в контейнер.",
data_dir_not_writable:
"Нет прав на запись в {path} — бэкапы, тарифы, логотипы и переводы не сохранятся.",
backups_dir_not_writable: "Каталог бэкапов {path} недоступен для записи.",
tariffs_config_invalid: "Файл тарифов {path} не читается: {error}",
locale_overrides_invalid: "Файл переводов {path} повреждён: {error}",
subscription_page_config_invalid: "Конфиг гайдов подписки не читается: {error}",
provider_not_configured:
"Провайдер {provider} включён, но не настроен — оплата через него не работает.",
provider_webhook_needs_base_url:
"Провайдеру {provider} нужен WEBHOOK_BASE_URL для приёма вебхуков, а он не задан.",
no_payment_methods: "Не включён ни один способ оплаты.",
mini_app_url_missing: "SUBSCRIPTION_MINI_APP_URL не задан — кнопка Mini App в боте не появится.",
mini_app_url_not_https: "SUBSCRIPTION_MINI_APP_URL должен начинаться с https:// (сейчас {url}).",
redis_not_configured:
"REDIS_URL не задан — состояния диалогов бота и кэш не переживут перезапуск.",
smtp_incomplete: "SMTP настроен не полностью — вход по email работать не будет.",
proxy_not_trusted:
"Запросы приходят через прокси {remote}, которого нет в TRUSTED_PROXIES — вебхуки платёжных провайдеров могут отклоняться по IP.",
bot_token_invalid: "Telegram отверг BOT_TOKEN — бот не работает.",
telegram_api_error: "Не удалось обратиться к Telegram API: {error}",
telegram_webhook_missing: "Вебхук Telegram не установлен — бот не получает обновления.",
telegram_webhook_mismatch: "Вебхук Telegram указывает на {actual}, ожидается {expected}.",
telegram_webhook_error: "Telegram сообщает об ошибке доставки вебхука: {error}",
telegram_webhook_pending: "В очереди Telegram скопилось {count} необработанных обновлений.",
panel_api_not_configured:
"PANEL_API_URL и PANEL_API_KEY не заданы — синхронизация и выдача подписок не работают.",
panel_api_unreachable: "Панель Remnawave недоступна по адресу {url}.",
};
const SECTION_FALLBACK_LABELS = {
settings: "Настройки",
payments: "Платежи",
backups: "Бэкапы",
tariffs: "Тарифы",
appearance: "Внешний вид",
translations: "Переводы",
users: "Пользователи",
};
function interpolate(template, params = {}) {
return String(template || "").replace(/\{(\w+)\}/g, (match, key) =>
params[key] !== undefined && params[key] !== null ? String(params[key]) : match
);
}
function alertText(alert) {
const fallback = interpolate(MESSAGE_FALLBACKS[alert.message_key] || alert.message_key, alert.params);
return at(`health_${alert.message_key}`, alert.params || {}, fallback);
}
function sectionLabel(id) {
return at(`nav_${id}`, {}, SECTION_FALLBACK_LABELS[id] || id);
}
$: alerts = $healthStore?.alerts || [];
$: healthLoading = $healthStore?.healthLoading;
$: isDashboard = section === "stats";
$: visibleAlerts = isDashboard
? alerts
: alerts.filter((alert) => (alert.sections || []).includes(section));
$: errorCount = visibleAlerts.filter((alert) => alert.severity === "error").length;
</script>
{#if visibleAlerts.length}
<div
class="admin-config-alerts"
class:admin-config-alerts-error={errorCount > 0}
role="alert"
aria-live="polite"
>
<div class="admin-config-alerts-head">
<span class="admin-config-alerts-title">
<TriangleAlert size={15} />
{at("health_title", {}, "Проблемы конфигурации")}
</span>
{#if isDashboard}
<AdminButton
onclick={() => healthStore.loadHealth({ refresh: true })}
disabled={healthLoading}
>
<RefreshCw size={13} />
{at("health_refresh", {}, "Проверить снова")}
</AdminButton>
{/if}
</div>
<ul class="admin-config-alerts-list">
{#each visibleAlerts as alert (alert.id)}
<li class="admin-config-alert admin-config-alert-{alert.severity}">
<span class="admin-config-alert-dot" aria-hidden="true"></span>
<span class="admin-config-alert-text">{alertText(alert)}</span>
{#if isDashboard && (alert.sections || []).length}
<span class="admin-config-alert-links">
{#each alert.sections as sectionId (sectionId)}
<button
type="button"
class="admin-config-alert-link"
on:click={() => onNavigate(sectionId)}
>
{sectionLabel(sectionId)}
</button>
{/each}
</span>
{/if}
</li>
{/each}
</ul>
</div>
{/if}
frontend/src/lib/admin/stores/healthStore.js
+35
View File
@@ -0,0 +1,35 @@
import { writable } from "svelte/store";
export function createHealthStore({ api }) {
const state = writable({
alerts: [],
checkedAt: null,
healthLoading: false,
healthError: "",
});
async function loadHealth({ refresh = false } = {}) {
state.update((s) => ({ ...s, healthLoading: true, healthError: "" }));
try {
const data = await api(`/admin/health${refresh ? "?refresh=1" : ""}`);
if (!data?.ok) {
state.update((s) => ({ ...s, healthError: data?.error || "load_failed" }));
} else {
state.update((s) => ({
...s,
alerts: Array.isArray(data.alerts) ? data.alerts : [],
checkedAt: data.checked_at || null,
}));
}
} catch (e) {
state.update((s) => ({ ...s, healthError: e?.message || String(e) }));
} finally {
state.update((s) => ({ ...s, healthLoading: false }));
}
}
return {
subscribe: state.subscribe,
loadHealth,
};
}
frontend/src/lib/webapp/mockApi.js
+23
View File
@@ -762,6 +762,29 @@ function demoApiResponse(path, cleanPath, options, context) {
}
if (cleanPath === "/admin/sync") return { ok: true, status: "queued" };
if (cleanPath === "/admin/health") {
return {
ok: true,
alerts: [
{
id: "provider_not_configured:wata",
severity: "error",
sections: ["settings"],
message_key: "provider_not_configured",
params: { provider: "Wata" },
},
{
id: "mini_app_url_missing",
severity: "warning",
sections: ["settings"],
message_key: "mini_app_url_missing",
params: {},
},
],
checked_at: new Date().toISOString(),
};
}
if (cleanPath === "/admin/payments") {
const page = paged(DEMO_DATASET.adminPayments || [], params, 25);
return {
frontend/src/styles/admin.css
+89
View File
@@ -372,6 +372,95 @@
width: 100%;
}
.admin-config-alerts {
display: flex;
flex-direction: column;
gap: 10px;
padding: 12px 14px;
border: 1px solid var(--warning-border);
border-radius: 12px;
background: var(--warning-soft);
color: var(--warning-text);
}
.admin-config-alerts-error {
border-color: var(--danger-border);
background: var(--danger-soft);
color: var(--danger-text);
}
.admin-config-alerts-head {
display: flex;
align-items: center;
justify-content: space-between;
gap: 12px;
flex-wrap: wrap;
}
.admin-config-alerts-title {
display: inline-flex;
align-items: center;
gap: 7px;
font-weight: 600;
font-size: 13px;
}
.admin-config-alerts-list {
display: flex;
flex-direction: column;
gap: 6px;
margin: 0;
padding: 0;
list-style: none;
}
.admin-config-alert {
display: flex;
align-items: baseline;
gap: 8px;
font-size: 12.5px;
line-height: 1.45;
}
.admin-config-alert-dot {
flex-shrink: 0;
width: 7px;
height: 7px;
border-radius: 50%;
background: var(--warning-text);
transform: translateY(-1px);
}
.admin-config-alert-error .admin-config-alert-dot {
background: var(--danger-text);
}
.admin-config-alert-text {
min-width: 0;
overflow-wrap: anywhere;
}
.admin-config-alert-links {
display: inline-flex;
gap: 6px;
flex-wrap: wrap;
}
.admin-config-alert-link {
border: 1px solid currentColor;
border-radius: 999px;
padding: 1px 9px;
font-size: 11px;
background: transparent;
color: inherit;
cursor: pointer;
opacity: 0.85;
}
.admin-config-alert-link:hover {
opacity: 1;
}
.admin-section-stage:has(.support-admin-layout) {
flex: 1 1 auto;
min-height: 0;
locales/en.json
+24
View File
@@ -1024,6 +1024,30 @@
"admin_section_backups_subtitle": "Archives, uploads, and database/compose restore",
"admin_section_settings_title": "App Settings",
"admin_section_settings_subtitle": "Overrides for .env, applied instantly",
"admin_health_title": "Configuration issues",
"admin_health_refresh": "Check again",
"admin_health_data_dir_missing": "Data directory ({path}) not found. Make sure the data volume is mounted into the container.",
"admin_health_data_dir_not_writable": "No write access to {path} — backups, tariffs, logos and translations cannot be saved.",
"admin_health_backups_dir_not_writable": "Backups directory {path} is not writable.",
"admin_health_tariffs_config_invalid": "Tariffs file {path} cannot be read: {error}",
"admin_health_locale_overrides_invalid": "Translations file {path} is corrupted: {error}",
"admin_health_subscription_page_config_invalid": "Subscription guides config cannot be read: {error}",
"admin_health_provider_not_configured": "Provider {provider} is enabled but not configured — payments through it will not work.",
"admin_health_provider_webhook_needs_base_url": "Provider {provider} requires WEBHOOK_BASE_URL for webhooks, but it is not set.",
"admin_health_no_payment_methods": "No payment method is enabled.",
"admin_health_mini_app_url_missing": "SUBSCRIPTION_MINI_APP_URL is not set — the Mini App button will not appear in the bot.",
"admin_health_mini_app_url_not_https": "SUBSCRIPTION_MINI_APP_URL must start with https:// (currently {url}).",
"admin_health_redis_not_configured": "REDIS_URL is not set — bot dialog states and caches will not survive a restart.",
"admin_health_smtp_incomplete": "SMTP is partially configured — email login will not work.",
"admin_health_proxy_not_trusted": "Requests arrive through proxy {remote} which is not in TRUSTED_PROXIES — payment provider webhooks may be rejected by IP.",
"admin_health_bot_token_invalid": "Telegram rejected BOT_TOKEN — the bot is not working.",
"admin_health_telegram_api_error": "Failed to reach the Telegram API: {error}",
"admin_health_telegram_webhook_missing": "Telegram webhook is not set — the bot does not receive updates.",
"admin_health_telegram_webhook_mismatch": "Telegram webhook points to {actual}, expected {expected}.",
"admin_health_telegram_webhook_error": "Telegram reports a webhook delivery error: {error}",
"admin_health_telegram_webhook_pending": "{count} unprocessed updates are queued in Telegram.",
"admin_health_panel_api_not_configured": "PANEL_API_URL and PANEL_API_KEY are not set — sync and subscription provisioning do not work.",
"admin_health_panel_api_unreachable": "Remnawave panel is unreachable at {url}.",
"admin_backups_load_failed": "Failed to load backups",
"admin_backups_upload_done": "Archive uploaded",
"admin_backups_upload_failed": "Failed to upload archive",
locales/ru.json
+24
View File
@@ -1024,6 +1024,30 @@
"admin_section_backups_subtitle": "Архивы, загрузка и восстановление БД/compose",
"admin_section_settings_title": "Настройки приложения",
"admin_section_settings_subtitle": "Оверрайды над .env, применяются мгновенно",
"admin_health_title": "Проблемы конфигурации",
"admin_health_refresh": "Проверить снова",
"admin_health_data_dir_missing": "Каталог data ({path}) не найден. Проверьте, что том data смонтирован в контейнер.",
"admin_health_data_dir_not_writable": "Нет прав на запись в {path} — бэкапы, тарифы, логотипы и переводы не сохранятся.",
"admin_health_backups_dir_not_writable": "Каталог бэкапов {path} недоступен для записи.",
"admin_health_tariffs_config_invalid": "Файл тарифов {path} не читается: {error}",
"admin_health_locale_overrides_invalid": "Файл переводов {path} повреждён: {error}",
"admin_health_subscription_page_config_invalid": "Конфиг гайдов подписки не читается: {error}",
"admin_health_provider_not_configured": "Провайдер {provider} включён, но не настроен — оплата через него не работает.",
"admin_health_provider_webhook_needs_base_url": "Провайдеру {provider} нужен WEBHOOK_BASE_URL для приёма вебхуков, а он не задан.",
"admin_health_no_payment_methods": "Не включён ни один способ оплаты.",
"admin_health_mini_app_url_missing": "SUBSCRIPTION_MINI_APP_URL не задан — кнопка Mini App в боте не появится.",
"admin_health_mini_app_url_not_https": "SUBSCRIPTION_MINI_APP_URL должен начинаться с https:// (сейчас {url}).",
"admin_health_redis_not_configured": "REDIS_URL не задан — состояния диалогов бота и кэш не переживут перезапуск.",
"admin_health_smtp_incomplete": "SMTP настроен не полностью — вход по email работать не будет.",
"admin_health_proxy_not_trusted": "Запросы приходят через прокси {remote}, которого нет в TRUSTED_PROXIES — вебхуки платёжных провайдеров могут отклоняться по IP.",
"admin_health_bot_token_invalid": "Telegram отверг BOT_TOKEN — бот не работает.",
"admin_health_telegram_api_error": "Не удалось обратиться к Telegram API: {error}",
"admin_health_telegram_webhook_missing": "Вебхук Telegram не установлен — бот не получает обновления.",
"admin_health_telegram_webhook_mismatch": "Вебхук Telegram указывает на {actual}, ожидается {expected}.",
"admin_health_telegram_webhook_error": "Telegram сообщает об ошибке доставки вебхука: {error}",
"admin_health_telegram_webhook_pending": "В очереди Telegram скопилось {count} необработанных обновлений.",
"admin_health_panel_api_not_configured": "PANEL_API_URL и PANEL_API_KEY не заданы — синхронизация и выдача подписок не работают.",
"admin_health_panel_api_unreachable": "Панель Remnawave недоступна по адресу {url}.",
"admin_backups_load_failed": "Не удалось загрузить бэкапы",
"admin_backups_upload_done": "Архив загружен",
"admin_backups_upload_failed": "Не удалось загрузить архив",
tests/test_admin_config_health.py
+351
View File
@@ -0,0 +1,351 @@
import json
import tempfile
import time
import unittest
from datetime import datetime, timezone
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch
from bot.services import config_health_service as health
def _settings(**overrides):
base = {
"BACKUP_DIR": "data/backups",
"TARIFFS_CONFIG_PATH": "data/tariffs.json",
"SUBSCRIPTION_MINI_APP_URL": "https://shop.example.com/app",
"REDIS_URL": "redis://redis:6379/0",
"SMTP_USERNAME": None,
"SMTP_PASSWORD": None,
"SMTP_FROM_EMAIL": None,
"email_auth_configured": False,
"WEBHOOK_BASE_URL": "https://shop.example.com",
"telegram_webhook_path": "/tg/webhook",
"PANEL_API_URL": "https://panel.example.com/api",
"PANEL_API_KEY": "panel-key",
"trusted_proxies": ["127.0.0.1", "172.16.0.0/12"],
}
base.update(overrides)
return SimpleNamespace(**base)
def _alert_ids(alerts):
return [alert.id for alert in alerts]
class DataDirAlertsTests(unittest.TestCase):
def test_missing_data_dir_reported_as_error(self):
with tempfile.TemporaryDirectory() as tmpdir:
alerts = health.data_dir_alerts(_settings(), app_root=Path(tmpdir))
self.assertEqual(_alert_ids(alerts), ["data_dir_missing"])
self.assertEqual(alerts[0].severity, "error")
self.assertIn("backups", alerts[0].sections)
def test_writable_data_dir_produces_no_alerts(self):
with tempfile.TemporaryDirectory() as tmpdir:
(Path(tmpdir) / "data").mkdir()
alerts = health.data_dir_alerts(_settings(), app_root=Path(tmpdir))
self.assertEqual(alerts, [])
def test_unwritable_data_dir_reported(self):
with tempfile.TemporaryDirectory() as tmpdir:
(Path(tmpdir) / "data").mkdir()
with patch.object(health, "_dir_is_writable", return_value=False):
alerts = health.data_dir_alerts(_settings(), app_root=Path(tmpdir))
self.assertIn("data_dir_not_writable", _alert_ids(alerts))
class ConfigFileAlertsTests(unittest.TestCase):
def test_invalid_tariffs_config_reported(self):
with tempfile.TemporaryDirectory() as tmpdir:
tariffs_path = Path(tmpdir) / "tariffs.json"
tariffs_path.write_text("{not json", encoding="utf-8")
settings = _settings(TARIFFS_CONFIG_PATH=str(tariffs_path))
with patch.object(health, "APP_ROOT", Path(tmpdir)):
alerts = health.config_file_alerts(settings)
self.assertIn("tariffs_config_invalid", _alert_ids(alerts))
def test_invalid_locale_overrides_reported(self):
with tempfile.TemporaryDirectory() as tmpdir:
data_dir = Path(tmpdir) / "data"
data_dir.mkdir()
(data_dir / "locales-overrides.json").write_text("{oops", encoding="utf-8")
settings = _settings(TARIFFS_CONFIG_PATH=str(Path(tmpdir) / "absent.json"))
with patch.object(health, "APP_ROOT", Path(tmpdir)):
alerts = health.config_file_alerts(settings)
self.assertIn("locale_overrides_invalid", _alert_ids(alerts))
def test_valid_files_produce_no_alerts(self):
with tempfile.TemporaryDirectory() as tmpdir:
data_dir = Path(tmpdir) / "data"
data_dir.mkdir()
(data_dir / "locales-overrides.json").write_text("{}", encoding="utf-8")
settings = _settings(TARIFFS_CONFIG_PATH=str(Path(tmpdir) / "absent.json"))
with patch.object(health, "APP_ROOT", Path(tmpdir)):
alerts = health.config_file_alerts(settings)
self.assertNotIn("tariffs_config_invalid", _alert_ids(alerts))
self.assertNotIn("locale_overrides_invalid", _alert_ids(alerts))
class PaymentProviderAlertsTests(unittest.TestCase):
@staticmethod
def _spec(
spec_id,
*,
enabled=True,
configured=True,
webhook_requires_base_url=False,
service_key=None,
):
return SimpleNamespace(
id=spec_id,
label=spec_id.title(),
service_key=service_key or f"{spec_id}_service",
webhook_requires_base_url=webhook_requires_base_url,
is_effectively_enabled=lambda settings: enabled,
is_service_configured=lambda app: configured,
)
def test_enabled_but_unconfigured_provider_reported(self):
specs = [self._spec("wata", configured=False)]
with patch("bot.payment_providers.iter_provider_specs", return_value=specs):
alerts = health.payment_provider_alerts(_settings(), app={})
self.assertEqual(_alert_ids(alerts), ["provider_not_configured:wata"])
self.assertEqual(alerts[0].message_key, "provider_not_configured")
self.assertEqual(alerts[0].params["provider"], "Wata")
def test_webhook_provider_without_base_url_reported(self):
specs = [self._spec("yookassa", webhook_requires_base_url=True)]
settings = _settings(WEBHOOK_BASE_URL=None)
with patch("bot.payment_providers.iter_provider_specs", return_value=specs):
alerts = health.payment_provider_alerts(settings, app={})
self.assertIn("provider_webhook_needs_base_url:yookassa", _alert_ids(alerts))
def test_no_enabled_providers_reported_as_warning(self):
specs = [self._spec("wata", enabled=False)]
with patch("bot.payment_providers.iter_provider_specs", return_value=specs):
alerts = health.payment_provider_alerts(_settings(), app={})
self.assertEqual(_alert_ids(alerts), ["no_payment_methods"])
self.assertEqual(alerts[0].severity, "warning")
def test_configured_enabled_provider_produces_no_alerts(self):
specs = [self._spec("wata")]
with patch("bot.payment_providers.iter_provider_specs", return_value=specs):
alerts = health.payment_provider_alerts(_settings(), app={})
self.assertEqual(alerts, [])
def test_shared_service_reported_once(self):
specs = [
self._spec("platega", configured=False, service_key="platega_service"),
self._spec("platega_crypto", configured=False, service_key="platega_service"),
]
with patch("bot.payment_providers.iter_provider_specs", return_value=specs):
alerts = health.payment_provider_alerts(_settings(), app={})
self.assertEqual(_alert_ids(alerts), ["provider_not_configured:platega"])
class SettingsAlertsTests(unittest.TestCase):
def test_clean_settings_produce_no_alerts(self):
self.assertEqual(health.settings_alerts(_settings()), [])
def test_missing_mini_app_url_reported(self):
alerts = health.settings_alerts(_settings(SUBSCRIPTION_MINI_APP_URL=None))
self.assertIn("mini_app_url_missing", _alert_ids(alerts))
def test_http_mini_app_url_reported_as_error(self):
alerts = health.settings_alerts(
_settings(SUBSCRIPTION_MINI_APP_URL="http://shop.example.com")
)
ids = _alert_ids(alerts)
self.assertIn("mini_app_url_not_https", ids)
self.assertEqual(alerts[ids.index("mini_app_url_not_https")].severity, "error")
def test_missing_redis_reported(self):
alerts = health.settings_alerts(_settings(REDIS_URL=None))
self.assertIn("redis_not_configured", _alert_ids(alerts))
def test_partial_smtp_reported(self):
alerts = health.settings_alerts(_settings(SMTP_USERNAME="mailer"))
self.assertIn("smtp_incomplete", _alert_ids(alerts))
def test_complete_smtp_not_reported(self):
alerts = health.settings_alerts(
_settings(SMTP_USERNAME="mailer", email_auth_configured=True)
)
self.assertNotIn("smtp_incomplete", _alert_ids(alerts))
class ProxyAlertsTests(unittest.TestCase):
def test_untrusted_proxy_reported(self):
request = SimpleNamespace(
remote="203.0.113.50",
headers={"X-Forwarded-For": "198.51.100.7"},
)
alerts = health.proxy_alerts(request, _settings())
self.assertEqual(_alert_ids(alerts), ["proxy_not_trusted"])
def test_trusted_proxy_not_reported(self):
request = SimpleNamespace(
remote="172.18.0.5",
headers={"X-Forwarded-For": "198.51.100.7"},
)
self.assertEqual(health.proxy_alerts(request, _settings()), [])
def test_direct_request_not_reported(self):
request = SimpleNamespace(remote="203.0.113.50", headers={})
self.assertEqual(health.proxy_alerts(request, _settings()), [])
class TelegramAlertsTests(unittest.IsolatedAsyncioTestCase):
@staticmethod
def _webhook_info(**overrides):
base = {
"url": "https://shop.example.com/tg/webhook",
"last_error_date": None,
"last_error_message": None,
"pending_update_count": 0,
}
base.update(overrides)
return SimpleNamespace(**base)
async def test_healthy_webhook_produces_no_alerts(self):
bot = SimpleNamespace(get_webhook_info=AsyncMock(return_value=self._webhook_info()))
self.assertEqual(await health.telegram_alerts(bot, _settings()), [])
async def test_missing_webhook_reported_as_error(self):
bot = SimpleNamespace(get_webhook_info=AsyncMock(return_value=self._webhook_info(url="")))
alerts = await health.telegram_alerts(bot, _settings())
self.assertEqual(_alert_ids(alerts), ["telegram_webhook_missing"])
self.assertEqual(alerts[0].severity, "error")
async def test_webhook_mismatch_reported(self):
bot = SimpleNamespace(
get_webhook_info=AsyncMock(
return_value=self._webhook_info(url="https://other.example.com/tg/webhook")
)
)
alerts = await health.telegram_alerts(bot, _settings())
self.assertEqual(_alert_ids(alerts), ["telegram_webhook_mismatch"])
async def test_recent_delivery_error_reported(self):
info = self._webhook_info(
last_error_date=datetime.now(timezone.utc),
last_error_message="SSL error",
)
bot = SimpleNamespace(get_webhook_info=AsyncMock(return_value=info))
alerts = await health.telegram_alerts(bot, _settings())
self.assertEqual(_alert_ids(alerts), ["telegram_webhook_error"])
self.assertEqual(alerts[0].params["error"], "SSL error")
async def test_stale_delivery_error_not_reported(self):
info = self._webhook_info(last_error_date=time.time() - 7200)
bot = SimpleNamespace(get_webhook_info=AsyncMock(return_value=info))
self.assertEqual(await health.telegram_alerts(bot, _settings()), [])
async def test_pending_updates_reported(self):
info = self._webhook_info(pending_update_count=500)
bot = SimpleNamespace(get_webhook_info=AsyncMock(return_value=info))
alerts = await health.telegram_alerts(bot, _settings())
self.assertEqual(_alert_ids(alerts), ["telegram_webhook_pending"])
async def test_unauthorized_token_reported_as_error(self):
class TelegramUnauthorizedError(Exception):
pass
bot = SimpleNamespace(
get_webhook_info=AsyncMock(side_effect=TelegramUnauthorizedError("401"))
)
alerts = await health.telegram_alerts(bot, _settings())
self.assertEqual(_alert_ids(alerts), ["bot_token_invalid"])
async def test_generic_api_error_reported_as_warning(self):
bot = SimpleNamespace(get_webhook_info=AsyncMock(side_effect=OSError("boom")))
alerts = await health.telegram_alerts(bot, _settings())
self.assertEqual(_alert_ids(alerts), ["telegram_api_error"])
self.assertEqual(alerts[0].severity, "warning")
class PanelAlertsTests(unittest.IsolatedAsyncioTestCase):
async def test_unconfigured_panel_reported(self):
settings = _settings(PANEL_API_URL=None, PANEL_API_KEY=None)
alerts = await health.panel_alerts(None, settings)
self.assertEqual(_alert_ids(alerts), ["panel_api_not_configured"])
async def test_unreachable_panel_reported(self):
panel_service = SimpleNamespace(get_system_stats=AsyncMock(return_value=None))
alerts = await health.panel_alerts(panel_service, _settings())
self.assertEqual(_alert_ids(alerts), ["panel_api_unreachable"])
async def test_healthy_panel_produces_no_alerts(self):
panel_service = SimpleNamespace(get_system_stats=AsyncMock(return_value={"cpu": 1}))
self.assertEqual(await health.panel_alerts(panel_service, _settings()), [])
class CollectAlertsTests(unittest.IsolatedAsyncioTestCase):
async def test_collect_sorts_errors_first_and_serializes(self):
settings = _settings()
request = SimpleNamespace(app={"settings": settings}, headers={}, remote="127.0.0.1")
warning = health.ConfigAlert(id="warn_alert", severity="warning", sections=("settings",))
error = health.ConfigAlert(id="error_alert", severity="error", sections=("backups",))
with (
patch.object(health, "local_alerts", return_value=[warning, error]),
patch.object(health, "network_alerts", AsyncMock(return_value=[])),
):
payload = await health.collect_config_alerts(request)
self.assertEqual([item["id"] for item in payload], ["error_alert", "warn_alert"])
self.assertEqual(payload[0]["message_key"], "error_alert")
self.assertEqual(payload[0]["sections"], ["backups"])
async def test_network_alerts_cached_between_calls(self):
settings = _settings()
app = {"settings": settings, "bot": None, "panel_service": None}
health._network_cache.clear()
with patch.object(health, "panel_alerts", AsyncMock(return_value=[])) as panel_mock:
await health.network_alerts(app, settings)
await health.network_alerts(app, settings)
self.assertEqual(panel_mock.await_count, 1)
await health.network_alerts(app, settings, refresh=True)
self.assertEqual(panel_mock.await_count, 2)
health._network_cache.clear()
class HealthLocaleKeysTests(unittest.TestCase):
def test_every_message_key_has_locale_entries(self):
root = Path(__file__).resolve().parents[1]
for language in ("ru", "en"):
messages = json.loads(
(root / "locales" / f"{language}.json").read_text(encoding="utf-8")
)
for suffix in ("title", "refresh", *health.ALL_MESSAGE_KEYS):
self.assertIn(
f"admin_health_{suffix}",
messages,
f"locales/{language}.json is missing admin_health_{suffix}",
)
def test_alert_ids_used_by_checks_are_known_message_keys(self):
known = set(health.ALL_MESSAGE_KEYS)
with tempfile.TemporaryDirectory() as tmpdir:
local = health.data_dir_alerts(_settings(), app_root=Path(tmpdir))
local += health.settings_alerts(_settings(SUBSCRIPTION_MINI_APP_URL=None, REDIS_URL=None))
for alert in local:
self.assertIn(alert.message_key or alert.id, known)
if __name__ == "__main__":
unittest.main()