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

"""Tests for app.services.auth_session — session CRUD + DB-role scope mapping (#2552).

Coverage:
  - create_session: INSERT with CAST(...) (never `:x::type`), commit, unique tokens.
  - get_session_user: valid/expired/не-active/missing-row + sliding refresh (only when
    last_seen_at is stale, best-effort — a refresh failure still returns the user).
  - get_user_by_username: found/not-found + состояние доступа как `AccessState`.
  - revoke_session / revoke_user_sessions: DELETE + commit.
  - get_db_role_scope: employee/manager/admin/unknown mapping.

All functions here take `db: Session` as a plain argument (no SessionLocal() opened
internally) — unit tests just pass a hand-rolled fake, mirroring the `_FakeSession`
pattern from tests/test_user_events.py but adapted for `.fetchone()`-based reads.

⚠️ ОБА РЕЖИМА РЕЕСТРА. Эпик «единый вход» вынес имена таблиц и имя/тип колонки
состояния доступа в `identity_store.identity_schema()`. Тесты, которые вообще
трогают SQL, прогоняются в ОБОИХ режимах (фикстура `identity_mode`): "tradein"
(дефолт, сегодняшний прод — `tradein_users`/`tradein_sessions`, boolean
`is_active`) и "auth" (`users`/`sessions`, text `access_state`). Ожидаемые имена
в ассертах берутся из `identity_schema()` — из того же словаря, что и у кода,
поэтому переименование таблиц не «разъезжает» тест с реальностью тихо;
поломка запроса ловится тем, что fake отдаёт строку ТОЛЬКО на ожидаемый SQL,
а сам SQL проверяется явными ассертами ниже.

Тесты БЕЗ фикстуры `identity_mode` намеренно идут в дефолтном режиме
(`_default_identity_mode` autouse) — это чистая логика без SQL.
"""

from __future__ import annotations

import os
import re
from datetime import UTC, datetime, timedelta
from types import SimpleNamespace
from typing import Any

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

import pytest

from app.services import auth_session as svc
from app.services.identity_store import AccessState, identity_schema
from tests.support.identity_modes import IDENTITY_MODES, column_value, use_identity_mode

# ---------------------------------------------------------------------------
# Режим реестра
# ---------------------------------------------------------------------------


@pytest.fixture(autouse=True)
def _default_identity_mode(monkeypatch: pytest.MonkeyPatch) -> None:
    """Каждый тест стартует в ДЕФОЛТНОМ режиме, даже если предыдущий его менял."""
    use_identity_mode(monkeypatch, "tradein")


@pytest.fixture(params=IDENTITY_MODES)
def identity_mode(request: pytest.FixtureRequest, monkeypatch: pytest.MonkeyPatch) -> str:
    """Тест прогоняется дважды: "tradein" (прод) и "auth" (после переезда)."""
    return use_identity_mode(monkeypatch, request.param)


@pytest.fixture
def auth_mode(monkeypatch: pytest.MonkeyPatch) -> str:
    """Только режим "auth" — для состояний, невыразимых булевой колонкой."""
    return use_identity_mode(monkeypatch, "auth")


# ---------------------------------------------------------------------------
# Fake DB session
# ---------------------------------------------------------------------------


class _FakeDB:
    """Minimal `Session` stand-in: queued `.fetchone()` results per `execute()` call,
    in call order. `execute()` beyond the queue returns a result with `fetchone()
    -> None`. Set `.raise_on_call = N` to make the Nth `execute()` (1-indexed) raise.
    """

    def __init__(self, rows: list[Any] | None = None) -> None:
        self._rows: list[Any] = list(rows or [])
        self.executed: list[tuple[str, dict[str, Any] | None]] = []
        self.committed = 0
        self.rolled_back = 0
        self.raise_on_call: int | None = None

    def execute(self, stmt: object, params: dict[str, Any] | None = None) -> SimpleNamespace:
        call_no = len(self.executed) + 1
        self.executed.append((str(stmt), params))
        if self.raise_on_call == call_no:
            raise RuntimeError("simulated DB failure")
        row = self._rows.pop(0) if self._rows else None
        return SimpleNamespace(fetchone=lambda: row)

    def commit(self) -> None:
        self.committed += 1

    def rollback(self) -> None:
        self.rolled_back += 1


# Часовой «аргумент не передан» — None здесь занят (это валидное сырое значение
# колонки: NULL, который to_access_state обязан трактовать как disabled).
_MISSING = object()


def _session_row(
    *,
    user_id: int = 1,
    expires_at: datetime | None = None,
    last_seen_at: datetime | None = None,
    username: str = "alice",
    role: str = "employee",
    access_state: AccessState = AccessState.ACTIVE,
    raw_access_state: object = _MISSING,
) -> SimpleNamespace:
    """Строка JOIN'а sessions×users, как её отдал бы драйвер.

    Колонка состояния всегда приезжает под алиасом `access_state` (`AS access_state`
    в реальном SELECT'е), а ЗНАЧЕНИЕ в ней — то, что лежит в БД текущего режима:
    boolean для `tradein_users.is_active`, text для `auth.users.access_state`.
    *raw_access_state* — обход таблицы состояний для проверки fail-closed на
    значении, которого код не знает.
    """
    now = datetime.now(UTC)
    return SimpleNamespace(
        user_id=user_id,
        expires_at=expires_at if expires_at is not None else now + timedelta(hours=1),
        last_seen_at=last_seen_at if last_seen_at is not None else now,
        username=username,
        role=role,
        display_name="Alice A.",
        org_name="Org LLC",
        email="alice@example.com",
        access_state=(
            column_value(access_state) if raw_access_state is _MISSING else raw_access_state
        ),
    )


def _user_row(
    *,
    user_id: int = 1,
    username: str = "alice",
    password_hash: str | None = "hash",
    role: str = "employee",
    access_state: AccessState = AccessState.ACTIVE,
) -> SimpleNamespace:
    return SimpleNamespace(
        id=user_id,
        username=username,
        password_hash=password_hash,
        role=role,
        access_state=column_value(access_state),
        display_name="Alice A.",
        org_name="Org LLC",
        email="alice@example.com",
    )


# ---------------------------------------------------------------------------
# create_session
# ---------------------------------------------------------------------------


def test_create_session_inserts_and_commits(identity_mode: str) -> None:
    db = _FakeDB()
    token = svc.create_session(db, user_id=42, ip="1.2.3.4", user_agent="pytest")

    assert db.committed == 1
    assert len(db.executed) == 1
    sql, params = db.executed[0]
    assert f"INSERT INTO {identity_schema().sessions_table}" in sql
    assert params is not None
    assert params["user_id"] == 42
    assert params["ip"] == "1.2.3.4"
    assert params["user_agent"] == "pytest"
    assert params["token"] == token
    assert isinstance(token, str)
    assert len(token) >= 32


def test_create_session_cast_not_doublecolon(identity_mode: str) -> None:
    db = _FakeDB()
    svc.create_session(db, user_id=1)
    sql, _ = db.executed[0]
    assert not re.search(r":\w+::\w", sql)
    assert "CAST(:ttl_hours AS integer)" in sql
    assert "CAST(:ip AS inet)" in sql


def test_create_session_tokens_are_unique() -> None:
    db = _FakeDB()
    t1 = svc.create_session(db, user_id=1)
    t2 = svc.create_session(db, user_id=1)
    assert t1 != t2


# ---------------------------------------------------------------------------
# get_session_user
# ---------------------------------------------------------------------------


def test_get_session_user_no_token_returns_none() -> None:
    db = _FakeDB()
    assert svc.get_session_user(db, "") is None
    assert db.executed == []


def test_get_session_user_missing_row_returns_none(identity_mode: str) -> None:
    schema = identity_schema()
    db = _FakeDB(rows=[None])
    assert svc.get_session_user(db, "tok") is None
    sql, params = db.executed[0]
    assert f"FROM {schema.sessions_table} s" in sql
    assert f"JOIN {schema.users_table} u" in sql
    # Колонка состояния — под именем текущей схемы и обязательно с алиасом:
    # без него вызывающий код читал бы то `is_active`, то `access_state`.
    assert f"u.{schema.access_state_column} AS access_state" in sql
    assert params == {"token": "tok"}


def test_get_session_user_expired_returns_none(identity_mode: str) -> None:
    now = datetime.now(UTC)
    db = _FakeDB(rows=[_session_row(expires_at=now - timedelta(minutes=1))])
    assert svc.get_session_user(db, "tok") is None
    # Никакого sliding-refresh UPDATE для невалидной сессии.
    assert len(db.executed) == 1


def test_get_session_user_disabled_returns_none(identity_mode: str) -> None:
    """Жёстко заблокированный аккаунт — сессия недействительна в обеих схемах."""
    db = _FakeDB(rows=[_session_row(access_state=AccessState.DISABLED)])
    assert svc.get_session_user(db, "tok") is None
    assert len(db.executed) == 1


def test_get_session_user_trial_expired_returns_none(auth_mode: str) -> None:
    """Пробный период истёк — УЖЕ ВЫДАННАЯ сессия гасится немедленно.

    Иначе сотрудник, залогиненный до истечения пробного доступа, продолжал бы
    работать, а sliding-refresh продлевал бы ему `expires_at` бесконечно —
    состояние `trial_expired` не наступило бы для него никогда.
    """
    db = _FakeDB(rows=[_session_row(access_state=AccessState.TRIAL_EXPIRED)])
    assert svc.get_session_user(db, "tok") is None
    # Ни UPDATE (sliding refresh), ни commit — сессия не продлевается.
    assert len(db.executed) == 1
    assert db.committed == 0


def test_get_session_user_unknown_state_returns_none(auth_mode: str) -> None:
    """Fail-closed: состояние, которого код не знает (миграция впереди кода),
    НЕ пускает. Обратный выбор молча раздавал бы доступ по новому значению."""
    db = _FakeDB(rows=[_session_row(raw_access_state="pending_review")])
    assert svc.get_session_user(db, "tok") is None
    assert len(db.executed) == 1


def test_get_session_user_valid_recent_no_refresh(identity_mode: str) -> None:
    """last_seen_at свежий (<5 мин) — sliding refresh НЕ триггерится."""
    now = datetime.now(UTC)
    db = _FakeDB(rows=[_session_row(last_seen_at=now - timedelta(minutes=1))])
    result = svc.get_session_user(db, "tok")

    assert result is not None
    assert result["username"] == "alice"
    assert result["role"] == "employee"
    assert result["display_name"] == "Alice A."
    assert result["org_name"] == "Org LLC"
    assert result["email"] == "alice@example.com"
    assert result["user_id"] == 1
    # Состояние доступа приезжает ЕДИНЫМ понятием, а не boolean/str по режимам;
    # сюда доходит только ACTIVE (не-active отсеян выше).
    assert result["access_state"] is AccessState.ACTIVE
    # Только 1 execute (SELECT) — никакого UPDATE.
    assert len(db.executed) == 1
    assert db.committed == 0


def test_get_session_user_stale_last_seen_triggers_refresh(identity_mode: str) -> None:
    """last_seen_at старше 5 минут — один UPDATE (sliding refresh) + commit."""
    now = datetime.now(UTC)
    db = _FakeDB(rows=[_session_row(last_seen_at=now - timedelta(minutes=10))])
    result = svc.get_session_user(db, "tok")

    assert result is not None
    assert len(db.executed) == 2
    update_sql, update_params = db.executed[1]
    assert f"UPDATE {identity_schema().sessions_table}" in update_sql
    assert "SET last_seen_at" in update_sql
    assert not re.search(r":\w+::\w", update_sql)
    assert "CAST(:ttl_hours AS integer)" in update_sql
    assert update_params == {"ttl_hours": 720, "token": "tok"}
    assert db.committed == 1


def test_get_session_user_refresh_failure_is_swallowed(identity_mode: str) -> None:
    """Sliding-refresh UPDATE падает — всё равно возвращаем валидного юзера
    (best-effort refresh, не часть решения "валидна ли сессия")."""
    now = datetime.now(UTC)
    db = _FakeDB(rows=[_session_row(last_seen_at=now - timedelta(minutes=10))])
    db.raise_on_call = 2

    result = svc.get_session_user(db, "tok")

    assert result is not None
    assert result["username"] == "alice"
    assert db.rolled_back == 1
    assert db.committed == 0


# ---------------------------------------------------------------------------
# get_user_by_username
# ---------------------------------------------------------------------------


def test_get_user_by_username_found(identity_mode: str) -> None:
    schema = identity_schema()
    db = _FakeDB(rows=[_user_row()])
    user = svc.get_user_by_username(db, "alice")

    assert user is not None
    assert user["username"] == "alice"
    assert user["password_hash"] == "hash"
    assert user["role"] == "employee"
    assert user["access_state"] is AccessState.ACTIVE
    sql, params = db.executed[0]
    assert f"FROM {schema.users_table}" in sql
    assert f"{schema.access_state_column} AS access_state" in sql
    assert params == {"username": "alice"}


def test_get_user_by_username_disabled_state_is_reported_not_hidden(identity_mode: str) -> None:
    """Строка отдаётся ВСЕГДА, состояние — отдельным полем.

    Login обязан отличать «нет такого логина» (None) от «есть, но доступ закрыт»
    (строка + не-ACTIVE): от этого зависит выбор события аудита, а прятать
    заблокированного за None означало бы потерять эту разницу.
    """
    db = _FakeDB(rows=[_user_row(access_state=AccessState.DISABLED)])
    user = svc.get_user_by_username(db, "alice")

    assert user is not None
    assert user["access_state"] is AccessState.DISABLED
    assert user["access_state"].can_sign_in is False


def test_get_user_by_username_trial_expired_state(auth_mode: str) -> None:
    db = _FakeDB(rows=[_user_row(access_state=AccessState.TRIAL_EXPIRED)])
    user = svc.get_user_by_username(db, "alice")

    assert user is not None
    assert user["access_state"] is AccessState.TRIAL_EXPIRED
    assert user["access_state"].can_sign_in is False


def test_get_user_by_username_not_found(identity_mode: str) -> None:
    db = _FakeDB(rows=[None])
    assert svc.get_user_by_username(db, "ghost") is None


# ---------------------------------------------------------------------------
# revoke_session / revoke_user_sessions
# ---------------------------------------------------------------------------


def test_revoke_session_deletes_and_commits(identity_mode: str) -> None:
    db = _FakeDB()
    svc.revoke_session(db, "tok")

    assert db.committed == 1
    sql, params = db.executed[0]
    assert f"DELETE FROM {identity_schema().sessions_table}" in sql
    assert "token" in sql
    assert params == {"token": "tok"}


def test_revoke_user_sessions_deletes_and_commits(identity_mode: str) -> None:
    db = _FakeDB()
    svc.revoke_user_sessions(db, 7)

    assert db.committed == 1
    sql, params = db.executed[0]
    assert f"DELETE FROM {identity_schema().sessions_table}" in sql
    assert "user_id" in sql
    assert params == {"user_id": 7}


# ---------------------------------------------------------------------------
# get_db_role_scope
# ---------------------------------------------------------------------------


def test_get_db_role_scope_employee_matches_legacy_pilot() -> None:
    paths, deny = svc.get_db_role_scope("employee")
    assert "/trade-in/**" in paths
    assert "/trade-in/api/v1/**" in paths
    assert "/admin/**" in deny
    assert "/api/v1/admin/**" in deny


def test_get_db_role_scope_manager_adds_team_path() -> None:
    paths, deny = svc.get_db_role_scope("manager")
    assert "/trade-in/**" in paths
    assert "/api/v1/team/**" in paths
    assert "/admin/**" in deny


# «Доля в продаже» и «Кэш» — внутренние инструменты (аналитика рынка / состояние
# кэшей и скраперов), клиентские роли их не видят (решение владельца 2026-07-31).
# В deny И страницы, И их API: один список гейтит пункт меню (Topbar через /me),
# страницу (RouteGuard) и серверные ручки (rbac_guard).
_INTERNAL_TOOL_DENY = (
    "/trade-in/sale-share/**",
    "/trade-in/cache/**",
    "/trade-in/api/v1/buildings/**",
    # Глоб, а не точный путь: точный обходится трейлинг-слэшем (см. NB в
    # app.services.auth_session над DB_ROLE_PATHS).
    "/trade-in/api/v1/trade-in/cache-stats/**",
)


def test_get_db_role_scope_client_roles_deny_internal_tools() -> None:
    for role in ("manager", "employee"):
        _, deny = svc.get_db_role_scope(role)
        for pattern in _INTERNAL_TOOL_DENY:
            assert pattern in deny, f"{role} deny missing {pattern}"


def test_get_db_role_scope_admin_full_access() -> None:
    paths, deny = svc.get_db_role_scope("admin")
    assert paths == ["/**"]
    assert deny == []


def test_get_db_role_scope_unknown_role_denies_all() -> None:
    paths, deny = svc.get_db_role_scope("ghost")
    assert paths == []
    assert deny == ["/**"]