"""Тесты для app/core/password.py — bcrypt hash/verify (#2550)."""

from __future__ import annotations

import asyncio
import os
import threading
import time
from concurrent.futures import ThreadPoolExecutor

# С #2665 password.py читает настройки (размер пула проверок) — значит тянет
# `Settings()`, которому нужен DATABASE_URL. В CI он в env (ci-tradein.yml),
# локально подставляем заглушку, как это делает tests/test_auth_api.py.
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

import pytest

from app.core import password as password_mod
from app.core.config import settings
from app.core.password import (
    PasswordVerifyOverloadedError,
    hash_password,
    verify_password,
    verify_password_bounded,
)


def test_roundtrip() -> None:
    """hash_password → verify_password с тем же паролем возвращает True."""
    hashed = hash_password("correct horse battery staple")
    assert verify_password("correct horse battery staple", hashed) is True


def test_wrong_password_returns_false() -> None:
    """Неверный пароль против валидного хеша → False."""
    hashed = hash_password("correct horse battery staple")
    assert verify_password("wrong password", hashed) is False


def test_hash_too_long_raises_value_error() -> None:
    """Пароль >72 байт в UTF-8 → ValueError в hash_password (нет silent truncation)."""
    long_password = "a" * 73
    with pytest.raises(ValueError):
        hash_password(long_password)


def test_hash_exactly_72_bytes_ok() -> None:
    """Ровно 72 байта — граничное значение, ещё допустимо."""
    password = "a" * 72
    hashed = hash_password(password)
    assert verify_password(password, hashed) is True


def test_hash_too_long_multibyte_raises_value_error() -> None:
    """40 кириллических символов = 80 байт UTF-8 (2 байта/символ) → ValueError.

    Проверяет, что лимит считается в байтах, а не в символах — иначе 40-символьный
    кириллический пароль (< 72 символов, но 80 байт) прошёл бы мимо guard'а.
    """
    long_cyrillic_password = "а" * 40
    assert len(long_cyrillic_password.encode("utf-8")) == 80
    with pytest.raises(ValueError):
        hash_password(long_cyrillic_password)


def test_verify_too_long_returns_false_not_raise() -> None:
    """verify_password на >72-байтовом пароле возвращает False, НЕ raise."""
    hashed = hash_password("some valid password")
    long_password = "a" * 73
    assert verify_password(long_password, hashed) is False


def test_hash_empty_raises_value_error() -> None:
    """Пустой пароль → ValueError в hash_password."""
    with pytest.raises(ValueError):
        hash_password("")


def test_verify_empty_returns_false() -> None:
    """Пустой пароль в verify_password → False (не raise)."""
    hashed = hash_password("some valid password")
    assert verify_password("", hashed) is False


def test_hash_is_unique_due_to_salt() -> None:
    """Два хеша одного пароля различаются (уникальная соль на каждый вызов)."""
    password = "correct horse battery staple"
    hash1 = hash_password(password)
    hash2 = hash_password(password)
    assert hash1 != hash2
    assert verify_password(password, hash1) is True
    assert verify_password(password, hash2) is True


def test_verify_malformed_hash_returns_false() -> None:
    """Некорректный (не-bcrypt) хеш в verify_password → False, не raise."""
    assert verify_password("some password", "not-a-bcrypt-hash") is False


# ---------------------------------------------------------------------------
# #2665 — verify_password_bounded: вне событийного цикла + потолок темпа
# ---------------------------------------------------------------------------


def test_verify_ceiling_defaults_stay_within_the_db_pool() -> None:
    """Дефолты — часть защиты, а не тюнинг. Стережём их здесь.

    Тест про темп (test_auth_api.py) вычисляет ожидаемый потолок из той же
    настройки, которую охраняет, поэтому подъём дефолта он не заметит. А
    наступит ослабление именно через настройку: не правкой кода и не ревью, а
    строчкой `LOGIN_PASSWORD_VERIFY_WORKERS=32` в `.env.runtime` под предлогом
    «входы тормозят». Пусть тогда краснеет хотя бы этот тест.
    """
    from app.core.db import engine

    # max_inflight ждущих ДЕРЖАТ по соединению к БД (сессия реестра открыта
    # после SELECT в get_user_by_username) — очередь обязана быть уже пула.
    assert (
        settings.login_password_verify_max_inflight < engine.pool.size() + engine.pool._max_overflow
    )
    assert settings.login_password_verify_workers == 1, (
        "потолок перебора = workers/282мс. Подъём — осознанное решение "
        "«во сколько раз ускоряем перебор», а не рефакторинг: правь вместе с тестом"
    )
    # ЛИТЕРАЛЫ, а не арифметика от настройки. Доля на ключ (#2714) считается как
    # max_inflight // 2, и сторож вида `cap == max_inflight // 2` был бы
    # тавтологией: подъём max_inflight до 64 он бы проспал, а вместе с ним —
    # возврат к «один адрес занимает всё» (доля 32 при очереди в 4 живых слота
    # ничего не делит). Поэтому здесь зафиксированы ОБА числа.
    assert settings.login_password_verify_max_inflight == 4, (
        "очередь 4 выбрана под QueuePool 5+10 и худшее ожидание 4/1×282мс ≈ 1.1с; "
        "меняешь — пересчитывай и долю на ключ ниже"
    )
    assert password_mod._per_key_slot_cap() == 2, (
        "один адрес держит не больше 2 слотов из 4: половина ёмкости обязана "
        "оставаться тем, кто приходит впервые (#2714)"
    )


async def test_bounded_gives_same_answer_as_sync() -> None:
    """Обёртка не меняет вердикт — она меняет только ГДЕ он считается."""
    hashed = hash_password("correct horse battery staple")
    key = "203.0.113.1"
    assert await verify_password_bounded("correct horse battery staple", hashed, key=key) is True
    assert await verify_password_bounded("wrong password", hashed, key=key) is False


async def test_bounded_runs_off_the_event_loop_thread(monkeypatch: pytest.MonkeyPatch) -> None:
    """bcrypt считается В ДРУГОМ ПОТОКЕ, а не в потоке событийного цикла.

    Замер на проде: сверка = 282 мс, и ровно столько цикл не обслуживал никого.
    Проверка идентичности потока — самая прямая формулировка «цикл свободен»;
    таймингом её подменять нельзя, тайминг в CI флейкует.
    """
    loop_thread = threading.get_ident()
    seen: list[int] = []

    def _spy(plain: str, hashed: str) -> bool:
        seen.append(threading.get_ident())
        return True

    monkeypatch.setattr(password_mod, "verify_password", _spy)
    assert await verify_password_bounded("x", "y", key="203.0.113.1") is True
    assert seen and seen[0] != loop_thread


async def test_bounded_rejects_surplus_instead_of_queueing(monkeypatch: pytest.MonkeyPatch) -> None:
    """Сверх лимита — немедленный отказ, а не ожидание в очереди.

    Ожидание выглядело бы безобиднее, но каждый ждущий запрос держит соединение
    к БД (сессия реестра открыта после SELECT), а в QueuePool их 5+10:
    неограниченная очередь выбрала бы пул и положила API — тем же концом, каким
    его клала блокировка цикла.
    """
    monkeypatch.setattr(settings, "login_password_verify_max_inflight", 2)

    def _slow(plain: str, hashed: str) -> bool:
        time.sleep(0.2)
        return False

    monkeypatch.setattr(password_mod, "verify_password", _slow)

    # У каждого запроса СВОЙ ключ: тест про ОБЩИЙ потолок, и отказывать здесь
    # обязан именно он. С одним ключом на всех первым сработал бы лимит доли
    # (#2714) — числа сошлись бы по другой причине, а поломка общего потолка
    # осталась бы незамеченной.
    results = await asyncio.gather(
        *(verify_password_bounded("x", "y", key=f"203.0.113.{i}") for i in range(6)),
        return_exceptions=True,
    )
    rejected = [r for r in results if isinstance(r, PasswordVerifyOverloadedError)]
    admitted = [r for r in results if r is False]
    assert len(admitted) == 2, results
    assert len(rejected) == 4, results

    # Слоты возвращаются: после отработки очереди вход снова доступен.
    assert await verify_password_bounded("x", "y", key="203.0.113.9") is False


async def test_bounded_slot_freed_by_the_work_not_by_cancellation(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Отмена запроса не возвращает слот раньше времени.

    Отмена снимает работу, которая ещё НЕ началась, — с ней проблем нет. Но уже
    начатую сверку она не забирает: поток занят ею все 282 мс. Освобождай мы
    слот по выходу из корутины, на это время он числился бы свободным, и
    одновременно работающих сверок стало бы больше, чем разрешено.
    """
    monkeypatch.setattr(settings, "login_password_verify_max_inflight", 1)
    started = threading.Event()
    finish = threading.Event()

    def _blocked(plain: str, hashed: str) -> bool:
        started.set()
        finish.wait(5)
        return False

    monkeypatch.setattr(password_mod, "verify_password", _blocked)

    task = asyncio.create_task(verify_password_bounded("x", "y", key="203.0.113.1"))
    await asyncio.to_thread(started.wait, 5)

    task.cancel()
    with pytest.raises(asyncio.CancelledError):
        await task

    # Работа всё ещё занимает поток — слот занят, следующий получает отказ.
    # Ключ ДРУГОЙ: отказ обязан прийти от общего потолка (max_inflight=1), а не
    # от доли на ключ — иначе тест проверял бы не тот механизм.
    with pytest.raises(PasswordVerifyOverloadedError):
        await verify_password_bounded("x", "y", key="203.0.113.2")

    finish.set()
    for _ in range(100):  # дать колбэку доехать до цикла
        await asyncio.sleep(0.01)
        if settings.login_password_verify_max_inflight > password_mod._verify_inflight:
            break
    # Отменённая работа вернула И общий слот, И слот своего ключа: тот же адрес
    # снова обслуживается (утечка по ключу при пуле в 1 поток была бы вечным
    # отказом именно этому адресу и больше ничем себя не проявила).
    assert await verify_password_bounded("x", "y", key="203.0.113.1") is False


async def test_bounded_frees_slot_when_verify_raises(monkeypatch: pytest.MonkeyPatch) -> None:
    """Исключение внутри сверки тоже возвращает слот — оба счётчика.

    `verify_password` глотает ValueError/TypeError сама, так что сюда доезжает
    только неожиданное (падение библиотеки, MemoryError). Пул из ОДНОГО потока
    не прощает: один невозвращённый слот — вечный 429 всем на входе, и внешне
    это выглядит не как ошибка bcrypt, а как «вход сломался неизвестно почему».
    """

    def _boom(plain: str, hashed: str) -> bool:
        raise MemoryError("bcrypt died")

    monkeypatch.setattr(password_mod, "verify_password", _boom)
    with pytest.raises(MemoryError):
        await verify_password_bounded("x", "y", key="10.0.0.3")

    monkeypatch.setattr(password_mod, "verify_password", lambda plain, hashed: False)
    assert await verify_password_bounded("x", "y", key="10.0.0.3") is False
    assert password_mod._verify_inflight == 0
    assert not password_mod._verify_inflight_by_key


async def test_one_key_cannot_take_more_than_its_share(monkeypatch: pytest.MonkeyPatch) -> None:
    """Один ключ занимает не больше своей доли — остальным ёмкость остаётся (#2714).

    Меряем именно ЭТО, а не латентность: с общим котлом слотов один источник
    выбирал его целиком, и вход с другого адреса получал 429 бессрочно —
    потолок темпа исправно работал против легитимных пользователей.
    """
    monkeypatch.setattr(settings, "login_password_verify_max_inflight", 4)
    assert password_mod._per_key_slot_cap() == 2  # 4 // 2 — исходные условия теста

    finish = threading.Event()

    def _blocked(plain: str, hashed: str) -> bool:
        finish.wait(5)
        return False

    monkeypatch.setattr(password_mod, "verify_password", _blocked)

    async def _wait_inflight(n: int) -> None:
        deadline = time.monotonic() + 5
        while password_mod._verify_inflight < n:
            assert (
                time.monotonic() < deadline
            ), f"слотов занято {password_mod._verify_inflight} < {n}"
            await asyncio.sleep(0.005)

    flood = [
        asyncio.create_task(verify_password_bounded("x", "y", key="10.0.0.1")) for _ in range(2)
    ]
    await _wait_inflight(2)

    # Третий с ТОГО ЖЕ адреса — отказ, хотя два слота из четырёх свободны.
    # Это ЦЕНА правки, а не побочный эффект: три одновременных входа из одного
    # офиса за NAT укладываются в окно одной сверки (282 мс), и третьему
    # сотруднику теперь отказывают при наполовину пустом пуле — до правки для
    # этого требовалось пятеро. Закрепляем явно, чтобы размен нельзя было
    # потерять молча: свои с ЧУЖИХ адресов за это получают 100% вместо 37%.
    assert password_mod._verify_inflight == 2
    assert settings.login_password_verify_max_inflight == 4
    with pytest.raises(PasswordVerifyOverloadedError):
        await verify_password_bounded("x", "y", key="10.0.0.1")

    # А с другого — пускают. Задачу ставим ДО finish.set() и ждём, пока она
    # займёт слот: иначе «пустили» означало бы только «флуд успел закончиться».
    legit = asyncio.create_task(verify_password_bounded("x", "y", key="10.0.0.2"))
    await _wait_inflight(3)

    finish.set()
    assert await legit is False, "вход с другого адреса обязан пройти во время флуда"
    assert [await f for f in flood] == [False, False]


def test_per_key_cap_never_rounds_down_to_zero(monkeypatch: pytest.MonkeyPatch) -> None:
    """При очереди в 1 слот доля не округляется в ноль.

    `1 // 2 == 0` означало бы «ни одному ключу нельзя ни одного слота» —
    молчаливый отказ ВСЕМ на входе, причём тем более незаметный, что настройка
    выглядит как безобидное ужесточение. `max(1, …)` — тот же страховочный пол,
    что `ge=1` у самой настройки, только от деления.
    """
    monkeypatch.setattr(settings, "login_password_verify_max_inflight", 1)
    assert password_mod._per_key_slot_cap() == 1


async def test_bounded_frees_slot_when_pool_refuses_work(monkeypatch: pytest.MonkeyPatch) -> None:
    """Пул не принял работу → слот отдан прямо здесь, колбэка ведь не будет.

    Единственный путь, где освобождение НЕ висит на future: `submit` бросает
    (пул закрыт на остановке процесса). Утечка тут стоила бы дорого — при пуле
    в один поток невозвращённый слот это вечный 429 всем на входе.
    """
    dead_pool = ThreadPoolExecutor(max_workers=1)
    dead_pool.shutdown()
    monkeypatch.setattr(password_mod, "_VERIFY_POOL", dead_pool)

    with pytest.raises(RuntimeError):
        await verify_password_bounded("x", "y", key="10.0.0.4")

    assert password_mod._verify_inflight == 0
    assert not password_mod._verify_inflight_by_key


async def test_cancelling_queued_work_returns_the_key_slot(monkeypatch: pytest.MonkeyPatch) -> None:
    """Отмена ЕЩЁ НЕ НАЧАТОЙ работы возвращает слот — и общий, и ключа.

    Ветка future другая, чем у отмены начатой работы (`cancel()` на очереди
    успевает, и работа не исполняется вовсе), поэтому проверяется отдельно:
    соседний тест про начатую работу эту не покрывает. Пул из одного потока —
    настоящий, так что второй запрос гарантированно ЖДЁТ в очереди.
    """
    monkeypatch.setattr(settings, "login_password_verify_max_inflight", 4)
    started = threading.Event()
    finish = threading.Event()

    def _blocked(plain: str, hashed: str) -> bool:
        started.set()
        finish.wait(5)
        return False

    monkeypatch.setattr(password_mod, "verify_password", _blocked)

    running = asyncio.create_task(verify_password_bounded("x", "y", key="10.0.0.5"))
    await asyncio.to_thread(started.wait, 5)

    queued = asyncio.create_task(verify_password_bounded("x", "y", key="10.0.0.6"))
    deadline = time.monotonic() + 5
    while password_mod._verify_inflight_by_key.get("10.0.0.6") != 1:
        assert time.monotonic() < deadline, "второй запрос не занял слот"
        await asyncio.sleep(0.005)

    queued.cancel()
    with pytest.raises(asyncio.CancelledError):
        await queued

    deadline = time.monotonic() + 5
    while "10.0.0.6" in password_mod._verify_inflight_by_key:
        assert time.monotonic() < deadline, "слот отменённой очереди не вернулся"
        await asyncio.sleep(0.005)

    finish.set()
    assert await running is False