forked from genewildish/Mainline
David Gwilliam
7185005f9b
- Add engine/effects/plugins/figment.py (native pipeline implementation) - Add engine/figment_render.py, engine/figment_trigger.py, engine/themes.py - Add 3 SVG assets in figments/ (Mexican/Aztec motif) - Add engine/display/backends/animation_report.py for debugging - Add engine/pipeline/adapters/frame_capture.py for frame capture - Add test-figment preset to presets.toml - Add cairosvg optional dependency to pyproject.toml - Update EffectPluginStage to support is_overlay attribute (for overlay effects) - Add comprehensive tests: test_figment_effect.py, test_figment_pipeline.py, test_figment_render.py - Remove obsolete test_ui_simple.py - Update TODO.md with test cleanup plan - Refactor test_adapters.py to use real components instead of mocks This completes the figment SVG overlay feature integration using the modern pipeline architecture, avoiding legacy effects_plugins. All tests pass (758 total).
273 lines
9.3 KiB
Python
273 lines
9.3 KiB
Python
"""
|
|
Tests for engine/pipeline/adapters.py - Stage adapters for the pipeline.
|
|
|
|
Tests Stage adapters that bridge existing components to the Stage interface.
|
|
Focuses on behavior testing rather than mock interactions.
|
|
"""
|
|
|
|
from unittest.mock import MagicMock
|
|
|
|
from engine.data_sources.sources import SourceItem
|
|
from engine.display.backends.null import NullDisplay
|
|
from engine.effects.plugins import discover_plugins
|
|
from engine.effects.registry import get_registry
|
|
from engine.pipeline.adapters import (
|
|
DataSourceStage,
|
|
DisplayStage,
|
|
EffectPluginStage,
|
|
PassthroughStage,
|
|
SourceItemsToBufferStage,
|
|
)
|
|
from engine.pipeline.core import PipelineContext
|
|
|
|
|
|
class TestDataSourceStage:
|
|
"""Test DataSourceStage adapter."""
|
|
|
|
def test_datasource_stage_properties(self):
|
|
"""DataSourceStage has correct name, category, and capabilities."""
|
|
mock_source = MagicMock()
|
|
stage = DataSourceStage(mock_source, name="headlines")
|
|
|
|
assert stage.name == "headlines"
|
|
assert stage.category == "source"
|
|
assert "source.headlines" in stage.capabilities
|
|
assert stage.dependencies == set()
|
|
|
|
def test_datasource_stage_process_calls_get_items(self):
|
|
"""DataSourceStage.process() calls source.get_items()."""
|
|
mock_items = [
|
|
SourceItem(content="Item 1", source="headlines", timestamp="12:00"),
|
|
]
|
|
mock_source = MagicMock()
|
|
mock_source.get_items.return_value = mock_items
|
|
|
|
stage = DataSourceStage(mock_source, name="headlines")
|
|
ctx = PipelineContext()
|
|
result = stage.process(None, ctx)
|
|
|
|
assert result == mock_items
|
|
mock_source.get_items.assert_called_once()
|
|
|
|
def test_datasource_stage_process_fallback(self):
|
|
"""DataSourceStage.process() returns data if no get_items method."""
|
|
mock_source = MagicMock(spec=[]) # No get_items method
|
|
stage = DataSourceStage(mock_source, name="headlines")
|
|
ctx = PipelineContext()
|
|
test_data = [{"content": "test"}]
|
|
|
|
result = stage.process(test_data, ctx)
|
|
assert result == test_data
|
|
|
|
|
|
class TestDisplayStage:
|
|
"""Test DisplayStage adapter using NullDisplay for real behavior."""
|
|
|
|
def test_display_stage_properties(self):
|
|
"""DisplayStage has correct name, category, and capabilities."""
|
|
display = NullDisplay()
|
|
stage = DisplayStage(display, name="terminal")
|
|
|
|
assert stage.name == "terminal"
|
|
assert stage.category == "display"
|
|
assert "display.output" in stage.capabilities
|
|
assert "render.output" in stage.dependencies
|
|
|
|
def test_display_stage_init_and_process(self):
|
|
"""DisplayStage initializes display and processes buffer."""
|
|
from engine.pipeline.params import PipelineParams
|
|
|
|
display = NullDisplay()
|
|
stage = DisplayStage(display, name="terminal")
|
|
|
|
ctx = PipelineContext()
|
|
ctx.params = PipelineParams()
|
|
ctx.params.viewport_width = 80
|
|
ctx.params.viewport_height = 24
|
|
|
|
# Initialize
|
|
result = stage.init(ctx)
|
|
assert result is True
|
|
|
|
# Process buffer
|
|
buffer = ["Line 1", "Line 2", "Line 3"]
|
|
output = stage.process(buffer, ctx)
|
|
assert output == buffer
|
|
|
|
# Verify display captured the buffer
|
|
assert display._last_buffer == buffer
|
|
|
|
def test_display_stage_skips_none_data(self):
|
|
"""DisplayStage.process() skips show() if data is None."""
|
|
display = NullDisplay()
|
|
stage = DisplayStage(display, name="terminal")
|
|
|
|
ctx = PipelineContext()
|
|
result = stage.process(None, ctx)
|
|
|
|
assert result is None
|
|
assert display._last_buffer is None
|
|
|
|
|
|
class TestPassthroughStage:
|
|
"""Test PassthroughStage adapter."""
|
|
|
|
def test_passthrough_stage_properties(self):
|
|
"""PassthroughStage has correct properties."""
|
|
stage = PassthroughStage(name="test")
|
|
|
|
assert stage.name == "test"
|
|
assert stage.category == "render"
|
|
assert stage.optional is True
|
|
assert "render.output" in stage.capabilities
|
|
assert "source" in stage.dependencies
|
|
|
|
def test_passthrough_stage_process_unchanged(self):
|
|
"""PassthroughStage.process() returns data unchanged."""
|
|
stage = PassthroughStage()
|
|
ctx = PipelineContext()
|
|
|
|
test_data = [
|
|
SourceItem(content="Line 1", source="test", timestamp="12:00"),
|
|
]
|
|
result = stage.process(test_data, ctx)
|
|
|
|
assert result == test_data
|
|
assert result is test_data
|
|
|
|
|
|
class TestSourceItemsToBufferStage:
|
|
"""Test SourceItemsToBufferStage adapter."""
|
|
|
|
def test_source_items_to_buffer_stage_properties(self):
|
|
"""SourceItemsToBufferStage has correct properties."""
|
|
stage = SourceItemsToBufferStage(name="custom-name")
|
|
|
|
assert stage.name == "custom-name"
|
|
assert stage.category == "render"
|
|
assert stage.optional is True
|
|
assert "render.output" in stage.capabilities
|
|
assert "source" in stage.dependencies
|
|
|
|
def test_source_items_to_buffer_stage_process_single_line(self):
|
|
"""SourceItemsToBufferStage converts single-line SourceItem."""
|
|
stage = SourceItemsToBufferStage()
|
|
ctx = PipelineContext()
|
|
|
|
items = [
|
|
SourceItem(content="Single line content", source="test", timestamp="12:00"),
|
|
]
|
|
result = stage.process(items, ctx)
|
|
|
|
assert isinstance(result, list)
|
|
assert len(result) >= 1
|
|
assert all(isinstance(line, str) for line in result)
|
|
assert "Single line content" in result[0]
|
|
|
|
def test_source_items_to_buffer_stage_process_multiline(self):
|
|
"""SourceItemsToBufferStage splits multiline SourceItem content."""
|
|
stage = SourceItemsToBufferStage()
|
|
ctx = PipelineContext()
|
|
|
|
content = "Line 1\nLine 2\nLine 3"
|
|
items = [
|
|
SourceItem(content=content, source="test", timestamp="12:00"),
|
|
]
|
|
result = stage.process(items, ctx)
|
|
|
|
# Should have at least 3 lines
|
|
assert len(result) >= 3
|
|
assert all(isinstance(line, str) for line in result)
|
|
|
|
def test_source_items_to_buffer_stage_process_multiple_items(self):
|
|
"""SourceItemsToBufferStage handles multiple SourceItems."""
|
|
stage = SourceItemsToBufferStage()
|
|
ctx = PipelineContext()
|
|
|
|
items = [
|
|
SourceItem(content="Item 1", source="test", timestamp="12:00"),
|
|
SourceItem(content="Item 2", source="test", timestamp="12:01"),
|
|
SourceItem(content="Item 3", source="test", timestamp="12:02"),
|
|
]
|
|
result = stage.process(items, ctx)
|
|
|
|
# Should have at least 3 lines (one per item, possibly more)
|
|
assert len(result) >= 3
|
|
assert all(isinstance(line, str) for line in result)
|
|
|
|
|
|
class TestEffectPluginStage:
|
|
"""Test EffectPluginStage adapter with real effect plugins."""
|
|
|
|
def test_effect_plugin_stage_properties(self):
|
|
"""EffectPluginStage has correct properties for real effects."""
|
|
discover_plugins()
|
|
registry = get_registry()
|
|
effect = registry.get("noise")
|
|
|
|
stage = EffectPluginStage(effect, name="noise")
|
|
|
|
assert stage.name == "noise"
|
|
assert stage.category == "effect"
|
|
assert stage.optional is False
|
|
assert "effect.noise" in stage.capabilities
|
|
|
|
def test_effect_plugin_stage_hud_special_handling(self):
|
|
"""EffectPluginStage has special handling for HUD effect."""
|
|
discover_plugins()
|
|
registry = get_registry()
|
|
hud_effect = registry.get("hud")
|
|
|
|
stage = EffectPluginStage(hud_effect, name="hud")
|
|
|
|
assert stage.stage_type == "overlay"
|
|
assert stage.is_overlay is True
|
|
assert stage.render_order == 100
|
|
|
|
def test_effect_plugin_stage_process_real_effect(self):
|
|
"""EffectPluginStage.process() calls real effect.process()."""
|
|
from engine.pipeline.params import PipelineParams
|
|
|
|
discover_plugins()
|
|
registry = get_registry()
|
|
effect = registry.get("noise")
|
|
|
|
stage = EffectPluginStage(effect, name="noise")
|
|
ctx = PipelineContext()
|
|
ctx.params = PipelineParams()
|
|
ctx.params.viewport_width = 80
|
|
ctx.params.viewport_height = 24
|
|
ctx.params.frame_number = 0
|
|
|
|
test_buffer = ["Line 1", "Line 2", "Line 3"]
|
|
result = stage.process(test_buffer, ctx)
|
|
|
|
# Should return a list (possibly modified buffer)
|
|
assert isinstance(result, list)
|
|
# Noise effect should preserve line count
|
|
assert len(result) == len(test_buffer)
|
|
|
|
def test_effect_plugin_stage_process_with_real_figment(self):
|
|
"""EffectPluginStage processes figment effect correctly."""
|
|
from engine.pipeline.params import PipelineParams
|
|
|
|
discover_plugins()
|
|
registry = get_registry()
|
|
figment = registry.get("figment")
|
|
|
|
stage = EffectPluginStage(figment, name="figment")
|
|
ctx = PipelineContext()
|
|
ctx.params = PipelineParams()
|
|
ctx.params.viewport_width = 80
|
|
ctx.params.viewport_height = 24
|
|
ctx.params.frame_number = 0
|
|
|
|
test_buffer = ["Line 1", "Line 2", "Line 3"]
|
|
result = stage.process(test_buffer, ctx)
|
|
|
|
# Figment is an overlay effect
|
|
assert stage.is_overlay is True
|
|
assert stage.stage_type == "overlay"
|
|
# Result should be a list
|
|
assert isinstance(result, list)
|