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

"""Integration tests for #2552 auth-core: POST /login, /logout, dual-mode GET /me,
and rbac_guard session-cookie resolution.

Uses the REAL `rbac_guard` (app.core.rbac) + REAL `auth.router` / `me.router` wired
into an isolated FastAPI test app (same pattern as tests/test_rbac.py), with an
in-memory fake DB standing in for `tradein_users`/`tradein_sessions`:
  - `app.core.rbac.SessionLocal` is monkeypatched (rbac_guard opens its own session,
    it's middleware — no FastAPI DI available there).
  - `app.core.db.get_db` is overridden via `app.dependency_overrides` (auth.py /
    me.py use `Depends(get_db)`, the idiomatic FastAPI-testable path).

Both point at the SAME `_Store` instance per test, so a session created by POST
/login is immediately visible to rbac_guard's own DB round trip on the next request.
"""

from __future__ import annotations

import os
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 fastapi import FastAPI
from fastapi.testclient import TestClient

from app.api.v1 import auth as auth_router
from app.api.v1 import me as me_router
from app.core import auth as auth_mod
from app.core import config
from app.core.db import get_db
from app.core.password import hash_password
from app.core.rbac import rbac_guard

# ---------------------------------------------------------------------------
# Fake DB backing tradein_users / tradein_sessions
# ---------------------------------------------------------------------------


class _Store:
    def __init__(self) -> None:
        self.users: dict[str, dict[str, Any]] = {}
        self.sessions: dict[str, dict[str, Any]] = {}
        self._next_id = 1

    def add_user(
        self,
        username: str,
        password_hash: str | None,
        *,
        role: str = "employee",
        is_active: bool = True,
        display_name: str | None = "Alice A.",
        org_name: str | None = "Org LLC",
        email: str | None = "alice@example.com",
    ) -> int:
        uid = self._next_id
        self._next_id += 1
        self.users[username] = {
            "id": uid,
            "username": username,
            "password_hash": password_hash,
            "role": role,
            "is_active": is_active,
            "display_name": display_name,
            "org_name": org_name,
            "email": email,
        }
        return uid

    def user_by_id(self, uid: int) -> dict[str, Any] | None:
        for u in self.users.values():
            if u["id"] == uid:
                return u
        return None

    def add_expired_session(self, token: str, user_id: int) -> None:
        now = datetime.now(UTC)
        self.sessions[token] = {
            "user_id": user_id,
            "expires_at": now - timedelta(minutes=1),
            "last_seen_at": now - timedelta(minutes=1),
        }


class _FakeDB:
    """Minimal Session stand-in dispatching on SQL text — see module docstring."""

    def __init__(self, store: _Store) -> None:
        self.store = store

    def __enter__(self) -> _FakeDB:
        return self

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

    def close(self) -> None:
        pass

    def commit(self) -> None:
        pass

    def rollback(self) -> None:
        pass

    def execute(self, stmt: object, params: dict[str, Any] | None = None) -> SimpleNamespace:
        sql = str(stmt)
        p = params or {}

        if "INSERT INTO tradein_sessions" in sql:
            now = datetime.now(UTC)
            self.store.sessions[p["token"]] = {
                "user_id": p["user_id"],
                "expires_at": now + timedelta(hours=p["ttl_hours"]),
                "last_seen_at": now,
            }
            return SimpleNamespace(fetchone=lambda: None)

        if "UPDATE tradein_sessions" in sql and "SET last_seen_at" in sql:
            sess = self.store.sessions.get(p["token"])
            if sess is not None:
                now = datetime.now(UTC)
                sess["last_seen_at"] = now
                sess["expires_at"] = now + timedelta(hours=p["ttl_hours"])
            return SimpleNamespace(fetchone=lambda: None)

        if "DELETE FROM tradein_sessions WHERE token" in sql:
            self.store.sessions.pop(p["token"], None)
            return SimpleNamespace(fetchone=lambda: None)

        if "DELETE FROM tradein_sessions WHERE user_id" in sql:
            uid = p["user_id"]
            for tok in [t for t, s in self.store.sessions.items() if s["user_id"] == uid]:
                del self.store.sessions[tok]
            return SimpleNamespace(fetchone=lambda: None)

        if "FROM tradein_sessions s" in sql and "JOIN tradein_users u" in sql:
            sess = self.store.sessions.get(p["token"])
            if sess is None:
                return SimpleNamespace(fetchone=lambda: None)
            user = self.store.user_by_id(sess["user_id"])
            if user is None:
                return SimpleNamespace(fetchone=lambda: None)
            row = SimpleNamespace(
                user_id=sess["user_id"],
                expires_at=sess["expires_at"],
                last_seen_at=sess["last_seen_at"],
                username=user["username"],
                role=user["role"],
                display_name=user["display_name"],
                org_name=user["org_name"],
                email=user["email"],
                is_active=user["is_active"],
            )
            return SimpleNamespace(fetchone=lambda: row)

        if "FROM tradein_users" in sql:
            user = self.store.users.get(p["username"])
            if user is None:
                return SimpleNamespace(fetchone=lambda: None)
            row = SimpleNamespace(**user)
            return SimpleNamespace(fetchone=lambda: row)

        raise AssertionError(f"unhandled fake SQL in test_auth_api: {sql!r}")


# ---------------------------------------------------------------------------
# Test app
# ---------------------------------------------------------------------------


def _build_test_app(store: _Store) -> FastAPI:
    app = FastAPI()
    app.middleware("http")(rbac_guard)
    app.include_router(auth_router.router, prefix="/api/v1/auth", tags=["auth"])
    app.include_router(me_router.router, prefix="/api/v1", tags=["me"])

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

    def _override_get_db():  # generator dependency — matches app.core.db.get_db shape
        yield _FakeDB(store)

    app.dependency_overrides[get_db] = _override_get_db
    return app


@pytest.fixture(autouse=True)
def _reset_state(monkeypatch: pytest.MonkeyPatch) -> None:
    auth_mod.reset_cache_for_tests()
    auth_router._LOGIN_LIMITER._hits.clear()
    monkeypatch.setattr(config.settings, "auth_mode", "dual")


@pytest.fixture
def store() -> _Store:
    return _Store()


@pytest.fixture
def client(store: _Store, monkeypatch: pytest.MonkeyPatch) -> TestClient:
    monkeypatch.setattr("app.core.rbac.SessionLocal", lambda: _FakeDB(store))
    # base_url=https:// — login sets the session cookie with Secure=True (real prod
    # behaviour, not weakened for tests); httpx's cookie jar silently drops Secure
    # cookies on a plain-http connection, so a plain http://testserver client would
    # never resend the cookie on subsequent requests within the same test.
    return TestClient(_build_test_app(store), base_url="https://testserver")


# ---------------------------------------------------------------------------
# POST /login
# ---------------------------------------------------------------------------


def test_login_happy_path_sets_cookie(client: TestClient, store: _Store) -> None:
    store.add_user("alice", hash_password("Secret123!"), role="employee")

    resp = client.post("/api/v1/auth/login", json={"username": "alice", "password": "Secret123!"})

    assert resp.status_code == 200, resp.text
    assert resp.json() == {"ok": True}
    cookie_name = config.settings.session_cookie_name
    assert cookie_name in resp.cookies
    assert resp.cookies[cookie_name]
    # Сессия реально создана в сторе под этим токеном.
    assert resp.cookies[cookie_name] in store.sessions


def test_login_wrong_password_401_and_logs_failed_event(
    client: TestClient, store: _Store, monkeypatch: pytest.MonkeyPatch
) -> None:
    store.add_user("alice", hash_password("Secret123!"), role="employee")
    events: list[dict[str, Any]] = []
    monkeypatch.setattr(auth_router, "schedule_event", lambda **kw: events.append(kw))

    resp = client.post("/api/v1/auth/login", json={"username": "alice", "password": "wrong"})

    assert resp.status_code == 401
    assert "detail" in resp.json()
    assert config.settings.session_cookie_name not in resp.cookies
    assert any(e["event_type"] == "login_failed" for e in events)
    failed = next(e for e in events if e["event_type"] == "login_failed")
    assert failed["username"] == "alice"
    # Raw-пароль никогда не попадает в событие.
    assert "wrong" not in str(failed)


def test_login_unknown_username_401_generic_message(client: TestClient) -> None:
    resp = client.post("/api/v1/auth/login", json={"username": "ghost", "password": "whatever"})
    assert resp.status_code == 401
    # НЕ раскрываем, что юзера не существует — то же сообщение, что и wrong-password.
    body_ghost = resp.json()["detail"]

    resp2 = client.post("/api/v1/auth/login", json={"username": "ghost2", "password": "x"})
    assert resp2.json()["detail"] == body_ghost


def test_login_inactive_user_401(client: TestClient, store: _Store) -> None:
    store.add_user("bob", hash_password("Secret123!"), role="employee", is_active=False)
    resp = client.post("/api/v1/auth/login", json={"username": "bob", "password": "Secret123!"})
    assert resp.status_code == 401


def test_login_null_password_hash_401(client: TestClient, store: _Store) -> None:
    store.add_user("carol", None, role="employee")
    resp = client.post("/api/v1/auth/login", json={"username": "carol", "password": "anything"})
    assert resp.status_code == 401


def test_login_rate_limit_429(client: TestClient, store: _Store) -> None:
    store.add_user("dave", hash_password("Secret123!"), role="employee")
    limit = config.settings.login_rate_limit

    for _ in range(limit):
        resp = client.post("/api/v1/auth/login", json={"username": "dave", "password": "wrong"})
        assert resp.status_code == 401

    resp = client.post("/api/v1/auth/login", json={"username": "dave", "password": "wrong"})
    assert resp.status_code == 429
    assert "Retry-After" in resp.headers


# ---------------------------------------------------------------------------
# POST /logout
# ---------------------------------------------------------------------------


def test_logout_revokes_session_and_clears_cookie(client: TestClient, store: _Store) -> None:
    store.add_user("alice", hash_password("Secret123!"), role="employee")
    login_resp = client.post(
        "/api/v1/auth/login", json={"username": "alice", "password": "Secret123!"}
    )
    token = login_resp.cookies[config.settings.session_cookie_name]
    assert token in store.sessions

    logout_resp = client.post("/api/v1/auth/logout")
    assert logout_resp.status_code == 200
    assert logout_resp.json() == {"ok": True}
    assert token not in store.sessions


# ---------------------------------------------------------------------------
# GET /me — session-first + dual-mode legacy fallback
# ---------------------------------------------------------------------------


def test_me_with_session_cookie_returns_db_role(client: TestClient, store: _Store) -> None:
    store.add_user(
        "alice",
        hash_password("Secret123!"),
        role="employee",
        display_name="Алиса",
        org_name="ООО Ромашка",
        email="alice@romashka.ru",
    )
    client.post("/api/v1/auth/login", json={"username": "alice", "password": "Secret123!"})

    resp = client.get("/api/v1/me")
    assert resp.status_code == 200, resp.text
    body = resp.json()
    assert body["username"] == "alice"
    assert body["role"] == "employee"
    assert "/trade-in/**" in body["allowed_paths"]
    assert "/admin/**" in body["deny_paths"]
    assert body["display_name"] == "Алиса"
    assert body["org"] == "ООО Ромашка"
    assert body["email"] == "alice@romashka.ru"


def test_me_manager_role_gets_team_path(client: TestClient, store: _Store) -> None:
    store.add_user("mgr", hash_password("Secret123!"), role="manager")
    client.post("/api/v1/auth/login", json={"username": "mgr", "password": "Secret123!"})

    resp = client.get("/api/v1/me")
    assert resp.status_code == 200
    assert "/api/v1/team/**" in resp.json()["allowed_paths"]


def test_me_without_cookie_dual_mode_legacy_still_works(client: TestClient) -> None:
    """Regression guard: без сессии, auth_mode=dual — legacy X-Authenticated-User
    путь через roles.yaml работает БЕЗ ИЗМЕНЕНИЙ."""
    resp = client.get("/api/v1/me", headers={"X-Authenticated-User": "admin"})
    assert resp.status_code == 200, resp.text
    body = resp.json()
    assert body["username"] == "admin"
    assert body["role"] == "admin"
    assert body["allowed_paths"] == ["/**"]


def test_me_no_cookie_no_header_401(client: TestClient) -> None:
    resp = client.get("/api/v1/me")
    assert resp.status_code == 401


def test_expired_session_falls_back_to_no_auth(client: TestClient, store: _Store) -> None:
    """Истёкшая сессия трактуется как отсутствие cookie: без legacy-заголовка — 401."""
    uid = store.add_user("alice", hash_password("Secret123!"), role="employee")
    store.add_expired_session("expired-tok", uid)

    client.cookies.set(config.settings.session_cookie_name, "expired-tok")
    resp = client.get("/api/v1/trade-in/dummy")
    assert resp.status_code == 401


def test_expired_session_falls_back_to_legacy_header_in_dual_mode(
    client: TestClient, store: _Store
) -> None:
    """Истёкшая сессия + валидный legacy header в dual-mode — header отрабатывает."""
    uid = store.add_user("alice", hash_password("Secret123!"), role="employee")
    store.add_expired_session("expired-tok", uid)

    client.cookies.set(config.settings.session_cookie_name, "expired-tok")
    resp = client.get(
        "/api/v1/trade-in/dummy",
        headers={"X-Authenticated-User": "kopylov"},
    )
    assert resp.status_code == 200
    assert resp.json() == {"ok": True}


# ---------------------------------------------------------------------------
# rbac_guard dual vs db_only
# ---------------------------------------------------------------------------


def test_db_only_mode_rejects_legacy_header_without_session(
    client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setattr(config.settings, "auth_mode", "db_only")

    resp = client.get(
        "/api/v1/trade-in/dummy",
        headers={"X-Authenticated-User": "admin"},
    )
    assert resp.status_code == 401
    assert "session" in resp.json()["detail"].lower()


def test_db_only_mode_accepts_valid_session(
    client: TestClient, store: _Store, monkeypatch: pytest.MonkeyPatch
) -> None:
    store.add_user("alice", hash_password("Secret123!"), role="employee")
    client.post("/api/v1/auth/login", json={"username": "alice", "password": "Secret123!"})

    monkeypatch.setattr(config.settings, "auth_mode", "db_only")

    resp = client.get("/api/v1/trade-in/dummy")
    assert resp.status_code == 200
    assert resp.json() == {"ok": True}


def test_session_user_can_reach_tradein_but_not_admin(client: TestClient, store: _Store) -> None:
    store.add_user("alice", hash_password("Secret123!"), role="employee")
    client.post("/api/v1/auth/login", json={"username": "alice", "password": "Secret123!"})

    ok = client.get("/api/v1/trade-in/dummy")
    assert ok.status_code == 200

    denied = client.get("/api/v1/admin/dummy")
    assert denied.status_code in (401, 403, 404)