feat(tests): improve coverage to 56%, add benchmark regression tests
- Add EffectPlugin ABC with @abstractmethod decorators for interface enforcement - Add runtime interface checking in discover_plugins() with issubclass() - Add EffectContext factory with sensible defaults - Standardize Display __init__ (remove redundant init in TerminalDisplay) - Document effect behavior when ticker_height=0 - Evaluate legacy effects: document coexistence, no deprecation needed - Research plugin patterns (VST, Python entry points) - Fix pysixel dependency (removed broken dependency) Test coverage improvements: - Add DisplayRegistry tests - Add MultiDisplay tests - Add SixelDisplay tests - Add controller._get_display tests - Add effects controller command handling tests - Add benchmark regression tests (@pytest.mark.benchmark) - Add pytest marker for benchmark tests in pyproject.toml Documentation updates: - Update AGENTS.md with 56% coverage stats and effect plugin docs - Update README.md with Sixel display mode and benchmark commands - Add new modules to architecture section
This commit is contained in:
48
AGENTS.md
48
AGENTS.md
@@ -159,6 +159,31 @@ mise run test-cov
|
||||
|
||||
The project uses pytest with strict marker enforcement. Test configuration is in `pyproject.toml` under `[tool.pytest.ini_options]`.
|
||||
|
||||
### Test Coverage Strategy
|
||||
|
||||
Current coverage: 56% (336 tests)
|
||||
|
||||
Key areas with lower coverage (acceptable for now):
|
||||
- **app.py** (8%): Main entry point - integration heavy, requires terminal
|
||||
- **scroll.py** (10%): Terminal-dependent rendering logic
|
||||
- **benchmark.py** (0%): Standalone benchmark tool, runs separately
|
||||
|
||||
Key areas with good coverage:
|
||||
- **display/backends/null.py** (95%): Easy to test headlessly
|
||||
- **display/backends/terminal.py** (96%): Uses mocking
|
||||
- **display/backends/multi.py** (100%): Simple forwarding logic
|
||||
- **effects/performance.py** (99%): Pure Python logic
|
||||
- **eventbus.py** (96%): Simple event system
|
||||
- **effects/controller.py** (95%): Effects command handling
|
||||
|
||||
Areas needing more tests:
|
||||
- **websocket.py** (48%): Network I/O, hard to test in CI
|
||||
- **ntfy.py** (50%): Network I/O, hard to test in CI
|
||||
- **mic.py** (61%): Audio I/O, hard to test in CI
|
||||
|
||||
Note: Terminal-dependent modules (scroll, layers render) are harder to test in CI.
|
||||
Performance regression tests are in `tests/test_benchmark.py` with `@pytest.mark.benchmark`.
|
||||
|
||||
## Architecture Notes
|
||||
|
||||
- **ntfy.py** and **mic.py** are standalone modules with zero internal dependencies
|
||||
@@ -169,13 +194,15 @@ The project uses pytest with strict marker enforcement. Test configuration is in
|
||||
|
||||
### Display System
|
||||
|
||||
- **Display abstraction** (`engine/display.py`): swap display backends via the Display protocol
|
||||
- `TerminalDisplay` - ANSI terminal output
|
||||
- `WebSocketDisplay` - broadcasts to web clients via WebSocket
|
||||
- `SixelDisplay` - renders to Sixel graphics (pure Python, no C dependency)
|
||||
- `MultiDisplay` - forwards to multiple displays simultaneously
|
||||
- **Display abstraction** (`engine/display/`): swap display backends via the Display protocol
|
||||
- `display/backends/terminal.py` - ANSI terminal output
|
||||
- `display/backends/websocket.py` - broadcasts to web clients via WebSocket
|
||||
- `display/backends/sixel.py` - renders to Sixel graphics (pure Python, no C dependency)
|
||||
- `display/backends/null.py` - headless display for testing
|
||||
- `display/backends/multi.py` - forwards to multiple displays simultaneously
|
||||
- `display/__init__.py` - DisplayRegistry for backend discovery
|
||||
|
||||
- **WebSocket display** (`engine/websocket_display.py`): real-time frame broadcasting to web browsers
|
||||
- **WebSocket display** (`engine/display/backends/websocket.py`): real-time frame broadcasting to web browsers
|
||||
- WebSocket server on port 8765
|
||||
- HTTP server on port 8766 (serves HTML client)
|
||||
- Client at `client/index.html` with ANSI color parsing and fullscreen support
|
||||
@@ -186,6 +213,15 @@ The project uses pytest with strict marker enforcement. Test configuration is in
|
||||
- `sixel` - Sixel graphics in supported terminals (iTerm2, mintty, etc.)
|
||||
- `both` - Terminal + WebSocket simultaneously
|
||||
|
||||
### Effect Plugin System
|
||||
|
||||
- **EffectPlugin ABC** (`engine/effects/types.py`): abstract base class for effects
|
||||
- All effects must inherit from EffectPlugin and implement `process()` and `configure()`
|
||||
- Runtime discovery via `effects_plugins/__init__.py` using `issubclass()` checks
|
||||
|
||||
- **EffectRegistry** (`engine/effects/registry.py`): manages registered effects
|
||||
- **EffectChain** (`engine/effects/chain.py`): chains effects in pipeline order
|
||||
|
||||
### Command & Control
|
||||
|
||||
- C&C uses separate ntfy topics for commands and responses
|
||||
|
||||
42
README.md
42
README.md
@@ -75,6 +75,7 @@ Mainline supports multiple display backends:
|
||||
|
||||
- **Terminal** (`--display terminal`): ANSI terminal output (default)
|
||||
- **WebSocket** (`--display websocket`): Stream to web browser clients
|
||||
- **Sixel** (`--display sixel`): Sixel graphics in supported terminals (iTerm2, mintty)
|
||||
- **Both** (`--display both`): Terminal + WebSocket simultaneously
|
||||
|
||||
WebSocket mode serves a web client at http://localhost:8766 with ANSI color support and fullscreen mode.
|
||||
@@ -131,10 +132,17 @@ engine/
|
||||
translate.py Google Translate wrapper + region detection
|
||||
render.py OTF → half-block pipeline (SSAA, gradient)
|
||||
effects/ plugin architecture for visual effects
|
||||
controller.py handles /effects commands
|
||||
chain.py effect pipeline chaining
|
||||
types.py EffectPlugin ABC, EffectConfig, EffectContext
|
||||
registry.py effect registration and lookup
|
||||
chain.py effect pipeline chaining
|
||||
controller.py handles /effects commands
|
||||
performance.py performance monitoring
|
||||
legacy.py legacy functional effects
|
||||
effects_plugins/ effect plugin implementations
|
||||
noise.py noise effect
|
||||
fade.py fade effect
|
||||
glitch.py glitch effect
|
||||
firehose.py firehose effect
|
||||
fetch.py RSS/Gutenberg fetching + cache
|
||||
ntfy.py NtfyPoller — standalone, zero internal deps
|
||||
mic.py MicMonitor — standalone, graceful fallback
|
||||
@@ -147,8 +155,15 @@ engine/
|
||||
controller.py coordinates ntfy/mic monitoring
|
||||
emitters.py background emitters
|
||||
types.py type definitions
|
||||
display.py Display protocol (Terminal, WebSocket, Multi)
|
||||
websocket_display.py WebSocket server for browser clients
|
||||
display/ Display backend system
|
||||
__init__.py DisplayRegistry, get_monitor
|
||||
backends/
|
||||
terminal.py ANSI terminal display
|
||||
websocket.py WebSocket server for browser clients
|
||||
sixel.py Sixel graphics (pure Python)
|
||||
null.py headless display for testing
|
||||
multi.py forwards to multiple displays
|
||||
benchmark.py performance benchmarking tool
|
||||
```
|
||||
|
||||
---
|
||||
@@ -172,17 +187,23 @@ With [mise](https://mise.jdx.dev/):
|
||||
```bash
|
||||
mise run test # run test suite
|
||||
mise run test-cov # run with coverage report
|
||||
|
||||
mise run lint # ruff check
|
||||
mise run lint-fix # ruff check --fix
|
||||
mise run format # ruff format
|
||||
|
||||
mise run run # terminal display
|
||||
mise run run-websocket # web display only
|
||||
mise run run-sixel # sixel graphics
|
||||
mise run run-both # terminal + web
|
||||
mise run run-client # both + open browser
|
||||
|
||||
mise run cmd # C&C command interface
|
||||
mise run cmd-stats # watch effects stats
|
||||
|
||||
mise run benchmark # run performance benchmarks
|
||||
mise run benchmark-json # save as JSON
|
||||
|
||||
mise run topics-init # initialize ntfy topics
|
||||
```
|
||||
|
||||
@@ -191,8 +212,21 @@ mise run topics-init # initialize ntfy topics
|
||||
```bash
|
||||
uv run pytest
|
||||
uv run pytest --cov=engine --cov-report=term-missing
|
||||
|
||||
# Run with mise
|
||||
mise run test
|
||||
mise run test-cov
|
||||
|
||||
# Run performance benchmarks
|
||||
mise run benchmark
|
||||
mise run benchmark-json
|
||||
|
||||
# Run benchmark hook mode (for CI)
|
||||
uv run python -m engine.benchmark --hook
|
||||
```
|
||||
|
||||
Performance regression tests are in `tests/test_benchmark.py` marked with `@pytest.mark.benchmark`.
|
||||
|
||||
### Linting
|
||||
|
||||
```bash
|
||||
|
||||
@@ -5,6 +5,7 @@ PLUGIN_DIR = Path(__file__).parent
|
||||
|
||||
def discover_plugins():
|
||||
from engine.effects.registry import get_registry
|
||||
from engine.effects.types import EffectPlugin
|
||||
|
||||
registry = get_registry()
|
||||
imported = {}
|
||||
@@ -22,11 +23,13 @@ def discover_plugins():
|
||||
attr = getattr(module, attr_name)
|
||||
if (
|
||||
isinstance(attr, type)
|
||||
and hasattr(attr, "name")
|
||||
and hasattr(attr, "process")
|
||||
and issubclass(attr, EffectPlugin)
|
||||
and attr is not EffectPlugin
|
||||
and attr_name.endswith("Effect")
|
||||
):
|
||||
plugin = attr()
|
||||
if not isinstance(plugin, EffectPlugin):
|
||||
continue
|
||||
registry.register(plugin)
|
||||
imported[plugin.name] = plugin
|
||||
except Exception:
|
||||
|
||||
@@ -3,7 +3,7 @@ import random
|
||||
from engine.effects.types import EffectConfig, EffectContext, EffectPlugin
|
||||
|
||||
|
||||
class FadeEffect:
|
||||
class FadeEffect(EffectPlugin):
|
||||
name = "fade"
|
||||
config = EffectConfig(enabled=True, intensity=1.0)
|
||||
|
||||
@@ -54,5 +54,5 @@ class FadeEffect:
|
||||
i += 1
|
||||
return "".join(result)
|
||||
|
||||
def configure(self, cfg: EffectConfig) -> None:
|
||||
self.config = cfg
|
||||
def configure(self, config: EffectConfig) -> None:
|
||||
self.config = config
|
||||
|
||||
@@ -7,7 +7,7 @@ from engine.sources import FEEDS, POETRY_SOURCES
|
||||
from engine.terminal import C_DIM, G_DIM, G_LO, RST, W_GHOST
|
||||
|
||||
|
||||
class FirehoseEffect:
|
||||
class FirehoseEffect(EffectPlugin):
|
||||
name = "firehose"
|
||||
config = EffectConfig(enabled=True, intensity=1.0)
|
||||
|
||||
@@ -68,5 +68,5 @@ class FirehoseEffect:
|
||||
color = random.choice([G_LO, C_DIM, W_GHOST])
|
||||
return f"{color}{text}{RST}"
|
||||
|
||||
def configure(self, cfg: EffectConfig) -> None:
|
||||
self.config = cfg
|
||||
def configure(self, config: EffectConfig) -> None:
|
||||
self.config = config
|
||||
|
||||
@@ -5,7 +5,7 @@ from engine.effects.types import EffectConfig, EffectContext, EffectPlugin
|
||||
from engine.terminal import C_DIM, DIM, G_DIM, G_LO, RST
|
||||
|
||||
|
||||
class GlitchEffect:
|
||||
class GlitchEffect(EffectPlugin):
|
||||
name = "glitch"
|
||||
config = EffectConfig(enabled=True, intensity=1.0)
|
||||
|
||||
@@ -33,5 +33,5 @@ class GlitchEffect:
|
||||
o = random.randint(0, w - n)
|
||||
return " " * o + f"{G_LO}{DIM}" + c * n + RST
|
||||
|
||||
def configure(self, cfg: EffectConfig) -> None:
|
||||
self.config = cfg
|
||||
def configure(self, config: EffectConfig) -> None:
|
||||
self.config = config
|
||||
|
||||
@@ -5,7 +5,7 @@ from engine.effects.types import EffectConfig, EffectContext, EffectPlugin
|
||||
from engine.terminal import C_DIM, G_DIM, G_LO, RST, W_GHOST
|
||||
|
||||
|
||||
class NoiseEffect:
|
||||
class NoiseEffect(EffectPlugin):
|
||||
name = "noise"
|
||||
config = EffectConfig(enabled=True, intensity=0.15)
|
||||
|
||||
@@ -32,5 +32,5 @@ class NoiseEffect:
|
||||
for _ in range(w)
|
||||
)
|
||||
|
||||
def configure(self, cfg: EffectConfig) -> None:
|
||||
self.config = cfg
|
||||
def configure(self, config: EffectConfig) -> None:
|
||||
self.config = config
|
||||
|
||||
@@ -11,10 +11,6 @@ class TerminalDisplay:
|
||||
width: int = 80
|
||||
height: int = 24
|
||||
|
||||
def __init__(self):
|
||||
self.width = 80
|
||||
self.height = 24
|
||||
|
||||
def init(self, width: int, height: int) -> None:
|
||||
from engine.terminal import CURSOR_OFF
|
||||
|
||||
|
||||
@@ -10,7 +10,12 @@ from engine.effects.legacy import (
|
||||
)
|
||||
from engine.effects.performance import PerformanceMonitor, get_monitor, set_monitor
|
||||
from engine.effects.registry import EffectRegistry, get_registry, set_registry
|
||||
from engine.effects.types import EffectConfig, EffectContext, PipelineConfig
|
||||
from engine.effects.types import (
|
||||
EffectConfig,
|
||||
EffectContext,
|
||||
PipelineConfig,
|
||||
create_effect_context,
|
||||
)
|
||||
|
||||
|
||||
def get_effect_chain():
|
||||
@@ -25,6 +30,7 @@ __all__ = [
|
||||
"EffectConfig",
|
||||
"EffectContext",
|
||||
"PipelineConfig",
|
||||
"create_effect_context",
|
||||
"get_registry",
|
||||
"set_registry",
|
||||
"get_effect_chain",
|
||||
|
||||
@@ -1,6 +1,14 @@
|
||||
"""
|
||||
Visual effects: noise, glitch, fade, ANSI-aware truncation, firehose, headline pool.
|
||||
Depends on: config, terminal, sources.
|
||||
|
||||
These are low-level functional implementations of visual effects. They are used
|
||||
internally by the EffectPlugin system (effects_plugins/*.py) and also directly
|
||||
by layers.py and scroll.py for rendering.
|
||||
|
||||
The plugin system provides a higher-level OOP interface with configuration
|
||||
support, while these legacy functions provide direct functional access.
|
||||
Both systems coexist - there are no current plans to deprecate the legacy functions.
|
||||
"""
|
||||
|
||||
import random
|
||||
|
||||
@@ -1,3 +1,24 @@
|
||||
"""
|
||||
Visual effects type definitions and base classes.
|
||||
|
||||
EffectPlugin Architecture:
|
||||
- Uses ABC (Abstract Base Class) for interface enforcement
|
||||
- Runtime discovery via directory scanning (effects_plugins/)
|
||||
- Configuration via EffectConfig dataclass
|
||||
- Context passed through EffectContext dataclass
|
||||
|
||||
Plugin System Research (see AGENTS.md for references):
|
||||
- VST: Standardized audio interfaces, chaining, presets (FXP/FXB)
|
||||
- Python Entry Points: Namespace packages, importlib.metadata discovery
|
||||
- Shadertoy: Shader-based with uniforms as context
|
||||
|
||||
Current gaps vs industry patterns:
|
||||
- No preset save/load system
|
||||
- No external plugin distribution via entry points
|
||||
- No plugin metadata (version, author, description)
|
||||
"""
|
||||
|
||||
from abc import ABC, abstractmethod
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any
|
||||
|
||||
@@ -22,15 +43,76 @@ class EffectConfig:
|
||||
params: dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
|
||||
class EffectPlugin:
|
||||
class EffectPlugin(ABC):
|
||||
"""Abstract base class for effect plugins.
|
||||
|
||||
Subclasses must define:
|
||||
- name: str - unique identifier for the effect
|
||||
- config: EffectConfig - current configuration
|
||||
|
||||
And implement:
|
||||
- process(buf, ctx) -> list[str]
|
||||
- configure(config) -> None
|
||||
|
||||
Effect Behavior with ticker_height=0:
|
||||
- NoiseEffect: Returns buffer unchanged (no ticker to apply noise to)
|
||||
- FadeEffect: Returns buffer unchanged (no ticker to fade)
|
||||
- GlitchEffect: Processes normally (doesn't depend on ticker_height)
|
||||
- FirehoseEffect: Returns buffer unchanged if no items in context
|
||||
|
||||
Effects should handle missing or zero context values gracefully by
|
||||
returning the input buffer unchanged rather than raising errors.
|
||||
"""
|
||||
|
||||
name: str
|
||||
config: EffectConfig
|
||||
|
||||
@abstractmethod
|
||||
def process(self, buf: list[str], ctx: EffectContext) -> list[str]:
|
||||
raise NotImplementedError
|
||||
"""Process the buffer with this effect applied.
|
||||
|
||||
Args:
|
||||
buf: List of lines to process
|
||||
ctx: Effect context with terminal state
|
||||
|
||||
Returns:
|
||||
Processed buffer (may be same object or new list)
|
||||
"""
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
def configure(self, config: EffectConfig) -> None:
|
||||
raise NotImplementedError
|
||||
"""Configure the effect with new settings.
|
||||
|
||||
Args:
|
||||
config: New configuration to apply
|
||||
"""
|
||||
...
|
||||
|
||||
|
||||
def create_effect_context(
|
||||
terminal_width: int = 80,
|
||||
terminal_height: int = 24,
|
||||
scroll_cam: int = 0,
|
||||
ticker_height: int = 0,
|
||||
mic_excess: float = 0.0,
|
||||
grad_offset: float = 0.0,
|
||||
frame_number: int = 0,
|
||||
has_message: bool = False,
|
||||
items: list | None = None,
|
||||
) -> EffectContext:
|
||||
"""Factory function to create EffectContext with sensible defaults."""
|
||||
return EffectContext(
|
||||
terminal_width=terminal_width,
|
||||
terminal_height=terminal_height,
|
||||
scroll_cam=scroll_cam,
|
||||
ticker_height=ticker_height,
|
||||
mic_excess=mic_excess,
|
||||
grad_offset=grad_offset,
|
||||
frame_number=frame_number,
|
||||
has_message=has_message,
|
||||
items=items or [],
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
|
||||
@@ -70,6 +70,9 @@ addopts = [
|
||||
"--tb=short",
|
||||
"-v",
|
||||
]
|
||||
markers = [
|
||||
"benchmark: marks tests as performance benchmarks (may be slow)",
|
||||
]
|
||||
filterwarnings = [
|
||||
"ignore::DeprecationWarning",
|
||||
]
|
||||
|
||||
100
tests/test_benchmark.py
Normal file
100
tests/test_benchmark.py
Normal file
@@ -0,0 +1,100 @@
|
||||
"""
|
||||
Tests for engine.benchmark module - performance regression tests.
|
||||
"""
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from engine.display import NullDisplay
|
||||
|
||||
|
||||
class TestBenchmarkNullDisplay:
|
||||
"""Performance tests for NullDisplay - regression tests."""
|
||||
|
||||
@pytest.mark.benchmark
|
||||
def test_null_display_minimum_fps(self):
|
||||
"""NullDisplay should meet minimum performance threshold."""
|
||||
import time
|
||||
|
||||
display = NullDisplay()
|
||||
display.init(80, 24)
|
||||
buffer = ["x" * 80 for _ in range(24)]
|
||||
|
||||
iterations = 1000
|
||||
start = time.perf_counter()
|
||||
for _ in range(iterations):
|
||||
display.show(buffer)
|
||||
elapsed = time.perf_counter() - start
|
||||
|
||||
fps = iterations / elapsed
|
||||
min_fps = 20000
|
||||
|
||||
assert fps >= min_fps, f"NullDisplay FPS {fps:.0f} below minimum {min_fps}"
|
||||
|
||||
@pytest.mark.benchmark
|
||||
def test_effects_minimum_throughput(self):
|
||||
"""Effects should meet minimum processing throughput."""
|
||||
import time
|
||||
|
||||
from effects_plugins import discover_plugins
|
||||
from engine.effects import EffectContext, get_registry
|
||||
|
||||
discover_plugins()
|
||||
registry = get_registry()
|
||||
effect = registry.get("noise")
|
||||
assert effect is not None, "Noise effect should be registered"
|
||||
|
||||
buffer = ["x" * 80 for _ in range(24)]
|
||||
ctx = EffectContext(
|
||||
terminal_width=80,
|
||||
terminal_height=24,
|
||||
scroll_cam=0,
|
||||
ticker_height=20,
|
||||
mic_excess=0.0,
|
||||
grad_offset=0.0,
|
||||
frame_number=0,
|
||||
has_message=False,
|
||||
)
|
||||
|
||||
iterations = 500
|
||||
start = time.perf_counter()
|
||||
for _ in range(iterations):
|
||||
effect.process(buffer, ctx)
|
||||
elapsed = time.perf_counter() - start
|
||||
|
||||
fps = iterations / elapsed
|
||||
min_fps = 10000
|
||||
|
||||
assert fps >= min_fps, (
|
||||
f"Effect processing FPS {fps:.0f} below minimum {min_fps}"
|
||||
)
|
||||
|
||||
|
||||
class TestBenchmarkWebSocketDisplay:
|
||||
"""Performance tests for WebSocketDisplay."""
|
||||
|
||||
@pytest.mark.benchmark
|
||||
def test_websocket_display_minimum_fps(self):
|
||||
"""WebSocketDisplay should meet minimum performance threshold."""
|
||||
import time
|
||||
|
||||
with patch("engine.display.backends.websocket.websockets", None):
|
||||
from engine.display import WebSocketDisplay
|
||||
|
||||
display = WebSocketDisplay()
|
||||
display.init(80, 24)
|
||||
buffer = ["x" * 80 for _ in range(24)]
|
||||
|
||||
iterations = 500
|
||||
start = time.perf_counter()
|
||||
for _ in range(iterations):
|
||||
display.show(buffer)
|
||||
elapsed = time.perf_counter() - start
|
||||
|
||||
fps = iterations / elapsed
|
||||
min_fps = 10000
|
||||
|
||||
assert fps >= min_fps, (
|
||||
f"WebSocketDisplay FPS {fps:.0f} below minimum {min_fps}"
|
||||
)
|
||||
@@ -5,7 +5,75 @@ Tests for engine.controller module.
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from engine import config
|
||||
from engine.controller import StreamController
|
||||
from engine.controller import StreamController, _get_display
|
||||
|
||||
|
||||
class TestGetDisplay:
|
||||
"""Tests for _get_display function."""
|
||||
|
||||
@patch("engine.controller.WebSocketDisplay")
|
||||
@patch("engine.controller.TerminalDisplay")
|
||||
def test_get_display_terminal(self, mock_terminal, mock_ws):
|
||||
"""returns TerminalDisplay for display=terminal."""
|
||||
mock_terminal.return_value = MagicMock()
|
||||
mock_ws.return_value = MagicMock()
|
||||
|
||||
cfg = config.Config(display="terminal")
|
||||
display = _get_display(cfg)
|
||||
|
||||
mock_terminal.assert_called()
|
||||
assert isinstance(display, MagicMock)
|
||||
|
||||
@patch("engine.controller.WebSocketDisplay")
|
||||
@patch("engine.controller.TerminalDisplay")
|
||||
def test_get_display_websocket(self, mock_terminal, mock_ws):
|
||||
"""returns WebSocketDisplay for display=websocket."""
|
||||
mock_ws_instance = MagicMock()
|
||||
mock_ws.return_value = mock_ws_instance
|
||||
mock_terminal.return_value = MagicMock()
|
||||
|
||||
cfg = config.Config(display="websocket")
|
||||
_get_display(cfg)
|
||||
|
||||
mock_ws.assert_called()
|
||||
mock_ws_instance.start_server.assert_called()
|
||||
mock_ws_instance.start_http_server.assert_called()
|
||||
|
||||
@patch("engine.controller.SixelDisplay")
|
||||
def test_get_display_sixel(self, mock_sixel):
|
||||
"""returns SixelDisplay for display=sixel."""
|
||||
mock_sixel.return_value = MagicMock()
|
||||
cfg = config.Config(display="sixel")
|
||||
_get_display(cfg)
|
||||
|
||||
mock_sixel.assert_called()
|
||||
|
||||
def test_get_display_unknown_returns_null(self):
|
||||
"""returns NullDisplay for unknown display mode."""
|
||||
cfg = config.Config(display="unknown")
|
||||
display = _get_display(cfg)
|
||||
|
||||
from engine.display import NullDisplay
|
||||
|
||||
assert isinstance(display, NullDisplay)
|
||||
|
||||
@patch("engine.controller.WebSocketDisplay")
|
||||
@patch("engine.controller.TerminalDisplay")
|
||||
@patch("engine.controller.MultiDisplay")
|
||||
def test_get_display_both(self, mock_multi, mock_terminal, mock_ws):
|
||||
"""returns MultiDisplay for display=both."""
|
||||
mock_terminal_instance = MagicMock()
|
||||
mock_ws_instance = MagicMock()
|
||||
mock_terminal.return_value = mock_terminal_instance
|
||||
mock_ws.return_value = mock_ws_instance
|
||||
|
||||
cfg = config.Config(display="both")
|
||||
_get_display(cfg)
|
||||
|
||||
mock_multi.assert_called()
|
||||
call_args = mock_multi.call_args[0][0]
|
||||
assert mock_terminal_instance in call_args
|
||||
assert mock_ws_instance in call_args
|
||||
|
||||
|
||||
class TestStreamController:
|
||||
@@ -68,6 +136,24 @@ class TestStreamController:
|
||||
assert mic_ok is False
|
||||
assert ntfy_ok is True
|
||||
|
||||
@patch("engine.controller.MicMonitor")
|
||||
def test_initialize_sources_cc_subscribed(self, mock_mic):
|
||||
"""initialize_sources subscribes C&C handler."""
|
||||
mock_mic_instance = MagicMock()
|
||||
mock_mic_instance.available = False
|
||||
mock_mic_instance.start.return_value = False
|
||||
mock_mic.return_value = mock_mic_instance
|
||||
|
||||
with patch("engine.controller.NtfyPoller") as mock_ntfy:
|
||||
mock_ntfy_instance = MagicMock()
|
||||
mock_ntfy_instance.start.return_value = True
|
||||
mock_ntfy.return_value = mock_ntfy_instance
|
||||
|
||||
controller = StreamController()
|
||||
controller.initialize_sources()
|
||||
|
||||
mock_ntfy_instance.subscribe.assert_called()
|
||||
|
||||
|
||||
class TestStreamControllerCleanup:
|
||||
"""Tests for StreamController cleanup."""
|
||||
|
||||
@@ -2,7 +2,10 @@
|
||||
Tests for engine.display module.
|
||||
"""
|
||||
|
||||
from engine.display import NullDisplay, TerminalDisplay
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from engine.display import DisplayRegistry, NullDisplay, TerminalDisplay
|
||||
from engine.display.backends.multi import MultiDisplay
|
||||
|
||||
|
||||
class TestDisplayProtocol:
|
||||
@@ -25,6 +28,66 @@ class TestDisplayProtocol:
|
||||
assert hasattr(display, "cleanup")
|
||||
|
||||
|
||||
class TestDisplayRegistry:
|
||||
"""Tests for DisplayRegistry class."""
|
||||
|
||||
def setup_method(self):
|
||||
"""Reset registry before each test."""
|
||||
DisplayRegistry._backends = {}
|
||||
DisplayRegistry._initialized = False
|
||||
|
||||
def test_register_adds_backend(self):
|
||||
"""register adds a backend to the registry."""
|
||||
DisplayRegistry.register("test", TerminalDisplay)
|
||||
assert DisplayRegistry.get("test") == TerminalDisplay
|
||||
|
||||
def test_register_case_insensitive(self):
|
||||
"""register is case insensitive."""
|
||||
DisplayRegistry.register("TEST", TerminalDisplay)
|
||||
assert DisplayRegistry.get("test") == TerminalDisplay
|
||||
|
||||
def test_get_returns_none_for_unknown(self):
|
||||
"""get returns None for unknown backend."""
|
||||
assert DisplayRegistry.get("unknown") is None
|
||||
|
||||
def test_list_backends_returns_all(self):
|
||||
"""list_backends returns all registered backends."""
|
||||
DisplayRegistry.register("a", TerminalDisplay)
|
||||
DisplayRegistry.register("b", NullDisplay)
|
||||
backends = DisplayRegistry.list_backends()
|
||||
assert "a" in backends
|
||||
assert "b" in backends
|
||||
|
||||
def test_create_returns_instance(self):
|
||||
"""create returns a display instance."""
|
||||
DisplayRegistry.register("test", NullDisplay)
|
||||
display = DisplayRegistry.create("test")
|
||||
assert isinstance(display, NullDisplay)
|
||||
|
||||
def test_create_returns_none_for_unknown(self):
|
||||
"""create returns None for unknown backend."""
|
||||
display = DisplayRegistry.create("unknown")
|
||||
assert display is None
|
||||
|
||||
def test_initialize_registers_defaults(self):
|
||||
"""initialize registers default backends."""
|
||||
DisplayRegistry.initialize()
|
||||
assert DisplayRegistry.get("terminal") == TerminalDisplay
|
||||
assert DisplayRegistry.get("null") == NullDisplay
|
||||
from engine.display.backends.sixel import SixelDisplay
|
||||
from engine.display.backends.websocket import WebSocketDisplay
|
||||
|
||||
assert DisplayRegistry.get("websocket") == WebSocketDisplay
|
||||
assert DisplayRegistry.get("sixel") == SixelDisplay
|
||||
|
||||
def test_initialize_idempotent(self):
|
||||
"""initialize can be called multiple times safely."""
|
||||
DisplayRegistry.initialize()
|
||||
DisplayRegistry._backends["custom"] = TerminalDisplay
|
||||
DisplayRegistry.initialize()
|
||||
assert "custom" in DisplayRegistry.list_backends()
|
||||
|
||||
|
||||
class TestTerminalDisplay:
|
||||
"""Tests for TerminalDisplay class."""
|
||||
|
||||
@@ -77,3 +140,62 @@ class TestNullDisplay:
|
||||
"""cleanup does nothing."""
|
||||
display = NullDisplay()
|
||||
display.cleanup()
|
||||
|
||||
|
||||
class TestMultiDisplay:
|
||||
"""Tests for MultiDisplay class."""
|
||||
|
||||
def test_init_stores_dimensions(self):
|
||||
"""init stores dimensions and forwards to displays."""
|
||||
mock_display1 = MagicMock()
|
||||
mock_display2 = MagicMock()
|
||||
multi = MultiDisplay([mock_display1, mock_display2])
|
||||
|
||||
multi.init(120, 40)
|
||||
|
||||
assert multi.width == 120
|
||||
assert multi.height == 40
|
||||
mock_display1.init.assert_called_once_with(120, 40)
|
||||
mock_display2.init.assert_called_once_with(120, 40)
|
||||
|
||||
def test_show_forwards_to_all_displays(self):
|
||||
"""show forwards buffer to all displays."""
|
||||
mock_display1 = MagicMock()
|
||||
mock_display2 = MagicMock()
|
||||
multi = MultiDisplay([mock_display1, mock_display2])
|
||||
|
||||
buffer = ["line1", "line2"]
|
||||
multi.show(buffer)
|
||||
|
||||
mock_display1.show.assert_called_once_with(buffer)
|
||||
mock_display2.show.assert_called_once_with(buffer)
|
||||
|
||||
def test_clear_forwards_to_all_displays(self):
|
||||
"""clear forwards to all displays."""
|
||||
mock_display1 = MagicMock()
|
||||
mock_display2 = MagicMock()
|
||||
multi = MultiDisplay([mock_display1, mock_display2])
|
||||
|
||||
multi.clear()
|
||||
|
||||
mock_display1.clear.assert_called_once()
|
||||
mock_display2.clear.assert_called_once()
|
||||
|
||||
def test_cleanup_forwards_to_all_displays(self):
|
||||
"""cleanup forwards to all displays."""
|
||||
mock_display1 = MagicMock()
|
||||
mock_display2 = MagicMock()
|
||||
multi = MultiDisplay([mock_display1, mock_display2])
|
||||
|
||||
multi.cleanup()
|
||||
|
||||
mock_display1.cleanup.assert_called_once()
|
||||
mock_display2.cleanup.assert_called_once()
|
||||
|
||||
def test_empty_displays_list(self):
|
||||
"""handles empty displays list gracefully."""
|
||||
multi = MultiDisplay([])
|
||||
multi.init(80, 24)
|
||||
multi.show(["test"])
|
||||
multi.clear()
|
||||
multi.cleanup()
|
||||
|
||||
@@ -5,8 +5,10 @@ Tests for engine.effects.controller module.
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from engine.effects.controller import (
|
||||
_format_stats,
|
||||
handle_effects_command,
|
||||
set_effect_chain_ref,
|
||||
show_effects_menu,
|
||||
)
|
||||
|
||||
|
||||
@@ -92,6 +94,29 @@ class TestHandleEffectsCommand:
|
||||
assert "Reordered pipeline" in result
|
||||
mock_chain_instance.reorder.assert_called_once_with(["noise", "fade"])
|
||||
|
||||
def test_reorder_failure(self):
|
||||
"""reorder returns error on failure."""
|
||||
with patch("engine.effects.controller.get_registry") as mock_registry:
|
||||
mock_registry.return_value.list_all.return_value = {}
|
||||
|
||||
with patch("engine.effects.controller._get_effect_chain") as mock_chain:
|
||||
mock_chain_instance = MagicMock()
|
||||
mock_chain_instance.reorder.return_value = False
|
||||
mock_chain.return_value = mock_chain_instance
|
||||
|
||||
result = handle_effects_command("/effects reorder bad")
|
||||
|
||||
assert "Failed to reorder" in result
|
||||
|
||||
def test_unknown_effect(self):
|
||||
"""unknown effect returns error."""
|
||||
with patch("engine.effects.controller.get_registry") as mock_registry:
|
||||
mock_registry.return_value.list_all.return_value = {}
|
||||
|
||||
result = handle_effects_command("/effects unknown on")
|
||||
|
||||
assert "Unknown effect" in result
|
||||
|
||||
def test_unknown_command(self):
|
||||
"""unknown command returns error."""
|
||||
result = handle_effects_command("/unknown")
|
||||
@@ -102,6 +127,105 @@ class TestHandleEffectsCommand:
|
||||
result = handle_effects_command("not a command")
|
||||
assert "Unknown command" in result
|
||||
|
||||
def test_invalid_intensity_value(self):
|
||||
"""invalid intensity value returns error."""
|
||||
with patch("engine.effects.controller.get_registry") as mock_registry:
|
||||
mock_plugin = MagicMock()
|
||||
mock_registry.return_value.get.return_value = mock_plugin
|
||||
mock_registry.return_value.list_all.return_value = {"noise": mock_plugin}
|
||||
|
||||
result = handle_effects_command("/effects noise intensity bad")
|
||||
|
||||
assert "Invalid intensity" in result
|
||||
|
||||
def test_missing_action(self):
|
||||
"""missing action returns usage."""
|
||||
with patch("engine.effects.controller.get_registry") as mock_registry:
|
||||
mock_plugin = MagicMock()
|
||||
mock_registry.return_value.get.return_value = mock_plugin
|
||||
mock_registry.return_value.list_all.return_value = {"noise": mock_plugin}
|
||||
|
||||
result = handle_effects_command("/effects noise")
|
||||
|
||||
assert "Usage" in result
|
||||
|
||||
def test_stats_command(self):
|
||||
"""stats command returns formatted stats."""
|
||||
with patch("engine.effects.controller.get_monitor") as mock_monitor:
|
||||
mock_monitor.return_value.get_stats.return_value = {
|
||||
"frame_count": 100,
|
||||
"pipeline": {"avg_ms": 1.5, "min_ms": 1.0, "max_ms": 2.0},
|
||||
"effects": {},
|
||||
}
|
||||
|
||||
result = handle_effects_command("/effects stats")
|
||||
|
||||
assert "Performance Stats" in result
|
||||
|
||||
def test_list_only_effects(self):
|
||||
"""list command works with just /effects."""
|
||||
with patch("engine.effects.controller.get_registry") as mock_registry:
|
||||
mock_plugin = MagicMock()
|
||||
mock_plugin.config.enabled = False
|
||||
mock_plugin.config.intensity = 0.5
|
||||
mock_registry.return_value.list_all.return_value = {"noise": mock_plugin}
|
||||
|
||||
with patch("engine.effects.controller._get_effect_chain") as mock_chain:
|
||||
mock_chain.return_value = None
|
||||
|
||||
result = handle_effects_command("/effects")
|
||||
|
||||
assert "noise: OFF" in result
|
||||
|
||||
|
||||
class TestShowEffectsMenu:
|
||||
"""Tests for show_effects_menu function."""
|
||||
|
||||
def test_returns_formatted_menu(self):
|
||||
"""returns formatted effects menu."""
|
||||
with patch("engine.effects.controller.get_registry") as mock_registry:
|
||||
mock_plugin = MagicMock()
|
||||
mock_plugin.config.enabled = True
|
||||
mock_plugin.config.intensity = 0.75
|
||||
mock_registry.return_value.list_all.return_value = {"noise": mock_plugin}
|
||||
|
||||
with patch("engine.effects.controller._get_effect_chain") as mock_chain:
|
||||
mock_chain_instance = MagicMock()
|
||||
mock_chain_instance.get_order.return_value = ["noise"]
|
||||
mock_chain.return_value = mock_chain_instance
|
||||
|
||||
result = show_effects_menu()
|
||||
|
||||
assert "EFFECTS MENU" in result
|
||||
assert "noise" in result
|
||||
|
||||
|
||||
class TestFormatStats:
|
||||
"""Tests for _format_stats function."""
|
||||
|
||||
def test_returns_error_when_no_monitor(self):
|
||||
"""returns error when monitor unavailable."""
|
||||
with patch("engine.effects.controller.get_monitor") as mock_monitor:
|
||||
mock_monitor.return_value.get_stats.return_value = {"error": "No data"}
|
||||
|
||||
result = _format_stats()
|
||||
|
||||
assert "No data" in result
|
||||
|
||||
def test_formats_pipeline_stats(self):
|
||||
"""formats pipeline stats correctly."""
|
||||
with patch("engine.effects.controller.get_monitor") as mock_monitor:
|
||||
mock_monitor.return_value.get_stats.return_value = {
|
||||
"frame_count": 50,
|
||||
"pipeline": {"avg_ms": 2.5, "min_ms": 2.0, "max_ms": 3.0},
|
||||
"effects": {"noise": {"avg_ms": 0.5, "min_ms": 0.4, "max_ms": 0.6}},
|
||||
}
|
||||
|
||||
result = _format_stats()
|
||||
|
||||
assert "Pipeline" in result
|
||||
assert "noise" in result
|
||||
|
||||
|
||||
class TestSetEffectChainRef:
|
||||
"""Tests for set_effect_chain_ref function."""
|
||||
|
||||
123
tests/test_sixel.py
Normal file
123
tests/test_sixel.py
Normal file
@@ -0,0 +1,123 @@
|
||||
"""
|
||||
Tests for engine.display.backends.sixel module.
|
||||
"""
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
|
||||
class TestSixelDisplay:
|
||||
"""Tests for SixelDisplay class."""
|
||||
|
||||
def test_init_stores_dimensions(self):
|
||||
"""init stores dimensions."""
|
||||
from engine.display.backends.sixel import SixelDisplay
|
||||
|
||||
display = SixelDisplay()
|
||||
display.init(80, 24)
|
||||
assert display.width == 80
|
||||
assert display.height == 24
|
||||
|
||||
def test_init_custom_cell_size(self):
|
||||
"""init accepts custom cell size."""
|
||||
from engine.display.backends.sixel import SixelDisplay
|
||||
|
||||
display = SixelDisplay(cell_width=12, cell_height=18)
|
||||
assert display.cell_width == 12
|
||||
assert display.cell_height == 18
|
||||
|
||||
def test_show_handles_empty_buffer(self):
|
||||
"""show handles empty buffer gracefully."""
|
||||
from engine.display.backends.sixel import SixelDisplay
|
||||
|
||||
display = SixelDisplay()
|
||||
display.init(80, 24)
|
||||
|
||||
with patch("engine.display.backends.sixel._encode_sixel") as mock_encode:
|
||||
mock_encode.return_value = ""
|
||||
display.show([])
|
||||
|
||||
def test_show_handles_pil_import_error(self):
|
||||
"""show gracefully handles missing PIL."""
|
||||
from engine.display.backends.sixel import SixelDisplay
|
||||
|
||||
display = SixelDisplay()
|
||||
display.init(80, 24)
|
||||
|
||||
with patch.dict("sys.modules", {"PIL": None}):
|
||||
display.show(["test line"])
|
||||
|
||||
def test_clear_sends_escape_sequence(self):
|
||||
"""clear sends clear screen escape sequence."""
|
||||
from engine.display.backends.sixel import SixelDisplay
|
||||
|
||||
display = SixelDisplay()
|
||||
|
||||
with patch("sys.stdout") as mock_stdout:
|
||||
display.clear()
|
||||
mock_stdout.buffer.write.assert_called()
|
||||
|
||||
def test_cleanup_does_nothing(self):
|
||||
"""cleanup does nothing."""
|
||||
from engine.display.backends.sixel import SixelDisplay
|
||||
|
||||
display = SixelDisplay()
|
||||
display.cleanup()
|
||||
|
||||
|
||||
class TestSixelAnsiParsing:
|
||||
"""Tests for ANSI parsing in SixelDisplay."""
|
||||
|
||||
def test_parse_empty_string(self):
|
||||
"""handles empty string."""
|
||||
from engine.display.backends.sixel import _parse_ansi
|
||||
|
||||
result = _parse_ansi("")
|
||||
assert len(result) > 0
|
||||
|
||||
def test_parse_plain_text(self):
|
||||
"""parses plain text without ANSI codes."""
|
||||
from engine.display.backends.sixel import _parse_ansi
|
||||
|
||||
result = _parse_ansi("hello world")
|
||||
assert len(result) == 1
|
||||
text, fg, bg, bold = result[0]
|
||||
assert text == "hello world"
|
||||
|
||||
def test_parse_with_color_codes(self):
|
||||
"""parses ANSI color codes."""
|
||||
from engine.display.backends.sixel import _parse_ansi
|
||||
|
||||
result = _parse_ansi("\033[31mred\033[0m")
|
||||
assert len(result) == 2
|
||||
|
||||
def test_parse_with_bold(self):
|
||||
"""parses bold codes."""
|
||||
from engine.display.backends.sixel import _parse_ansi
|
||||
|
||||
result = _parse_ansi("\033[1mbold\033[0m")
|
||||
assert len(result) == 2
|
||||
|
||||
def test_parse_256_color(self):
|
||||
"""parses 256 color codes."""
|
||||
from engine.display.backends.sixel import _parse_ansi
|
||||
|
||||
result = _parse_ansi("\033[38;5;196mred\033[0m")
|
||||
assert len(result) == 2
|
||||
|
||||
|
||||
class TestSixelEncoding:
|
||||
"""Tests for Sixel encoding."""
|
||||
|
||||
def test_encode_empty_image(self):
|
||||
"""handles empty image."""
|
||||
from engine.display.backends.sixel import _encode_sixel
|
||||
|
||||
with patch("PIL.Image.Image") as mock_image:
|
||||
mock_img_instance = MagicMock()
|
||||
mock_img_instance.convert.return_value = mock_img_instance
|
||||
mock_img_instance.size = (0, 0)
|
||||
mock_img_instance.load.return_value = {}
|
||||
mock_image.return_value = mock_img_instance
|
||||
|
||||
result = _encode_sixel(mock_img_instance)
|
||||
assert result == ""
|
||||
Reference in New Issue
Block a user