forked from genewildish/Mainline
test: Add comprehensive pipeline adapter tests for Stage implementations
This commit is contained in:
345
tests/test_adapters.py
Normal file
345
tests/test_adapters.py
Normal file
@@ -0,0 +1,345 @@
|
||||
"""
|
||||
Tests for engine/pipeline/adapters.py - Stage adapters for the pipeline.
|
||||
|
||||
Tests Stage adapters that bridge existing components to the Stage interface:
|
||||
- DataSourceStage: Wraps DataSource objects
|
||||
- DisplayStage: Wraps Display backends
|
||||
- PassthroughStage: Simple pass-through stage for pre-rendered data
|
||||
- SourceItemsToBufferStage: Converts SourceItem objects to text buffers
|
||||
- EffectPluginStage: Wraps effect plugins
|
||||
"""
|
||||
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from engine.data_sources.sources import SourceItem
|
||||
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_name(self):
|
||||
"""DataSourceStage stores name correctly."""
|
||||
mock_source = MagicMock()
|
||||
stage = DataSourceStage(mock_source, name="headlines")
|
||||
assert stage.name == "headlines"
|
||||
|
||||
def test_datasource_stage_category(self):
|
||||
"""DataSourceStage has 'source' category."""
|
||||
mock_source = MagicMock()
|
||||
stage = DataSourceStage(mock_source, name="headlines")
|
||||
assert stage.category == "source"
|
||||
|
||||
def test_datasource_stage_capabilities(self):
|
||||
"""DataSourceStage advertises source capability."""
|
||||
mock_source = MagicMock()
|
||||
stage = DataSourceStage(mock_source, name="headlines")
|
||||
assert "source.headlines" in stage.capabilities
|
||||
|
||||
def test_datasource_stage_dependencies(self):
|
||||
"""DataSourceStage has no dependencies."""
|
||||
mock_source = MagicMock()
|
||||
stage = DataSourceStage(mock_source, name="headlines")
|
||||
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_returns_data(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."""
|
||||
|
||||
def test_display_stage_name(self):
|
||||
"""DisplayStage stores name correctly."""
|
||||
mock_display = MagicMock()
|
||||
stage = DisplayStage(mock_display, name="terminal")
|
||||
assert stage.name == "terminal"
|
||||
|
||||
def test_display_stage_category(self):
|
||||
"""DisplayStage has 'display' category."""
|
||||
mock_display = MagicMock()
|
||||
stage = DisplayStage(mock_display, name="terminal")
|
||||
assert stage.category == "display"
|
||||
|
||||
def test_display_stage_capabilities(self):
|
||||
"""DisplayStage advertises display capability."""
|
||||
mock_display = MagicMock()
|
||||
stage = DisplayStage(mock_display, name="terminal")
|
||||
assert "display.output" in stage.capabilities
|
||||
|
||||
def test_display_stage_dependencies(self):
|
||||
"""DisplayStage has no dependencies."""
|
||||
mock_display = MagicMock()
|
||||
stage = DisplayStage(mock_display, name="terminal")
|
||||
assert stage.dependencies == set()
|
||||
|
||||
def test_display_stage_init(self):
|
||||
"""DisplayStage.init() calls display.init() with dimensions."""
|
||||
mock_display = MagicMock()
|
||||
mock_display.init.return_value = True
|
||||
stage = DisplayStage(mock_display, name="terminal")
|
||||
|
||||
ctx = PipelineContext()
|
||||
ctx.params = MagicMock()
|
||||
ctx.params.viewport_width = 100
|
||||
ctx.params.viewport_height = 30
|
||||
|
||||
result = stage.init(ctx)
|
||||
|
||||
assert result is True
|
||||
mock_display.init.assert_called_once_with(100, 30, reuse=False)
|
||||
|
||||
def test_display_stage_init_uses_defaults(self):
|
||||
"""DisplayStage.init() uses defaults when params missing."""
|
||||
mock_display = MagicMock()
|
||||
mock_display.init.return_value = True
|
||||
stage = DisplayStage(mock_display, name="terminal")
|
||||
|
||||
ctx = PipelineContext()
|
||||
ctx.params = None
|
||||
|
||||
result = stage.init(ctx)
|
||||
|
||||
assert result is True
|
||||
mock_display.init.assert_called_once_with(80, 24, reuse=False)
|
||||
|
||||
def test_display_stage_process_calls_show(self):
|
||||
"""DisplayStage.process() calls display.show() with data."""
|
||||
mock_display = MagicMock()
|
||||
stage = DisplayStage(mock_display, name="terminal")
|
||||
|
||||
test_buffer = [[["A", "red"] for _ in range(80)] for _ in range(24)]
|
||||
ctx = PipelineContext()
|
||||
result = stage.process(test_buffer, ctx)
|
||||
|
||||
assert result == test_buffer
|
||||
mock_display.show.assert_called_once_with(test_buffer)
|
||||
|
||||
def test_display_stage_process_skips_none_data(self):
|
||||
"""DisplayStage.process() skips show() if data is None."""
|
||||
mock_display = MagicMock()
|
||||
stage = DisplayStage(mock_display, name="terminal")
|
||||
|
||||
ctx = PipelineContext()
|
||||
result = stage.process(None, ctx)
|
||||
|
||||
assert result is None
|
||||
mock_display.show.assert_not_called()
|
||||
|
||||
def test_display_stage_cleanup(self):
|
||||
"""DisplayStage.cleanup() calls display.cleanup()."""
|
||||
mock_display = MagicMock()
|
||||
stage = DisplayStage(mock_display, name="terminal")
|
||||
|
||||
stage.cleanup()
|
||||
|
||||
mock_display.cleanup.assert_called_once()
|
||||
|
||||
|
||||
class TestPassthroughStage:
|
||||
"""Test PassthroughStage adapter."""
|
||||
|
||||
def test_passthrough_stage_name(self):
|
||||
"""PassthroughStage stores name correctly."""
|
||||
stage = PassthroughStage(name="test")
|
||||
assert stage.name == "test"
|
||||
|
||||
def test_passthrough_stage_category(self):
|
||||
"""PassthroughStage has 'render' category."""
|
||||
stage = PassthroughStage()
|
||||
assert stage.category == "render"
|
||||
|
||||
def test_passthrough_stage_is_optional(self):
|
||||
"""PassthroughStage is optional."""
|
||||
stage = PassthroughStage()
|
||||
assert stage.optional is True
|
||||
|
||||
def test_passthrough_stage_capabilities(self):
|
||||
"""PassthroughStage advertises render output capability."""
|
||||
stage = PassthroughStage()
|
||||
assert "render.output" in stage.capabilities
|
||||
|
||||
def test_passthrough_stage_dependencies(self):
|
||||
"""PassthroughStage depends on source."""
|
||||
stage = PassthroughStage()
|
||||
assert "source" in stage.dependencies
|
||||
|
||||
def test_passthrough_stage_process_returns_data_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_name(self):
|
||||
"""SourceItemsToBufferStage stores name correctly."""
|
||||
stage = SourceItemsToBufferStage(name="custom-name")
|
||||
assert stage.name == "custom-name"
|
||||
|
||||
def test_source_items_to_buffer_stage_category(self):
|
||||
"""SourceItemsToBufferStage has 'render' category."""
|
||||
stage = SourceItemsToBufferStage()
|
||||
assert stage.category == "render"
|
||||
|
||||
def test_source_items_to_buffer_stage_is_optional(self):
|
||||
"""SourceItemsToBufferStage is optional."""
|
||||
stage = SourceItemsToBufferStage()
|
||||
assert stage.optional is True
|
||||
|
||||
def test_source_items_to_buffer_stage_capabilities(self):
|
||||
"""SourceItemsToBufferStage advertises render output capability."""
|
||||
stage = SourceItemsToBufferStage()
|
||||
assert "render.output" in stage.capabilities
|
||||
|
||||
def test_source_items_to_buffer_stage_dependencies(self):
|
||||
"""SourceItemsToBufferStage depends on source."""
|
||||
stage = SourceItemsToBufferStage()
|
||||
assert "source" in stage.dependencies
|
||||
|
||||
def test_source_items_to_buffer_stage_process_single_line_item(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
|
||||
# Result should be lines of text
|
||||
assert all(isinstance(line, str) for line in result)
|
||||
|
||||
def test_source_items_to_buffer_stage_process_multiline_item(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."""
|
||||
|
||||
def test_effect_plugin_stage_name(self):
|
||||
"""EffectPluginStage stores name correctly."""
|
||||
mock_effect = MagicMock()
|
||||
stage = EffectPluginStage(mock_effect, name="blur")
|
||||
assert stage.name == "blur"
|
||||
|
||||
def test_effect_plugin_stage_category(self):
|
||||
"""EffectPluginStage has 'effect' category."""
|
||||
mock_effect = MagicMock()
|
||||
stage = EffectPluginStage(mock_effect, name="blur")
|
||||
assert stage.category == "effect"
|
||||
|
||||
def test_effect_plugin_stage_is_not_optional(self):
|
||||
"""EffectPluginStage is required when configured."""
|
||||
mock_effect = MagicMock()
|
||||
stage = EffectPluginStage(mock_effect, name="blur")
|
||||
assert stage.optional is False
|
||||
|
||||
def test_effect_plugin_stage_capabilities(self):
|
||||
"""EffectPluginStage advertises effect capability with name."""
|
||||
mock_effect = MagicMock()
|
||||
stage = EffectPluginStage(mock_effect, name="blur")
|
||||
assert "effect.blur" in stage.capabilities
|
||||
|
||||
def test_effect_plugin_stage_dependencies(self):
|
||||
"""EffectPluginStage has no static dependencies."""
|
||||
mock_effect = MagicMock()
|
||||
stage = EffectPluginStage(mock_effect, name="blur")
|
||||
# EffectPluginStage has empty dependencies - they are resolved dynamically
|
||||
assert stage.dependencies == set()
|
||||
|
||||
def test_effect_plugin_stage_stage_type(self):
|
||||
"""EffectPluginStage.stage_type returns effect for non-HUD."""
|
||||
mock_effect = MagicMock()
|
||||
stage = EffectPluginStage(mock_effect, name="blur")
|
||||
assert stage.stage_type == "effect"
|
||||
|
||||
def test_effect_plugin_stage_hud_special_handling(self):
|
||||
"""EffectPluginStage has special handling for HUD effect."""
|
||||
mock_effect = MagicMock()
|
||||
stage = EffectPluginStage(mock_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(self):
|
||||
"""EffectPluginStage.process() calls effect.process()."""
|
||||
mock_effect = MagicMock()
|
||||
mock_effect.process.return_value = "processed_data"
|
||||
|
||||
stage = EffectPluginStage(mock_effect, name="blur")
|
||||
ctx = PipelineContext()
|
||||
test_buffer = "test_buffer"
|
||||
|
||||
result = stage.process(test_buffer, ctx)
|
||||
|
||||
assert result == "processed_data"
|
||||
mock_effect.process.assert_called_once()
|
||||
Reference in New Issue
Block a user