David Gwilliam
2d28e92594
Reviewed-on: #53 Co-authored-by: David Gwilliam <dhgwilliam@gmail.com> Co-committed-by: David Gwilliam <dhgwilliam@gmail.com>
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
|