David Gwilliam
2d28e92594
Reviewed-on: #53 Co-authored-by: David Gwilliam <dhgwilliam@gmail.com> Co-committed-by: David Gwilliam <dhgwilliam@gmail.com>
105 lines
3.2 KiB
Python
105 lines
3.2 KiB
Python
"""
|
|
Tests to verify figment rendering in the pipeline.
|
|
"""
|
|
|
|
from engine.effects.plugins import discover_plugins
|
|
from engine.effects.registry import get_registry
|
|
from engine.effects.types import EffectConfig
|
|
from engine.pipeline import Pipeline, PipelineConfig
|
|
from engine.pipeline.adapters import (
|
|
EffectPluginStage,
|
|
SourceItemsToBufferStage,
|
|
create_stage_from_display,
|
|
)
|
|
from engine.pipeline.controller import PipelineRunner
|
|
|
|
|
|
def test_figment_renders_in_pipeline():
|
|
"""Verify figment renders overlay in pipeline."""
|
|
# Discover plugins
|
|
discover_plugins()
|
|
|
|
# Get figment plugin
|
|
registry = get_registry()
|
|
figment = registry.get("figment")
|
|
|
|
# Configure with short interval for testing
|
|
config = EffectConfig(
|
|
enabled=True,
|
|
intensity=1.0,
|
|
params={
|
|
"interval_secs": 0.1, # 100ms
|
|
"display_secs": 1.0,
|
|
"figment_dir": "figments",
|
|
},
|
|
)
|
|
figment.configure(config)
|
|
|
|
# Create pipeline
|
|
pipeline = Pipeline(
|
|
config=PipelineConfig(
|
|
source="empty",
|
|
display="null",
|
|
camera="feed",
|
|
effects=["figment"],
|
|
)
|
|
)
|
|
|
|
# Add source stage
|
|
from engine.data_sources.sources import EmptyDataSource
|
|
from engine.pipeline.adapters import DataSourceStage
|
|
|
|
empty_source = EmptyDataSource(width=80, height=24)
|
|
pipeline.add_stage("source", DataSourceStage(empty_source, name="empty"))
|
|
|
|
# Add render stage
|
|
pipeline.add_stage("render", SourceItemsToBufferStage(name="items-to-buffer"))
|
|
|
|
# Add figment effect stage
|
|
pipeline.add_stage("effect_figment", EffectPluginStage(figment, name="figment"))
|
|
|
|
# Add display stage
|
|
from engine.display import DisplayRegistry
|
|
|
|
display = DisplayRegistry.create("null")
|
|
display.init(0, 0)
|
|
pipeline.add_stage("display", create_stage_from_display(display, "null"))
|
|
|
|
# Build and initialize pipeline
|
|
pipeline.build()
|
|
assert pipeline.initialize()
|
|
|
|
# Use PipelineRunner to step through frames
|
|
runner = PipelineRunner(pipeline)
|
|
runner.start()
|
|
|
|
# Run pipeline until figment renders (or timeout)
|
|
figment_rendered = False
|
|
for i in range(30):
|
|
runner.step()
|
|
|
|
# Check if figment rendered by inspecting the display's internal buffer
|
|
# The null display stores the last rendered buffer
|
|
if hasattr(display, "_last_buffer") and display._last_buffer:
|
|
buffer = display._last_buffer
|
|
# Check if buffer contains ANSI escape codes (indicating figment overlay)
|
|
# Figment adds overlay lines at the end of the buffer
|
|
for line in buffer:
|
|
if "\x1b[" in line:
|
|
figment_rendered = True
|
|
print(f"Figment rendered at frame {i}")
|
|
# Print first few lines containing escape codes
|
|
for j, line in enumerate(buffer[:10]):
|
|
if "\x1b[" in line:
|
|
print(f"Line {j}: {repr(line[:80])}")
|
|
break
|
|
if figment_rendered:
|
|
break
|
|
|
|
assert figment_rendered, "Figment did not render in 30 frames"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
test_figment_renders_in_pipeline()
|
|
print("Test passed!")
|