gendesign/backend/tests/test_auth_session_service.py

"""Резолв сессии общего реестра — `app/services/auth_session.py` (эпик «единый вход»).

Слой ниже guard'а: «что считать валидной сессией» и «когда продлевать». Через guard
эти правила проверяются end-to-end в `tests/test_auth_session_guard.py`; здесь —
поштучно, включая ветки, до которых из guard'а дотянуться дорого (sliding refresh,
сбой продления, исключения БД).

Дублёр сессии БД (`FakeAuthDb`) намеренно ОДИН на оба файла и живёт в guard-тестах:
разъехавшиеся двойники — типовой способ получить два зелёных теста при одном сломанном
поведении. Прецедент кросс-импорта внутри пакета tests — `tests/integration/*`.
"""

from __future__ import annotations

import logging
from datetime import UTC, datetime, timedelta

import pytest
from pydantic import SecretStr

from app.core import auth_db
from app.core.auth_db import AuthDatabaseNotConfiguredError
from app.core.config import settings
from app.services import auth_session as svc
from app.services.auth_session import AccessState, SessionUser, get_session_user, to_access_state
from tests.test_auth_session_guard import FakeAuthDb, _install_auth_db, _Row

_TOKEN = "tok-1"
_USER = "user1"


def _row(
    *,
    expires_in: timedelta = timedelta(days=7),
    last_seen_ago: timedelta | None = timedelta(seconds=30),
    username: str = _USER,
    access_state: str = "active",
) -> _Row:
    now = datetime.now(UTC)
    return _Row(
        expires_at=now + expires_in,
        last_seen_at=None if last_seen_ago is None else now - last_seen_ago,
        username=username,
        access_state=access_state,
    )


# ---------------------------------------------------------------------------
# AccessState / to_access_state — fail-closed
# ---------------------------------------------------------------------------


def test_only_active_can_sign_in() -> None:
    assert AccessState.ACTIVE.can_sign_in is True
    assert AccessState.TRIAL_EXPIRED.can_sign_in is False
    assert AccessState.DISABLED.can_sign_in is False


def test_to_access_state_known_values() -> None:
    assert to_access_state("active") is AccessState.ACTIVE
    assert to_access_state("trial_expired") is AccessState.TRIAL_EXPIRED
    assert to_access_state("disabled") is AccessState.DISABLED


@pytest.mark.parametrize("value", ["frozen", "", None, 42])
def test_to_access_state_unknown_is_disabled_with_warning(
    value: object, caplog: pytest.LogCaptureFixture
) -> None:
    """Неизвестное/NULL/чужой тип → disabled + WARNING.

    Миграции БД `auth` применяет деплой «Птицы», то есть новое состояние может
    появиться в базе раньше, чем код о нём узнает. Обратный выбор («не disabled =
    пускаем») означал бы, что такая миграция молча раздаёт доступ.
    """
    with caplog.at_level(logging.WARNING, logger="app.services.auth_session"):
        assert to_access_state(value) is AccessState.DISABLED
    assert caplog.records


# ---------------------------------------------------------------------------
# get_session_user — что считается валидной сессией
# ---------------------------------------------------------------------------


def test_valid_session_resolves_to_user() -> None:
    db = FakeAuthDb({_TOKEN: _row()})

    assert get_session_user(db, _TOKEN) == SessionUser(
        username=_USER, access_state=AccessState.ACTIVE
    )
    assert db.select_tokens == [_TOKEN]


def test_empty_token_short_circuits_without_query() -> None:
    db = FakeAuthDb({_TOKEN: _row()})

    assert get_session_user(db, "") is None
    assert db.select_tokens == []


def test_unknown_token_returns_none() -> None:
    db = FakeAuthDb()

    assert get_session_user(db, "never-issued") is None
    assert db.updates == []


def test_expired_session_returns_none_and_is_not_refreshed() -> None:
    """Истёкшая сессия не воскресает sliding-refresh'ем — иначе TTL был бы вечным."""
    db = FakeAuthDb(
        {_TOKEN: _row(expires_in=timedelta(seconds=-1), last_seen_ago=timedelta(days=1))}
    )

    assert get_session_user(db, _TOKEN) is None
    assert db.updates == []


def test_null_expires_at_returns_none() -> None:
    """`expires_at IS NULL` (колонку ослабили) → сессии нет, а не TypeError в auth-пути."""
    row = _row()
    row.expires_at = None
    db = FakeAuthDb({_TOKEN: row})

    assert get_session_user(db, _TOKEN) is None


@pytest.mark.parametrize("access_state", ["disabled", "trial_expired", "unheard_of"])
def test_non_active_user_returns_none_immediately(access_state: str) -> None:
    """Блокировка в реестре бьёт сразу, не дожидаясь expires_at (иначе до 30 дней)."""
    db = FakeAuthDb({_TOKEN: _row(access_state=access_state)})

    assert get_session_user(db, _TOKEN) is None
    assert db.updates == []


# ---------------------------------------------------------------------------
# Sliding refresh
# ---------------------------------------------------------------------------


def test_refresh_skipped_for_recent_session() -> None:
    """Свежий last_seen_at → ни одного UPDATE: иначе каждый API-запрос бил бы в БД."""
    db = FakeAuthDb({_TOKEN: _row(last_seen_ago=timedelta(seconds=30))})

    assert get_session_user(db, _TOKEN) is not None
    assert db.updates == []
    assert db.commits == 0


@pytest.mark.parametrize(
    "last_seen_ago", [timedelta(minutes=5), timedelta(hours=3), None], ids=["at-5m", "3h", "null"]
)
def test_refresh_extends_after_interval(last_seen_ago: timedelta | None) -> None:
    """>= 5 минут (и NULL) → один UPDATE на обе колонки + commit.

    TTL берётся из настроек и обязан совпадать с «Мерой»: продлевает сессию тот
    продукт, в котором кликнули последним, и срок жизни не должен от этого зависеть.
    """
    db = FakeAuthDb({_TOKEN: _row(last_seen_ago=last_seen_ago)})

    assert get_session_user(db, _TOKEN) is not None
    assert db.updates == [{"ttl_hours": settings.session_ttl_hours, "token": _TOKEN}]
    assert db.commits == 1


def test_refresh_uses_configured_ttl(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(settings, "session_ttl_hours", 12)
    db = FakeAuthDb({_TOKEN: _row(last_seen_ago=timedelta(hours=1))})

    get_session_user(db, _TOKEN)

    assert db.updates == [{"ttl_hours": 12, "token": _TOKEN}]


def test_refresh_failure_does_not_block_valid_session(caplog: pytest.LogCaptureFixture) -> None:
    """Продление — best-effort: сбой логируется и откатывается, юзер всё равно валиден.

    Иначе read-only реплика или блокировка строки разлогинивала бы всех, у кого
    сессия старше пяти минут.
    """
    db = FakeAuthDb({_TOKEN: _row(last_seen_ago=timedelta(hours=1))}, fail_refresh=True)

    with caplog.at_level(logging.WARNING, logger="app.services.auth_session"):
        user = get_session_user(db, _TOKEN)

    assert user == SessionUser(username=_USER, access_state=AccessState.ACTIVE)
    assert db.rollbacks == 1
    assert any("sliding refresh failed" in r.getMessage() for r in caplog.records)
    # В сообщении не должно быть ни username, ни токена: лог — не место для связки
    # «кто именно» + «когда», а разбор идёт по времени.
    assert not any(_USER in r.getMessage() or _TOKEN in r.getMessage() for r in caplog.records)


def test_select_failure_is_not_swallowed() -> None:
    """Сбой SELECT'а летит наружу: решение «что делать со сломанным реестром» — не здесь.

    Проглоти резолвер ошибку — вызывающий получил бы «сессии нет», то есть отказ
    выглядел бы как «просто не залогинен», а откат на trusted-header — как норма.
    """
    db = FakeAuthDb()

    def _boom(*_a: object, **_k: object) -> None:
        raise RuntimeError("auth registry is down")

    db.execute = _boom  # type: ignore[method-assign]

    with pytest.raises(RuntimeError, match="auth registry is down"):
        get_session_user(db, _TOKEN)


# ---------------------------------------------------------------------------
# resolve_session_token — точка входа guard'а
# ---------------------------------------------------------------------------


def test_resolve_returns_none_without_touching_db_when_flag_off(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """🔒 Инвариант «выключенный флаг = ни одного коннекта» держится этим модулем.

    Он не полагается на то, что вызывающий сам не позовёт резолв: даже с валидным
    токеном соединение не открывается.
    """
    monkeypatch.setattr(settings, "auth_mode", "legacy")
    _install_auth_db(monkeypatch, None)  # открытие сессии → RuntimeError

    assert svc.resolve_session_token(_TOKEN) is None


@pytest.mark.parametrize("token", [None, ""])
def test_resolve_returns_none_for_empty_token(
    monkeypatch: pytest.MonkeyPatch, token: str | None
) -> None:
    monkeypatch.setattr(settings, "auth_mode", "dual")
    _install_auth_db(monkeypatch, None)

    assert svc.resolve_session_token(token) is None


def test_resolve_opens_registry_session_when_flag_on(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(settings, "auth_mode", "dual")
    db = FakeAuthDb({_TOKEN: _row()})
    _install_auth_db(monkeypatch, db)

    assert svc.resolve_session_token(_TOKEN) == SessionUser(
        username=_USER, access_state=AccessState.ACTIVE
    )
    assert db.select_tokens == [_TOKEN]


def test_resolve_propagates_not_configured_error(monkeypatch: pytest.MonkeyPatch) -> None:
    """Флаг включён, DSN пуст → исключение наружу, а не «сессия не найдена».

    Тихий None здесь означал бы либо массовый отказ доступа под видом «не залогинен»,
    либо (в guard'е) бессрочную раздачу прав в обход реестра. Настоящий `auth_db` не
    подменяется — проверяется именно связка сервис ↔ конфигурация.
    """
    monkeypatch.setattr(settings, "auth_mode", "dual")
    monkeypatch.setattr(settings, "auth_database_url", "")
    monkeypatch.setattr(settings, "auth_db_password", SecretStr(""))
    auth_db.reset_auth_db()
    try:
        with pytest.raises(AuthDatabaseNotConfiguredError):
            svc.resolve_session_token(_TOKEN)
    finally:
        auth_db.reset_auth_db()


def test_expiry_is_also_filtered_by_db_clock() -> None:
    """Срок годности отсекается ЧАСАМИ БД, а не только часами процесса.

    Асимметрия, которую это закрывает: решение «жива ли сессия» принимал Python
    (`datetime.now(UTC)`), а продление писало `expires_at = now() + interval` часами
    СЕРВЕРА. Отставание часов приложения давало бы не «сессия проживёт на дельту
    дольше», а необратимое воскрешение: строку, которую БД уже считает мёртвой, Python
    пропускал бы, тут же срабатывал sliding-refresh и отодвигал expires_at на полный
    TTL от серверного now(). Секунда расхождения → +30 дней жизни.

    Форма запроса проверяется дублёром (`_assert_select_shape`), поэтому здесь
    достаточно одного прохода: потеряется `AND s.expires_at > now()` — тест покраснеет.
    """
    db = FakeAuthDb({_TOKEN: _row()})

    assert get_session_user(db, _TOKEN) is not None


def test_rollback_failure_does_not_break_the_resolve(caplog: pytest.LogCaptureFixture) -> None:
    """Сбой самого rollback'а (оборванный коннект) не отменяет валидную сессию.

    Иначе «best-effort продление» переставало быть best-effort: исключение улетало бы
    из get_session_user наружу, и валидный юзер получал бы вместо доступа ERROR в
    GlitchTip и деградацию на легаси-заголовок (а в db_only — отказ).
    """
    db = FakeAuthDb({_TOKEN: _row(last_seen_ago=timedelta(hours=1))}, fail_refresh=True)

    def _dead_connection() -> None:
        raise RuntimeError("server closed the connection unexpectedly")

    db.rollback = _dead_connection  # type: ignore[method-assign]

    with caplog.at_level(logging.WARNING, logger="app.services.auth_session"):
        user = get_session_user(db, _TOKEN)

    assert user == SessionUser(username=_USER, access_state=AccessState.ACTIVE)
    assert any("rollback" in r.getMessage() for r in caplog.records)