test_sessions.py•9.91 kB
"""
Unit tests for session manager.
"""
from uuid import UUID
import sys
import pytest
from mcp_debug_tool.schemas import SessionStatus, StartSessionRequest
from mcp_debug_tool.sessions import SessionManager
@pytest.fixture
def workspace_root(tmp_path):
"""Create temporary workspace directory."""
return tmp_path
@pytest.fixture
def session_manager(workspace_root):
"""Create session manager instance."""
return SessionManager(workspace_root)
@pytest.fixture
def sample_script(workspace_root):
"""Create a sample Python script."""
script = workspace_root / "test_script.py"
script.write_text("""
def main():
x = 1
y = 2
return x + y
if __name__ == "__main__":
main()
""")
return script
def test_create_session(session_manager, sample_script):
"""Test session creation."""
request = StartSessionRequest(entry="test_script.py", pythonPath=sys.executable)
response = session_manager.create_session(request)
assert response.sessionId
# Verify it's a valid UUID
UUID(response.sessionId)
# Verify session is stored
session = session_manager.get_session(response.sessionId)
assert session.status == SessionStatus.IDLE
assert session.entry == sample_script
def test_create_session_with_args_env(session_manager, sample_script):
"""Test session creation with args and env."""
request = StartSessionRequest(
entry="test_script.py",
args=["--flag", "value"],
env={"VAR": "test"},
pythonPath=sys.executable,
)
response = session_manager.create_session(request)
session = session_manager.get_session(response.sessionId)
assert session.args == ["--flag", "value"]
assert session.env == {"VAR": "test"}
def test_create_session_invalid_entry(session_manager):
"""Test session creation with non-existent entry."""
request = StartSessionRequest(entry="nonexistent.py", pythonPath=sys.executable)
with pytest.raises(FileNotFoundError):
session_manager.create_session(request)
def test_get_session_not_found(session_manager):
"""Test getting non-existent session."""
with pytest.raises(KeyError):
session_manager.get_session("invalid-id")
def test_get_state(session_manager, sample_script):
"""Test getting session state."""
request = StartSessionRequest(entry="test_script.py", pythonPath=sys.executable)
response = session_manager.create_session(request)
state = session_manager.get_state(response.sessionId)
assert state.status == SessionStatus.IDLE
assert state.lastBreakpoint is None
assert state.timings is not None
def test_end_session(session_manager, sample_script):
"""Test ending a session."""
request = StartSessionRequest(entry="test_script.py", pythonPath=sys.executable)
response = session_manager.create_session(request)
session_id = response.sessionId
end_response = session_manager.end_session(session_id)
assert end_response.ended is True
# Verify session is removed
with pytest.raises(KeyError):
session_manager.get_session(session_id)
def test_update_status(session_manager, sample_script):
"""Test session status update."""
request = StartSessionRequest(entry="test_script.py", pythonPath=sys.executable)
response = session_manager.create_session(request)
session = session_manager.get_session(response.sessionId)
initial_updated = session.updated_at
session.update_status(SessionStatus.RUNNING)
assert session.status == SessionStatus.RUNNING
assert session.updated_at > initial_updated
def test_update_breakpoint(session_manager, sample_script):
"""Test breakpoint update."""
request = StartSessionRequest(entry="test_script.py", pythonPath=sys.executable)
response = session_manager.create_session(request)
session = session_manager.get_session(response.sessionId)
# First hit
session.update_breakpoint("test.py", 10)
assert session.last_breakpoint is not None
assert session.last_breakpoint.file == "test.py"
assert session.last_breakpoint.line == 10
assert session.last_breakpoint.hitCount == 1
# Same breakpoint again
session.update_breakpoint("test.py", 10)
assert session.last_breakpoint.hitCount == 2
# Different breakpoint
session.update_breakpoint("test.py", 20)
assert session.last_breakpoint.line == 20
assert session.last_breakpoint.hitCount == 1
def test_update_timings(session_manager, sample_script):
"""Test timing updates."""
request = StartSessionRequest(entry="test_script.py", pythonPath=sys.executable)
response = session_manager.create_session(request)
session = session_manager.get_session(response.sessionId)
session.update_timings(100.5)
assert session.timings.lastRunMs == 100.5
assert session.timings.totalCpuTimeMs == 100.5
session.update_timings(50.2)
assert session.timings.lastRunMs == 50.2
assert session.timings.totalCpuTimeMs == pytest.approx(150.7)
class TestSessionStateTransitions:
"""Tests for session state transitions."""
def test_idle_to_running_transition(self, session_manager, sample_script):
"""Test transition from idle to running state."""
request = StartSessionRequest(entry="test_script.py", pythonPath=sys.executable)
response = session_manager.create_session(request)
session_id = response.sessionId
session = session_manager.get_session(session_id)
assert session.status == SessionStatus.IDLE
session.update_status(SessionStatus.RUNNING)
assert session.status == SessionStatus.RUNNING
def test_running_to_paused_transition(self, session_manager, sample_script):
"""Test transition from running to paused state."""
request = StartSessionRequest(entry="test_script.py", pythonPath=sys.executable)
response = session_manager.create_session(request)
session_id = response.sessionId
session = session_manager.get_session(session_id)
session.update_status(SessionStatus.RUNNING)
session.update_status(SessionStatus.PAUSED)
assert session.status == SessionStatus.PAUSED
def test_paused_to_running_transition(self, session_manager, sample_script):
"""Test transition from paused back to running (continue operation)."""
request = StartSessionRequest(entry="test_script.py", pythonPath=sys.executable)
response = session_manager.create_session(request)
session_id = response.sessionId
session = session_manager.get_session(session_id)
session.update_status(SessionStatus.RUNNING)
session.update_status(SessionStatus.PAUSED)
session.update_status(SessionStatus.RUNNING)
assert session.status == SessionStatus.RUNNING
def test_running_to_completed_transition(self, session_manager, sample_script):
"""Test transition from running to completed state."""
request = StartSessionRequest(entry="test_script.py", pythonPath=sys.executable)
response = session_manager.create_session(request)
session_id = response.sessionId
session = session_manager.get_session(session_id)
session.update_status(SessionStatus.RUNNING)
session.update_status(SessionStatus.COMPLETED)
assert session.status == SessionStatus.COMPLETED
def test_any_state_to_error_transition(self, session_manager, sample_script):
"""Test transition to error state from any state."""
request = StartSessionRequest(entry="test_script.py", pythonPath=sys.executable)
response = session_manager.create_session(request)
session_id = response.sessionId
session = session_manager.get_session(session_id)
# Idle to error
session.update_status(SessionStatus.ERROR)
assert session.status == SessionStatus.ERROR
# Running to error
session.update_status(SessionStatus.RUNNING)
session.update_status(SessionStatus.ERROR)
assert session.status == SessionStatus.ERROR
def test_state_updates_timestamp(self, session_manager, sample_script):
"""Test that state updates refresh the updated_at timestamp."""
import time
request = StartSessionRequest(entry="test_script.py", pythonPath=sys.executable)
response = session_manager.create_session(request)
session_id = response.sessionId
session = session_manager.get_session(session_id)
initial_time = session.updated_at
# Wait a bit and update status
time.sleep(0.01)
session.update_status(SessionStatus.RUNNING)
assert session.updated_at > initial_time
def test_breakpoint_update_updates_timestamp(
self, session_manager, sample_script
):
"""Test that breakpoint updates refresh the updated_at timestamp."""
import time
request = StartSessionRequest(entry="test_script.py", pythonPath=sys.executable)
response = session_manager.create_session(request)
session_id = response.sessionId
session = session_manager.get_session(session_id)
initial_time = session.updated_at
time.sleep(0.01)
session.update_breakpoint("test.py", 10)
assert session.updated_at > initial_time
def test_timings_update_updates_timestamp(self, session_manager, sample_script):
"""Test that timing updates refresh the updated_at timestamp."""
import time
request = StartSessionRequest(entry="test_script.py", pythonPath=sys.executable)
response = session_manager.create_session(request)
session_id = response.sessionId
session = session_manager.get_session(session_id)
initial_time = session.updated_at
time.sleep(0.01)
session.update_timings(100.0)
assert session.updated_at > initial_time