forked from genewildish/Mainline
David Gwilliam
238bac1bb2
- Implements pipeline hot-rebuild with state preservation (issue #43) - Adds auto-injection of MVP stages for missing capabilities - Adds radial camera mode for polar coordinate scanning - Adds afterimage and motionblur effects using framebuffer history - Adds comprehensive acceptance tests for camera modes and pipeline rebuild - Updates presets.toml with new effect configurations Related to: #35 (Pipeline Mutation API epic) Closes: #43, #44, #45
406 lines
13 KiB
Python
406 lines
13 KiB
Python
"""
|
|
Integration tests for pipeline hot-rebuild and state preservation.
|
|
|
|
Tests:
|
|
1. Viewport size control via --viewport flag
|
|
2. NullDisplay recording and save/load functionality
|
|
3. Pipeline state preservation during hot-rebuild
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
|
|
from engine.display import DisplayRegistry
|
|
from engine.display.backends.null import NullDisplay
|
|
from engine.display.backends.replay import ReplayDisplay
|
|
from engine.effects import get_registry
|
|
from engine.fetch import load_cache
|
|
from engine.pipeline import Pipeline, PipelineConfig, PipelineContext
|
|
from engine.pipeline.adapters import (
|
|
EffectPluginStage,
|
|
FontStage,
|
|
ViewportFilterStage,
|
|
create_stage_from_display,
|
|
create_stage_from_effect,
|
|
)
|
|
from engine.pipeline.params import PipelineParams
|
|
|
|
|
|
@pytest.fixture
|
|
def viewport_dims():
|
|
"""Small viewport dimensions for testing."""
|
|
return (40, 15)
|
|
|
|
|
|
@pytest.fixture
|
|
def items():
|
|
"""Load cached source items."""
|
|
items = load_cache()
|
|
if not items:
|
|
pytest.skip("No fixture cache available")
|
|
return items
|
|
|
|
|
|
@pytest.fixture
|
|
def null_display(viewport_dims):
|
|
"""Create a NullDisplay for testing."""
|
|
display = DisplayRegistry.create("null")
|
|
display.init(viewport_dims[0], viewport_dims[1])
|
|
return display
|
|
|
|
|
|
@pytest.fixture
|
|
def pipeline_with_null_display(items, null_display):
|
|
"""Create a pipeline with NullDisplay for testing."""
|
|
import engine.effects.plugins as effects_plugins
|
|
|
|
effects_plugins.discover_plugins()
|
|
|
|
width, height = null_display.width, null_display.height
|
|
params = PipelineParams()
|
|
params.viewport_width = width
|
|
params.viewport_height = height
|
|
|
|
config = PipelineConfig(
|
|
source="fixture",
|
|
display="null",
|
|
camera="scroll",
|
|
effects=["noise", "fade"],
|
|
)
|
|
|
|
pipeline = Pipeline(config=config, context=PipelineContext())
|
|
|
|
from engine.camera import Camera
|
|
from engine.data_sources.sources import ListDataSource
|
|
from engine.pipeline.adapters import CameraClockStage, CameraStage, DataSourceStage
|
|
|
|
list_source = ListDataSource(items, name="fixture")
|
|
pipeline.add_stage("source", DataSourceStage(list_source, name="fixture"))
|
|
|
|
# Add camera stages (required by ViewportFilterStage)
|
|
camera = Camera.scroll(speed=0.3)
|
|
camera.set_canvas_size(200, 200)
|
|
pipeline.add_stage("camera_update", CameraClockStage(camera, name="camera-clock"))
|
|
pipeline.add_stage("camera", CameraStage(camera, name="scroll"))
|
|
|
|
pipeline.add_stage("viewport_filter", ViewportFilterStage(name="viewport-filter"))
|
|
pipeline.add_stage("font", FontStage(name="font"))
|
|
|
|
effect_registry = get_registry()
|
|
for effect_name in config.effects:
|
|
effect = effect_registry.get(effect_name)
|
|
if effect:
|
|
pipeline.add_stage(
|
|
f"effect_{effect_name}",
|
|
create_stage_from_effect(effect, effect_name),
|
|
)
|
|
|
|
pipeline.add_stage("display", create_stage_from_display(null_display, "null"))
|
|
pipeline.build()
|
|
|
|
if not pipeline.initialize():
|
|
pytest.fail("Failed to initialize pipeline")
|
|
|
|
ctx = pipeline.context
|
|
ctx.params = params
|
|
ctx.set("display", null_display)
|
|
ctx.set("items", items)
|
|
ctx.set("pipeline", pipeline)
|
|
ctx.set("pipeline_order", pipeline.execution_order)
|
|
ctx.set("camera_y", 0)
|
|
|
|
yield pipeline, params, null_display
|
|
|
|
pipeline.cleanup()
|
|
null_display.cleanup()
|
|
|
|
|
|
class TestNullDisplayRecording:
|
|
"""Tests for NullDisplay recording functionality."""
|
|
|
|
def test_null_display_initialization(self, viewport_dims):
|
|
"""NullDisplay initializes with correct dimensions."""
|
|
display = NullDisplay()
|
|
display.init(viewport_dims[0], viewport_dims[1])
|
|
assert display.width == viewport_dims[0]
|
|
assert display.height == viewport_dims[1]
|
|
|
|
def test_start_stop_recording(self, null_display):
|
|
"""NullDisplay can start and stop recording."""
|
|
assert not null_display._is_recording
|
|
|
|
null_display.start_recording()
|
|
assert null_display._is_recording is True
|
|
|
|
null_display.stop_recording()
|
|
assert null_display._is_recording is False
|
|
|
|
def test_record_frames(self, null_display, pipeline_with_null_display):
|
|
"""NullDisplay records frames when recording is enabled."""
|
|
pipeline, params, display = pipeline_with_null_display
|
|
|
|
display.start_recording()
|
|
assert len(display._recorded_frames) == 0
|
|
|
|
for frame in range(5):
|
|
params.frame_number = frame
|
|
pipeline.context.params = params
|
|
pipeline.execute([])
|
|
|
|
assert len(display._recorded_frames) == 5
|
|
|
|
def test_get_frames(self, null_display, pipeline_with_null_display):
|
|
"""NullDisplay.get_frames() returns recorded buffers."""
|
|
pipeline, params, display = pipeline_with_null_display
|
|
|
|
display.start_recording()
|
|
|
|
for frame in range(3):
|
|
params.frame_number = frame
|
|
pipeline.context.params = params
|
|
pipeline.execute([])
|
|
|
|
frames = display.get_frames()
|
|
assert len(frames) == 3
|
|
assert all(isinstance(f, list) for f in frames)
|
|
|
|
def test_clear_recording(self, null_display, pipeline_with_null_display):
|
|
"""NullDisplay.clear_recording() clears recorded frames."""
|
|
pipeline, params, display = pipeline_with_null_display
|
|
|
|
display.start_recording()
|
|
for frame in range(3):
|
|
params.frame_number = frame
|
|
pipeline.context.params = params
|
|
pipeline.execute([])
|
|
|
|
assert len(display._recorded_frames) == 3
|
|
|
|
display.clear_recording()
|
|
assert len(display._recorded_frames) == 0
|
|
|
|
def test_save_load_recording(self, null_display, pipeline_with_null_display):
|
|
"""NullDisplay can save and load recordings."""
|
|
pipeline, params, display = pipeline_with_null_display
|
|
|
|
display.start_recording()
|
|
for frame in range(3):
|
|
params.frame_number = frame
|
|
pipeline.context.params = params
|
|
pipeline.execute([])
|
|
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
|
|
temp_path = f.name
|
|
|
|
try:
|
|
display.save_recording(temp_path)
|
|
|
|
with open(temp_path) as f:
|
|
data = json.load(f)
|
|
|
|
assert data["version"] == 1
|
|
assert data["display"] == "null"
|
|
assert data["frame_count"] == 3
|
|
assert len(data["frames"]) == 3
|
|
|
|
display2 = NullDisplay()
|
|
display2.load_recording(temp_path)
|
|
assert len(display2._recorded_frames) == 3
|
|
|
|
finally:
|
|
Path(temp_path).unlink(missing_ok=True)
|
|
|
|
|
|
class TestReplayDisplay:
|
|
"""Tests for ReplayDisplay functionality."""
|
|
|
|
def test_replay_display_initialization(self, viewport_dims):
|
|
"""ReplayDisplay initializes correctly."""
|
|
display = ReplayDisplay()
|
|
display.init(viewport_dims[0], viewport_dims[1])
|
|
assert display.width == viewport_dims[0]
|
|
assert display.height == viewport_dims[1]
|
|
|
|
def test_set_and_get_frames(self):
|
|
"""ReplayDisplay can set and retrieve frames."""
|
|
display = ReplayDisplay()
|
|
frames = [
|
|
{"buffer": ["line1", "line2"], "width": 40, "height": 15},
|
|
{"buffer": ["line3", "line4"], "width": 40, "height": 15},
|
|
]
|
|
display.set_frames(frames)
|
|
|
|
frame = display.get_next_frame()
|
|
assert frame == ["line1", "line2"]
|
|
|
|
frame = display.get_next_frame()
|
|
assert frame == ["line3", "line4"]
|
|
|
|
frame = display.get_next_frame()
|
|
assert frame is None
|
|
|
|
def test_replay_loop_mode(self):
|
|
"""ReplayDisplay can loop playback."""
|
|
display = ReplayDisplay()
|
|
display.set_loop(True)
|
|
frames = [
|
|
{"buffer": ["frame1"], "width": 40, "height": 15},
|
|
{"buffer": ["frame2"], "width": 40, "height": 15},
|
|
]
|
|
display.set_frames(frames)
|
|
|
|
assert display.get_next_frame() == ["frame1"]
|
|
assert display.get_next_frame() == ["frame2"]
|
|
assert display.get_next_frame() == ["frame1"]
|
|
assert display.get_next_frame() == ["frame2"]
|
|
|
|
def test_replay_seek_and_reset(self):
|
|
"""ReplayDisplay supports seek and reset."""
|
|
display = ReplayDisplay()
|
|
frames = [
|
|
{"buffer": [f"frame{i}"], "width": 40, "height": 15} for i in range(5)
|
|
]
|
|
display.set_frames(frames)
|
|
|
|
display.seek(3)
|
|
assert display.get_next_frame() == ["frame3"]
|
|
|
|
display.reset()
|
|
assert display.get_next_frame() == ["frame0"]
|
|
|
|
|
|
class TestPipelineHotRebuild:
|
|
"""Tests for pipeline hot-rebuild and state preservation."""
|
|
|
|
def test_pipeline_runs_with_null_display(self, pipeline_with_null_display):
|
|
"""Pipeline executes successfully with NullDisplay."""
|
|
pipeline, params, display = pipeline_with_null_display
|
|
|
|
for frame in range(5):
|
|
params.frame_number = frame
|
|
pipeline.context.params = params
|
|
result = pipeline.execute([])
|
|
|
|
assert result.success
|
|
assert display._last_buffer is not None
|
|
|
|
def test_effect_toggle_during_execution(self, pipeline_with_null_display):
|
|
"""Effects can be toggled during pipeline execution."""
|
|
pipeline, params, display = pipeline_with_null_display
|
|
|
|
params.frame_number = 0
|
|
pipeline.context.params = params
|
|
pipeline.execute([])
|
|
buffer1 = display._last_buffer
|
|
|
|
fade_stage = pipeline.get_stage("effect_fade")
|
|
assert fade_stage is not None
|
|
assert isinstance(fade_stage, EffectPluginStage)
|
|
|
|
fade_stage._enabled = False
|
|
fade_stage._effect.config.enabled = False
|
|
|
|
params.frame_number = 1
|
|
pipeline.context.params = params
|
|
pipeline.execute([])
|
|
buffer2 = display._last_buffer
|
|
|
|
assert buffer1 != buffer2
|
|
|
|
def test_state_preservation_across_rebuild(self, pipeline_with_null_display):
|
|
"""Pipeline state is preserved across hot-rebuild events."""
|
|
pipeline, params, display = pipeline_with_null_display
|
|
|
|
for frame in range(5):
|
|
params.frame_number = frame
|
|
pipeline.context.params = params
|
|
pipeline.execute([])
|
|
|
|
camera_y_before = pipeline.context.get("camera_y")
|
|
|
|
fade_stage = pipeline.get_stage("effect_fade")
|
|
if fade_stage and isinstance(fade_stage, EffectPluginStage):
|
|
fade_stage.set_enabled(not fade_stage.is_enabled())
|
|
fade_stage._effect.config.enabled = fade_stage.is_enabled()
|
|
|
|
params.frame_number = 5
|
|
pipeline.context.params = params
|
|
pipeline.execute([])
|
|
|
|
pipeline.context.get("camera_y")
|
|
|
|
assert camera_y_before is not None
|
|
|
|
|
|
class TestViewportControl:
|
|
"""Tests for viewport size control."""
|
|
|
|
def test_viewport_dimensions_applied(self, items):
|
|
"""Viewport dimensions are correctly applied to pipeline."""
|
|
width, height = 40, 15
|
|
|
|
display = DisplayRegistry.create("null")
|
|
display.init(width, height)
|
|
|
|
params = PipelineParams()
|
|
params.viewport_width = width
|
|
params.viewport_height = height
|
|
|
|
config = PipelineConfig(
|
|
source="fixture",
|
|
display="null",
|
|
camera="scroll",
|
|
effects=[],
|
|
)
|
|
|
|
pipeline = Pipeline(config=config, context=PipelineContext())
|
|
|
|
from engine.camera import Camera
|
|
from engine.data_sources.sources import ListDataSource
|
|
from engine.pipeline.adapters import (
|
|
CameraClockStage,
|
|
CameraStage,
|
|
DataSourceStage,
|
|
)
|
|
|
|
list_source = ListDataSource(items, name="fixture")
|
|
pipeline.add_stage("source", DataSourceStage(list_source, name="fixture"))
|
|
|
|
# Add camera stages (required by ViewportFilterStage)
|
|
camera = Camera.scroll(speed=0.3)
|
|
camera.set_canvas_size(200, 200)
|
|
pipeline.add_stage(
|
|
"camera_update", CameraClockStage(camera, name="camera-clock")
|
|
)
|
|
pipeline.add_stage("camera", CameraStage(camera, name="scroll"))
|
|
|
|
pipeline.add_stage(
|
|
"viewport_filter", ViewportFilterStage(name="viewport-filter")
|
|
)
|
|
pipeline.add_stage("font", FontStage(name="font"))
|
|
pipeline.add_stage("display", create_stage_from_display(display, "null"))
|
|
pipeline.build()
|
|
|
|
assert pipeline.initialize()
|
|
|
|
ctx = pipeline.context
|
|
ctx.params = params
|
|
ctx.set("display", display)
|
|
ctx.set("items", items)
|
|
ctx.set("pipeline", pipeline)
|
|
ctx.set("camera_y", 0)
|
|
|
|
result = pipeline.execute(items)
|
|
|
|
assert result.success
|
|
assert display._last_buffer is not None
|
|
|
|
pipeline.cleanup()
|
|
display.cleanup()
|