Spaces:
Paused
Paused
File size: 1,766 Bytes
d958e80 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 | 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
|