raspberry-pi-mixer/tests/test_session.py

"""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)