David Gwilliam
7c26150408
- tests/test_canvas.py: 33 tests for Canvas (2D rendering surface) - tests/test_firehose.py: 5 tests for FirehoseEffect - tests/test_pipeline_order.py: 3 tests for execution order verification - tests/test_renderer.py: 22 tests for ANSI parsing and PIL rendering These tests provide solid coverage for foundational modules.
119 lines
4.0 KiB
Python
119 lines
4.0 KiB
Python
"""Tests for pipeline execution order verification."""
|
|
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
from engine.pipeline import Pipeline, Stage, discover_stages
|
|
from engine.pipeline.core import DataType
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_registry():
|
|
"""Reset stage registry before each test."""
|
|
from engine.pipeline.registry import StageRegistry
|
|
|
|
StageRegistry._discovered = False
|
|
StageRegistry._categories.clear()
|
|
StageRegistry._instances.clear()
|
|
discover_stages()
|
|
yield
|
|
StageRegistry._discovered = False
|
|
StageRegistry._categories.clear()
|
|
StageRegistry._instances.clear()
|
|
|
|
|
|
def _create_mock_stage(name: str, category: str, capabilities: set, dependencies: set):
|
|
"""Helper to create a mock stage."""
|
|
mock = MagicMock(spec=Stage)
|
|
mock.name = name
|
|
mock.category = category
|
|
mock.stage_type = category
|
|
mock.render_order = 0
|
|
mock.is_overlay = False
|
|
mock.inlet_types = {DataType.ANY}
|
|
mock.outlet_types = {DataType.TEXT_BUFFER}
|
|
mock.capabilities = capabilities
|
|
mock.dependencies = dependencies
|
|
mock.process = lambda data, ctx: data
|
|
mock.init = MagicMock(return_value=True)
|
|
mock.cleanup = MagicMock()
|
|
mock.is_enabled = MagicMock(return_value=True)
|
|
mock.set_enabled = MagicMock()
|
|
mock._enabled = True
|
|
return mock
|
|
|
|
|
|
def test_pipeline_execution_order_linear():
|
|
"""Verify stages execute in linear order based on dependencies."""
|
|
pipeline = Pipeline()
|
|
pipeline.build(auto_inject=False)
|
|
|
|
source = _create_mock_stage("source", "source", {"source"}, set())
|
|
render = _create_mock_stage("render", "render", {"render"}, {"source"})
|
|
effect = _create_mock_stage("effect", "effect", {"effect"}, {"render"})
|
|
display = _create_mock_stage("display", "display", {"display"}, {"effect"})
|
|
|
|
pipeline.add_stage("source", source, initialize=False)
|
|
pipeline.add_stage("render", render, initialize=False)
|
|
pipeline.add_stage("effect", effect, initialize=False)
|
|
pipeline.add_stage("display", display, initialize=False)
|
|
|
|
pipeline._rebuild()
|
|
|
|
assert pipeline.execution_order == [
|
|
"source",
|
|
"render",
|
|
"effect",
|
|
"display",
|
|
]
|
|
|
|
|
|
def test_pipeline_effects_chain_order():
|
|
"""Verify effects execute in config order when chained."""
|
|
pipeline = Pipeline()
|
|
pipeline.build(auto_inject=False)
|
|
|
|
# Source and render
|
|
source = _create_mock_stage("source", "source", {"source"}, set())
|
|
render = _create_mock_stage("render", "render", {"render"}, {"source"})
|
|
|
|
# Effects chain: effect_a → effect_b → effect_c
|
|
effect_a = _create_mock_stage("effect_a", "effect", {"effect_a"}, {"render"})
|
|
effect_b = _create_mock_stage("effect_b", "effect", {"effect_b"}, {"effect_a"})
|
|
effect_c = _create_mock_stage("effect_c", "effect", {"effect_c"}, {"effect_b"})
|
|
|
|
# Display
|
|
display = _create_mock_stage("display", "display", {"display"}, {"effect_c"})
|
|
|
|
for stage in [source, render, effect_a, effect_b, effect_c, display]:
|
|
pipeline.add_stage(stage.name, stage, initialize=False)
|
|
|
|
pipeline._rebuild()
|
|
|
|
effect_names = [
|
|
name for name in pipeline.execution_order if name.startswith("effect_")
|
|
]
|
|
assert effect_names == ["effect_a", "effect_b", "effect_c"]
|
|
|
|
|
|
def test_pipeline_overlay_executes_after_regular_effects():
|
|
"""Overlay stages should execute after all regular effects."""
|
|
pipeline = Pipeline()
|
|
pipeline.build(auto_inject=False)
|
|
|
|
effect = _create_mock_stage("effect1", "effect", {"effect1"}, {"render"})
|
|
overlay = _create_mock_stage("overlay_test", "overlay", {"overlay"}, {"effect1"})
|
|
display = _create_mock_stage("display", "display", {"display"}, {"overlay"})
|
|
|
|
for stage in [effect, overlay, display]:
|
|
pipeline.add_stage(stage.name, stage, initialize=False)
|
|
|
|
pipeline._rebuild()
|
|
|
|
names = pipeline.execution_order
|
|
idx_effect = names.index("effect1")
|
|
idx_overlay = names.index("overlay_test")
|
|
idx_display = names.index("display")
|
|
assert idx_effect < idx_overlay < idx_display
|