Hugging Face's logo

Spaces:
SilverElixir
/
Lumen
Running

App Files Community
Lumen / tests /test_bot_admin.py
SilverElixir
Pin M2 webhook retry and H2 mention gate with failing-first tests
b711b21
Raw History Blame Contribute Delete
33.3 kB
"""
test_bot_admin.py — Админка и webhook: гейты секретов, healthcheck, /export_state, /logs, /stats.
Выделено из test_bot.py (P2 аудита); общие фейки — в bot_test_helpers.py.
"""
import asyncio
import bot
import logging
import os
import threading
import time
import pytest
from unittest.mock import Mock
from tests.bot_test_helpers import (
_FakeAdminRequest,
_FakeWebhookRequest,
_run_webhook_handler,
)
def test_check_bot_token_auth_accepts_correct_bearer_header():
original = bot.BOT_TOKEN
bot.BOT_TOKEN = "real-secret-token"
try:
req = _FakeAdminRequest(headers={"Authorization": "Bearer real-secret-token"})
assert bot._check_bot_token_auth(req) is True
finally:
bot.BOT_TOKEN = original
def test_check_bot_token_auth_rejects_query_param_regression():
# РЕГРЕССИЯ (код-ревью): раньше BOT_TOKEN читался из ?bot_token=... в URL — GET-
# запрос с секретом в query-строке попадает в access-логи прокси/историю браузера
# (CWE-598). Теперь query-параметр должен полностью ИГНОРИРОВАТЬСЯ — единственный
# легитимный путь — заголовок Authorization: Bearer.
original = bot.BOT_TOKEN
bot.BOT_TOKEN = "real-secret-token"
try:
req = _FakeAdminRequest(headers={}, query_params={"bot_token": "real-secret-token"})
assert bot._check_bot_token_auth(req) is False
finally:
bot.BOT_TOKEN = original
def test_check_bot_token_auth_rejects_wrong_or_missing_header():
original = bot.BOT_TOKEN
bot.BOT_TOKEN = "real-secret-token"
try:
assert bot._check_bot_token_auth(_FakeAdminRequest(headers={"Authorization": "Bearer wrong"})) is False
assert bot._check_bot_token_auth(_FakeAdminRequest(headers={})) is False
# Без префикса "Bearer " — тоже отказ, даже если сам токен совпадает.
assert bot._check_bot_token_auth(_FakeAdminRequest(headers={"Authorization": "real-secret-token"})) is False
finally:
bot.BOT_TOKEN = original
def test_check_admin_key_accepts_correct_bearer_header():
original = bot.ADMIN_PANEL_KEY
bot.ADMIN_PANEL_KEY = "real-admin-key"
try:
req = _FakeAdminRequest(headers={"Authorization": "Bearer real-admin-key"})
assert bot._check_admin_key(req) is True
finally:
bot.ADMIN_PANEL_KEY = original
def test_check_admin_key_rejects_query_param_regression():
# РЕГРЕССИЯ (аудит техдолга): раньше ADMIN_PANEL_KEY читался из ?key=... в URL —
# та же уязвимость (CWE-598), что уже была исправлена для BOT_TOKEN в /admin_keys
# (см. test_check_bot_token_auth_rejects_query_param_regression), но не была
# применена к /diag/webhook_url/export_state. Query-параметр больше не должен
# приниматься вообще, даже если значение верное.
original = bot.ADMIN_PANEL_KEY
bot.ADMIN_PANEL_KEY = "real-admin-key"
try:
req = _FakeAdminRequest(headers={}, query_params={"key": "real-admin-key"})
assert bot._check_admin_key(req) is False
finally:
bot.ADMIN_PANEL_KEY = original
def test_check_admin_key_rejects_wrong_or_missing_key():
original = bot.ADMIN_PANEL_KEY
bot.ADMIN_PANEL_KEY = "real-admin-key"
try:
assert bot._check_admin_key(_FakeAdminRequest(headers={"Authorization": "Bearer wrong"})) is False
assert bot._check_admin_key(_FakeAdminRequest(headers={})) is False
assert bot._check_admin_key(_FakeAdminRequest(headers={"Authorization": "real-admin-key"})) is False
finally:
bot.ADMIN_PANEL_KEY = original
def test_bearer_non_ascii_returns_false_not_500():
# compare_digest на str падает TypeError на не-ASCII: отказ без исключения.
original = bot.ADMIN_PANEL_KEY
bot.ADMIN_PANEL_KEY = "real-admin-key"
try:
assert bot._check_bearer_token(_FakeAdminRequest(headers={"Authorization": "Bearer café"}), "real-admin-key") is False
assert bot._check_admin_key(_FakeAdminRequest(headers={"Authorization": "Bearer café"})) is False
finally:
bot.ADMIN_PANEL_KEY = original
def test_webhook_non_ascii_secret_returns_false_not_500():
original = bot.WEBHOOK_SECRET
bot.WEBHOOK_SECRET = "real-webhook-secret"
try:
req = _FakeWebhookRequest(headers={"X-Telegram-Bot-Api-Secret-Token": "café"}, body={"update_id": 1})
result = asyncio.run(_run_webhook_handler(req))
assert result == {"ok": False}
finally:
bot.WEBHOOK_SECRET = original
def test_webhook_handler_tracks_dispatched_task_for_shutdown():
original_secret = bot.WEBHOOK_SECRET
original_bot_obj = bot.bot
original_process = bot._process_raw_update
bot.WEBHOOK_SECRET = "real-webhook-secret"
bot.bot = object()
bot._inflight_tasks.clear()
async def fake_process(raw_update):
await asyncio.sleep(0)
bot._process_raw_update = fake_process
try:
req = _FakeWebhookRequest(
headers={"X-Telegram-Bot-Api-Secret-Token": "real-webhook-secret"},
body={"update_id": 1},
)
asyncio.run(_run_webhook_handler(req))
# _run_webhook_handler уже дожидается одного тика планировщика — к этому
# моменту короткая fake_process должна была завершиться и самоудалиться
# из набора (см. test_track_inflight_task_registers_and_self_removes_on_completion).
assert not bot._inflight_tasks
finally:
bot.WEBHOOK_SECRET = original_secret
bot.bot = original_bot_obj
bot._process_raw_update = original_process
def test_healthcheck_reports_not_ready_when_bot_or_client_uninitialized():
original_bot, original_client = bot.bot, bot.client
bot.bot = None
bot.client = None
try:
result = asyncio.run(bot.healthcheck())
assert result == {"status": "starting", "ready": False}
finally:
bot.bot, bot.client = original_bot, original_client
def test_healthcheck_reports_ready_when_initialized():
original_bot, original_client = bot.bot, bot.client
bot.bot = object()
bot.client = object()
try:
result = asyncio.run(bot.healthcheck())
assert result == {"status": "ok", "ready": True}
finally:
bot.bot, bot.client = original_bot, original_client
def test_export_state_rejects_missing_or_wrong_key():
original = bot.ADMIN_PANEL_KEY
bot.ADMIN_PANEL_KEY = "real-admin-key"
try:
result = asyncio.run(bot.export_state(_FakeAdminRequest(headers={"Authorization": "Bearer wrong"})))
# Отказ — честный 401, а не 200 с телом {"error": ...}: иначе брутфорс
# ключа отличался от успеха только телом ответа (аудит 26.09.2026).
assert result.status_code == 401
assert b"error" in result.body
finally:
bot.ADMIN_PANEL_KEY = original
def test_export_state_logs_denied_attempt(caplog):
import logging
original = bot.ADMIN_PANEL_KEY
bot.ADMIN_PANEL_KEY = "real-admin-key"
try:
with caplog.at_level(logging.WARNING, logger="bot"):
asyncio.run(bot.export_state(_FakeAdminRequest(headers={})))
assert any("[admin] Denied GET /export_state" in r.getMessage() for r in caplog.records)
finally:
bot.ADMIN_PANEL_KEY = original
def test_fastapi_schema_endpoints_are_disabled():
# Публичная схема на HF Space описывала все эндпоинты бесплатно.
assert bot.app.docs_url is None
assert bot.app.redoc_url is None
assert bot.app.openapi_url is None
def test_export_state_returns_detached_snapshot():
# Аудит A11-05: экспорт отдавал живые ссылки на мутабельные объекты —
# запись между возвратом и сериализацией давала несогласованный бэкап.
original = bot.ADMIN_PANEL_KEY
bot.ADMIN_PANEL_KEY = "real-admin-key"
chat_id = 999980
state = bot.get_state(chat_id)
state["history"].append({"role": "user", "content": "hi"})
bot.GLOBAL_QUOTA.setdefault("groq", {})["snaptest"] = {"used": 1}
try:
result = asyncio.run(bot.export_state(_FakeAdminRequest(headers={"Authorization": "Bearer real-admin-key"})))
bot.GLOBAL_QUOTA["groq"]["snaptest"]["used"] = 999
state["history"].append({"role": "user", "content": "after"})
assert result["global_quota"]["groq"]["snaptest"]["used"] == 1
assert all(m["content"] != "after" for m in result["chats"][str(chat_id)]["history"])
finally:
bot.GLOBAL_QUOTA.get("groq", {}).pop("snaptest", None)
bot.chat_state.pop(chat_id, None)
bot.ADMIN_PANEL_KEY = original
def test_export_state_rejects_query_param_regression():
# См. test_check_admin_key_rejects_query_param_regression — /export_state — самый
# чувствительный из трёх эндпоинтов (отдаёт ПОЛНЫЕ истории всех чатов), поэтому
# регрессия здесь проверяется отдельно, а не только на уровне _check_admin_key.
original = bot.ADMIN_PANEL_KEY
bot.ADMIN_PANEL_KEY = "real-admin-key"
try:
result = asyncio.run(bot.export_state(_FakeAdminRequest(headers={}, query_params={"key": "real-admin-key"})))
assert result.status_code == 401
finally:
bot.ADMIN_PANEL_KEY = original
def test_export_state_returns_chats_and_quota_with_valid_key():
original = bot.ADMIN_PANEL_KEY
bot.ADMIN_PANEL_KEY = "real-admin-key"
chat_id = 999411
bot.chat_state[chat_id] = {
"image_model": bot.DEFAULT_POLLINATIONS_IMAGE_MODEL, "history": [{"role": "user", "content": "hi"}],
"quota": {}, "recent_media_ids": {}, "last_activity": 0.0,
}
try:
result = asyncio.run(bot.export_state(_FakeAdminRequest(headers={"Authorization": "Bearer real-admin-key"})))
assert str(chat_id) in result["chats"]
assert result["chats"][str(chat_id)]["history"] == [{"role": "user", "content": "hi"}]
assert "global_quota" in result
assert "exported_at" in result
finally:
bot.ADMIN_PANEL_KEY = original
bot.chat_state.pop(chat_id, None)
def test_webhook_handler_accepts_valid_secret_and_dispatches_update():
original_secret = bot.WEBHOOK_SECRET
original_bot_obj = bot.bot
original_process = bot._process_raw_update
bot.WEBHOOK_SECRET = "real-webhook-secret"
bot.bot = object() # не-None достаточно, чтобы пройти проверку "бот уже инициализирован"
calls = []
async def fake_process(raw_update):
calls.append(raw_update)
bot._process_raw_update = fake_process
try:
req = _FakeWebhookRequest(
headers={"X-Telegram-Bot-Api-Secret-Token": "real-webhook-secret"},
body={"update_id": 1},
)
result = asyncio.run(_run_webhook_handler(req))
assert result == {"ok": True}
assert calls == [{"update_id": 1}]
finally:
bot.WEBHOOK_SECRET = original_secret
bot.bot = original_bot_obj
bot._process_raw_update = original_process
def test_webhook_handler_rejects_invalid_or_missing_secret_and_does_not_dispatch():
original_secret = bot.WEBHOOK_SECRET
original_process = bot._process_raw_update
bot.WEBHOOK_SECRET = "real-webhook-secret"
calls = []
async def fake_process(raw_update):
calls.append(raw_update)
bot._process_raw_update = fake_process
try:
for bad_headers in (
{"X-Telegram-Bot-Api-Secret-Token": "wrong-secret"},
{},
):
req = _FakeWebhookRequest(headers=bad_headers, body={"update_id": 1})
result = asyncio.run(_run_webhook_handler(req))
assert result == {"ok": False}
assert calls == []
finally:
bot.WEBHOOK_SECRET = original_secret
bot._process_raw_update = original_process
def test_webhook_handler_rejects_oversized_body_before_reading():
# Тело сверх капа: отказ до разбора, сырые байты не читаем целиком.
import lumen_admin
original_secret = bot.WEBHOOK_SECRET
bot.WEBHOOK_SECRET = "real-webhook-secret"
try:
req = _FakeWebhookRequest(
headers={
"X-Telegram-Bot-Api-Secret-Token": "real-webhook-secret",
"Content-Length": str(lumen_admin.WEBHOOK_MAX_BODY_BYTES + 1),
},
body={"update_id": 1},
)
assert asyncio.run(_run_webhook_handler(req)) == {"ok": False}
assert req.body_calls == 0
# Заголовок соврал (маленький), тело большое — тоже отказ.
req2 = _FakeWebhookRequest(
headers={"X-Telegram-Bot-Api-Secret-Token": "real-webhook-secret"},
body={"update_id": 2},
)
req2._raw = b"x" * (lumen_admin.WEBHOOK_MAX_BODY_BYTES + 1)
assert asyncio.run(_run_webhook_handler(req2)) == {"ok": False}
finally:
bot.WEBHOOK_SECRET = original_secret
def test_webhook_denied_log_is_throttled(caplog):
# Гроздья отклонений — одной строкой, а не WARNING на каждый запрос в очередь.
import logging
import lumen_admin
orig_count = lumen_admin._webhook_denied_count
orig_last = lumen_admin._webhook_denied_last_log
lumen_admin._webhook_denied_count = 0
# Ноль monotonic вместо прошлого: на свежей CI-машине uptime < 60с и первый
# отказ молча копится (прод 07.10.2026) — уводим метку в прошлое явно.
lumen_admin._webhook_denied_last_log = time.monotonic() - lumen_admin._WEBHOOK_DENIED_LOG_INTERVAL_SEC - 1.0
original_secret = bot.WEBHOOK_SECRET
bot.WEBHOOK_SECRET = "real-webhook-secret"
try:
with caplog.at_level(logging.WARNING, logger="bot"):
for _ in range(3):
req = _FakeWebhookRequest(
headers={"X-Telegram-Bot-Api-Secret-Token": "wrong"},
body={"update_id": 1},
)
assert asyncio.run(_run_webhook_handler(req)) == {"ok": False}
denied = [r for r in caplog.records if "Rejected" in r.getMessage()]
assert len(denied) == 1
finally:
bot.WEBHOOK_SECRET = original_secret
lumen_admin._webhook_denied_count = orig_count
lumen_admin._webhook_denied_last_log = orig_last
def test_webhook_handler_returns_503_when_inflight_full(monkeypatch):
# Переполнение фоновых задач — 503 с повтором, апдейт не теряется молча.
import lumen_admin
original_secret = bot.WEBHOOK_SECRET
original_bot_obj = bot.bot
original_process = bot._process_raw_update
bot.WEBHOOK_SECRET = "real-webhook-secret"
bot.bot = object()
calls = []
async def fake_process(raw_update):
calls.append(raw_update)
bot._process_raw_update = fake_process
monkeypatch.setattr(lumen_admin, "WEBHOOK_MAX_INFLIGHT_TASKS", 0)
try:
req = _FakeWebhookRequest(
headers={"X-Telegram-Bot-Api-Secret-Token": "real-webhook-secret"},
body={"update_id": 1},
)
result = asyncio.run(_run_webhook_handler(req))
assert result.status_code == 503
assert calls == []
finally:
bot.WEBHOOK_SECRET = original_secret
bot.bot = original_bot_obj
bot._process_raw_update = original_process
@pytest.mark.parametrize(("env_value", "expected"), [
("abc", 25.0), ("", 25.0), ("0", 25.0), ("-5", 25.0), ("30", 30.0),
])
def test_diag_budget_sec_falls_back_on_garbage(monkeypatch, env_value, expected):
# Голый float() давал 500 на мусоре в env, а 0 молча ронял все зонды.
import lumen_admin
monkeypatch.setenv("DIAG_TOTAL_BUDGET_SEC", env_value)
assert lumen_admin._diag_budget_sec() == expected
monkeypatch.delenv("DIAG_TOTAL_BUDGET_SEC", raising=False)
assert lumen_admin._diag_budget_sec() == 25.0
def test_webhook_handler_drops_update_when_bot_not_yet_initialized():
# Апдейт может прийти раньше, чем main() успеет создать глобальный bot (Bot/
# genai.Client создаются уже после старта uvicorn) — отвечаем 503, чтобы
# Telegram повторил апдейт, а не считаем дроп успехом (AUD-E-003).
original_secret = bot.WEBHOOK_SECRET
original_bot_obj = bot.bot
original_process = bot._process_raw_update
bot.WEBHOOK_SECRET = "real-webhook-secret"
bot.bot = None
calls = []
async def fake_process(raw_update):
calls.append(raw_update)
bot._process_raw_update = fake_process
try:
req = _FakeWebhookRequest(
headers={"X-Telegram-Bot-Api-Secret-Token": "real-webhook-secret"},
body={"update_id": 1},
)
result = asyncio.run(_run_webhook_handler(req))
assert result.status_code == 503
assert calls == []
finally:
bot.WEBHOOK_SECRET = original_secret
bot.bot = original_bot_obj
bot._process_raw_update = original_process
def test_webhook_handler_retries_broken_body_instead_of_swallowing():
# Битое тело раньше отвечало ok:true и Telegram считал дроп успехом (аудит M2).
original_secret = bot.WEBHOOK_SECRET
original_bot_obj = bot.bot
original_process = bot._process_raw_update
bot.WEBHOOK_SECRET = "real-webhook-secret"
bot.bot = object()
calls = []
async def fake_process(raw_update):
calls.append(raw_update)
bot._process_raw_update = fake_process
try:
req = _FakeWebhookRequest(
headers={"X-Telegram-Bot-Api-Secret-Token": "real-webhook-secret"},
body={"update_id": 1},
)
req._raw = b"{broken-json"
result = asyncio.run(_run_webhook_handler(req))
assert result.status_code == 503
assert calls == []
finally:
bot.WEBHOOK_SECRET = original_secret
bot.bot = original_bot_obj
bot._process_raw_update = original_process
def test_allowed_updates_contains_only_real_telegram_types():
# guest_message — валидное поле Update (guest mode, Bot API; проверено по
# core.telegram.org/bots/api 2026-09-21). Удаление отсюда было ошибкой
# аудита AUD-J-001 и ломало гостевой режим — этот тест её ловит.
assert "guest_message" in bot.ALLOWED_UPDATES
assert "message" in bot.ALLOWED_UPDATES
def test_admin_secrets_are_independent_of_bot_token_when_seed_set():
# РЕГРЕССИЯ (аудит техдолга): раньше WEBHOOK_SECRET/ADMIN_PANEL_KEY выводились
# ИСКЛЮЧИТЕЛЬНО из BOT_TOKEN — компрометация токена компрометировала оба сразу,
# и ни один нельзя было ротировать независимо. Теперь можно задать отдельную соль.
#
# Прежняя версия теста считала sha256 дважды и сравнивала соль саму с собой,
# то есть НЕ вызывала код бота и проходила при любой поломке вывода ключей
# (враждебное ревью 27.09.2026). Теперь проверяем настоящий инвариант модуля.
import hashlib
seed = bot._ADMIN_SECRET_SEED
assert seed, "соль обязана быть непустой, иначе ключи предсказуемы"
# Ключи — реальные производные текущей соли, а не литералы в тесте.
assert bot.WEBHOOK_SECRET == hashlib.sha256(seed.encode()).hexdigest()[:32]
assert bot.ADMIN_PANEL_KEY == hashlib.sha256(seed.encode() + b"admin_panel").hexdigest()[:24]
# Два ключа не совпадают, и ни один не равен самому токену.
assert bot.WEBHOOK_SECRET != bot.ADMIN_PANEL_KEY
assert bot.WEBHOOK_SECRET != bot.BOT_TOKEN
assert bot.ADMIN_PANEL_KEY != bot.BOT_TOKEN
# Соль из ADMIN_SECRET_SEED, когда она задана, — ключи не выводятся из BOT_TOKEN.
configured_seed = os.environ.get("ADMIN_SECRET_SEED", "").strip()
if configured_seed:
assert seed == configured_seed
def test_admin_secret_seed_falls_back_to_bot_token_when_unset():
# Без ADMIN_SECRET_SEED поведение идентично прежнему (соль = BOT_TOKEN) — не
# ломает существующие деплои, которые эту переменную не настраивали.
# Прежняя версия ждала `or "default"`, а код делает `or secrets.token_hex(32)`
# (bot.py) — ветка «всё пусто» не проверялась и ожидание было неверным.
import re
seed_env = os.environ.get("ADMIN_SECRET_SEED", "").strip()
if seed_env:
assert bot._ADMIN_SECRET_SEED == seed_env
elif bot.BOT_TOKEN:
assert bot._ADMIN_SECRET_SEED == bot.BOT_TOKEN
else:
# Оба пустые: случайная соль, а не литерал "default".
assert re.fullmatch(r"[0-9a-f]{64}", bot._ADMIN_SECRET_SEED)
# ─────────────── lifecycle: токен, часовой цикл, старт (S7) ───────────────
def test_require_bot_token_exits_on_empty_token(monkeypatch):
# A1-1: пустой токен раньше умирал внутри Bot() с TokenValidationError и
# рестартами — теперь чистый выход с понятной причиной.
monkeypatch.setattr(bot, "BOT_TOKEN", "")
with pytest.raises(SystemExit):
bot._require_bot_token()
def test_require_bot_token_returns_configured_token(monkeypatch):
# Сторож: непустой токен молча проходит дальше в Bot().
monkeypatch.setattr(bot, "BOT_TOKEN", "123:abc")
assert bot._require_bot_token() == "123:abc"
def test_secret_log_formatter_survives_interpreter_teardown():
# Финализация обнуляет глобалы модуля: форматирование не должно падать.
import logging
fmt = bot._SecretLogFormatter()
rec = logging.LogRecord("bot", logging.WARNING, __file__, 1, "hello %s", ("world",), None)
orig = bot._SECRET_NAMES
bot._SECRET_NAMES = None
try:
assert fmt.format(rec) == "hello world"
finally:
bot._SECRET_NAMES = orig
def _sleep_then_cancel(loop_sleeps):
# Первые loop_sleeps+1 вызовов sleep проходят (1.5с старта + тики),
# дальше CancelledError останавливает бесконечный цикл стартапа.
state = {"n": 0}
async def _fake_sleep(delay):
state["n"] += 1
if state["n"] > loop_sleeps + 1:
raise asyncio.CancelledError
return _fake_sleep
def _patch_startup(monkeypatch, tg_impl, loader=None):
monkeypatch.setattr(bot, "load_state_from_disk", loader or Mock())
monkeypatch.setattr(bot, "_check_temporary_free_models_expiry", Mock())
monkeypatch.setattr(bot, "_check_unconfirmed_model_quotas", Mock())
monkeypatch.setattr(bot, "_check_scheduled_removals_due", Mock())
monkeypatch.setattr(bot, "telegram_api_call", tg_impl)
monkeypatch.setattr(bot, "_cleanup_rate_limit_dict", Mock())
monkeypatch.setattr(bot, "_evict_orphan_chat_locks", Mock())
monkeypatch.setattr(bot, "_reset_quota_if_new_day", Mock())
# getMe в try_setup переписывает globals — снапшот для отката.
monkeypatch.setattr(bot, "BOT_USERNAME", bot.BOT_USERNAME)
monkeypatch.setattr(bot, "OPENROUTER_HTTP_REFERER", bot.OPENROUTER_HTTP_REFERER)
async def _fake_tg_ok(method, *args, **kwargs):
if method == "getMe":
return {"username": "testbot"}
return {}
def test_hourly_tick_survives_failing_step(monkeypatch):
# A1-2: упавшая чистка раньше убивала часовой цикл навсегда — теперь
# второй тик всё равно наступает.
monkeypatch.setattr(bot.asyncio, "sleep", _sleep_then_cancel(2))
_patch_startup(monkeypatch, _fake_tg_ok)
monkeypatch.setattr(
bot, "_cleanup_rate_limit_dict",
Mock(side_effect=[RuntimeError("boom"), None]),
)
with pytest.raises(asyncio.CancelledError):
asyncio.run(bot._webhook_startup())
# Упавший шаг пропускает остаток своего тика, но цикл живёт: чистка звалась
# дважды (второй тик наступил), evict — один раз (первый тик оборвался).
assert bot._cleanup_rate_limit_dict.call_count == 2
assert bot._evict_orphan_chat_locks.call_count == 1
def test_startup_loads_state_off_loop(monkeypatch):
# A1-3: синхронное чтение диска раньше стопорило loop на старте.
seen = {}
monkeypatch.setattr(bot.asyncio, "sleep", _sleep_then_cancel(0))
_patch_startup(monkeypatch, _fake_tg_ok,
loader=lambda: seen.setdefault("thread", threading.current_thread()))
with pytest.raises(asyncio.CancelledError):
asyncio.run(bot._webhook_startup())
assert seen["thread"] is not threading.main_thread()
def test_startup_logs_commands_failure_honestly(monkeypatch, caplog):
# A1-4: провал всех языков раньше логировался как успех (for-else без break).
async def _fake_tg_fail_commands(method, *args, **kwargs):
if method == "getMe":
return {"username": "testbot"}
if method == "setMyCommands":
raise RuntimeError("boom")
return {}
monkeypatch.setattr(bot.asyncio, "sleep", _sleep_then_cancel(0))
_patch_startup(monkeypatch, _fake_tg_fail_commands)
with caplog.at_level(logging.INFO, logger="bot"), pytest.raises(asyncio.CancelledError):
asyncio.run(bot._webhook_startup())
assert "were not set for any language" in caplog.text
assert "Bot commands set successfully" not in caplog.text
def test_startup_logs_commands_success(monkeypatch, caplog):
# Сторож: хотя бы один язык прошёл — лог успеха на месте.
monkeypatch.setattr(bot.asyncio, "sleep", _sleep_then_cancel(0))
_patch_startup(monkeypatch, _fake_tg_ok)
with caplog.at_level(logging.INFO, logger="bot"), pytest.raises(asyncio.CancelledError):
asyncio.run(bot._webhook_startup())
assert "Bot commands set successfully" in caplog.text
def test_message_mentions_bot_without_chat_returns_false():
# A1-6: chat None раньше давал AttributeError, апдейт глотался хендлером.
class _NoChatMessage:
chat = None
text = "@somebot hello"
caption = None
reply_to_message = None
assert bot.message_mentions_bot(_NoChatMessage()) is False
def test_webhook_setup_retries_then_raises(monkeypatch):
# Аудит D4: провал setWebhook без ретрая давал тихий мёртвый бот — теперь
# ретраи с бэкоффом и явный отказ вместо молчания.
sleeps = []
async def fast_sleep(delay):
sleeps.append(delay)
async def always_fail(method, *args, **kwargs):
if method == "getMe":
return {"username": "testbot"}
raise RuntimeError("boom")
monkeypatch.setattr(bot.asyncio, "sleep", fast_sleep)
monkeypatch.setattr(bot, "SETWEBHOOK_MAX_ATTEMPTS", 3)
_patch_startup(monkeypatch, always_fail)
with pytest.raises(RuntimeError, match="setWebhook"):
asyncio.run(bot._webhook_startup())
assert sleeps == [1.5, 5.0, 10.0]
def test_webhook_setup_succeeds_after_retries(monkeypatch):
# Сторож: два провала подряд — третий регистрирует, стартап живёт дальше.
calls = {"hook": 0}
sleeps = []
async def fast_sleep(delay):
sleeps.append(delay)
if len(sleeps) > 3:
raise asyncio.CancelledError
async def flaky(method, *args, **kwargs):
if method == "getMe":
return {"username": "testbot"}
if method == "setWebhook":
calls["hook"] += 1
if calls["hook"] < 3:
raise RuntimeError("boom")
return True
return {}
monkeypatch.setattr(bot.asyncio, "sleep", fast_sleep)
_patch_startup(monkeypatch, flaky)
with pytest.raises(asyncio.CancelledError):
asyncio.run(bot._webhook_startup())
assert calls["hook"] == 3
assert sleeps == [1.5, 5.0, 10.0, 3600]
def test_main_exits_nonzero_when_server_dies_first(monkeypatch, caplog):
# Падение uvicorn первым (порт занят) раньше терялось: процесс тихо выходил
# с кодом 0. Теперь исключение сервера логируется и даёт SystemExit(1).
from types import SimpleNamespace
from unittest.mock import AsyncMock
import lumen_chat_state
monkeypatch.setattr(bot, "_require_bot_token", lambda: "123:abc")
monkeypatch.setattr(bot, "Bot", lambda *a, **k: SimpleNamespace())
monkeypatch.setattr(bot, "IPv4AiohttpSession", lambda *a, **k: SimpleNamespace())
monkeypatch.setattr(bot.genai, "Client", lambda *a, **k: SimpleNamespace())
async def _never():
await asyncio.sleep(3600)
monkeypatch.setattr(bot, "_webhook_startup", _never)
monkeypatch.setattr(bot, "_flush_dirty_state", _never)
monkeypatch.setattr(bot, "_drain_inflight_tasks", AsyncMock())
monkeypatch.setattr(bot, "_close_sessions", AsyncMock())
monkeypatch.setattr(bot.uvicorn, "Config", lambda *a, **k: SimpleNamespace())
class _DeadServer:
should_exit = False
async def serve(self):
raise RuntimeError("port busy")
monkeypatch.setattr(bot.uvicorn, "Server", lambda cfg: _DeadServer())
orig_bot, orig_client = bot.bot, bot.client
orig_dirty, orig_deletions = set(bot._dirty_chat_ids), set(bot._pending_chat_deletions)
orig_index_dirty, orig_quota_dirty = lumen_chat_state._index_dirty, lumen_chat_state._quota_dirty
bot._dirty_chat_ids.clear()
bot._pending_chat_deletions.clear()
lumen_chat_state._index_dirty = False
lumen_chat_state._quota_dirty = False
try:
with caplog.at_level(logging.ERROR, logger="bot"), pytest.raises(SystemExit) as exc_info:
asyncio.run(bot.main())
assert exc_info.value.code == 1
assert "[serve] HTTP server failed" in caplog.text
finally:
bot.bot, bot.client = orig_bot, orig_client
bot._dirty_chat_ids.clear()
bot._dirty_chat_ids.update(orig_dirty)
bot._pending_chat_deletions.clear()
bot._pending_chat_deletions.update(orig_deletions)
lumen_chat_state._index_dirty = orig_index_dirty
lumen_chat_state._quota_dirty = orig_quota_dirty
def test_probe_url_returns_ok_shape_and_redacts_secret_on_error():
# Общий зонд /diag и /selftest: успех — статус, неуспех — текст без секрета.
class _FakeResp:
status = 200
async def __aenter__(self):
return self
async def __aexit__(self, *args):
return False
class _FakeSession:
def get(self, url, **kwargs):
return _FakeResp()
class _BoomSession:
def get(self, url, **kwargs):
raise RuntimeError("net down, token abc123")
ok = asyncio.run(bot.probe_url(_FakeSession(), "http://x"))
assert ok["ok"] is True and ok["status"] == 200 and ok["elapsed_sec"] >= 0
err = asyncio.run(bot.probe_url(_BoomSession(), "http://x", redact="abc123"))
assert err["ok"] is False
assert "abc123" not in err["error"] and "<TOKEN>" in err["error"]