Trixy7/tests/test_audio_processing.py

# -*- coding: utf-8 -*-
"""
Tests für Audio Processing Pipeline.

Testet:
- AudioProcessingContext
- AudioProcessor Interface
- AudioProcessingPipeline
- Hilfsfunktionen (apply_volume, mix_chunks, fade_chunk)
"""

import pytest
from unittest.mock import MagicMock

from trixy_core.audio.processing import (
    AudioProcessingContext,
    AudioProcessor,
    ProcessorPriority,
    AudioProcessingPipeline,
)
from trixy_core.audio.processing.processor import (
    apply_volume,
    mix_chunks,
    fade_chunk,
)


# =============================================================================
# Test Processor Implementation
# =============================================================================

class MockProcessor(AudioProcessor):
    """Mock-Prozessor für Unit-Tests."""

    def __init__(
        self,
        processor_id: str = "test",
        name: str = "Test",
        priority: int = ProcessorPriority.DEFAULT,
        enabled: bool = True,
        modify: bool = False,
    ):
        super().__init__(processor_id, name, priority, enabled)
        self.modify = modify
        self.process_count = 0
        self.last_context = None
        self.track_start_called = False
        self.track_end_called = False
        self.state_changes = []

    def process(self, chunk: bytes, context: AudioProcessingContext) -> bytes:
        self.process_count += 1
        self.last_context = context

        if self.modify:
            # Invertiere alle Bytes als Modifikation
            return bytes([255 - b for b in chunk])
        return chunk

    def on_track_start(self, context: AudioProcessingContext) -> None:
        self.track_start_called = True

    def on_track_end(self, context: AudioProcessingContext) -> None:
        self.track_end_called = True

    def on_state_change(self, state_name: str, value: bool) -> None:
        self.state_changes.append((state_name, value))

    def reset(self) -> None:
        self.process_count = 0
        self.last_context = None
        self.track_start_called = False
        self.track_end_called = False
        self.state_changes.clear()


# =============================================================================
# Tests für AudioProcessingContext
# =============================================================================

class TestAudioProcessingContext:
    """Tests für AudioProcessingContext."""

    def test_default_values(self):
        """Testet Standard-Werte."""
        context = AudioProcessingContext()

        assert context.position_ms == 0
        assert context.duration_ms == 0
        assert context.sample_rate == 44100
        assert context.channels == 2
        assert context.is_wakeword_active is False
        assert context.is_conversation_active is False
        assert context.volume == 1.0

    def test_remaining_ms(self):
        """Testet remaining_ms Berechnung."""
        context = AudioProcessingContext(
            position_ms=30000,
            duration_ms=180000,  # 3 Minuten
        )

        assert context.remaining_ms == 150000

    def test_remaining_ms_past_end(self):
        """Testet remaining_ms wenn Position > Duration."""
        context = AudioProcessingContext(
            position_ms=200000,
            duration_ms=180000,
        )

        assert context.remaining_ms == 0

    def test_progress(self):
        """Testet progress Berechnung."""
        context = AudioProcessingContext(
            position_ms=90000,
            duration_ms=180000,
        )

        assert context.progress == 0.5

    def test_progress_zero_duration(self):
        """Testet progress bei Duration 0."""
        context = AudioProcessingContext(duration_ms=0)
        assert context.progress == 0.0

    def test_is_near_end(self):
        """Testet is_near_end."""
        context = AudioProcessingContext(
            position_ms=176000,
            duration_ms=180000,
        )
        assert context.is_near_end is True

        context.position_ms = 170000
        assert context.is_near_end is False

    def test_should_duck(self):
        """Testet should_duck bei verschiedenen Zuständen."""
        context = AudioProcessingContext()
        assert context.should_duck is False

        context.is_wakeword_active = True
        assert context.should_duck is True

        context.is_wakeword_active = False
        context.is_conversation_active = True
        assert context.should_duck is True

        context.is_conversation_active = False
        context.is_tts_playing = True
        assert context.should_duck is True

    def test_copy(self):
        """Testet copy mit Updates."""
        original = AudioProcessingContext(
            position_ms=1000,
            duration_ms=5000,
        )

        copied = original.copy(position_ms=2000)

        assert copied.position_ms == 2000
        assert copied.duration_ms == 5000
        assert original.position_ms == 1000


# =============================================================================
# Tests für AudioProcessor
# =============================================================================

class TestAudioProcessor:
    """Tests für AudioProcessor Interface."""

    def test_processor_creation(self):
        """Testet Prozessor-Erstellung."""
        processor = MockProcessor(
            processor_id="test_proc",
            name="Test Processor",
            priority=ProcessorPriority.DUCKING,
        )

        assert processor.id == "test_proc"
        assert processor.name == "Test Processor"
        assert processor.priority == ProcessorPriority.DUCKING
        assert processor.enabled is True

    def test_processor_enable_disable(self):
        """Testet Aktivieren/Deaktivieren."""
        processor = MockProcessor()

        assert processor.enabled is True

        processor.enabled = False
        assert processor.enabled is False

        processor.enabled = True
        assert processor.enabled is True

    def test_process_called(self):
        """Testet, dass process aufgerufen wird."""
        processor = MockProcessor()
        chunk = b'\x00\x01\x02\x03'
        context = AudioProcessingContext()

        result = processor.process(chunk, context)

        assert processor.process_count == 1
        assert processor.last_context is context
        assert result == chunk

    def test_process_modifies_chunk(self):
        """Testet Chunk-Modifikation."""
        processor = MockProcessor(modify=True)
        chunk = b'\x00\x01\x02\x03'
        context = AudioProcessingContext()

        result = processor.process(chunk, context)

        assert result != chunk
        assert result == bytes([255, 254, 253, 252])

    def test_get_config(self):
        """Testet get_config."""
        processor = MockProcessor(
            processor_id="config_test",
            name="Config Test",
            priority=100,
        )

        config = processor.get_config()

        assert config["id"] == "config_test"
        assert config["name"] == "Config Test"
        assert config["priority"] == 100
        assert config["enabled"] is True


# =============================================================================
# Tests für ProcessorPriority
# =============================================================================

class TestProcessorPriority:
    """Tests für ProcessorPriority Enum."""

    def test_priority_order(self):
        """Testet, dass Prioritäten korrekt sortiert sind."""
        assert ProcessorPriority.FIRST < ProcessorPriority.ANALYSIS
        assert ProcessorPriority.ANALYSIS < ProcessorPriority.DUCKING
        assert ProcessorPriority.DUCKING < ProcessorPriority.CROSSFADE
        assert ProcessorPriority.CROSSFADE < ProcessorPriority.VOLUME
        assert ProcessorPriority.VOLUME < ProcessorPriority.LAST


# =============================================================================
# Tests für AudioProcessingPipeline
# =============================================================================

class TestAudioProcessingPipeline:
    """Tests für AudioProcessingPipeline."""

    @pytest.fixture
    def pipeline(self):
        """Erstellt eine leere Pipeline."""
        return AudioProcessingPipeline()

    def test_empty_pipeline(self, pipeline):
        """Testet leere Pipeline."""
        assert pipeline.count == 0
        assert len(pipeline.processors) == 0

    def test_add_processor(self, pipeline):
        """Testet Hinzufügen eines Prozessors."""
        processor = MockProcessor()
        pipeline.add(processor)

        assert pipeline.count == 1
        assert processor in pipeline.processors

    def test_add_replaces_existing(self, pipeline):
        """Testet, dass Prozessor mit gleicher ID ersetzt wird."""
        processor1 = MockProcessor(processor_id="test", name="First")
        processor2 = MockProcessor(processor_id="test", name="Second")

        pipeline.add(processor1)
        pipeline.add(processor2)

        assert pipeline.count == 1
        assert pipeline.get("test").name == "Second"

    def test_processors_sorted_by_priority(self, pipeline):
        """Testet Sortierung nach Priorität."""
        p_high = MockProcessor("high", priority=100)
        p_low = MockProcessor("low", priority=900)
        p_mid = MockProcessor("mid", priority=500)

        pipeline.add(p_low)
        pipeline.add(p_high)
        pipeline.add(p_mid)

        processors = pipeline.processors
        assert processors[0].id == "high"
        assert processors[1].id == "mid"
        assert processors[2].id == "low"

    def test_remove_processor(self, pipeline):
        """Testet Entfernen eines Prozessors."""
        processor = MockProcessor()
        pipeline.add(processor)

        result = pipeline.remove("test")

        assert result is True
        assert pipeline.count == 0

    def test_remove_nonexistent(self, pipeline):
        """Testet Entfernen nicht existierender ID."""
        result = pipeline.remove("nonexistent")
        assert result is False

    def test_get_processor(self, pipeline):
        """Testet Abrufen eines Prozessors."""
        processor = MockProcessor(processor_id="find_me")
        pipeline.add(processor)

        found = pipeline.get("find_me")
        assert found is processor

        not_found = pipeline.get("unknown")
        assert not_found is None

    def test_enable_disable(self, pipeline):
        """Testet enable/disable."""
        processor = MockProcessor(enabled=True)
        pipeline.add(processor)

        pipeline.disable("test")
        assert processor.enabled is False

        pipeline.enable("test")
        assert processor.enabled is True

    def test_clear(self, pipeline):
        """Testet clear."""
        pipeline.add(MockProcessor("a"))
        pipeline.add(MockProcessor("b"))
        pipeline.add(MockProcessor("c"))

        assert pipeline.count == 3

        pipeline.clear()

        assert pipeline.count == 0

    def test_contains(self, pipeline):
        """Testet __contains__."""
        processor = MockProcessor(processor_id="contained")
        pipeline.add(processor)

        assert "contained" in pipeline
        assert "not_contained" not in pipeline

    def test_process_empty_pipeline(self, pipeline):
        """Testet process mit leerer Pipeline."""
        chunk = b'\x00\x01\x02\x03'
        result = pipeline.process(chunk)

        assert result == chunk

    def test_process_single_processor(self, pipeline):
        """Testet process mit einem Prozessor."""
        processor = MockProcessor(modify=True)
        pipeline.add(processor)

        chunk = b'\x00\x01\x02\x03'
        result = pipeline.process(chunk)

        assert processor.process_count == 1
        assert result != chunk

    def test_process_chain(self, pipeline):
        """Testet Verarbeitungskette."""
        p1 = MockProcessor("p1", priority=100)
        p2 = MockProcessor("p2", priority=200)
        p3 = MockProcessor("p3", priority=300)

        pipeline.add(p2)
        pipeline.add(p3)
        pipeline.add(p1)

        chunk = b'\x00\x01\x02\x03'
        pipeline.process(chunk)

        assert p1.process_count == 1
        assert p2.process_count == 1
        assert p3.process_count == 1

    def test_process_skips_disabled(self, pipeline):
        """Testet, dass deaktivierte Prozessoren übersprungen werden."""
        enabled = MockProcessor("enabled", enabled=True)
        disabled = MockProcessor("disabled", enabled=False)

        pipeline.add(enabled)
        pipeline.add(disabled)

        chunk = b'\x00\x01\x02\x03'
        pipeline.process(chunk)

        assert enabled.process_count == 1
        assert disabled.process_count == 0

    def test_process_creates_context_if_none(self, pipeline):
        """Testet automatische Kontext-Erstellung."""
        processor = MockProcessor()
        pipeline.add(processor)

        pipeline.process(b'\x00')

        assert processor.last_context is not None

    def test_process_uses_provided_context(self, pipeline):
        """Testet Verwendung des übergebenen Kontexts."""
        processor = MockProcessor()
        pipeline.add(processor)

        context = AudioProcessingContext(position_ms=5000)
        pipeline.process(b'\x00', context)

        assert processor.last_context.position_ms == 5000

    def test_state_management(self, pipeline):
        """Testet Zustandsverwaltung."""
        processor = MockProcessor()
        pipeline.add(processor)

        pipeline.set_wakeword_active(True)
        pipeline.set_conversation_active(True)
        pipeline.set_tts_playing(True)

        assert ("wakeword_active", True) in processor.state_changes
        assert ("conversation_active", True) in processor.state_changes
        assert ("tts_playing", True) in processor.state_changes

    def test_track_notifications(self, pipeline):
        """Testet Track-Start/End-Benachrichtigungen."""
        processor = MockProcessor()
        pipeline.add(processor)

        context = AudioProcessingContext()

        pipeline.notify_track_start(context)
        assert processor.track_start_called is True

        pipeline.notify_track_end(context)
        assert processor.track_end_called is True

    def test_reset_all(self, pipeline):
        """Testet reset_all."""
        processor = MockProcessor()
        processor.process_count = 10
        pipeline.add(processor)

        pipeline.set_wakeword_active(True)
        pipeline.reset_all()

        assert processor.process_count == 0
        # Pipeline-Zustand auch zurückgesetzt
        context = pipeline._create_default_context()
        assert context.is_wakeword_active is False

    def test_get_status(self, pipeline):
        """Testet get_status."""
        pipeline.add(MockProcessor("a", priority=100, enabled=True))
        pipeline.add(MockProcessor("b", priority=200, enabled=False))
        pipeline.set_wakeword_active(True)

        status = pipeline.get_status()

        assert status["processor_count"] == 2
        assert status["active_count"] == 1
        assert status["is_wakeword_active"] is True
        assert len(status["processors"]) == 2


# =============================================================================
# Tests für Hilfsfunktionen
# =============================================================================

class TestAudioHelpers:
    """Tests für Audio-Hilfsfunktionen."""

    def test_apply_volume_no_change(self):
        """Testet apply_volume bei Volume 1.0."""
        chunk = b'\x00\x10\x00\x20'
        result = apply_volume(chunk, 1.0)
        assert result == chunk

    def test_apply_volume_mute(self):
        """Testet apply_volume bei Volume 0.0."""
        chunk = b'\x00\x10\x00\x20'
        result = apply_volume(chunk, 0.0)
        assert result == b'\x00\x00\x00\x00'

    def test_apply_volume_half(self):
        """Testet apply_volume bei Volume 0.5."""
        # 16-bit Sample: 0x1000 = 4096
        import struct
        chunk = struct.pack("<h", 4096)
        result = apply_volume(chunk, 0.5)
        value = struct.unpack("<h", result)[0]
        assert value == 2048

    def test_mix_chunks_equal(self):
        """Testet mix_chunks bei 50/50."""
        import struct
        chunk1 = struct.pack("<h", 1000)
        chunk2 = struct.pack("<h", 3000)

        result = mix_chunks(chunk1, chunk2, 0.5)
        value = struct.unpack("<h", result)[0]

        # 0.5 * 1000 + 0.5 * 3000 = 2000
        assert value == 2000

    def test_mix_chunks_only_first(self):
        """Testet mix_chunks mit mix=0.0."""
        import struct
        chunk1 = struct.pack("<h", 1000)
        chunk2 = struct.pack("<h", 3000)

        result = mix_chunks(chunk1, chunk2, 0.0)
        value = struct.unpack("<h", result)[0]

        assert value == 1000

    def test_mix_chunks_only_second(self):
        """Testet mix_chunks mit mix=1.0."""
        import struct
        chunk1 = struct.pack("<h", 1000)
        chunk2 = struct.pack("<h", 3000)

        result = mix_chunks(chunk1, chunk2, 1.0)
        value = struct.unpack("<h", result)[0]

        assert value == 3000

    def test_fade_chunk_in(self):
        """Testet fade_chunk fade_in."""
        import struct
        # 4 Samples
        chunk = struct.pack("<4h", 1000, 1000, 1000, 1000)

        result = fade_chunk(chunk, fade_in=True)
        values = struct.unpack("<4h", result)

        # Erstes Sample sollte 0 sein, letztes 1000
        assert values[0] == 0
        assert values[-1] == 1000

    def test_fade_chunk_out(self):
        """Testet fade_chunk fade_out."""
        import struct
        chunk = struct.pack("<4h", 1000, 1000, 1000, 1000)

        result = fade_chunk(chunk, fade_out=True)
        values = struct.unpack("<4h", result)

        # Erstes Sample sollte 1000 sein, letztes 0
        assert values[0] == 1000
        assert values[-1] == 0

    def test_fade_chunk_no_fade(self):
        """Testet fade_chunk ohne Fade."""
        chunk = b'\x00\x10\x00\x20'
        result = fade_chunk(chunk)
        assert result == chunk