""" test_lumen_typing_pace.py — юнит-тесты на lumen_typing_pace.py: самокалибрующаяся оценка скорости "печати" при стриминге (speed_key/get_typing_speed/ record_observed_speed) и раскладка "довывода" остатка на шаги (catchup_reveal_steps). lumen_typing_pace.py не имеет зависимостей от Telegram/Gemini/OpenRouter/рантайм- состояния бота (тот же принцип, что и у lumen_formatting.py/lumen_security.py/ lumen_router_config.py — см. их тестовые файлы) — тестируется здесь напрямую (import lumen_typing_pace), без импорта bot.py. _speed_ema — module-level state; каждый тест использует свой уникальный ключ (через speed_key с уникальным model_id), чтобы тесты не зависели от порядка выполнения друг друга и не требовали ручного сброса общего состояния. Запуск: pytest test_lumen_typing_pace.py -v """ import lumen_typing_pace # ─────────────────────────── speed_key ─────────────────────────── def test_speed_key_combines_provider_and_model(): assert lumen_typing_pace.speed_key("gemini", "gemini-3.6-flash") == "gemini:gemini-3.6-flash" assert lumen_typing_pace.speed_key("openrouter", "nvidia/nemotron-nano-9b-v2:free") == "openrouter:nvidia/nemotron-nano-9b-v2:free" # ─────────────────────────── get_typing_speed / record_observed_speed ─────────────────────────── def test_get_typing_speed_returns_default_when_no_samples_yet(): key = lumen_typing_pace.speed_key("gemini", "__test_never_seen_model__") assert lumen_typing_pace.get_typing_speed(key) == lumen_typing_pace.DEFAULT_CHARS_PER_SEC def test_record_observed_speed_updates_ema_towards_observed_value(): key = lumen_typing_pace.speed_key("openrouter", "__test_model_ema__") # Первый замер — EMA сразу принимает значение наблюдения (нет предыдущего). lumen_typing_pace.record_observed_speed(key, elapsed_sec=1.0, chars_len=100) # 100 симв/сек first = lumen_typing_pace.get_typing_speed(key) assert first == 100.0 # Второй замер намного медленнее — EMA должна сдвинуться К нему, но не # перескочить мгновенно на него целиком (сглаживание, см. _EMA_ALPHA). lumen_typing_pace.record_observed_speed(key, elapsed_sec=1.0, chars_len=40) # 40 симв/сек second = lumen_typing_pace.get_typing_speed(key) assert 40.0 < second < 100.0 def test_record_observed_speed_clamps_extreme_burst_before_averaging(): # РЕГРЕССИЯ на реальный сценарий: бэкенд присвоил ответ ОДНИМ куском за доли # секунды — сырая наблюдённая "скорость" была бы в тысячи симв/сек. Без # зажима EMA улетела бы в небеса, и следующий ответ той же модели "мигал" бы # мгновенно вместо плавного набора — именно то, что эта функция должна # предотвращать (см. докстринг record_observed_speed). key = lumen_typing_pace.speed_key("openrouter", "__test_model_burst__") lumen_typing_pace.record_observed_speed(key, elapsed_sec=0.01, chars_len=5000) # ~500000 симв/сек assert lumen_typing_pace.get_typing_speed(key) <= lumen_typing_pace.MAX_CHARS_PER_SEC def test_record_observed_speed_clamps_extremely_slow_observation(): key = lumen_typing_pace.speed_key("openrouter", "__test_model_slow__") lumen_typing_pace.record_observed_speed(key, elapsed_sec=100.0, chars_len=10) # 0.1 симв/сек assert lumen_typing_pace.get_typing_speed(key) >= lumen_typing_pace.MIN_CHARS_PER_SEC def test_record_observed_speed_ignores_non_positive_inputs(): key = lumen_typing_pace.speed_key("gemini", "__test_model_noop__") lumen_typing_pace.record_observed_speed(key, elapsed_sec=0.0, chars_len=100) lumen_typing_pace.record_observed_speed(key, elapsed_sec=1.0, chars_len=0) lumen_typing_pace.record_observed_speed(key, elapsed_sec=-1.0, chars_len=100) # Ни один из вызовов не должен был создать запись — до сих пор дефолт. assert lumen_typing_pace.get_typing_speed(key) == lumen_typing_pace.DEFAULT_CHARS_PER_SEC def test_get_typing_speed_different_models_are_independent(): key_fast = lumen_typing_pace.speed_key("gemini", "__test_fast_model__") key_slow = lumen_typing_pace.speed_key("openrouter", "__test_slow_model__") lumen_typing_pace.record_observed_speed(key_fast, elapsed_sec=1.0, chars_len=200) lumen_typing_pace.record_observed_speed(key_slow, elapsed_sec=1.0, chars_len=50) assert lumen_typing_pace.get_typing_speed(key_fast) > lumen_typing_pace.get_typing_speed(key_slow) # ─────────────────────────── catchup_reveal_steps ─────────────────────────── def test_catchup_reveal_steps_empty_when_nothing_remaining(): assert lumen_typing_pace.catchup_reveal_steps(0, 100.0, 0.5, 6) == [] assert lumen_typing_pace.catchup_reveal_steps(-5, 100.0, 0.5, 6) == [] def test_catchup_reveal_steps_normal_case_reaches_exact_total(): # 100 симв. остатка, 50 симв/сек, тик 0.5с -> по 25 симв. за тик. steps = lumen_typing_pace.catchup_reveal_steps(100, 50.0, 0.5, 6) assert steps == [25, 50, 75, 100] assert steps[-1] == 100 # последний шаг всегда доводит ровно до конца def test_catchup_reveal_steps_respects_max_ticks_ceiling(): # РЕГРЕССИЯ на ключевое требование: сколько бы шагов ни потребовалось при # заниженной оценке скорости, число шагов никогда не превышает max_ticks — # иначе очень длинный ответ мог бы "допечатываться" неприлично долго. steps = lumen_typing_pace.catchup_reveal_steps(1000, 10.0, 0.5, 6) assert len(steps) <= 6 # Но последний шаг всё равно обязан довести до конца целиком (форсированный # рывок), а не оставить хвост навсегда невидимым. assert steps[-1] == 1000 def test_catchup_reveal_steps_cumulative_and_monotonic(): steps = lumen_typing_pace.catchup_reveal_steps(237, 90.0, 0.5, 6) assert steps == sorted(steps) assert steps[-1] == 237 def test_catchup_reveal_steps_single_tick_when_fast_enough(): # Скорость достаточно высокая, чтобы весь остаток поместился в один тик. steps = lumen_typing_pace.catchup_reveal_steps(20, 200.0, 0.5, 6) assert steps == [20] # ─────────────────────────── blend_arrival_speed / display_speed_for ─────────────────────────── def test_blend_arrival_speed_first_sample_sets_estimate(): # Первый кусок: оценки ещё нет — мгновенная скорость становится ею (в границах). assert lumen_typing_pace.blend_arrival_speed(None, 100, 1.0) == 100.0 # Выбросы жмутся в границы, а не утаскивают оценку. assert lumen_typing_pace.blend_arrival_speed(None, 10000, 0.01) == lumen_typing_pace.MAX_CHARS_PER_SEC assert lumen_typing_pace.blend_arrival_speed(None, 1, 100.0) == lumen_typing_pace.MIN_CHARS_PER_SEC def test_blend_arrival_speed_ignores_non_positive_inputs(): assert lumen_typing_pace.blend_arrival_speed(90.0, 0, 1.0) == 90.0 assert lumen_typing_pace.blend_arrival_speed(90.0, 50, 0.0) == 90.0 assert lumen_typing_pace.blend_arrival_speed(None, 50, 0.0) is None def test_blend_arrival_speed_moves_toward_new_sample(): # 100 симв/сек было, приехало 200 симв/сек — оценка сдвинулась вверх, но не прыгнула целиком. updated = lumen_typing_pace.blend_arrival_speed(100.0, 200, 1.0) assert 100.0 < updated < 200.0 def test_display_speed_for_prefers_measured_arrival(): key = lumen_typing_pace.speed_key("gemini", "__test_display_model__") assert lumen_typing_pace.display_speed_for(150.0, key) == 150.0 # Без замера прихода — глобальная EMA модели (затравка на первые куски). assert lumen_typing_pace.display_speed_for(None, key) == lumen_typing_pace.get_typing_speed(key)