gendesign/backend/tests/test_auth_db.py

"""DSN и ленивый engine БД `auth` — `app/core/config.py` + `app/core/auth_db.py`.

Эпик «единый вход», стадия 3. Три группы:

  1. Дефолты. Они ЧАСТЬ КОНТРАКТА PR, а не декорация: пока Caddy basic_auth стоит,
     прод обязан вести себя ровно как до эпика — флаг выключен, DSN не сконфигурирован,
     engine не создаётся, отсутствие AUTH_* в окружении не роняет старт.
  2. Сборка DSN из частей: приоритет явного URL, экранирование секрета, пустые
     значения переменных → прод-дефолты (а не мусорный DSN и не падение на импорте).
  3. `auth_db`: ленивость, кеш, внятная ошибка вместо утечки пароля.

Сеть здесь не нужна: `create_engine` пул создаёт лениво и к серверу не ходит.
"""

from __future__ import annotations

import os
import subprocess
import sys
from collections.abc import Iterator
from pathlib import Path

import pytest
from pydantic import SecretStr, ValidationError
from sqlalchemy.engine import make_url

from app.core import auth_db
from app.core.auth_db import AuthDatabaseNotConfiguredError, AuthDatabaseUnreachableError
from app.core.config import Settings, settings

_AUTH_ENV_VARS = (
    "AUTH_MODE",
    "AUTH_DATABASE_URL",
    "AUTH_DB_PASSWORD",
    "AUTH_DB_HOST",
    "AUTH_DB_PORT",
    "AUTH_DB_NAME",
    "AUTH_DB_USER",
    "SESSION_COOKIE_NAME",
    "SESSION_TTL_HOURS",
)

# Заведомо синтаксически корректный DSN на несуществующий хост: engine по нему
# создаётся, но соединение не открывается (пул ленивый), поэтому тесты офлайновы.
_OFFLINE_DSN = "postgresql+psycopg://auth_app:pw@127.0.0.1:1/auth"


@pytest.fixture
def clean_env(monkeypatch: pytest.MonkeyPatch) -> None:
    """Ни одной AUTH_*/SESSION_* переменной — тест дефолтов не зависит от машины."""
    for name in _AUTH_ENV_VARS:
        monkeypatch.delenv(name, raising=False)


@pytest.fixture(autouse=True)
def _reset_engine_cache() -> Iterator[None]:
    """Ни один тест не оставляет за собой закешированный engine БД `auth`."""
    auth_db.reset_auth_db()
    yield
    auth_db.reset_auth_db()


# ---------------------------------------------------------------------------
# 1. Дефолты
# ---------------------------------------------------------------------------


def test_session_mode_is_off_and_unconfigured_by_default(clean_env: None) -> None:
    """Свежие настройки без AUTH_*: режим legacy, DSN пуст — и это НЕ ошибка."""
    fresh = Settings()

    assert fresh.auth_mode == "legacy"
    assert fresh.auth_session_enabled is False
    assert fresh.resolved_auth_database_url == ""


def test_live_settings_singleton_is_off() -> None:
    """Тот же инвариант на настоящем синглтоне, которым пользуется приложение."""
    assert settings.auth_mode == "legacy"
    assert settings.auth_session_enabled is False


@pytest.mark.parametrize("raw", ["", "   ", "LEGACY", " legacy "])
def test_blank_or_odd_case_auth_mode_falls_back_to_legacy(
    clean_env: None, monkeypatch: pytest.MonkeyPatch, raw: str
) -> None:
    """`AUTH_MODE=` (или регистр/пробелы) → legacy, а не ValidationError на импорте.

    `settings = Settings()` выполняется на уровне модуля: невалидное значение уронило бы
    ИМПОРТ конфига и увело контейнер в restart-loop. Сценарий бытовой — ops копирует
    блок AUTH_* в .env.runtime и заполняет только пароль.
    """
    monkeypatch.setenv("AUTH_MODE", raw)

    assert Settings().auth_mode == "legacy"


def test_meaningful_garbage_in_auth_mode_still_fails(
    clean_env: None, monkeypatch: pytest.MonkeyPatch
) -> None:
    """`AUTH_MODE=off` — опечатка со смыслом, и она обязана падать.

    Молча трактовать её как legacy значило бы тихо оставить продукт на trusted-header
    после того, как последний PR эпика снимет popup.
    """
    monkeypatch.setenv("AUTH_MODE", "off")

    with pytest.raises(ValidationError):
        Settings()


@pytest.mark.parametrize(
    ("mode", "enabled"), [("legacy", False), ("dual", True), ("db_only", True)]
)
def test_auth_session_enabled_is_derived_from_mode(
    clean_env: None, monkeypatch: pytest.MonkeyPatch, mode: str, enabled: bool
) -> None:
    """Свойство `auth_session_enabled` = «режим не legacy» — единый выключатель реестра.

    Оно и закорачивает `resolve_session_token` / `require_auth_db_configured`; разница
    dual vs db_only ему не видна и не должна быть (она про фолбэк в rbac_guard).
    """
    monkeypatch.setenv("AUTH_MODE", mode)

    assert Settings().auth_session_enabled is enabled


def test_default_host_is_this_stacks_postgres(clean_env: None) -> None:
    """🪤 Дефолт хоста — `postgres`, и это ЗЕРКАЛЬНО «Мере», а не копия с неё.

    У «Меры» дефолт `gendesign-postgres`, потому что внутри её стека имя `postgres`
    занято её собственным контейнером. У «Птицы» наоборот: её стек главный, сервис
    `postgres` корневого docker-compose.prod.yml и есть сервер с БД `auth`. Алиас
    `gendesign-postgres` живёт только во внешней сети `shared`, куда входят не все
    сервисы (beat — нет), поэтому дефолтом он быть не может.
    """
    fresh = Settings()

    assert fresh.auth_db_host == "postgres"
    assert fresh.auth_db_host != "gendesign-postgres"
    assert fresh.auth_db_port == 5432
    assert fresh.auth_db_name == "auth"
    assert fresh.auth_db_user == "auth_app"


def test_cookie_defaults_match_the_other_product(clean_env: None) -> None:
    """Имя куки и TTL обязаны совпадать с «Мерой» — иначе общая сессия не общая.

    Имя историческое («tradein_» уже ни о чём не говорит); переименование
    разлогинивает всех сразу в обоих продуктах, поэтому оно закреплено тестом.
    """
    fresh = Settings()

    assert fresh.session_cookie_name == "tradein_session"
    assert fresh.session_ttl_hours == 720


# ---------------------------------------------------------------------------
# 2. Сборка DSN
# ---------------------------------------------------------------------------


def test_explicit_url_wins_over_parts(clean_env: None) -> None:
    """AUTH_DATABASE_URL — аварийный обход (другой хост, sslmode): выигрывает всегда."""
    fresh = Settings(
        auth_database_url="  postgresql+psycopg://u:p@elsewhere:6432/auth?sslmode=require  ",
        auth_db_password=SecretStr("ignored"),
        auth_db_host="postgres",
    )

    assert (
        fresh.resolved_auth_database_url
        == "postgresql+psycopg://u:p@elsewhere:6432/auth?sslmode=require"
    )


def test_dsn_built_from_password_and_defaults(clean_env: None) -> None:
    """Включение на проде = одна переменная: пароль + прод-дефолты остальных частей."""
    fresh = Settings(auth_db_password=SecretStr("s3cret"))

    assert (
        fresh.resolved_auth_database_url
        == "postgresql+psycopg://auth_app:s3cret@postgres:5432/auth"
    )


def test_password_special_chars_survive_round_trip(clean_env: None) -> None:
    """Пароль экранируется: `@`/`/`/`:`/`#`/`%` иначе порвали бы URL по его грамматике.

    Проверяем не наличие процентов в строке, а РАЗБОР обратно: важно, что SQLAlchemy
    видит тот же пароль и, главное, тот же хост/базу. Незакавыченный `@` молча увёл бы
    подключение на другой хост.
    """
    raw = "p@ss:w/rd#1%zz?x"
    url = make_url(Settings(auth_db_password=SecretStr(raw)).resolved_auth_database_url)

    assert url.password == raw
    assert url.host == "postgres"
    assert url.port == 5432
    assert url.database == "auth"
    assert url.username == "auth_app"


def test_username_is_quoted_too(clean_env: None) -> None:
    url = make_url(
        Settings(
            auth_db_password=SecretStr("pw"), auth_db_user="odd:user@name"
        ).resolved_auth_database_url
    )

    assert url.username == "odd:user@name"
    assert url.host == "postgres"


def test_password_whitespace_is_preserved_not_stripped(clean_env: None) -> None:
    """Ведущий/хвостовой пробел может быть частью настоящего пароля — не режем."""
    url = make_url(Settings(auth_db_password=SecretStr(" pw ")).resolved_auth_database_url)

    assert url.password == " pw "


def test_blank_password_means_not_configured(clean_env: None) -> None:
    """Пробельная строка — опечатка в .env, а не пароль: «не сконфигурировано»."""
    assert Settings(auth_db_password=SecretStr("   ")).resolved_auth_database_url == ""
    assert Settings(auth_db_password=SecretStr("")).resolved_auth_database_url == ""


def test_blank_parts_fall_back_to_defaults(clean_env: None) -> None:
    """`AUTH_DB_HOST=` в .env.runtime не должен давать DSN вида `...@:5432/auth`.

    Сценарий бытовой: ops копирует блок AUTH_DB_* целиком и заполняет только пароль.
    """
    url = make_url(
        Settings(
            auth_db_password=SecretStr("pw"),
            auth_db_host="   ",
            auth_db_name="",
            auth_db_user=" ",
        ).resolved_auth_database_url
    )

    assert (url.host, url.database, url.username) == ("postgres", "auth", "auth_app")


@pytest.mark.parametrize("blank", ["", "   "])
def test_blank_port_does_not_break_import(clean_env: None, blank: str) -> None:
    """`AUTH_DB_PORT=` → дефолт, а НЕ ValidationError.

    `settings = Settings()` выполняется на уровне модуля: падение здесь уводило бы
    контейнер в restart-loop — причём в дефолтном режиме, где к БД `auth` не идёт ни
    одного обращения.
    """
    assert Settings(auth_db_port=blank).auth_db_port == 5432


def test_non_blank_garbage_port_still_fails(clean_env: None) -> None:
    """`AUTH_DB_PORT=abc` — опечатка со смыслом, её глушить нельзя."""
    with pytest.raises(ValueError):
        Settings(auth_db_port="abc")


def test_password_is_not_printed_by_repr_or_dump(clean_env: None) -> None:
    """SecretStr: пароль не утекает в `repr(settings)` / `model_dump()`.

    Сегодня их никто не рендерит, но появиться такой рендер (лог старта, /debug) может
    тихо — а рядом с обычным str-полем это была бы утечка секрета в открытый лог.
    """
    fresh = Settings(auth_db_password=SecretStr("s3cret"))

    assert "s3cret" not in repr(fresh)
    assert "s3cret" not in str(fresh.model_dump())
    assert fresh.auth_db_password.get_secret_value() == "s3cret"


# ---------------------------------------------------------------------------
# 3. auth_db: ленивость, кеш, ошибки
# ---------------------------------------------------------------------------


def test_import_without_any_auth_env_does_not_build_engine() -> None:
    """Импорт в чистом окружении: ошибки нет, engine не создан, DSN пуст.

    Проверяется отдельным процессом, потому что в текущем модуль импортирован давно и
    любое утверждение про «на импорте» было бы про уже случившийся импорт. Это отличие
    от `app.core.db`, где engine создаётся в теле модуля: сделай мы так же, приложение
    падало бы на старте везде, где реестр не сконфигурирован — локально, в pytest, на
    любом стенде. Ровно тот контракт, который держит дефолтное поведение прода.
    """
    env = {k: v for k, v in os.environ.items() if k not in _AUTH_ENV_VARS}
    code = (
        "from app.core import auth_db\n"
        "from app.core.config import settings\n"
        "print(auth_db._engine, repr(settings.resolved_auth_database_url), "
        "settings.auth_session_enabled)\n"
    )
    proc = subprocess.run(
        [sys.executable, "-c", code],
        cwd=Path(__file__).resolve().parents[1],
        env=env,
        capture_output=True,
        text=True,
    )

    assert proc.returncode == 0, proc.stderr
    assert proc.stdout.strip() == "None '' False"


def test_unconfigured_registry_raises_with_actionable_message(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Пустой DSN → явная ошибка с инструкцией, а не «сессия не найдена»."""
    monkeypatch.setattr(settings, "auth_database_url", "")
    monkeypatch.setattr(settings, "auth_db_password", SecretStr(""))

    with pytest.raises(AuthDatabaseNotConfiguredError) as excinfo:
        auth_db.get_auth_engine()

    assert "AUTH_MODE" in str(excinfo.value)
    assert "AUTH_DB_PASSWORD" in str(excinfo.value)


@pytest.mark.parametrize(
    "broken",
    [
        "not-a-dsn-at-all",
        # «Почти URL»: разбор доходит до int(port) и падает, унося в текст ошибки
        # кусок пароля, съехавший на позицию порта.
        "postgresql+psycopg://u:pa@ss@host:wo/auth",
    ],
)
def test_malformed_dsn_does_not_leak_into_the_error(
    monkeypatch: pytest.MonkeyPatch, broken: str
) -> None:
    """Битый DSN → своя ошибка БЕЗ самого DSN и без исходного traceback.

    Текст ошибки SQLAlchemy цитирует строку целиком, а в ней пароль роли auth_app.
    `from None` обязателен: без него исходная ошибка печаталась бы в traceback как
    «During handling of the above exception...» — то есть пароль всё равно оказался бы
    в логе.
    """
    monkeypatch.setattr(settings, "auth_database_url", broken)

    with pytest.raises(AuthDatabaseNotConfiguredError) as excinfo:
        auth_db.get_auth_engine()

    assert broken not in str(excinfo.value)
    assert "pa@ss" not in str(excinfo.value)
    assert excinfo.value.__suppress_context__ is True
    assert excinfo.value.__cause__ is None


def test_engine_is_built_once_and_reused(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(settings, "auth_database_url", _OFFLINE_DSN)

    first = auth_db.get_auth_engine()
    second = auth_db.get_auth_engine()

    assert first is second
    assert auth_db.get_auth_session_factory().kw["bind"] is first
    assert first.url.database == "auth"


def test_reset_drops_the_cached_engine(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(settings, "auth_database_url", _OFFLINE_DSN)
    first = auth_db.get_auth_engine()

    auth_db.reset_auth_db()

    assert auth_db._engine is None
    assert auth_db.get_auth_engine() is not first


# ---------------------------------------------------------------------------
# require_auth_db_configured — fail-fast на старте (lifespan)
# ---------------------------------------------------------------------------


def test_startup_check_is_noop_while_flag_is_off(monkeypatch: pytest.MonkeyPatch) -> None:
    """🔒 Дефолт: пустой DSN на старте — не ошибка, и engine не создаётся.

    Ровно то, что произойдёт на проде сразу после мержа этого PR.
    """
    monkeypatch.setattr(settings, "auth_mode", "legacy")
    monkeypatch.setattr(settings, "auth_database_url", "")
    monkeypatch.setattr(settings, "auth_db_password", SecretStr(""))

    auth_db.require_auth_db_configured()

    assert auth_db._engine is None


def test_startup_check_fails_fast_when_enabled_without_dsn(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Флаг включили, DSN не задали → контейнер не стартует.

    Иначе пустой DSN ловил бы `except` в guard'е, и сломанная конфигурация выглядела бы
    как «ни у кого нет сессии» — сутками, при живом приложении и 200-х в ответах.
    """
    monkeypatch.setattr(settings, "auth_mode", "dual")
    monkeypatch.setattr(settings, "auth_database_url", "")
    monkeypatch.setattr(settings, "auth_db_password", SecretStr(""))

    with pytest.raises(AuthDatabaseNotConfiguredError):
        auth_db.require_auth_db_configured()


def test_startup_check_builds_engine_and_probes_connection(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Режим включён и реестр отвечает → engine готов ещё до первого запроса.

    Проба соединения подменена: поднимать Postgres ради этого теста незачем, важно, что
    она вызывается ИМЕННО на том engine, который останется закешированным.
    """
    monkeypatch.setattr(settings, "auth_mode", "dual")
    monkeypatch.setattr(settings, "auth_database_url", _OFFLINE_DSN)
    probed: list[object] = []
    monkeypatch.setattr(auth_db, "_probe_connection", probed.append)

    auth_db.require_auth_db_configured()

    assert auth_db._engine is not None
    assert probed == [auth_db._engine]


def test_startup_check_fails_when_dsn_parses_but_connection_does_not(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """🔴 Смысл проверки: DSN разобрался — это ещё НЕ «реестр рабочий».

    `create_engine` к серверу не ходит, поэтому одна лишь сборка engine отлавливала бы
    ровно два случая (DSN пуст / не парсится). Весь вероятный класс ошибок — неверный
    AUTH_DB_PASSWORD, опечатка в хосте, не созданная БД `auth`, отозванная роль
    auth_app, нет сети — проходил бы мимо, контейнер стартовал бы зелёным, `/health`
    отвечал бы 200, а каждый запрос с кукой молча деградировал бы на легаси-заголовок.
    Сутками. Ровно то, что комментарий в app/main.py обещает НЕ допускать.
    """
    monkeypatch.setattr(settings, "auth_mode", "dual")
    monkeypatch.setattr(settings, "auth_database_url", _OFFLINE_DSN)

    def _refused(_engine: object) -> None:
        raise OSError("connection to server at 127.0.0.1, port 1 failed: Connection refused")

    monkeypatch.setattr(auth_db, "_probe_connection", _refused)

    with pytest.raises(AuthDatabaseUnreachableError) as excinfo:
        auth_db.require_auth_db_configured()

    # Причина сохранена в цепочке — ради неё проверка и делается; DSN (в нём пароль) в
    # наш текст не подставляется.
    assert isinstance(excinfo.value.__cause__, OSError)
    assert "AUTH_MODE" in str(excinfo.value)
    assert _OFFLINE_DSN not in str(excinfo.value)


def test_startup_check_does_not_probe_while_flag_is_off(monkeypatch: pytest.MonkeyPatch) -> None:
    """🔒 legacy: ни коннекта, ни пробы — даже если DSN задан и валиден."""
    monkeypatch.setattr(settings, "auth_mode", "legacy")
    monkeypatch.setattr(settings, "auth_database_url", _OFFLINE_DSN)

    def _boom(_engine: object) -> None:
        raise AssertionError("в режиме legacy соединение с реестром недопустимо")

    monkeypatch.setattr(auth_db, "_probe_connection", _boom)

    auth_db.require_auth_db_configured()

    assert auth_db._engine is None


def test_engine_has_short_timeouts(monkeypatch: pytest.MonkeyPatch) -> None:
    """Реестр — не критический путь: его сбой обязан деградировать за секунды.

    Без `connect_timeout` дропнутые SYN (фаервол молча глотает пакеты) держали бы
    попытку до TCP-таймаута ОС — на Linux ~130 с, и так на КАЖДОМ checkout'е, потому
    что включён `pool_pre_ping`. `pool_timeout` по дефолту 30 с — в auth-пути столько
    ждать свободный коннект незачем.
    """
    monkeypatch.setattr(settings, "auth_database_url", _OFFLINE_DSN)
    captured: dict[str, object] = {}
    real_create_engine = auth_db.create_engine

    def _spy(dsn: str, **kwargs: object) -> object:
        captured.update(kwargs)
        return real_create_engine(dsn, **kwargs)  # type: ignore[arg-type]

    monkeypatch.setattr(auth_db, "create_engine", _spy)

    auth_db.get_auth_engine()

    assert captured["pool_timeout"] == 3
    assert captured["pool_pre_ping"] is True
    connect_args = captured["connect_args"]
    assert isinstance(connect_args, dict)
    assert connect_args["connect_timeout"] == 3
    assert "statement_timeout=3000" in connect_args["options"]