test_continue_to_next_breakpoint.py•8.41 kB
"""
Integration tests for continue_to_next_breakpoint functionality.
Tests the full flow of continuing within the same session.
"""
import pytest
import sys
from mcp_debug_tool.schemas import BreakpointRequest, StartSessionRequest
from mcp_debug_tool.sessions import SessionManager
@pytest.fixture
def workspace_root(tmp_path):
"""Create a temporary workspace."""
return tmp_path
@pytest.fixture
def multi_breakpoint_script(workspace_root):
"""Create a test script with multiple breakpoint locations."""
script_path = workspace_root / "multi_bp_script.py"
script_path.write_text("""def process(data):
result = []
for item in data: # Line 3 - first breakpoint
doubled = item * 2
result.append(doubled) # Line 6 - second breakpoint
return result
items = [1, 2, 3]
final = process(items) # Line 10 - third breakpoint
print(final)
""")
return script_path
@pytest.fixture
def session_manager(workspace_root):
"""Create a session manager."""
return SessionManager(workspace_root)
class TestContinueToNextBreakpoint:
"""Integration tests for continue functionality."""
def test_continue_from_paused_state_works(
self, session_manager, multi_breakpoint_script
):
"""Test that session must be in paused state to continue."""
# Create session
create_request = StartSessionRequest(
entry=multi_breakpoint_script.relative_to(
session_manager.workspace_root
).as_posix(),
pythonPath=sys.executable,
)
create_response = session_manager.create_session(create_request)
session_id = create_response.sessionId
# Run to first breakpoint
bp1_request = BreakpointRequest(
file=multi_breakpoint_script.relative_to(
session_manager.workspace_root
).as_posix(),
line=3,
)
response1 = session_manager.run_to_breakpoint(session_id, bp1_request)
assert response1.hit is True
# Get session state - should be paused
state1 = session_manager.get_state(session_id)
assert state1.status == "paused"
# Continue to second breakpoint
bp2_request = BreakpointRequest(
file=multi_breakpoint_script.relative_to(
session_manager.workspace_root
).as_posix(),
line=6,
)
response2 = session_manager.continue_execution(session_id, bp2_request)
# Response could be hit or not depending on execution, but should not error
assert response2.error is None or response2.completed is True
# Clean up
session_manager.end_session(session_id)
def test_continue_from_non_paused_state_fails(
self, session_manager, multi_breakpoint_script
):
"""Test that continue fails if session is not in paused state."""
# Create session
create_request = StartSessionRequest(
entry=multi_breakpoint_script.relative_to(
session_manager.workspace_root
).as_posix(),
pythonPath=sys.executable,
)
create_response = session_manager.create_session(create_request)
session_id = create_response.sessionId
# Try to continue without first running to breakpoint
bp_request = BreakpointRequest(
file=multi_breakpoint_script.relative_to(
session_manager.workspace_root
).as_posix(),
line=6,
)
response = session_manager.continue_execution(session_id, bp_request)
assert response.hit is False
assert response.error is not None
assert "InvalidStateError" in response.error.type
# Clean up
session_manager.end_session(session_id)
def test_continue_validates_breakpoint_location(
self, session_manager, multi_breakpoint_script
):
"""Test that continue validates the breakpoint location."""
# Create session
create_request = StartSessionRequest(
entry=multi_breakpoint_script.relative_to(
session_manager.workspace_root
).as_posix(),
pythonPath=sys.executable,
)
create_response = session_manager.create_session(create_request)
session_id = create_response.sessionId
# Run to first breakpoint
bp1_request = BreakpointRequest(
file=multi_breakpoint_script.relative_to(
session_manager.workspace_root
).as_posix(),
line=3,
)
session_manager.run_to_breakpoint(session_id, bp1_request)
# Try to continue with invalid line (beyond file length)
bp2_invalid = BreakpointRequest(
file=multi_breakpoint_script.relative_to(
session_manager.workspace_root
).as_posix(),
line=1000,
)
with pytest.raises(ValueError, match="exceeds file length"):
session_manager.continue_execution(session_id, bp2_invalid)
# Clean up
session_manager.end_session(session_id)
def test_continue_updates_session_state(
self, session_manager, multi_breakpoint_script
):
"""Test that continue updates session state appropriately."""
# Create session
create_request = StartSessionRequest(
entry=multi_breakpoint_script.relative_to(
session_manager.workspace_root
).as_posix(),
pythonPath=sys.executable,
)
create_response = session_manager.create_session(create_request)
session_id = create_response.sessionId
# Run to first breakpoint
bp1_request = BreakpointRequest(
file=multi_breakpoint_script.relative_to(
session_manager.workspace_root
).as_posix(),
line=3,
)
session_manager.run_to_breakpoint(session_id, bp1_request)
state_before = session_manager.get_state(session_id)
assert state_before.status == "paused"
# Continue
bp2_request = BreakpointRequest(
file=multi_breakpoint_script.relative_to(
session_manager.workspace_root
).as_posix(),
line=6,
)
session_manager.continue_execution(session_id, bp2_request)
# State should be either paused (if hit) or completed (if execution finished)
state_after = session_manager.get_state(session_id)
assert state_after.status in ("paused", "completed")
# Clean up
session_manager.end_session(session_id)
def test_session_timings_update_on_continue(
self, session_manager, workspace_root
):
"""Test that session timings are updated on continue."""
# Create simple script
script_path = workspace_root / "simple.py"
script_path.write_text("""x = 1 # Line 1
y = 2 # Line 2
z = 3 # Line 3
""")
# Create session
create_request = StartSessionRequest(
entry=script_path.relative_to(session_manager.workspace_root).as_posix(),
pythonPath=sys.executable,
)
create_response = session_manager.create_session(create_request)
session_id = create_response.sessionId
# Run to first breakpoint
bp1_request = BreakpointRequest(
file=script_path.relative_to(session_manager.workspace_root).as_posix(),
line=1,
)
session_manager.run_to_breakpoint(session_id, bp1_request)
# Get timing after first run
state1 = session_manager.get_state(session_id)
first_run_ms = state1.timings.lastRunMs
assert first_run_ms is not None
# Continue to second breakpoint
bp2_request = BreakpointRequest(
file=script_path.relative_to(session_manager.workspace_root).as_posix(),
line=2,
)
session_manager.continue_execution(session_id, bp2_request)
# Get timing after continue
state2 = session_manager.get_state(session_id)
second_run_ms = state2.timings.lastRunMs
# Timings should be recorded (even if 0 due to fast execution)
assert second_run_ms is not None
assert state2.timings.totalCpuTimeMs >= 0
# Clean up
session_manager.end_session(session_id)