Cheng Zhou
112 lines
3.7 KiB
Python
112 lines
3.7 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
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.services.user_login_sessions import get_login_summaries, 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_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)
|