Spaces:
Running
Running
File size: 7,988 Bytes
9a08963 9983566 398ccda 9a08963 398ccda 9a08963 9983566 9a08963 398ccda 9a08963 398ccda 9a08963 9983566 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 | """
conftest.py — настройка окружения ПЕРЕД импортом bot.py тестами.
bot.py на уровне модуля читает переменные окружения (BOT_TOKEN, GEMINI_API_KEY
и т.д.) и пишет лог-файл по пути из BOT_LOG_PATH (по умолчанию /app/bot.log).
Чтобы тесты можно было гонять где угодно (не только внутри Docker-контейнера
HF Spaces, где /app существует и доступен на запись), здесь выставляются
безопасные заглушки ДО того, как pytest соберёт и импортирует тестовые модули.
Ничего из этого не требует сети и не создаёт реальных Bot()/genai.Client() —
они создаются только внутри main(), которая тестами не вызывается.
"""
import os
import tempfile
os.environ.setdefault("BOT_LOG_PATH", os.path.join(tempfile.gettempdir(), "lumen_test_bot.log"))
os.environ.setdefault("BOT_TOKEN", "test-token-not-real")
os.environ.setdefault("GEMINI_API_KEY", "test-key-not-real")
import copy
import pytest
from types import SimpleNamespace
from unittest.mock import AsyncMock
import bot
import lumen_limits
import lumen_router_config
@pytest.fixture(autouse=True)
def _bot_global_state_guard():
"""Снимок наиболее часто вручную сохраняемых модульных глобалов bot.py перед
каждым тестом и восстановление после (аудит техдолга, август 2026).
Карантин моделей (lumen_router_config._QUARANTINE) тоже снимается: пустые
ответы в тестах честно пишут в счётчик, без сброса один и тот же фейковый
ID за три теста ушёл бы в карантин и начал фильтровать маршруты.
Десятки тестов в test_bot_*.py вручную сохраняли/восстанавливали
bot.chat_state/bot.GLOBAL_QUOTA/bot.client/bot.bot в try/finally — рабочий, но
повторяющийся бойлерплейт и потенциальный источник тонких утечек между тестами
при росте сьюта (забытый finally молча "протравливает" состояние в следующие
тесты). Это ДОПОЛНИТЕЛЬНАЯ страховочная сетка, а не замена — существующие тесты
со своим ручным cleanup продолжают работать точно так же, как раньше (двойное
восстановление безвредно); новые тесты могут полагаться только на эту фикстуру
и не писать собственный try/finally для этих четырёх глобалов.
Намеренно НЕ снимается снимок вообще всего модульного состояния (_dirty_chat_ids,
TELEGRAM_API_BASE_URL и т.п.) — тесты, которые их трогают, уже сами аккуратно
восстанавливают именно то, что меняют (см. test_flush_dirty_state_once_* и
аналогичные); расширять фикстуру на них без реальной необходимости было бы
накоплением сложности "на будущее", а не решением подтверждённой проблемы."""
import bot as _bot_module
chat_state_snapshot = copy.deepcopy(_bot_module.chat_state)
quota_snapshot = copy.deepcopy(_bot_module.GLOBAL_QUOTA)
quarantine_snapshot = copy.deepcopy(lumen_router_config._QUARANTINE)
client_snapshot = _bot_module.client
bot_snapshot = _bot_module.bot
yield
_bot_module.chat_state.clear()
_bot_module.chat_state.update(chat_state_snapshot)
_bot_module.GLOBAL_QUOTA.clear()
_bot_module.GLOBAL_QUOTA.update(quota_snapshot)
lumen_router_config._QUARANTINE.clear()
lumen_router_config._QUARANTINE.update(quarantine_snapshot)
_bot_module.client = client_snapshot
_bot_module.bot = bot_snapshot
@pytest.fixture(autouse=True)
def _instant_typing_pace():
"""Стриминг теперь искусственно "довыводит" остаток уже полученного, но ещё не
показанного текста (см. STREAM_TYPING_*/_run_streaming_reply в bot.py и
lumen_typing_pace.py) — несколько await bot._typing_sleep(...) между правками
сообщения, чтобы визуально было похоже на живой набор текста. В тестах
реальное ожидание не нужно и заметно замедлило бы весь сьют без единой пользы —
bot._typing_sleep существует именно как точка подмены (тот же приём, что и
bot._get_http_session/bot._openrouter_stream_pieces и т.п. в других местах),
здесь она безусловно патчится на no-op для КАЖДОГО теста, а не только тех,
что явно тестируют стриминг."""
import bot as _bot_module
original_typing_sleep = _bot_module._typing_sleep
async def _instant_sleep(_seconds: float) -> None:
return None
_bot_module._typing_sleep = _instant_sleep
yield
_bot_module._typing_sleep = original_typing_sleep
@pytest.fixture(autouse=True)
def _instant_tikwm_throttle():
"""Тот же приём и та же причина, что и у _instant_typing_pace выше, только для
троттлинга запросов к TikWM API (см. _tikwm_throttle/_fetch_tikwm_media_data в
lumen_tiktok.py, добавлено при отладке 403-ошибок TikWM 11 августа 2026) —
_tikwm_last_request_ts персистентен на весь процесс, без сброса между тестами
накопленное состояние заставляло бы КАЖДЫЙ следующий тест, трогающий TikTok,
реально ждать до _TIKWM_MIN_INTERVAL_SEC секунд без единой пользы для теста."""
import lumen_tiktok as _tiktok_module
original_sleep = _tiktok_module._sleep
original_last_ts = _tiktok_module._tikwm_last_request_ts
async def _instant_sleep(_seconds: float) -> None:
return None
_tiktok_module._sleep = _instant_sleep
_tiktok_module._tikwm_last_request_ts = None
yield
_tiktok_module._sleep = original_sleep
_tiktok_module._tikwm_last_request_ts = original_last_ts
@pytest.fixture
def rate_guard_setup(monkeypatch):
def _setup():
monkeypatch.setattr(lumen_limits, "user_rate_limits", {})
monkeypatch.setattr(lumen_limits, "RATE_LIMIT_MAX_REQUESTS", 2)
monkeypatch.setattr(bot, "get_state", lambda chat_id: {})
monkeypatch.setattr(bot, "is_guest_message", lambda message: False)
monkeypatch.setattr(bot, "message_mentions_bot", lambda message: False)
monkeypatch.setattr(bot, "_safe_reply", AsyncMock())
monkeypatch.setattr(bot, "_tg_call", AsyncMock())
monkeypatch.setattr(bot, "inline_draw", AsyncMock())
monkeypatch.setattr(bot, "inline_tts", AsyncMock())
return SimpleNamespace(
text="", caption=None, chat=SimpleNamespace(id=123, type=bot.ChatType.PRIVATE),
from_user=SimpleNamespace(id=456), reply=AsyncMock(), reply_to_message=None,
)
return _setup
|