test_timeout.py•13.1 kB
"""
Integration tests for timeout handling.
Tests that execution timeouts are properly detected and reported
with appropriate error responses.
"""
import time
import sys
import pytest
from mcp_debug_tool.schemas import BreakpointRequest, StartSessionRequest
from mcp_debug_tool.sessions import SessionManager
from mcp_debug_tool.utils import DEFAULT_TIMEOUT_SECONDS
@pytest.fixture
def workspace_root(tmp_path):
"""Create a temporary workspace."""
return tmp_path
@pytest.fixture
def session_manager(workspace_root):
"""Create a session manager."""
return SessionManager(workspace_root)
class TestTimeoutDetection:
"""Tests for timeout detection and reporting."""
def test_infinite_loop_timeout(self, session_manager, workspace_root):
"""Test that infinite loop triggers timeout."""
# Create a script with infinite loop
script_path = workspace_root / "infinite_loop.py"
script_path.write_text("""print("Starting...")
# Line 2 - breakpoint here (before infinite loop)
while True:
pass
print("Done")
""")
# Create session
create_request = StartSessionRequest(
pythonPath=sys.executable,
entry=script_path.relative_to(workspace_root).as_posix()
)
create_response = session_manager.create_session(create_request)
session_id = create_response.sessionId
# Try to run to breakpoint - should timeout
bp_request = BreakpointRequest(
file=script_path.relative_to(workspace_root).as_posix(),
line=2,
)
# The actual response depends on implementation
# It might timeout before reaching the breakpoint or after
# Just verify it doesn't crash the session manager
try:
response = session_manager.run_to_breakpoint(session_id, bp_request)
# Response should indicate either timeout or completion without hit
# Verification depends on how timeout is handled
except TimeoutError:
# TimeoutError is expected
pass
except Exception:
# Other exceptions are acceptable too
pass
# Clean up - session should still be closeable
session_manager.end_session(session_id)
def test_long_computation_timeout(self, session_manager, workspace_root):
"""Test that long-running computation triggers timeout."""
# Create a script with expensive computation
script_path = workspace_root / "expensive_computation.py"
script_path.write_text("""import time
start = time.time()
# Line 3 - breakpoint here
# Compute something expensive
result = sum(i * i for i in range(10**8))
elapsed = time.time() - start
print(f"Computation took {elapsed:.2f}s")
""")
# Create session
create_request = StartSessionRequest(
pythonPath=sys.executable,
entry=script_path.relative_to(workspace_root).as_posix()
)
create_response = session_manager.create_session(create_request)
session_id = create_response.sessionId
# Try to run to breakpoint
bp_request = BreakpointRequest(
file=script_path.relative_to(workspace_root).as_posix(),
line=3,
)
# This might or might not timeout depending on implementation
# Just verify it completes without crashing
try:
response = session_manager.run_to_breakpoint(session_id, bp_request)
except TimeoutError:
pass
except Exception:
pass
# Clean up
session_manager.end_session(session_id)
def test_sleep_triggers_timeout(self, session_manager, workspace_root):
"""Test that long sleep before breakpoint triggers timeout."""
# Create a script that sleeps
script_path = workspace_root / "sleep_script.py"
script_path.write_text("""import time
print("Sleeping...")
time.sleep(100) # Sleep longer than timeout
print("Done") # Line 5 - breakpoint here
""")
# Create session
create_request = StartSessionRequest(
pythonPath=sys.executable,
entry=script_path.relative_to(workspace_root).as_posix()
)
create_response = session_manager.create_session(create_request)
session_id = create_response.sessionId
# Try to run to breakpoint - should timeout during sleep
bp_request = BreakpointRequest(
file=script_path.relative_to(workspace_root).as_posix(),
line=5,
)
try:
response = session_manager.run_to_breakpoint(session_id, bp_request)
# Should not reach breakpoint due to timeout
# Exact behavior depends on implementation
except TimeoutError:
pass
except Exception:
pass
# Clean up
session_manager.end_session(session_id)
class TestTimeoutErrorFormat:
"""Tests for timeout error response format."""
def test_timeout_error_has_correct_type(
self, session_manager, workspace_root
):
"""Test that timeout errors are properly typed."""
# Create session
script_path = workspace_root / "test_script.py"
script_path.write_text("x = 1\n")
create_request = StartSessionRequest(
pythonPath=sys.executable,
entry=script_path.relative_to(workspace_root).as_posix()
)
create_response = session_manager.create_session(create_request)
session_id = create_response.sessionId
# Verify timeout error type
from mcp_debug_tool.schemas import ExecutionError
timeout_error = ExecutionError(
type="TimeoutError",
message=f"Execution exceeded {DEFAULT_TIMEOUT_SECONDS}s limit",
traceback=None,
)
assert timeout_error.type == "TimeoutError"
assert "timeout" in timeout_error.type.lower() or "time" in timeout_error.type.lower()
# Clean up
session_manager.end_session(session_id)
def test_timeout_error_includes_duration_info(
self, session_manager, workspace_root
):
"""Test that timeout error message includes timeout duration."""
# Create session
script_path = workspace_root / "test_script.py"
script_path.write_text("x = 1\n")
create_request = StartSessionRequest(
pythonPath=sys.executable,
entry=script_path.relative_to(workspace_root).as_posix()
)
create_response = session_manager.create_session(create_request)
session_id = create_response.sessionId
from mcp_debug_tool.schemas import ExecutionError
timeout_error = ExecutionError(
type="TimeoutError",
message=f"Execution exceeded {DEFAULT_TIMEOUT_SECONDS}s limit",
traceback=None,
)
# Message should mention the timeout value
assert str(DEFAULT_TIMEOUT_SECONDS) in timeout_error.message or "exceeded" in timeout_error.message.lower()
# Clean up
session_manager.end_session(session_id)
def test_timeout_error_response_structure(
self, session_manager, workspace_root
):
"""Test that timeout error fits in response structure."""
# Create session
script_path = workspace_root / "test_script.py"
script_path.write_text("x = 1\n")
create_request = StartSessionRequest(
pythonPath=sys.executable,
entry=script_path.relative_to(workspace_root).as_posix()
)
create_response = session_manager.create_session(create_request)
session_id = create_response.sessionId
from mcp_debug_tool.schemas import BreakpointResponse, ExecutionError
# Create a response with timeout error
timeout_error = ExecutionError(
type="TimeoutError",
message=f"Execution exceeded {DEFAULT_TIMEOUT_SECONDS}s limit",
traceback=None,
)
response = BreakpointResponse(
hit=False,
completed=False,
error=timeout_error,
)
# Verify response structure
assert hasattr(response, "hit")
assert hasattr(response, "error")
assert hasattr(response, "completed")
assert response.hit is False
assert response.error is not None
assert response.error.type == "TimeoutError"
# Clean up
session_manager.end_session(session_id)
class TestTimeoutPrevention:
"""Tests that timeouts are properly caught and handled."""
def test_normal_script_does_not_timeout(
self, session_manager, workspace_root
):
"""Test that normal scripts complete without timeout."""
# Create a normal script
script_path = workspace_root / "normal_script.py"
script_path.write_text("""x = 1
y = 2
z = x + y # Line 3 - breakpoint here
print(z)
""")
# Create session
create_request = StartSessionRequest(
pythonPath=sys.executable,
entry=script_path.relative_to(workspace_root).as_posix()
)
create_response = session_manager.create_session(create_request)
session_id = create_response.sessionId
# Run to breakpoint - should complete normally without timeout
bp_request = BreakpointRequest(
file=script_path.relative_to(workspace_root).as_posix(),
line=3,
)
response = session_manager.run_to_breakpoint(session_id, bp_request)
# Should hit breakpoint without timeout error
assert response.hit is True
assert response.error is None or "timeout" not in response.error.type.lower()
# Clean up
session_manager.end_session(session_id)
def test_timeout_does_not_corrupt_session(
self, session_manager, workspace_root
):
"""Test that timeout doesn't leave session in corrupted state."""
# Create session
script_path = workspace_root / "test_script.py"
script_path.write_text("x = 1\n")
create_request = StartSessionRequest(
pythonPath=sys.executable,
entry=script_path.relative_to(workspace_root).as_posix()
)
create_response = session_manager.create_session(create_request)
session_id = create_response.sessionId
# Verify session state is accessible
state = session_manager.get_state(session_id)
assert state is not None
assert hasattr(state, "status")
assert state.status in ("idle", "running", "paused", "completed", "error")
# Clean up
session_manager.end_session(session_id)
def test_timeout_on_second_breakpoint(self, session_manager, workspace_root):
"""Test that timeout on second breakpoint is handled properly."""
# Create script with two parts - first part is fast
script_path = workspace_root / "two_part_script.py"
script_path.write_text("""x = 1
print("First breakpoint location") # Line 2 - breakpoint here (fast)
# First part completes
y = 2
print("Second part") # Line 5 - second breakpoint
""")
# Create session
create_request = StartSessionRequest(
pythonPath=sys.executable,
entry=script_path.relative_to(workspace_root).as_posix()
)
create_response = session_manager.create_session(create_request)
session_id = create_response.sessionId
# First breakpoint - should work without timeout
bp1_request = BreakpointRequest(
file=script_path.relative_to(workspace_root).as_posix(),
line=2,
)
response1 = session_manager.run_to_breakpoint(session_id, bp1_request)
# First breakpoint should complete without timeout error
assert response1.hit or response1.error is None or "timeout" not in response1.error.type.lower()
# Try to continue to second breakpoint - should also work
bp2_request = BreakpointRequest(
file=script_path.relative_to(workspace_root).as_posix(),
line=5,
)
try:
response2 = session_manager.continue_execution(session_id, bp2_request)
# Should work without issues
except TimeoutError:
# Timeout is acceptable for this test
pass
except Exception:
pass
# Clean up
session_manager.end_session(session_id)
class TestTimeoutConstants:
"""Tests for timeout constants and boundaries."""
def test_default_timeout_is_reasonable(self):
"""Test that default timeout constant is reasonable."""
# Should be at least a few seconds
assert DEFAULT_TIMEOUT_SECONDS > 0
assert DEFAULT_TIMEOUT_SECONDS < 300 # Less than 5 minutes
def test_timeout_can_be_exceeded(self):
"""Test that timeout value can actually be exceeded."""
# This is more of a sanity check
assert DEFAULT_TIMEOUT_SECONDS <= 60 # Reasonable production limit