ctms/backend/tests/test_user_login_sessions.py

import uuid
from datetime import datetime, timedelta, timezone
from pathlib import Path
from types import SimpleNamespace

import pytest

from app.core.config import settings
from app.crud import user as user_crud
from app.models.user import User, UserStatus
from app.models.user_login_session import UserLoginSession
from app.schemas.user import UserLoginActivityRead
from app.services.user_login_sessions import (
    create_login_session,
    get_login_summaries,
    login_activity_payload,
    session_id_from_payload,
)


def test_legacy_session_id_is_stable_without_storing_a_token():
    payload = {"sub": "00000000-0000-0000-0000-000000000101", "orig_iat": 1_700_000_000, "client_type": "web"}

    first = session_id_from_payload(payload)
    second = session_id_from_payload(payload)

    assert first == second
    assert isinstance(first, uuid.UUID)


def test_session_heartbeat_returns_server_observed_client_ip():
    auth_source = (Path(__file__).resolve().parents[1] / "app" / "api" / "v1" / "auth.py").read_text()

    assert "resolve_client_ip(request)" in auth_source
    assert '"client_ip": resolve_client_ip(request)' in auth_source


@pytest.mark.asyncio
async def test_login_session_stores_server_observed_login_ip(db_session):
    user = User(
        id=uuid.uuid4(),
        email="login-ip@example.com",
        password_hash="hash",
        full_name="Login IP",
        clinical_department="IT",
        status=UserStatus.ACTIVE,
    )
    db_session.add(user)
    await db_session.commit()
    request = SimpleNamespace(
        headers={
            "x-ctms-client-type": "desktop",
            "x-ctms-client-platform": "macos",
            "x-ctms-client-version": "0.1.0",
        },
        client=SimpleNamespace(host="203.0.113.18"),
    )

    session = await create_login_session(
        db_session,
        session_id=uuid.uuid4(),
        user_id=user.id,
        request=request,
    )

    assert session.login_ip == "203.0.113.18"
    assert session.client_type == "desktop"


def test_login_activity_payload_uses_server_authoritative_status_and_local_ip_location(monkeypatch):
    now = datetime.now(timezone.utc)
    monkeypatch.setattr(settings, "USER_SESSION_ONLINE_SECONDS", 300)
    session = UserLoginSession(
        id=uuid.uuid4(),
        user_id=uuid.uuid4(),
        client_type="web",
        login_ip="127.0.0.1",
        login_at=now - timedelta(minutes=10),
        last_seen_at=now - timedelta(seconds=30),
    )

    online = login_activity_payload(session, reference_time=now)
    assert online["activity_status"] == "ONLINE"
    assert online["login_ip"] == "127.0.0.1"
    assert online["ip_location"] == "本机"
    assert UserLoginActivityRead.model_validate(online).activity_status == "ONLINE"

    session.last_seen_at = now - timedelta(minutes=6)
    assert login_activity_payload(session, reference_time=now)["activity_status"] == "OFFLINE"

    session.ended_at = now - timedelta(minutes=1)
    assert login_activity_payload(session, reference_time=now)["activity_status"] == "ENDED"


@pytest.mark.asyncio
async def test_login_summary_marks_recent_unended_sessions_online(db_session, monkeypatch):
    now = datetime.now(timezone.utc)
    monkeypatch.setattr(settings, "USER_SESSION_ONLINE_SECONDS", 300)
    user = User(
        id=uuid.uuid4(),
        email="login-summary@example.com",
        password_hash="hash",
        full_name="Login Summary",
        clinical_department="IT",
        status=UserStatus.ACTIVE,
    )
    db_session.add(user)
    db_session.add_all(
        [
            UserLoginSession(
                id=uuid.uuid4(),
                user_id=user.id,
                client_type="desktop",
                login_at=now - timedelta(minutes=10),
                last_seen_at=now - timedelta(seconds=20),
            ),
            UserLoginSession(
                id=uuid.uuid4(),
                user_id=user.id,
                client_type="web",
                login_at=now - timedelta(days=1),
                last_seen_at=now - timedelta(days=1),
                ended_at=now - timedelta(days=1),
                end_reason="logout",
            ),
        ]
    )
    await db_session.commit()

    summaries = await get_login_summaries(db_session, [user.id])

    assert summaries[user.id].status == "ONLINE"
    assert summaries[user.id].active_session_count == 1
    assert summaries[user.id].client_type == "desktop"


@pytest.mark.asyncio
async def test_user_list_filters_online_and_offline_accounts(db_session, monkeypatch):
    now = datetime.now(timezone.utc)
    monkeypatch.setattr(settings, "USER_SESSION_ONLINE_SECONDS", 300)
    online_user = User(
        id=uuid.uuid4(),
        email="online-filter@example.com",
        password_hash="hash",
        full_name="Online Filter",
        clinical_department="IT",
        status=UserStatus.ACTIVE,
    )
    offline_user = User(
        id=uuid.uuid4(),
        email="offline-filter@example.com",
        password_hash="hash",
        full_name="Offline Filter",
        clinical_department="IT",
        status=UserStatus.ACTIVE,
    )
    db_session.add_all([online_user, offline_user])
    db_session.add(
        UserLoginSession(
            id=uuid.uuid4(),
            user_id=online_user.id,
            client_type="web",
            login_at=now - timedelta(minutes=1),
            last_seen_at=now - timedelta(seconds=10),
        )
    )
    await db_session.commit()

    online = await user_crud.list_users(db_session, login_status="ONLINE")
    offline = await user_crud.list_users(db_session, login_status="OFFLINE")

    assert online_user.id in {user.id for user in online}
    assert offline_user.id not in {user.id for user in online}
    assert offline_user.id in {user.id for user in offline}
    assert await user_crud.count_users(db_session, login_status="ONLINE") == len(online)