gendesign/tradein-mvp/backend/tests/test_identity_store.py

"""Tests for app.services.identity_store + app.core.auth_db — эпик «единый вход».

`identity_store` — единственное место, знающее, В КАКОЙ БД и В КАКИХ ТАБЛИЦАХ
живёт identity. Всё остальное (auth_session, rbac, роуты) спрашивает у него, и
поэтому ошибка ЗДЕСЬ — это ошибка сразу везде.

Главное, что пинят эти тесты (⚠️ ограничение PR: после мержа прод обязан
работать ТОЧНО как сейчас):

  1. ДЕФОЛТ = старое поведение. `IDENTITY_STORE` не задан → `tradein_users` /
     `tradein_sessions`, boolean-колонка, сессия из `app.core.db.SessionLocal`.
  2. При дефолте код НЕ ТРОГАЕТ БД `auth` вообще: engine не строится, пустой
     `AUTH_DATABASE_URL` не ошибка. На проде роль `auth_app` ещё без пароля и
     DSN не заведён — любое обращение туда было бы отказом входа.
  3. `IDENTITY_STORE=auth` + пустой DSN → ЯВНАЯ `AuthDatabaseNotConfiguredError`,
     а не тихий фолбэк на tradein-таблицы и не пустой результат. Молчаливая
     деградация auth-пути читалась бы как «неверный пароль» у всех сразу.
  4. `get_identity_db` в дефолтном режиме отдаёт ТОТ ЖЕ объект `Session`, что и
     `get_db` — «Команда» пишет строку сотрудника и его квоту одной транзакцией.
     Регрессия здесь дала бы состояние «сотрудник создан, квота нет».
  5. Литералы значений состояния (`True`/`'active'`/...) — пин по таблице
     значений, а не round-trip через `to_access_state`: инверсия
     `access_state_param` обязана быть видна.
"""

from __future__ import annotations

import os
from typing import Annotated, Any

os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

import pytest
from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient
from pydantic import SecretStr
from sqlalchemy import Engine

from app.core import auth_db, config
from app.core.db import get_db
from app.core.rbac import rbac_guard
from app.services import identity_store
from app.services.identity_store import (
    AccessState,
    access_state_param,
    get_identity_db,
    identity_schema,
    identity_session,
    to_access_state,
)
from tests.support.identity_modes import IDENTITY_MODES, use_identity_mode

_FAKE_AUTH_DSN = "postgresql+psycopg://auth_app:secret@localhost:5432/auth"


@pytest.fixture(autouse=True)
def _clean_identity_state(monkeypatch: pytest.MonkeyPatch):
    """Дефолтный режим + пустой DSN + сброшенный engine до И после теста.

    Engine БД `auth` живёт в module-global, а не в `settings`, поэтому
    monkeypatch его не откатывает — держим сброс явно с обеих сторон, иначе
    построенный здесь engine утёк бы в любой следующий тест сьюта.
    """
    auth_db.reset_auth_db()
    monkeypatch.setattr(config.settings, "identity_store", "tradein")
    monkeypatch.setattr(config.settings, "auth_database_url", "")
    # Второй источник DSN: при пустом AUTH_DATABASE_URL он собирается из
    # AUTH_DB_PASSWORD + частей (см. Settings.resolved_auth_database_url). Не
    # обнули его здесь — и заданная в окружении переменная сделала бы реестр
    # «сконфигурированным»: тесты про «пустой DSN → явная ошибка» позеленели бы
    # мимо проверяемого поведения.
    # SecretStr, а не "": поле объявлено `SecretStr`, а `validate_assignment` у
    # Settings выключен — monkeypatch кладёт значение КАК ЕСТЬ, без приведения
    # типа, и голая строка уронила бы резолвер на `.get_secret_value()`.
    monkeypatch.setattr(config.settings, "auth_db_password", SecretStr(""))
    yield
    auth_db.reset_auth_db()


class _FakeSession:
    """Session-заглушка: тестам здесь важна ИДЕНТИЧНОСТЬ объекта, не поведение."""

    def __enter__(self) -> _FakeSession:
        return self

    def __exit__(self, *exc: object) -> bool:
        return False

    def close(self) -> None:
        pass


# ---------------------------------------------------------------------------
# identity_schema — имена, попадающие прямо в SQL
# ---------------------------------------------------------------------------


def test_settings_defaults_are_legacy_mode(monkeypatch: pytest.MonkeyPatch) -> None:
    """⚠️ ГЛАВНЫЙ ИНВАРИАНТ PR, пин НАПРЯМУЮ по классу настроек.

    Все остальные identity-тесты работают под autouse-фикстурой, которая
    ПРИНУДИТЕЛЬНО выставляет `identity_store="tradein"` — то есть проверяют
    поведение при уже выбранном режиме, а не сам дефолт. Перевернись
    `Field(default=...)` в config.py — они бы этого не заметили, и прод молча
    ушёл бы в БД `auth`, где ещё нет ни пароля роли `auth_app`, ни данных.

    Поэтому здесь настройки конструируются заново, минуя `config.settings`:
      * `_env_file=None` — не читать локальный `.env` (дев-машина или CI могут
        держать там свои значения; пиним ДЕФОЛТ КОДА, а не окружение);
      * `delenv` обеих переменных — то же самое для переменных процесса.
    Останется ровно то, что записано литералом в `Settings`.
    """
    monkeypatch.delenv("IDENTITY_STORE", raising=False)
    monkeypatch.delenv("AUTH_DATABASE_URL", raising=False)

    fresh = config.Settings(_env_file=None)  # type: ignore[call-arg]

    assert fresh.identity_store == "tradein", (
        "дефолт IDENTITY_STORE обязан остаться 'tradein': прод после мержа должен "
        "работать ТОЧНО как сейчас, на tradein_users/tradein_sessions"
    )
    assert fresh.auth_database_url == "", (
        "AUTH_DATABASE_URL обязан быть пуст по умолчанию: на проде DSN роли "
        "auth_app ещё не заведён, и пустое значение не должно ронять старт"
    )


def test_default_schema_is_todays_production(monkeypatch: pytest.MonkeyPatch) -> None:
    """Без переменной окружения — ровно сегодняшние таблицы «Меры»."""
    schema = identity_schema()
    assert schema.store == "tradein"
    assert schema.users_table == "tradein_users"
    assert schema.sessions_table == "tradein_sessions"
    assert schema.access_state_column == "is_active"
    assert schema.access_state_sql_type == "boolean"


def test_auth_schema_points_at_shared_registry(monkeypatch: pytest.MonkeyPatch) -> None:
    """В БД `auth` таблицы без префикса продукта — реестр общий на «Меру» и «Птицу»."""
    use_identity_mode(monkeypatch, "auth")
    schema = identity_schema()
    assert schema.store == "auth"
    assert schema.users_table == "users"
    assert schema.sessions_table == "sessions"
    assert schema.access_state_column == "access_state"
    assert schema.access_state_sql_type == "text"


def test_schema_is_read_per_call_not_cached_at_import(monkeypatch: pytest.MonkeyPatch) -> None:
    """Флаг читается на КАЖДОМ вызове: переключение не требует перезагрузки модулей."""
    assert identity_schema().users_table == "tradein_users"
    use_identity_mode(monkeypatch, "auth")
    assert identity_schema().users_table == "users"


def test_unknown_store_raises_instead_of_silent_fallback(monkeypatch: pytest.MonkeyPatch) -> None:
    """Значение вне словаря — ошибка, а не «ну возьмём tradein».

    Недостижимо через настройки (`Literal` валидируется pydantic), но молчаливый
    фолбэк здесь означал бы поход не в ту БД.
    """
    monkeypatch.setattr(config.settings, "identity_store", "elsewhere")
    with pytest.raises(ValueError, match="elsewhere"):
        identity_schema()


@pytest.mark.parametrize("mode", IDENTITY_MODES)
def test_table_names_never_come_from_outside(monkeypatch: pytest.MonkeyPatch, mode: str) -> None:
    """Имена таблиц — только из фиксированного словаря (защита от SQL-инъекции по имени).

    Имя таблицы нельзя передать bind-параметром, оно склеивается в строку запроса,
    поэтому единственный допустимый источник — `_SCHEMAS`. Тест пинит, что весь
    набор значений конечен и не содержит ничего, кроме идентификаторов.
    """
    use_identity_mode(monkeypatch, mode)
    schema = identity_schema()
    for name in (schema.users_table, schema.sessions_table, schema.access_state_column):
        assert name.replace("_", "").isalnum(), name
    assert schema.access_state_sql_type in ("boolean", "text")


# ---------------------------------------------------------------------------
# AccessState / to_access_state — ОДНО понятие состояния на обе схемы
# ---------------------------------------------------------------------------


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_boolean_column_maps_to_active_disabled() -> None:
    """Булев `tradein_users.is_active` — ровно два состояния, `trial_expired` там нет."""
    assert to_access_state(True) is AccessState.ACTIVE
    assert to_access_state(False) is AccessState.DISABLED


def test_text_column_maps_by_value() -> 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", ["", "ACTIVE", "pending_review", None, 1, 0, object()])
def test_unrecognized_state_is_fail_closed(value: object) -> None:
    """Неизвестное значение / NULL / неожиданный тип → `disabled`.

    Обратный выбор («пускать всё, что не disabled») означал бы, что состояние,
    добавленное миграцией РАНЬШЕ кода, молча раздаёт доступ. NB: `1`/`0` — это
    int, а не bool, и в булевой схеме они не появляются; сюда они попадают как
    «неожиданный тип» и тоже блокируются.
    """
    assert to_access_state(value) is AccessState.DISABLED


# ---------------------------------------------------------------------------
# access_state_param — обратное направление (ЗАПИСЬ)
# ---------------------------------------------------------------------------


def test_write_value_in_boolean_schema() -> None:
    """Литералы, а не round-trip: инверсия функции обязана быть видна прямо здесь."""
    assert access_state_param(AccessState.ACTIVE) is True
    assert access_state_param(AccessState.DISABLED) is False


def test_write_value_in_text_schema(monkeypatch: pytest.MonkeyPatch) -> None:
    use_identity_mode(monkeypatch, "auth")
    assert access_state_param(AccessState.ACTIVE) == "active"
    assert access_state_param(AccessState.TRIAL_EXPIRED) == "trial_expired"
    assert access_state_param(AccessState.DISABLED) == "disabled"


def test_trial_expired_is_not_silently_downgraded_in_boolean_schema() -> None:
    """`trial_expired` в булевой схеме — ошибка вызывающего, НЕ тихий `False`.

    Тихое приведение превратило бы «пробный период истёк» в жёсткую блокировку:
    клиент увидел бы «неверный логин или пароль» вместо экрана пробного периода.
    """
    with pytest.raises(ValueError, match="trial_expired"):
        access_state_param(AccessState.TRIAL_EXPIRED)


# ---------------------------------------------------------------------------
# Где физически берётся сессия реестра
# ---------------------------------------------------------------------------


def test_default_mode_uses_product_session_and_never_builds_auth_engine(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Дефолт: та же `SessionLocal`, что у всего приложения; БД `auth` не трогается.

    Это буквально «прод после мержа работает как сейчас»: `AUTH_DATABASE_URL` на
    проде пуст, и его отсутствие не должно ни ронять старт, ни всплывать в
    рантайме.
    """
    opened: list[_FakeSession] = []

    def _session_local() -> _FakeSession:
        s = _FakeSession()
        opened.append(s)
        return s

    monkeypatch.setattr(identity_store, "SessionLocal", _session_local)

    with identity_session() as db:
        assert db is opened[0]

    assert len(opened) == 1
    assert auth_db._engine is None
    assert auth_db._session_factory is None


def test_auth_mode_without_dsn_raises_instead_of_silent_fallback(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """`IDENTITY_STORE=auth` + пустой DSN → явная ошибка, и НИ ОДНОГО запроса в tradein.

    Тихий фолбэк на `tradein_users` был бы худшим исходом: вход бы «работал», но
    в реестре, который к тому моменту считается неактуальным.
    """
    use_identity_mode(monkeypatch, "auth")

    def _must_not_be_called() -> _FakeSession:
        raise AssertionError("режим auth не имеет права открывать сессию БД tradein")

    monkeypatch.setattr(identity_store, "SessionLocal", _must_not_be_called)

    with pytest.raises(auth_db.AuthDatabaseNotConfiguredError, match="AUTH_DATABASE_URL"):
        with identity_session():
            pass


def test_auth_engine_is_lazy_cached_and_resettable(monkeypatch: pytest.MonkeyPatch) -> None:
    """Engine строится при ПЕРВОМ обращении, кешируется, сбрасывается `reset_auth_db`.

    `create_engine` к серверу не ходит (пул ленивый), поэтому тест не требует
    живой БД — проверяется именно кеширование, из-за которого два одновременных
    первых запроса иначе создали бы два независимых пула.
    """
    use_identity_mode(monkeypatch, "auth")
    monkeypatch.setattr(config.settings, "auth_database_url", _FAKE_AUTH_DSN)

    assert auth_db._engine is None  # ленивость: до первого обращения ничего нет
    engine = auth_db.get_auth_engine()
    assert isinstance(engine, Engine)
    assert auth_db.get_auth_engine() is engine
    assert auth_db.get_auth_session_factory() is auth_db.get_auth_session_factory()

    auth_db.reset_auth_db()
    assert auth_db._engine is None
    assert auth_db.get_auth_engine() is not engine


def test_blank_dsn_is_not_configured(monkeypatch: pytest.MonkeyPatch) -> None:
    """DSN из одних пробелов = не задан (иначе `create_engine('')` дал бы мутную ошибку)."""
    use_identity_mode(monkeypatch, "auth")
    monkeypatch.setattr(config.settings, "auth_database_url", "   ")
    with pytest.raises(auth_db.AuthDatabaseNotConfiguredError):
        auth_db.get_auth_engine()


# ---------------------------------------------------------------------------
# get_identity_db — FastAPI-зависимость: ОДНА транзакция в дефолте, две в auth
# ---------------------------------------------------------------------------


def _probe_app() -> FastAPI:
    """Мини-приложение с обеими зависимостями сразу — как у роутов «Команды»."""
    app = FastAPI()

    @app.get("/probe")
    async def probe(
        db: Annotated[Any, Depends(get_db)],
        identity_db: Annotated[Any, Depends(get_identity_db)],
    ) -> dict[str, bool]:
        return {"same_session": db is identity_db}

    return app


def test_default_mode_shares_one_session_with_get_db(monkeypatch: pytest.MonkeyPatch) -> None:
    """`db is identity_db` в дефолте — не экономия коннекта, а требование прода.

    «Команда» пишет строку сотрудника (реестр) и его квоту (`account_quota_overrides`,
    продуктовая таблица) В ОДНОЙ транзакции. Две сессии = две транзакции =
    состояние «сотрудник создан, квота нет» на ровном месте.
    """
    app = _probe_app()
    app.dependency_overrides[get_db] = lambda: iter([_FakeSession()])

    resp = TestClient(app).get("/probe")

    assert resp.status_code == 200, resp.text
    assert resp.json() == {"same_session": True}


def test_auth_mode_yields_separate_registry_session(monkeypatch: pytest.MonkeyPatch) -> None:
    """В режиме `auth` БД физически разные → и сессии обязаны быть разными объектами.

    `db is not identity_db` — рантайм-признак «БД разные», по которому `team.py`
    решает, коммитить ли вторую транзакцию.
    """
    use_identity_mode(monkeypatch, "auth")
    registry_session = _FakeSession()

    from contextlib import contextmanager

    @contextmanager
    def _fake_auth_session():
        yield registry_session

    monkeypatch.setattr(auth_db, "auth_session", _fake_auth_session)

    app = _probe_app()
    app.dependency_overrides[get_db] = lambda: iter([_FakeSession()])

    resp = TestClient(app).get("/probe")

    assert resp.status_code == 200, resp.text
    assert resp.json() == {"same_session": False}


# ---------------------------------------------------------------------------
# Сломанная конфигурация не роняет запрос (rbac_guard)
# ---------------------------------------------------------------------------


def _guarded_app() -> FastAPI:
    app = FastAPI()
    app.middleware("http")(rbac_guard)

    @app.get("/api/v1/trade-in/dummy")
    async def dummy() -> dict[str, bool]:
        return {"ok": True}

    return app


def test_misconfigured_auth_store_degrades_to_401_not_500(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """`IDENTITY_STORE=auth` без DSN + запрос С КУКОЙ → 401, а не 500.

    `AuthDatabaseNotConfiguredError` обрабатывается тем же путём, что и любой
    сбой БД: резолв сессии не состоялся, дальше решает `auth_mode`. Сознательно
    не отличается от «БД недоступна» — обе ситуации это сломанная конфигурация
    реестра, и ни одна не имеет права отдавать 500 (или, тем более, пускать).
    """
    use_identity_mode(monkeypatch, "auth")
    client = TestClient(_guarded_app(), base_url="https://testserver")
    client.cookies.set(config.settings.session_cookie_name, "some-token")

    resp = client.get("/api/v1/trade-in/dummy")

    assert resp.status_code == 401
    # Legacy trusted-header путь (dual-mode) при этом продолжает работать —
    # сломанный реестр не отрезает существующих пользователей Caddy.
    #
    # ⚠️ Это поведение УЖЕ НЕДОСТИЖИМО в реальном процессе: до такого состояния
    # приложение не доживает, потому что lifespan падает на старте (см.
    # `test_lifespan_fails_fast_when_auth_store_has_no_dsn` ниже). Тест держит
    # guard'а от 500-ки/анонимного доступа как второй рубеж — на случай, если
    # DSN сломается уже ПОСЛЕ успешного старта.
    fallback = client.get("/api/v1/trade-in/dummy", headers={"X-Authenticated-User": "kopylov"})
    assert fallback.status_code == 200, fallback.text


# ---------------------------------------------------------------------------
# Boot-time guard: сломанный реестр не должен ЖИТЬ на legacy-пути
# ---------------------------------------------------------------------------


def _run_lifespan(monkeypatch: pytest.MonkeyPatch) -> None:
    """Прогоняет lifespan приложения до `yield` и обратно.

    FDW-bootstrap выключен: он ходит в продуктовую БД, которой в юнит-тестах
    нет. К проверяемому здесь он отношения не имеет (и в самом lifespan обёрнут
    в try/except), а без заглушки тест ждал бы таймаута коннекта.
    """
    import asyncio

    from app import main as app_main

    monkeypatch.setattr(app_main, "ensure_fdw_user_mapping", lambda db: None)

    async def _cycle() -> None:
        async with app_main.lifespan(app_main.app):
            pass

    asyncio.run(_cycle())


def test_lifespan_fails_fast_when_auth_store_has_no_dsn(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """`IDENTITY_STORE=auth` + пустой DSN → контейнер НЕ поднимается.

    Почему не «работает как-нибудь»: `rbac_guard` ловит
    `AuthDatabaseNotConfiguredError` вместе с любым другим сбоем резолва сессии
    и уходит в legacy trusted-header ветку. Продуктовая БД при этом жива, и
    такой деплой способен работать сутками, раздавая права из roles.yaml всем,
    кого пропустил Caddy basic_auth, — включая аккаунты, у которых в реестре
    `access_state='disabled'`/`'trial_expired'`. Ошибка КОНФИГУРАЦИИ обязана
    убивать старт, а не деградировать в тихий обход реестра.
    """
    use_identity_mode(monkeypatch, "auth")
    monkeypatch.setattr(config.settings, "auth_database_url", "")

    with pytest.raises(auth_db.AuthDatabaseNotConfiguredError):
        _run_lifespan(monkeypatch)


def test_lifespan_does_not_touch_auth_db_in_default_mode(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Дефолтный режим: старт НЕ обращается к БД `auth` и пустой DSN не мешает.

    Ровно ограничение PR — сегодняшний прод (`IDENTITY_STORE` не задан,
    `AUTH_DATABASE_URL` нет вовсе) обязан подниматься как раньше.
    """
    from app import main as app_main

    calls: list[str] = []
    monkeypatch.setattr(app_main, "get_auth_engine", lambda: calls.append("built"))

    _run_lifespan(monkeypatch)

    assert calls == [], "в дефолтном режиме engine БД `auth` не должен строиться на старте"