alex
6c8cfe9157
Implements a resilient, responsive web terminal that survives network blips, provides instant typing feedback, and restores scrollback on reconnect. Backend changes: - Add heartbeat tracking (15s ping interval, 60s idle timeout) - Add message batching (16ms flush window) for efficient I/O - Add termios echo detection and set_echo_state control messages - Add graceful session_ended notification before close - Add ping/pong protocol support Frontend changes: - Rewrite TerminalComponent with status bar, connection indicator, session-ended overlay, reconnect banner, and ResizeObserver - Add useTerminalConnection hook with: - Exponential backoff auto-reconnect (1s → 30s max, 10 attempts) - Heartbeat/ping-pong with latency tracking - Local echo for printable ASCII with server deduplication - Resize debounce (200ms) + throttle (500ms) - Scrollback serialization via xterm-addon-serialize - Ctrl+Shift+R manual reconnect shortcut - Add WebSocket protocol types and encoding utilities - Add xterm-addon-serialize dependency Tests: - 16 backend unit tests (TerminalSession + TerminalManager) - 13 frontend hook tests (connection lifecycle, reconnect, resize, scrollback, callbacks) Quality gates: - Frontend typecheck: clean - Frontend lint: clean - Frontend tests: 48 passed - Backend unit tests: 101 passed - Backend ruff: clean SDD artifacts: openspec/changes/responsive-terminal/
113 lines
4.0 KiB
Python
113 lines
4.0 KiB
Python
"""Unit tests for TerminalManager."""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from src.services.terminal_manager import TerminalManager
|
|
from src.services.terminal_session import TerminalSession
|
|
|
|
|
|
@pytest.fixture
|
|
def manager():
|
|
return TerminalManager()
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_websocket():
|
|
ws = AsyncMock()
|
|
ws.send_bytes = AsyncMock()
|
|
ws.send_json = AsyncMock()
|
|
ws.close = AsyncMock()
|
|
ws.receive = AsyncMock()
|
|
return ws
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_session():
|
|
session = MagicMock(spec=TerminalSession)
|
|
session.session_id = "sess-123"
|
|
session.is_alive.return_value = True
|
|
session._closed = False
|
|
session.read_output = AsyncMock(return_value=b"")
|
|
session.write_input = AsyncMock()
|
|
session.resize = AsyncMock()
|
|
session.close = AsyncMock()
|
|
session.get_exit_reason.return_value = None
|
|
return session
|
|
|
|
|
|
class TestCreateSession:
|
|
@patch("src.services.terminal_manager.asyncio.create_task")
|
|
@patch("src.services.terminal_manager.uuid.uuid4", return_value="sess-123")
|
|
async def test_create_session_registers_and_starts_loops(
|
|
self, mock_uuid, mock_create_task, manager, mock_websocket
|
|
):
|
|
instance_id = __import__("uuid").uuid4()
|
|
mock_sess = MagicMock()
|
|
mock_sess.session_id = "sess-123"
|
|
mock_sess.is_alive.return_value = True
|
|
mock_sess._closed = False
|
|
mock_sess.start = AsyncMock()
|
|
mock_sess.read_output = AsyncMock(return_value=b"")
|
|
mock_sess.write_input = AsyncMock()
|
|
mock_sess.resize = AsyncMock()
|
|
mock_sess.close = AsyncMock()
|
|
mock_sess.get_exit_reason.return_value = None
|
|
|
|
with (
|
|
patch.object(manager, "_read_loop", new=AsyncMock()),
|
|
patch.object(manager, "_write_loop", new=AsyncMock()),
|
|
patch.object(manager, "_heartbeat_loop", new=AsyncMock()),
|
|
patch(
|
|
"src.services.terminal_manager.TerminalSession",
|
|
return_value=mock_sess,
|
|
),
|
|
):
|
|
session = await manager.create_session(
|
|
instance_id, "container-abc", mock_websocket
|
|
)
|
|
assert session.session_id == "sess-123"
|
|
assert "sess-123" in manager._sessions
|
|
assert "sess-123" in manager._last_client_message
|
|
|
|
|
|
class TestHandleControlMessage:
|
|
async def test_handle_resize(self, manager, mock_session, mock_websocket):
|
|
ctrl = {"type": "resize", "cols": 120, "rows": 40}
|
|
await manager._handle_control_message(mock_session, mock_websocket, ctrl)
|
|
mock_session.resize.assert_awaited_once_with(120, 40)
|
|
|
|
async def test_handle_ping(self, manager, mock_session, mock_websocket):
|
|
ctrl = {"type": "ping", "id": 42}
|
|
await manager._handle_control_message(mock_session, mock_websocket, ctrl)
|
|
mock_websocket.send_json.assert_awaited_once_with({"type": "pong", "id": 42})
|
|
|
|
async def test_handle_unknown_type(self, manager, mock_session, mock_websocket):
|
|
ctrl = {"type": "unknown", "data": "test"}
|
|
await manager._handle_control_message(mock_session, mock_websocket, ctrl)
|
|
mock_websocket.send_json.assert_not_awaited()
|
|
mock_session.resize.assert_not_awaited()
|
|
|
|
|
|
class TestCleanupSession:
|
|
async def test_cleanup_removes_session(self, manager, mock_session):
|
|
manager._sessions["sess-123"] = mock_session
|
|
manager._last_client_message["sess-123"] = 123.0
|
|
|
|
await manager._cleanup_session(mock_session)
|
|
assert "sess-123" not in manager._sessions
|
|
assert "sess-123" not in manager._last_client_message
|
|
mock_session.close.assert_awaited_once()
|
|
|
|
|
|
class TestCloseAll:
|
|
async def test_close_all_clears_sessions(self, manager, mock_session):
|
|
manager._sessions["sess-123"] = mock_session
|
|
manager._last_client_message["sess-123"] = 123.0
|
|
|
|
await manager.close_all()
|
|
assert len(manager._sessions) == 0
|
|
assert len(manager._last_client_message) == 0
|
|
mock_session.close.assert_awaited_once()
|