tests/test_lumen_typing_pace.py

"""
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)