"""Tests for session management module. Tests cover: - MixerSession: capture, serialize, deserialize, apply - SessionManager: save, load, list, delete, rename, duplicate, auto-save - Setlist: create, serialize, activate, navigate - SnapshotEngine: capture, recall, next/prev, persistence """ from __future__ import annotations import json import os import tempfile import time from pathlib import Path import pytest from src.mixer.dsp_engine import DSPEngine, DSPEngineConfig from src.session.mixer_session import MixerSession, SessionMetadata, SESSION_SCHEMA_VERSION from src.session.session_manager import SessionManager from src.session.setlist import Setlist, SetlistEntry, TransitionType from src.session.snapshot import Snapshot, SnapshotEngine, SnapshotMetadata # ── Fixtures ───────────────────────────────────────────────────────────────── @pytest.fixture def engine(): """Create a minimal DSP engine for testing.""" config = DSPEngineConfig( num_channels=8, num_aux=2, num_subgroups=1, num_vca=1, osc_enabled=False, jack_routing_enabled=False, automation_enabled=False, ) eng = DSPEngine(config) eng.start(connect_osc=False) yield eng eng.stop() @pytest.fixture def tmp_sessions_dir(): """Create a temporary sessions directory.""" with tempfile.TemporaryDirectory() as td: yield Path(td) # ── MixerSession tests ────────────────────────────────────────────────────── class TestMixerSession: """Test MixerSession model.""" def test_create_empty(self): """Create an empty session.""" session = MixerSession() assert session.version == SESSION_SCHEMA_VERSION assert session.metadata.name == "Untitled Session" assert session.channels == [] assert session.master == {} def test_from_engine(self, engine): """Capture engine state into a session.""" session = MixerSession.from_engine(engine, name="Test Mix") assert session.metadata.name == "Test Mix" assert session.channel_count == 8 assert session.master # has bus state assert session.routing # has routing matrix assert not session.has_plugins # no OSC, no plugins assert session.settings["num_channels"] == 8 def test_serialize_roundtrip(self, engine): """Serialize and deserialize a session.""" session = MixerSession.from_engine(engine, name="Roundtrip") json_str = session.to_json() # Parse back restored = MixerSession.from_json(json_str) assert restored.metadata.name == "Roundtrip" assert restored.channel_count == 8 assert restored.version == SESSION_SCHEMA_VERSION def test_serialize_json_valid(self, engine): """Serialized session is valid JSON.""" session = MixerSession.from_engine(engine) json_str = session.to_json() data = json.loads(json_str) assert "version" in data assert "metadata" in data assert "channels" in data assert "master" in data assert "routing" in data def test_apply_to_engine(self, engine): """Apply a session to the engine.""" # Change a channel parameter ch0 = engine.channels[0] ch0.state.volume = -10.0 ch0.state.mute = True # Capture session = MixerSession.from_engine(engine, name="Modified") # Reset engine ch0.state.volume = 0.0 ch0.state.mute = False # Apply assert session.apply_to(engine) # Verify assert engine.channels[0].state.volume == -10.0 assert engine.channels[0].state.mute is True def test_from_dict_to_dict(self, engine): """Dict roundtrip.""" session = MixerSession.from_engine(engine, name="Dict Test") data = session.to_dict() restored = MixerSession.from_dict(data) assert restored.metadata.name == "Dict Test" def test_metadata_fields(self): """Test session metadata fields.""" meta = SessionMetadata( name="My Song", tempo=140.0, key="D minor", time_signature="3/4", lyrics="Test lyrics", genre="Rock", tags=["live", "soundcheck"], notes="Test notes", ) session = MixerSession(metadata=meta) data = session.to_dict() assert data["metadata"]["tempo"] == 140.0 assert data["metadata"]["key"] == "D minor" assert data["metadata"]["tags"] == ["live", "soundcheck"] def test_file_save_load(self, engine, tmp_path): """Save and load a session from file.""" session = MixerSession.from_engine(engine, name="File Test") filepath = tmp_path / "test_session.json" session.save_to_file(filepath) assert filepath.exists() loaded = MixerSession.from_file(filepath) assert loaded.metadata.name == "File Test" assert loaded.channel_count == 8 def test_custom_metadata(self): """Custom metadata fields are preserved.""" meta = SessionMetadata(custom={"venue": "Madison Square Garden", "set": 2}) session = MixerSession(metadata=meta) data = session.to_dict() assert data["metadata"]["custom"]["venue"] == "Madison Square Garden" # ── SessionManager tests ──────────────────────────────────────────────────── class TestSessionManager: """Test SessionManager.""" def test_save_and_list(self, engine, tmp_sessions_dir): """Save a session and list it.""" mgr = SessionManager(sessions_dir=tmp_sessions_dir, auto_save_enabled=False) mgr.save(engine, "Test Session") sessions = mgr.list_sessions() assert len(sessions) == 1 assert sessions[0]["name"] == "Test Session" def test_load_session(self, engine, tmp_sessions_dir): """Save and then load a session.""" mgr = SessionManager(sessions_dir=tmp_sessions_dir, auto_save_enabled=False) # Modify engine state engine.channels[0].state.volume = -5.0 mgr.save(engine, "Load Test") # Reset engine.channels[0].state.volume = 0.0 # Load and apply ok = mgr.load_and_apply(engine, "Load Test") assert ok assert engine.channels[0].state.volume == -5.0 def test_delete_session(self, engine, tmp_sessions_dir): """Delete a session.""" mgr = SessionManager(sessions_dir=tmp_sessions_dir, auto_save_enabled=False) mgr.save(engine, "Delete Me") assert mgr.exists("Delete Me") assert mgr.delete("Delete Me") assert not mgr.exists("Delete Me") def test_rename_session(self, engine, tmp_sessions_dir): """Rename a session.""" mgr = SessionManager(sessions_dir=tmp_sessions_dir, auto_save_enabled=False) mgr.save(engine, "Old Name") assert mgr.rename("Old Name", "New Name") assert mgr.exists("New Name") assert not mgr.exists("Old Name") def test_duplicate_session(self, engine, tmp_sessions_dir): """Duplicate a session.""" mgr = SessionManager(sessions_dir=tmp_sessions_dir, auto_save_enabled=False) engine.channels[0].state.volume = -12.0 mgr.save(engine, "Original") dup = mgr.duplicate("Original", "Copy") assert dup is not None assert dup.metadata.name == "Copy" # Load copy into reset engine engine.channels[0].state.volume = 0.0 ok = mgr.load_and_apply(engine, "Copy") assert ok assert engine.channels[0].state.volume == -12.0 def test_exists(self, engine, tmp_sessions_dir): """Check session existence.""" mgr = SessionManager(sessions_dir=tmp_sessions_dir, auto_save_enabled=False) assert not mgr.exists("No Such Session") mgr.save(engine, "Exists") assert mgr.exists("Exists") def test_session_count(self, engine, tmp_sessions_dir): """Session count is accurate.""" mgr = SessionManager(sessions_dir=tmp_sessions_dir, auto_save_enabled=False) assert mgr.session_count == 0 mgr.save(engine, "One") mgr.save(engine, "Two") assert mgr.session_count == 2 def test_list_nonexistent_dir(self): """Listing when sessions dir cannot be created returns empty.""" # Use a path under /dev/null which exists as a file, not a dir mgr = SessionManager(sessions_dir=Path("/dev/null/sessions"), auto_save_enabled=False) assert mgr.list_sessions() == [] def test_get_session_info(self, engine, tmp_sessions_dir): """Get session info without loading full state.""" mgr = SessionManager(sessions_dir=tmp_sessions_dir, auto_save_enabled=False) mgr.save(engine, "Info Test") info = mgr.get_session_info("Info Test") assert info is not None assert info["name"] == "Info Test" assert info["channel_count"] == 8 def test_auto_save_disabled(self, engine, tmp_sessions_dir): """Auto-save can be disabled.""" mgr = SessionManager(sessions_dir=tmp_sessions_dir, auto_save_enabled=False) mgr.set_auto_save_engine(engine) mgr.disable_auto_save() assert not mgr.auto_save_enabled def test_export_import(self, engine, tmp_sessions_dir, tmp_path): """Export and import a session.""" mgr = SessionManager(sessions_dir=tmp_sessions_dir, auto_save_enabled=False) mgr.save(engine, "Export Test") export_path = mgr.export_session("Export Test", target_dir=tmp_path) assert export_path is not None assert export_path.exists() # Import into a new manager mgr2 = SessionManager( sessions_dir=tmp_path / "imported", auto_save_enabled=False, ) imported = mgr2.import_session(export_path) assert imported is not None assert imported.metadata.name == "Export Test" def test_load_nonexistent(self, tmp_sessions_dir): """Loading a nonexistent session returns None.""" mgr = SessionManager(sessions_dir=tmp_sessions_dir, auto_save_enabled=False) assert mgr.load("No Such Session") is None def test_delete_nonexistent(self, tmp_sessions_dir): """Deleting a nonexistent session returns False.""" mgr = SessionManager(sessions_dir=tmp_sessions_dir, auto_save_enabled=False) assert not mgr.delete("No Such Session") # ── Setlist tests ─────────────────────────────────────────────────────────── class TestSetlist: """Test Setlist model.""" def test_create_empty(self): """Create an empty setlist.""" sl = Setlist(name="Test Setlist") assert sl.name == "Test Setlist" assert sl.entry_count == 0 assert not sl.is_active def test_add_entries(self): """Add entries to a setlist.""" sl = Setlist(name="Gig") sl.add_entry("Soundcheck", transition=TransitionType.CUT) sl.add_entry("Set 1", transition=TransitionType.CROSSFADE, transition_duration=3.0) sl.add_entry("Set 2", transition=TransitionType.WAIT, wait_for_trigger=True) assert sl.entry_count == 3 assert sl.entries[0].session_name == "Soundcheck" assert sl.entries[1].transition == TransitionType.CROSSFADE assert sl.entries[2].wait_for_trigger is True def test_remove_entry(self): """Remove an entry.""" sl = Setlist(name="Test") sl.add_entry("A") sl.add_entry("B") assert sl.remove_entry(0) assert sl.entry_count == 1 assert sl.entries[0].session_name == "B" def test_move_entry(self): """Move an entry.""" sl = Setlist(name="Test") sl.add_entry("First") sl.add_entry("Second") assert sl.move_entry(0, 1) assert sl.entries[0].session_name == "Second" assert sl.entries[1].session_name == "First" def test_serialize_roundtrip(self): """Serialize and deserialize a setlist.""" sl = Setlist(name="Tour", description="Summer 2026") sl.add_entry("Opening", transition=TransitionType.CUT) sl.add_entry("Main Set", transition=TransitionType.CROSSFADE) json_str = sl.to_json() restored = Setlist.from_json(json_str) assert restored.name == "Tour" assert restored.description == "Summer 2026" assert restored.entry_count == 2 def test_file_save_load(self, tmp_path): """Save and load setlist from file.""" sl = Setlist(name="File Test") sl.add_entry("Test Session") filepath = sl.save_to(tmp_path) assert filepath.exists() loaded = Setlist.load_from(filepath) assert loaded.name == "File Test" assert loaded.entry_count == 1 def test_progress(self): """Progress tracking.""" sl = Setlist(name="Test") sl.add_entry("A") sl.add_entry("B") sl.add_entry("C") assert sl.progress == (0, 3) # not active def test_activate_empty(self, engine, tmp_sessions_dir): """Activating empty setlist fails.""" mgr = SessionManager(sessions_dir=tmp_sessions_dir, auto_save_enabled=False) sl = Setlist(name="Empty") assert not sl.activate(mgr, engine) def test_stop(self): """Stopping a setlist.""" sl = Setlist(name="Test") sl._active = True sl._current_index = 0 sl.stop() assert not sl.is_active assert sl._current_index == -1 def test_clear(self): """Clearing a setlist.""" sl = Setlist(name="Test") sl.add_entry("A") sl.add_entry("B") sl.clear() assert sl.entry_count == 0 class TestSetlistEntry: """Test SetlistEntry.""" def test_defaults(self): """Default values.""" entry = SetlistEntry(session_name="Test") assert entry.transition == TransitionType.CUT assert entry.transition_duration == 2.0 assert not entry.wait_for_trigger def test_serialize(self): """Serialize entry.""" entry = SetlistEntry( session_name="My Session", display_name="My Display", transition=TransitionType.CROSSFADE, transition_duration=3.5, wait_for_trigger=True, notes="Test note", ) data = entry.to_dict() assert data["transition"] == "crossfade" assert data["transition_duration"] == 3.5 restored = SetlistEntry.from_dict(data) assert restored.transition == TransitionType.CROSSFADE # ── Snapshot tests ────────────────────────────────────────────────────────── class TestSnapshot: """Test Snapshot model.""" def test_from_engine(self, engine): """Capture a snapshot from engine.""" snap = Snapshot.from_engine(engine, name="Test Snap") assert snap.metadata.name == "Test Snap" assert len(snap.channels) == 8 assert snap.master def test_apply_to_engine(self, engine): """Apply snapshot to engine.""" engine.channels[0].state.volume = -20.0 snap = Snapshot.from_engine(engine, name="Loud") engine.channels[0].state.volume = 0.0 assert snap.apply_to(engine) assert engine.channels[0].state.volume == -20.0 def test_serialize(self, engine): """Serialize snapshot.""" snap = Snapshot.from_engine(engine, name="Serial Snap", notes="test", category="verse") data = snap.to_dict() assert data["metadata"]["name"] == "Serial Snap" assert data["metadata"]["notes"] == "test" assert data["metadata"]["category"] == "verse" restored = Snapshot.from_dict(data) assert restored.metadata.name == "Serial Snap" class TestSnapshotEngine: """Test SnapshotEngine.""" def test_capture_and_recall(self, engine): """Capture a snapshot and recall it.""" se = SnapshotEngine() engine.channels[0].state.volume = -15.0 se.capture(engine, "Verse") # Modify engine.channels[0].state.volume = 0.0 # Recall ok = se.recall(engine, "Verse") assert ok assert engine.channels[0].state.volume == -15.0 def test_list_snapshots(self, engine): """List snapshots.""" se = SnapshotEngine() se.capture(engine, "Intro", category="intro") se.capture(engine, "Chorus", category="chorus") snaps = se.list_snapshots() assert len(snaps) == 2 assert snaps[0]["name"] in ("Intro", "Chorus") def test_delete(self, engine): """Delete a snapshot.""" se = SnapshotEngine() se.capture(engine, "To Delete") assert se.delete("To Delete") assert se.snapshot_count == 0 def test_rename(self, engine): """Rename a snapshot.""" se = SnapshotEngine() se.capture(engine, "Old") assert se.rename("Old", "New") assert se.snapshot_count == 1 assert not se.recall(engine, "Old") assert se.recall(engine, "New") def test_next_previous(self, engine): """Next and previous snapshot navigation.""" se = SnapshotEngine() se.capture(engine, "A") se.capture(engine, "B") se.capture(engine, "C") # Next ok, name = se.next(engine) assert ok assert name is not None # Previous ok, name = se.previous(engine) assert ok assert name is not None def test_clear(self, engine): """Clear all snapshots.""" se = SnapshotEngine() se.capture(engine, "One") se.capture(engine, "Two") se.clear() assert se.snapshot_count == 0 def test_categories(self, engine): """Snapshot categories.""" se = SnapshotEngine() se.capture(engine, "Intro", category="intro") se.capture(engine, "Verse 1", category="verse") se.capture(engine, "Chorus", category="chorus") cats = se.categories assert "intro" in cats assert "verse" in cats assert "chorus" in cats def test_get_by_category(self, engine): """Filter snapshots by category.""" se = SnapshotEngine() se.capture(engine, "Intro", category="intro") se.capture(engine, "Chorus", category="chorus") intros = se.get_by_category("intro") assert len(intros) == 1 assert intros[0].metadata.name == "Intro" def test_persist_to_disk(self, engine, tmp_path): """Snapshots persist to disk and reload.""" snap_dir = tmp_path / "snapshots" se = SnapshotEngine(storage_dir=snap_dir) se.capture(engine, "Persistent") # Create a new engine reading from the same dir se2 = SnapshotEngine(storage_dir=snap_dir) assert se2.snapshot_count == 1 assert se2.recall(engine, "Persistent") def test_recall_nonexistent(self, engine): """Recalling nonexistent snapshot fails.""" se = SnapshotEngine() assert not se.recall(engine, "No Such Snap") def test_empty_next_prev(self, engine): """Next/prev on empty snapshot engine.""" se = SnapshotEngine() ok, name = se.next(engine) assert not ok ok, name = se.previous(engine) assert not ok def test_go_to_index(self, engine): """Go to snapshot by index.""" se = SnapshotEngine() se.capture(engine, "First") se.capture(engine, "Second") assert se.go_to(engine, 0) assert se.go_to(engine, 1) assert not se.go_to(engine, 99) def test_duplicate_name(self, engine): """Capturing with same name overwrites by default.""" se = SnapshotEngine() se.capture(engine, "Same") se.capture(engine, "Same") assert se.snapshot_count == 1 def test_no_overwrite(self, engine): """Capturing with overwrite=False raises on duplicate.""" se = SnapshotEngine() se.capture(engine, "Unique") with pytest.raises(KeyError): se.capture(engine, "Unique", overwrite=False)