tests/test_session_manager.py

import pytest
from sandbox.session_manager import SessionManager
from sandbox.models import CreateSessionRequest, SessionStatus, SandboxConfig
import time


@pytest.fixture
def session_manager():
    """Create session manager for testing"""
    try:
        manager = SessionManager(SandboxConfig())
        yield manager
        # Cleanup all sessions
        manager.shutdown()
    except RuntimeError as e:
        pytest.skip(f"Docker not available: {e}")


def test_create_session(session_manager):
    """Test creating a new session"""
    request = CreateSessionRequest(
        metadata={"test": "true", "user": "pytest"},
        timeout_minutes=30
    )
    
    session = session_manager.create_session(request)
    
    assert session.session_id is not None
    assert session.container_id is not None
    assert session.volume_name is not None
    assert session.status in [SessionStatus.CREATING, SessionStatus.READY]
    assert session.metadata["test"] == "true"
    assert session.timeout_minutes == 30


def test_get_session(session_manager):
    """Test retrieving a session"""
    request = CreateSessionRequest(metadata={"test": "get"})
    session = session_manager.create_session(request)
    
    # Get session by ID
    retrieved = session_manager.get_session(session.session_id)
    
    assert retrieved is not None
    assert retrieved.session_id == session.session_id
    assert retrieved.container_id == session.container_id


def test_get_nonexistent_session(session_manager):
    """Test getting a session that doesn't exist"""
    session = session_manager.get_session("nonexistent-id")
    assert session is None


def test_list_sessions(session_manager):
    """Test listing all sessions"""
    # Create multiple sessions
    request1 = CreateSessionRequest(metadata={"num": 1})
    request2 = CreateSessionRequest(metadata={"num": 2})
    
    session1 = session_manager.create_session(request1)
    session2 = session_manager.create_session(request2)
    
    # List sessions
    sessions = session_manager.list_sessions()
    
    assert len(sessions) >= 2
    session_ids = [s.session_id for s in sessions]
    assert session1.session_id in session_ids
    assert session2.session_id in session_ids


def test_destroy_session(session_manager):
    """Test destroying a session"""
    request = CreateSessionRequest(metadata={"test": "destroy"})
    session = session_manager.create_session(request)
    
    # Destroy session
    result = session_manager.destroy_session(session.session_id)
    assert result is True
    
    # Verify it's gone
    retrieved = session_manager.get_session(session.session_id)
    assert retrieved is None


def test_destroy_nonexistent_session(session_manager):
    """Test destroying a session that doesn't exist"""
    result = session_manager.destroy_session("nonexistent-id")
    assert result is False


def test_update_activity(session_manager):
    """Test updating session activity timestamp"""
    request = CreateSessionRequest()
    session = session_manager.create_session(request)
    
    original_activity = session.last_activity
    
    # Wait a bit
    time.sleep(0.1)
    
    # Update activity
    session_manager.update_activity(session.session_id)
    
    # Get updated session
    updated = session_manager.get_session(session.session_id)
    
    assert updated.last_activity > original_activity


def test_session_timeout_cleanup(session_manager):
    """Test that idle sessions are cleaned up"""
    # Create session with very short timeout
    request = CreateSessionRequest(timeout_minutes=1)  # 1 minute
    session = session_manager.create_session(request)
    
    # Manually trigger cleanup (simulate timeout by modifying activity time)
    import datetime
    with session_manager._lock:
        # Set activity to 2 minutes ago
        session_manager.sessions[session.session_id].last_activity = \
            datetime.datetime.utcnow() - datetime.timedelta(minutes=2)
    
    # Trigger cleanup
    session_manager._cleanup_idle_sessions()
    
    # Session should be gone
    retrieved = session_manager.get_session(session.session_id)
    assert retrieved is None


def test_concurrent_sessions(session_manager):
    """Test creating multiple concurrent sessions"""
    sessions = []
    num_sessions = 3
    
    for i in range(num_sessions):
        request = CreateSessionRequest(metadata={"index": i})
        session = session_manager.create_session(request)
        sessions.append(session)
    
    # Verify all exist
    all_sessions = session_manager.list_sessions()
    assert len(all_sessions) >= num_sessions
    
    # Cleanup
    for session in sessions:
        session_manager.destroy_session(session.session_id)


def test_session_persistence(session_manager):
    """Test that session containers are actually running"""
    import docker
    
    request = CreateSessionRequest(metadata={"test": "persistence"})
    session = session_manager.create_session(request)
    
    # Verify container exists and is running
    client = docker.from_env()
    try:
        container = client.containers.get(session.container_id)
        assert container.status == 'running'
    except docker.errors.NotFound:
        pytest.fail(f"Container {session.container_id} not found")