David Gwilliam
c57617bb3d
- Replace estimate_block_height (PIL-based) with estimate_simple_height (word wrap) - Update viewport filter tests to match new height-based filtering (~4 items vs 24) - Fix CI task duplication in mise.toml (remove redundant depends) Closes #38 Closes #36
396 lines
16 KiB
Python
396 lines
16 KiB
Python
"""
|
|
Tests for engine.display.backends.websocket module.
|
|
"""
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from engine.display.backends.websocket import WebSocketDisplay
|
|
|
|
|
|
class TestWebSocketDisplayImport:
|
|
"""Test that websocket module can be imported."""
|
|
|
|
def test_import_does_not_error(self):
|
|
"""Module imports without error."""
|
|
from engine.display import backends
|
|
|
|
assert backends is not None
|
|
|
|
|
|
class TestWebSocketDisplayInit:
|
|
"""Tests for WebSocketDisplay initialization."""
|
|
|
|
def test_default_init(self):
|
|
"""Default initialization sets correct defaults."""
|
|
with patch("engine.display.backends.websocket.websockets", None):
|
|
display = WebSocketDisplay()
|
|
assert display.host == "0.0.0.0"
|
|
assert display.port == 8765
|
|
assert display.http_port == 8766
|
|
assert display.width == 80
|
|
assert display.height == 24
|
|
|
|
def test_custom_init(self):
|
|
"""Custom initialization uses provided values."""
|
|
with patch("engine.display.backends.websocket.websockets", None):
|
|
display = WebSocketDisplay(host="localhost", port=9000, http_port=9001)
|
|
assert display.host == "localhost"
|
|
assert display.port == 9000
|
|
assert display.http_port == 9001
|
|
|
|
def test_is_available_when_websockets_present(self):
|
|
"""is_available returns True when websockets is available."""
|
|
pytest.importorskip("websockets")
|
|
display = WebSocketDisplay()
|
|
assert display.is_available() is True
|
|
|
|
@pytest.mark.skipif(
|
|
pytest.importorskip("websockets") is not None, reason="websockets is available"
|
|
)
|
|
def test_is_available_when_websockets_missing(self):
|
|
"""is_available returns False when websockets is not available."""
|
|
display = WebSocketDisplay()
|
|
assert display.is_available() is False
|
|
|
|
|
|
class TestWebSocketDisplayProtocol:
|
|
"""Test that WebSocketDisplay satisfies Display protocol."""
|
|
|
|
def test_websocket_display_is_display(self):
|
|
"""WebSocketDisplay satisfies Display protocol."""
|
|
with patch("engine.display.backends.websocket.websockets", MagicMock()):
|
|
display = WebSocketDisplay()
|
|
assert hasattr(display, "init")
|
|
assert hasattr(display, "show")
|
|
assert hasattr(display, "clear")
|
|
assert hasattr(display, "cleanup")
|
|
|
|
|
|
class TestWebSocketDisplayMethods:
|
|
"""Tests for WebSocketDisplay methods."""
|
|
|
|
def test_init_stores_dimensions(self):
|
|
"""init stores terminal dimensions."""
|
|
with patch("engine.display.backends.websocket.websockets", MagicMock()):
|
|
display = WebSocketDisplay()
|
|
display.init(100, 40)
|
|
assert display.width == 100
|
|
assert display.height == 40
|
|
|
|
@pytest.mark.skip(reason="port binding conflict in CI environment")
|
|
def test_client_count_initially_zero(self):
|
|
"""client_count returns 0 when no clients connected."""
|
|
with patch("engine.display.backends.websocket.websockets", MagicMock()):
|
|
display = WebSocketDisplay()
|
|
assert display.client_count() == 0
|
|
|
|
def test_get_ws_port(self):
|
|
"""get_ws_port returns configured port."""
|
|
with patch("engine.display.backends.websocket.websockets", MagicMock()):
|
|
display = WebSocketDisplay(port=9000)
|
|
assert display.get_ws_port() == 9000
|
|
|
|
def test_get_http_port(self):
|
|
"""get_http_port returns configured port."""
|
|
with patch("engine.display.backends.websocket.websockets", MagicMock()):
|
|
display = WebSocketDisplay(http_port=9001)
|
|
assert display.get_http_port() == 9001
|
|
|
|
def test_frame_delay_defaults_to_zero(self):
|
|
"""get_frame_delay returns 0 by default."""
|
|
with patch("engine.display.backends.websocket.websockets", MagicMock()):
|
|
display = WebSocketDisplay()
|
|
assert display.get_frame_delay() == 0.0
|
|
|
|
def test_set_frame_delay(self):
|
|
"""set_frame_delay stores the value."""
|
|
with patch("engine.display.backends.websocket.websockets", MagicMock()):
|
|
display = WebSocketDisplay()
|
|
display.set_frame_delay(0.05)
|
|
assert display.get_frame_delay() == 0.05
|
|
|
|
|
|
class TestWebSocketDisplayCallbacks:
|
|
"""Tests for WebSocketDisplay callback methods."""
|
|
|
|
def test_set_client_connected_callback(self):
|
|
"""set_client_connected_callback stores callback."""
|
|
with patch("engine.display.backends.websocket.websockets", MagicMock()):
|
|
display = WebSocketDisplay()
|
|
callback = MagicMock()
|
|
display.set_client_connected_callback(callback)
|
|
assert display._client_connected_callback is callback
|
|
|
|
def test_set_client_disconnected_callback(self):
|
|
"""set_client_disconnected_callback stores callback."""
|
|
with patch("engine.display.backends.websocket.websockets", MagicMock()):
|
|
display = WebSocketDisplay()
|
|
callback = MagicMock()
|
|
display.set_client_disconnected_callback(callback)
|
|
assert display._client_disconnected_callback is callback
|
|
|
|
|
|
class TestWebSocketDisplayUnavailable:
|
|
"""Tests when WebSocket support is unavailable."""
|
|
|
|
@pytest.mark.skipif(
|
|
pytest.importorskip("websockets") is not None, reason="websockets is available"
|
|
)
|
|
def test_start_server_noop_when_unavailable(self):
|
|
"""start_server does nothing when websockets unavailable."""
|
|
display = WebSocketDisplay()
|
|
display.start_server()
|
|
assert display._server_thread is None
|
|
|
|
@pytest.mark.skipif(
|
|
pytest.importorskip("websockets") is not None, reason="websockets is available"
|
|
)
|
|
def test_start_http_server_noop_when_unavailable(self):
|
|
"""start_http_server does nothing when websockets unavailable."""
|
|
display = WebSocketDisplay()
|
|
display.start_http_server()
|
|
assert display._http_thread is None
|
|
|
|
@pytest.mark.skipif(
|
|
pytest.importorskip("websockets") is not None, reason="websockets is available"
|
|
)
|
|
def test_show_noops_when_unavailable(self):
|
|
"""show does nothing when websockets unavailable."""
|
|
display = WebSocketDisplay()
|
|
display.show(["line1", "line2"])
|
|
|
|
|
|
class TestWebSocketUIPanelIntegration:
|
|
"""Tests for WebSocket-UIPanel integration for remote control."""
|
|
|
|
def test_set_controller_stores_controller(self):
|
|
"""set_controller stores the controller reference."""
|
|
with patch("engine.display.backends.websocket.websockets", MagicMock()):
|
|
display = WebSocketDisplay()
|
|
mock_controller = MagicMock()
|
|
display.set_controller(mock_controller)
|
|
assert display._controller is mock_controller
|
|
|
|
def test_set_command_callback_stores_callback(self):
|
|
"""set_command_callback stores the callback."""
|
|
with patch("engine.display.backends.websocket.websockets", MagicMock()):
|
|
display = WebSocketDisplay()
|
|
callback = MagicMock()
|
|
display.set_command_callback(callback)
|
|
assert display._command_callback is callback
|
|
|
|
def test_get_state_snapshot_returns_none_without_controller(self):
|
|
"""_get_state_snapshot returns None when no controller is set."""
|
|
with patch("engine.display.backends.websocket.websockets", MagicMock()):
|
|
display = WebSocketDisplay()
|
|
assert display._get_state_snapshot() is None
|
|
|
|
def test_get_state_snapshot_returns_controller_state(self):
|
|
"""_get_state_snapshot returns state from controller."""
|
|
with patch("engine.display.backends.websocket.websockets", MagicMock()):
|
|
display = WebSocketDisplay()
|
|
|
|
# Create mock controller with expected attributes
|
|
mock_controller = MagicMock()
|
|
mock_controller.stages = {
|
|
"test_stage": MagicMock(
|
|
enabled=True, params={"intensity": 0.5}, selected=False
|
|
)
|
|
}
|
|
mock_controller._current_preset = "demo"
|
|
mock_controller._presets = ["demo", "test"]
|
|
mock_controller.selected_stage = "test_stage"
|
|
|
|
display.set_controller(mock_controller)
|
|
state = display._get_state_snapshot()
|
|
|
|
assert state is not None
|
|
assert "stages" in state
|
|
assert "test_stage" in state["stages"]
|
|
assert state["stages"]["test_stage"]["enabled"] is True
|
|
assert state["stages"]["test_stage"]["params"] == {"intensity": 0.5}
|
|
assert state["preset"] == "demo"
|
|
assert state["presets"] == ["demo", "test"]
|
|
assert state["selected_stage"] == "test_stage"
|
|
|
|
def test_get_state_snapshot_handles_missing_attributes(self):
|
|
"""_get_state_snapshot handles controller without all attributes."""
|
|
with patch("engine.display.backends.websocket.websockets", MagicMock()):
|
|
display = WebSocketDisplay()
|
|
|
|
# Create mock controller without stages attribute using spec
|
|
# This prevents MagicMock from auto-creating the attribute
|
|
mock_controller = MagicMock(spec=[]) # Empty spec means no attributes
|
|
|
|
display.set_controller(mock_controller)
|
|
state = display._get_state_snapshot()
|
|
|
|
assert state == {}
|
|
|
|
def test_broadcast_state_sends_to_clients(self):
|
|
"""broadcast_state sends state update to all connected clients."""
|
|
with patch("engine.display.backends.websocket.websockets", MagicMock()):
|
|
display = WebSocketDisplay()
|
|
|
|
# Mock client with send method
|
|
mock_client = MagicMock()
|
|
mock_client.send = MagicMock()
|
|
display._clients.add(mock_client)
|
|
|
|
test_state = {"test": "state"}
|
|
display.broadcast_state(test_state)
|
|
|
|
# Verify send was called with JSON containing state
|
|
mock_client.send.assert_called_once()
|
|
call_args = mock_client.send.call_args[0][0]
|
|
assert '"type": "state"' in call_args
|
|
assert '"test"' in call_args
|
|
|
|
def test_broadcast_state_noop_when_no_clients(self):
|
|
"""broadcast_state does nothing when no clients connected."""
|
|
with patch("engine.display.backends.websocket.websockets", MagicMock()):
|
|
display = WebSocketDisplay()
|
|
display._clients.clear()
|
|
|
|
# Should not raise error
|
|
display.broadcast_state({"test": "state"})
|
|
|
|
|
|
class TestWebSocketHTTPServerPath:
|
|
"""Tests for WebSocket HTTP server client directory path calculation."""
|
|
|
|
def test_client_dir_path_calculation(self):
|
|
"""Client directory path is correctly calculated from websocket.py location."""
|
|
import os
|
|
|
|
# Use the actual websocket.py file location, not the test file
|
|
websocket_module = __import__(
|
|
"engine.display.backends.websocket", fromlist=["WebSocketDisplay"]
|
|
)
|
|
websocket_file = websocket_module.__file__
|
|
parts = websocket_file.split(os.sep)
|
|
|
|
if "engine" in parts:
|
|
engine_idx = parts.index("engine")
|
|
project_root = os.sep.join(parts[:engine_idx])
|
|
client_dir = os.path.join(project_root, "client")
|
|
else:
|
|
# Fallback calculation (shouldn't happen in normal test runs)
|
|
client_dir = os.path.join(
|
|
os.path.dirname(
|
|
os.path.dirname(os.path.dirname(os.path.dirname(websocket_file)))
|
|
),
|
|
"client",
|
|
)
|
|
|
|
# Verify the client directory exists and contains expected files
|
|
assert os.path.exists(client_dir), f"Client directory not found: {client_dir}"
|
|
assert "index.html" in os.listdir(client_dir), (
|
|
"index.html not found in client directory"
|
|
)
|
|
assert "editor.html" in os.listdir(client_dir), (
|
|
"editor.html not found in client directory"
|
|
)
|
|
|
|
# Verify the path is correct (should be .../Mainline/client)
|
|
assert client_dir.endswith("client"), (
|
|
f"Client dir should end with 'client': {client_dir}"
|
|
)
|
|
assert "Mainline" in client_dir, (
|
|
f"Client dir should contain 'Mainline': {client_dir}"
|
|
)
|
|
|
|
def test_http_server_directory_serves_client_files(self):
|
|
"""HTTP server directory correctly serves client files."""
|
|
import os
|
|
|
|
# Use the actual websocket.py file location, not the test file
|
|
websocket_module = __import__(
|
|
"engine.display.backends.websocket", fromlist=["WebSocketDisplay"]
|
|
)
|
|
websocket_file = websocket_module.__file__
|
|
parts = websocket_file.split(os.sep)
|
|
|
|
if "engine" in parts:
|
|
engine_idx = parts.index("engine")
|
|
project_root = os.sep.join(parts[:engine_idx])
|
|
client_dir = os.path.join(project_root, "client")
|
|
else:
|
|
client_dir = os.path.join(
|
|
os.path.dirname(
|
|
os.path.dirname(os.path.dirname(os.path.dirname(websocket_file)))
|
|
),
|
|
"client",
|
|
)
|
|
|
|
# Verify the handler would be able to serve files from this directory
|
|
# We can't actually instantiate the handler without a valid request,
|
|
# but we can verify the directory is accessible
|
|
assert os.access(client_dir, os.R_OK), (
|
|
f"Client directory not readable: {client_dir}"
|
|
)
|
|
|
|
# Verify key files exist
|
|
index_path = os.path.join(client_dir, "index.html")
|
|
editor_path = os.path.join(client_dir, "editor.html")
|
|
|
|
assert os.path.exists(index_path), f"index.html not found at: {index_path}"
|
|
assert os.path.exists(editor_path), f"editor.html not found at: {editor_path}"
|
|
|
|
# Verify files are readable
|
|
assert os.access(index_path, os.R_OK), "index.html not readable"
|
|
assert os.access(editor_path, os.R_OK), "editor.html not readable"
|
|
|
|
def test_old_buggy_path_does_not_find_client_directory(self):
|
|
"""The old buggy path (3 dirname calls) should NOT find the client directory.
|
|
|
|
This test verifies that the old buggy behavior would have failed.
|
|
The old code used:
|
|
client_dir = os.path.join(
|
|
os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "client"
|
|
)
|
|
|
|
This would resolve to: .../engine/client (which doesn't exist)
|
|
Instead of: .../Mainline/client (which does exist)
|
|
"""
|
|
import os
|
|
|
|
# Use the actual websocket.py file location
|
|
websocket_module = __import__(
|
|
"engine.display.backends.websocket", fromlist=["WebSocketDisplay"]
|
|
)
|
|
websocket_file = websocket_module.__file__
|
|
|
|
# OLD BUGGY CODE: 3 dirname calls
|
|
old_buggy_client_dir = os.path.join(
|
|
os.path.dirname(os.path.dirname(os.path.dirname(websocket_file))), "client"
|
|
)
|
|
|
|
# This path should NOT exist (it's the buggy path)
|
|
assert not os.path.exists(old_buggy_client_dir), (
|
|
f"Old buggy path should not exist: {old_buggy_client_dir}\n"
|
|
f"If this assertion fails, the bug may have been fixed elsewhere or "
|
|
f"the test needs updating."
|
|
)
|
|
|
|
# The buggy path should be .../engine/client, not .../Mainline/client
|
|
assert old_buggy_client_dir.endswith("engine/client"), (
|
|
f"Old buggy path should end with 'engine/client': {old_buggy_client_dir}"
|
|
)
|
|
|
|
# Verify that going up one more level (4 dirname calls) finds the correct path
|
|
correct_client_dir = os.path.join(
|
|
os.path.dirname(
|
|
os.path.dirname(os.path.dirname(os.path.dirname(websocket_file)))
|
|
),
|
|
"client",
|
|
)
|
|
assert os.path.exists(correct_client_dir), (
|
|
f"Correct path should exist: {correct_client_dir}"
|
|
)
|
|
assert "index.html" in os.listdir(correct_client_dir), (
|
|
f"index.html should exist in correct path: {correct_client_dir}"
|
|
)
|