forked from genewildish/Mainline
- Add PipelineIntrospectionSource that renders live ASCII DAG with metrics - Add PipelineMetricsSensor exposing pipeline performance as sensor values - Add PipelineIntrospectionDemo controller with 3-phase animation: - Phase 1: Toggle effects one at a time (3s each) - Phase 2: LFO drives intensity default→max→min→default - Phase 3: All effects with shared LFO (infinite loop) - Add pipeline-inspect preset - Add get_frame_times() to Pipeline for sparkline data - Add tests for new components - Update mise.toml with pipeline-inspect preset task
168 lines
5.0 KiB
Python
168 lines
5.0 KiB
Python
"""
|
|
Tests for PipelineIntrospectionDemo.
|
|
"""
|
|
|
|
from engine.pipeline.pipeline_introspection_demo import (
|
|
DemoConfig,
|
|
DemoPhase,
|
|
PhaseState,
|
|
PipelineIntrospectionDemo,
|
|
)
|
|
|
|
|
|
class MockPipeline:
|
|
"""Mock pipeline for testing."""
|
|
|
|
pass
|
|
|
|
|
|
class MockEffectConfig:
|
|
"""Mock effect config."""
|
|
|
|
def __init__(self):
|
|
self.enabled = False
|
|
self.intensity = 0.5
|
|
|
|
|
|
class MockEffect:
|
|
"""Mock effect for testing."""
|
|
|
|
def __init__(self, name):
|
|
self.name = name
|
|
self.config = MockEffectConfig()
|
|
|
|
|
|
class MockRegistry:
|
|
"""Mock effect registry."""
|
|
|
|
def __init__(self, effects):
|
|
self._effects = {e.name: e for e in effects}
|
|
|
|
def get(self, name):
|
|
return self._effects.get(name)
|
|
|
|
|
|
class TestDemoPhase:
|
|
"""Tests for DemoPhase enum."""
|
|
|
|
def test_phases_exist(self):
|
|
"""All three phases exist."""
|
|
assert DemoPhase.PHASE_1_TOGGLE is not None
|
|
assert DemoPhase.PHASE_2_LFO is not None
|
|
assert DemoPhase.PHASE_3_SHARED_LFO is not None
|
|
|
|
|
|
class TestDemoConfig:
|
|
"""Tests for DemoConfig."""
|
|
|
|
def test_defaults(self):
|
|
"""Default config has sensible values."""
|
|
config = DemoConfig()
|
|
assert config.effect_cycle_duration == 3.0
|
|
assert config.gap_duration == 1.0
|
|
assert config.lfo_duration == 4.0
|
|
assert config.phase_2_effect_duration == 4.0
|
|
assert config.phase_3_lfo_duration == 6.0
|
|
|
|
|
|
class TestPhaseState:
|
|
"""Tests for PhaseState."""
|
|
|
|
def test_defaults(self):
|
|
"""PhaseState initializes correctly."""
|
|
state = PhaseState(phase=DemoPhase.PHASE_1_TOGGLE, start_time=0.0)
|
|
assert state.phase == DemoPhase.PHASE_1_TOGGLE
|
|
assert state.start_time == 0.0
|
|
assert state.current_effect_index == 0
|
|
|
|
|
|
class TestPipelineIntrospectionDemo:
|
|
"""Tests for PipelineIntrospectionDemo."""
|
|
|
|
def test_basic_init(self):
|
|
"""Demo initializes with defaults."""
|
|
demo = PipelineIntrospectionDemo(pipeline=None)
|
|
assert demo.phase == DemoPhase.PHASE_1_TOGGLE
|
|
assert demo.effect_names == ["noise", "fade", "glitch", "firehose"]
|
|
|
|
def test_init_with_custom_effects(self):
|
|
"""Demo initializes with custom effects."""
|
|
demo = PipelineIntrospectionDemo(pipeline=None, effect_names=["noise", "fade"])
|
|
assert demo.effect_names == ["noise", "fade"]
|
|
|
|
def test_phase_display(self):
|
|
"""phase_display returns correct string."""
|
|
demo = PipelineIntrospectionDemo(pipeline=None)
|
|
assert "Phase 1" in demo.phase_display
|
|
|
|
def test_shared_oscillator_created(self):
|
|
"""Shared oscillator is created."""
|
|
demo = PipelineIntrospectionDemo(pipeline=None)
|
|
assert demo.shared_oscillator is not None
|
|
assert demo.shared_oscillator.name == "demo-lfo"
|
|
|
|
|
|
class TestPipelineIntrospectionDemoUpdate:
|
|
"""Tests for update method."""
|
|
|
|
def test_update_returns_dict(self):
|
|
"""update() returns a dict with expected keys."""
|
|
demo = PipelineIntrospectionDemo(pipeline=None)
|
|
result = demo.update()
|
|
assert "phase" in result
|
|
assert "phase_display" in result
|
|
assert "effect_states" in result
|
|
|
|
def test_update_phase_1_structure(self):
|
|
"""Phase 1 has correct structure."""
|
|
demo = PipelineIntrospectionDemo(pipeline=None)
|
|
result = demo.update()
|
|
assert result["phase"] == "PHASE_1_TOGGLE"
|
|
assert "current_effect" in result
|
|
|
|
def test_effect_states_structure(self):
|
|
"""effect_states has correct structure."""
|
|
demo = PipelineIntrospectionDemo(pipeline=None)
|
|
result = demo.update()
|
|
states = result["effect_states"]
|
|
for name in demo.effect_names:
|
|
assert name in states
|
|
assert "enabled" in states[name]
|
|
assert "intensity" in states[name]
|
|
|
|
|
|
class TestPipelineIntrospectionDemoPhases:
|
|
"""Tests for phase transitions."""
|
|
|
|
def test_phase_1_initial(self):
|
|
"""Starts in phase 1."""
|
|
demo = PipelineIntrospectionDemo(pipeline=None)
|
|
assert demo.phase == DemoPhase.PHASE_1_TOGGLE
|
|
|
|
def test_shared_oscillator_not_started_initially(self):
|
|
"""Shared oscillator not started in phase 1."""
|
|
demo = PipelineIntrospectionDemo(pipeline=None)
|
|
assert demo.shared_oscillator is not None
|
|
# The oscillator.start() is called when transitioning to phase 3
|
|
|
|
|
|
class TestPipelineIntrospectionDemoCleanup:
|
|
"""Tests for cleanup method."""
|
|
|
|
def test_cleanup_no_error(self):
|
|
"""cleanup() runs without error."""
|
|
demo = PipelineIntrospectionDemo(pipeline=None)
|
|
demo.cleanup() # Should not raise
|
|
|
|
def test_cleanup_resets_effects(self):
|
|
"""cleanup() resets effects."""
|
|
demo = PipelineIntrospectionDemo(pipeline=None)
|
|
demo._apply_effect_states(
|
|
{
|
|
"noise": {"enabled": True, "intensity": 1.0},
|
|
"fade": {"enabled": True, "intensity": 1.0},
|
|
}
|
|
)
|
|
demo.cleanup()
|
|
# If we had a mock registry, we could verify effects were reset
|