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_login_summary_counts_same_client_and_ip_as_one_active_source(db_session, monkeypatch): now = datetime.now(timezone.utc) monkeypatch.setattr(settings, "USER_SESSION_ONLINE_SECONDS", 300) user = User( id=uuid.uuid4(), email="deduplicated-login-source@example.com", password_hash="hash", full_name="Deduplicated Login Source", clinical_department="IT", status=UserStatus.ACTIVE, ) db_session.add(user) db_session.add_all( [ UserLoginSession( id=uuid.uuid4(), user_id=user.id, client_type="web", login_ip="192.168.97.1", login_at=now - timedelta(minutes=2), last_seen_at=now - timedelta(seconds=20), ), UserLoginSession( id=uuid.uuid4(), user_id=user.id, client_type="web", login_ip="192.168.97.1", login_at=now - timedelta(minutes=1), last_seen_at=now - timedelta(seconds=10), ), UserLoginSession( id=uuid.uuid4(), user_id=user.id, client_type="desktop", login_ip="192.168.97.1", login_at=now - timedelta(minutes=1), last_seen_at=now - timedelta(seconds=5), ), ] ) 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 == 2 @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)