tests/test_sessions.py

import pytest
from datetime import datetime, timedelta

from app.session_store import create_session, get_session, delete_session, attach_worker, detach_worker
from app.models import WorkerState, WorkerRuntimeType
from app.security import generate_worker_id


def test_create_session():
    session = create_session(owner_label="test-owner")
    assert session.session_id.startswith("sess_")
    assert session.owner_label == "test-owner"
    assert session.status == "active"


def test_session_expires():
    from app.config import Settings
    session = create_session()
    ttl = Settings.get_session_ttl_seconds()
    assert session.expires_at > session.created_at
    assert (session.expires_at - session.created_at).total_seconds() == ttl


def test_worker_attaches_to_session():
    session = create_session()
    worker = WorkerState(
        worker_id=generate_worker_id(),
        session_id=session.session_id,
        runtime_type=WorkerRuntimeType.SAFARI_WASM,
    )
    ok = attach_worker(session.session_id, worker)
    assert ok is True
    workers = [w for w in session.workers.values()]
    assert len(workers) == 1


def test_worker_detaches_from_session():
    session = create_session()
    worker = WorkerState(
        worker_id=generate_worker_id(),
        session_id=session.session_id,
        runtime_type=WorkerRuntimeType.SAFARI_WASM,
    )
    attach_worker(session.session_id, worker)
    ok = detach_worker(session.session_id, worker.worker_id)
    assert ok is True
    assert len(session.workers) == 0


def test_invalid_join_token_rejected():
    from app.security import verify_worker_join_token
    assert verify_worker_join_token("sess_test", "bad") is False
    assert verify_worker_join_token("sess_test", "a" * 32) is True