X-Financial/server/tests/test_auth_session_endpoints.py

from __future__ import annotations

from collections.abc import Generator
from datetime import UTC, datetime, timedelta

import pytest
from fastapi import HTTPException
from fastapi.testclient import TestClient
from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.pool import StaticPool

from app.api.deps import CurrentUserContext, get_db, require_platform_admin_user
from app.db.base import Base
from app.main import create_app
from app.models.auth_session import AuthSession
from app.models.user_session_metric import UserSessionMetric
from app.schemas.settings import SettingsWrite
from app.services.auth_sessions import AuthSessionService
from app.services.settings import SettingsService


def build_client() -> tuple[TestClient, sessionmaker[Session]]:
    engine = create_engine(
        "sqlite+pysqlite:///:memory:",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    Base.metadata.create_all(bind=engine)
    session_factory = sessionmaker(bind=engine, autoflush=False, autocommit=False)
    with session_factory() as db:
        settings_service = SettingsService(db)
        payload = settings_service.get_settings_snapshot().model_dump()
        payload["adminForm"]["adminAccount"] = "auth-admin"
        payload["adminForm"]["newPassword"] = "safe-admin-password"
        payload["adminForm"]["confirmPassword"] = "safe-admin-password"
        settings_service.save_settings_snapshot(SettingsWrite(**payload))

    app = create_app()

    def override_db() -> Generator[Session, None, None]:
        with session_factory() as db:
            yield db

    app.dependency_overrides[get_db] = override_db
    return TestClient(app), session_factory


def login_admin(client: TestClient) -> dict:
    response = client.post(
        "/api/v1/auth/login",
        json={"username": "auth-admin", "password": "safe-admin-password"},
    )
    assert response.status_code == 200
    return response.json()


def test_login_issues_opaque_bearer_token_and_me_uses_server_session() -> None:
    client, session_factory = build_client()
    payload = login_admin(client)
    token = payload["accessToken"]

    assert payload["tokenType"] == "Bearer"
    assert payload["expiresAt"]
    with session_factory() as db:
        auth_session = db.scalars(select(AuthSession)).one()
        assert auth_session.token_hash == AuthSessionService.hash_token(token)
        assert auth_session.token_hash != token

    response = client.get(
        "/api/v1/auth/me",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert response.status_code == 200
    assert response.json()["username"] == "auth-admin"
    assert response.json()["isAdmin"] is True


def test_forged_identity_headers_no_longer_authenticate() -> None:
    client, _ = build_client()
    response = client.get(
        "/api/v1/auth/me",
        headers={
            "X-Auth-Username": "superadmin",
            "X-Auth-Role-Codes": "manager",
            "X-Auth-Is-Admin": "true",
        },
    )

    assert response.status_code == 401
    assert response.headers["www-authenticate"] == "Bearer"


def test_expired_and_revoked_tokens_are_rejected() -> None:
    client, session_factory = build_client()
    first_payload = login_admin(client)
    first_token = first_payload["accessToken"]

    with session_factory() as db:
        auth_session = db.scalars(
            select(AuthSession).where(
                AuthSession.token_hash == AuthSessionService.hash_token(first_token)
            )
        ).one()
        auth_session.expires_at = datetime.now(UTC) - timedelta(seconds=1)
        db.commit()

    expired_response = client.get(
        "/api/v1/auth/me",
        headers={"Authorization": f"Bearer {first_token}"},
    )
    assert expired_response.status_code == 401

    second_payload = login_admin(client)
    second_token = second_payload["accessToken"]
    logout_response = client.post(
        "/api/v1/auth/logout",
        headers={"Authorization": f"Bearer {second_token}"},
        json={"sessionId": second_payload["sessionId"], "reason": "manual"},
    )
    assert logout_response.status_code == 200
    with session_factory() as db:
        metric_session = db.scalars(
            select(UserSessionMetric).where(
                UserSessionMetric.session_id == second_payload["sessionId"]
            )
        ).one()
        assert metric_session.status == "closed"

    revoked_response = client.get(
        "/api/v1/auth/me",
        headers={"Authorization": f"Bearer {second_token}"},
    )
    assert revoked_response.status_code == 401


def test_manager_role_is_not_platform_admin() -> None:
    manager = CurrentUserContext(
        tenant_id="default",
        username="manager@example.com",
        name="Manager",
        role_codes=["manager"],
        is_admin=False,
    )

    with pytest.raises(HTTPException) as exc_info:
        require_platform_admin_user(manager)

    assert exc_info.value.status_code == 403