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