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/inactive/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.
  - 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.
"""

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")

from app.services import auth_session as svc

# ---------------------------------------------------------------------------
# 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


def _session_row(
    *,
    user_id: int = 1,
    expires_at: datetime | None = None,
    last_seen_at: datetime | None = None,
    username: str = "alice",
    role: str = "employee",
    is_active: bool = True,
) -> SimpleNamespace:
    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",
        is_active=is_active,
    )


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


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


def test_create_session_inserts_and_commits() -> 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 "INSERT INTO tradein_sessions" 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() -> 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() -> None:
    db = _FakeDB(rows=[None])
    assert svc.get_session_user(db, "tok") is None
    sql, params = db.executed[0]
    assert "FROM tradein_sessions s" in sql
    assert "JOIN tradein_users u" in sql
    assert params == {"token": "tok"}


def test_get_session_user_expired_returns_none() -> 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_inactive_returns_none() -> None:
    db = _FakeDB(rows=[_session_row(is_active=False)])
    assert svc.get_session_user(db, "tok") is None
    assert len(db.executed) == 1


def test_get_session_user_valid_recent_no_refresh() -> 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
    # Только 1 execute (SELECT) — никакого UPDATE.
    assert len(db.executed) == 1
    assert db.committed == 0


def test_get_session_user_stale_last_seen_triggers_refresh() -> 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 "UPDATE tradein_sessions" 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() -> 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() -> None:
    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["is_active"] is True
    sql, params = db.executed[0]
    assert "FROM tradein_users" in sql
    assert params == {"username": "alice"}


def test_get_user_by_username_not_found() -> 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() -> None:
    db = _FakeDB()
    svc.revoke_session(db, "tok")

    assert db.committed == 1
    sql, params = db.executed[0]
    assert "DELETE FROM tradein_sessions" in sql
    assert "token" in sql
    assert params == {"token": "tok"}


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

    assert db.committed == 1
    sql, params = db.executed[0]
    assert "DELETE FROM tradein_sessions" 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 == ["/**"]