kithara-studio/backend/tests/test_waveform.py

"""Tests for waveform generation module."""

import asyncio
import sys
from pathlib import Path
from unittest.mock import AsyncMock, patch, MagicMock

import pytest
import pytest_asyncio
import numpy as np

# Ensure src is importable
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src"))

from waveform import generate_waveform_svg, get_waveform_path, DOWNSAMPLE_TARGET


class TestGetWaveformPath:
    """Tests for get_waveform_path (sync path lookup)."""

    def test_returns_path_when_svg_exists(self, tmp_path):
        from waveform import WAVEFORM_DIR
        import waveform

        orig = waveform.WAVEFORM_DIR
        try:
            waveform.WAVEFORM_DIR = tmp_path
            # Create a fake SVG
            (tmp_path / "test-clip-id.svg").write_text("<svg></svg>")

            result = get_waveform_path("test-clip-id")
            assert result is not None
            assert result.exists()
            assert result.name == "test-clip-id.svg"
        finally:
            waveform.WAVEFORM_DIR = orig

    def test_returns_none_when_svg_absent(self, tmp_path):
        import waveform

        orig = waveform.WAVEFORM_DIR
        try:
            waveform.WAVEFORM_DIR = tmp_path
            result = get_waveform_path("nonexistent-id")
            assert result is None
        finally:
            waveform.WAVEFORM_DIR = orig


class TestGenerateWaveformSvg:
    """Tests for async waveform SVG generation."""

    @pytest.mark.asyncio
    async def test_raises_file_not_found_for_missing_audio(self, tmp_path):
        import waveform

        orig_clips = waveform.CLIPS_DIR
        orig_wave = waveform.WAVEFORM_DIR
        try:
            waveform.CLIPS_DIR = tmp_path / "clips"
            waveform.CLIPS_DIR.mkdir(exist_ok=True)
            waveform.WAVEFORM_DIR = tmp_path / "waveforms"
            waveform.WAVEFORM_DIR.mkdir(exist_ok=True)

            with pytest.raises(FileNotFoundError, match="Audio file not found"):
                await generate_waveform_svg("no-such-id", "no-such-file.mp3")
        finally:
            waveform.CLIPS_DIR = orig_clips
            waveform.WAVEFORM_DIR = orig_wave

    @pytest.mark.asyncio
    async def test_generates_svg_for_audio_file(self, tmp_path):
        """Test that generate_waveform_svg produces a valid SVG from a real audio file.

        This requires ffmpeg to be available on the system, since the function
        uses asyncio.create_subprocess_exec to call ffmpeg.
        """
        import subprocess

        # Skip if ffmpeg is not available
        try:
            subprocess.run(["ffmpeg", "-version"], capture_output=True, check=True)
        except (FileNotFoundError, subprocess.CalledProcessError):
            pytest.skip("ffmpeg not available on this system")

        import waveform

        clips_dir = tmp_path / "clips"
        clips_dir.mkdir()
        waveforms_dir = tmp_path / "waveforms"
        waveforms_dir.mkdir()

        orig_clips = waveform.CLIPS_DIR
        orig_wave = waveform.WAVEFORM_DIR
        try:
            waveform.CLIPS_DIR = clips_dir
            waveform.WAVEFORM_DIR = waveforms_dir

            # Generate a 2-second silent WAV with ffmpeg, then convert to MP3
            raw_wav = clips_dir / "test-clip.wav"
            mp3_path = clips_dir / "test-clip.mp3"
            subprocess.run(
                [
                    "ffmpeg", "-y", "-f", "lavfi",
                    "-i", "anullsrc=r=22050:cl=mono",
                    "-t", "2", "-q:a", "9",
                    str(mp3_path)
                ],
                capture_output=True, check=True,
            )

            result = await generate_waveform_svg("test-clip", "test-clip.mp3")

            assert result is not None
            svg_path = Path(result)
            assert svg_path.exists()
            content = svg_path.read_text()
            assert "<svg" in content
            assert "</svg>" in content
            # Should have rect elements (waveform bars)
            assert "<rect" in content
        finally:
            waveform.CLIPS_DIR = orig_clips
            waveform.WAVEFORM_DIR = orig_wave

    @pytest.mark.asyncio
    async def test_svg_has_cacheable_filename(self, tmp_path):
        """Verify SVG filename matches clip_id pattern for cacheability."""
        import waveform

        clips_dir = tmp_path / "clips"
        clips_dir.mkdir()
        waveforms_dir = tmp_path / "waveforms"
        waveforms_dir.mkdir()

        clip_id = "abc-123-def"

        # We just need to verify the naming convention
        expected_path = waveforms_dir / f"{clip_id}.svg"
        # If get_waveform_path returns None, the naming is correct but file doesn't exist yet
        # If it returns a path, it should match this pattern
        orig_wave = waveform.WAVEFORM_DIR
        try:
            waveform.WAVEFORM_DIR = waveforms_dir
            result = get_waveform_path(clip_id)
            # File doesn't exist yet, so should be None
            assert result is None
            # But the naming convention {clip_id}.svg is clear from the code
        finally:
            waveform.WAVEFORM_DIR = orig_wave


class TestDownsampling:
    """Tests for the early-downsample optimization."""

    def test_downsample_constant_is_reasonable(self):
        """The DOWNSAMPLE_TARGET should be large enough for visible waveforms
        but small enough to reduce memory."""
        assert DOWNSAMPLE_TARGET > 0
        assert DOWNSAMPLE_TARGET <= 5000  # 5K samples max for waveform
        assert DOWNSAMPLE_TARGET >= 500   # Need enough resolution for 200 bars

    def test_numpy_linspace_produces_correct_count(self):
        """Verify the downsampling logic produces exactly DOWNSAMPLE_TARGET samples."""
        # Simulate a 120s track at 22050 Hz = ~2.65M samples
        audio_data = np.random.randn(2_650_000).astype(np.float32)

        if len(audio_data) > DOWNSAMPLE_TARGET:
            # Use len-1 as upper bound to match the fixed waveform.py
            indices = np.linspace(0, len(audio_data) - 1, DOWNSAMPLE_TARGET, dtype=int)
            downsampled = audio_data[indices]

        assert len(downsampled) == DOWNSAMPLE_TARGET