"""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` не должен строиться на старте"