SonicForgeStudio/tests/test_vst_engine.py

import os
import json
import pytest
import numpy as np
from unittest.mock import patch, MagicMock

from app.core.vst_engine import PluginManager, midi_note_to_freq, render_midi_events_to_audio


class TestPluginManager:
    def test_init(self):
        pm = PluginManager()
        if os.name == "nt":
            pf = os.environ.get("ProgramFiles", r"C:\Program Files")
            assert pm.vst_dir == os.path.join(pf, "Common Files", "VST3")
            root = os.environ.get("APPDATA") or os.path.expanduser("~")
            assert pm.sf_dir == os.path.join(root, "SonicForgeDAW", "soundfonts")
        else:
            assert pm.vst_dir == "/opt/daw_engine/vst3"
            assert pm.sf_dir == "/opt/daw_engine/soundfonts"

    def test_midi_events_to_messages(self):
        events = [{"note": 60, "start_beat": 0, "duration_beats": 1, "velocity": 100}]
        bpm = 120
        sr = 44100
        msgs = PluginManager.midi_events_to_messages(events, bpm, sr)
        beat_sec = 60.0 / 120
        # pedalboard >= 0.9: message = (bytes raw MIDI, timestamp_seconds)
        assert msgs[0] == (bytes([0x90, 60, 100]), 0.0)
        assert msgs[1] == (bytes([0x80, 60, 0]), beat_sec)

    def test_list_available_empty(self):
        pm = PluginManager(vst_dir="/tmp/nonexistent_vst_dir_xyz", sf_dir="/tmp/nonexistent_sf_dir_xyz")
        available = pm.list_available()
        assert "vst_instruments" in available
        assert "soundfonts" in available
        assert available["vst_instruments"] == []
        assert available["soundfonts"] == []

    def test_validate_sf2_header_valid(self):
        # Valid RIFF + sfbk header
        valid = b'RIFF' + b'\x00' * 4 + b'sfbk' + b'\x00' * 100
        assert PluginManager.validate_sf2_header(valid) is True

    def test_validate_sf2_header_invalid_no_riff(self):
        invalid = b'XXXX' + b'\x00' * 4 + b'sfbk' + b'\x00' * 100
        assert PluginManager.validate_sf2_header(invalid) is False

    def test_validate_sf2_header_invalid_no_sfbk(self):
        invalid = b'RIFF' + b'\x00' * 4 + b'XXXX' + b'\x00' * 100
        assert PluginManager.validate_sf2_header(invalid) is False

    def test_validate_sf2_header_too_short(self):
        assert PluginManager.validate_sf2_header(b'RIFF') is False


class TestFxScanCache:
    """list_scan_all cache TTL + invalidate — fix MASTERING PANEL không hiển thị
    VST FX đã scan trong Plugin Manager (cache kẹt [] vĩnh viễn)."""

    SCAN_JSON = json.dumps({"plugins": [
        {"path": "C:\\VST3\\Ozone Pro.vst3", "name": "Ozone Pro",
         "type": "VST3", "is_fx": True, "is_instrument": False},
    ]})

    def _scan_json(self, d):
        """JSON scan với path plugin THẬT tồn tại trên đĩa — list_scan_all lọc
        entry plugin đã xóa khỏi đĩa (os.path.exists) trước khi cache; path ảo
        (C:\\VST3\\Ozone Pro.vst3) bị lọc hết → test không thể assert entry."""
        return json.dumps({"plugins": [
            {"path": os.path.join(d, "Ozone Pro.vst3"), "name": "Ozone Pro",
             "type": "VST3", "is_fx": True, "is_instrument": False},
        ]})

    def _pm(self, tmp_path):
        d = str(tmp_path)
        return PluginManager(vst_dir=d, sf_dir=d, extra_vst_dirs=[])

    def test_scan_caches_with_bridge(self, tmp_path):
        pm = self._pm(tmp_path)
        # Plugin phải tồn tại trên đĩa — list_scan_all lọc entry chết trước cache.
        open(os.path.join(str(tmp_path), "Ozone Pro.vst3"), "w").close()
        with patch("app.core.native_render.find_fx_bridge_exe", return_value="fx_bridge.exe"), \
             patch("subprocess.run", return_value=MagicMock(
                 returncode=0, stdout=self._scan_json(str(tmp_path)), stderr="")) as mrun:
            r1 = pm.list_scan_all()
            assert len(r1) == 1 and r1[0]["name"] == "Ozone Pro"
            assert pm._fx_scan_all is not None and len(pm._fx_scan_all) == 1
            # Lần 2 trong TTL → KHÔNG chạy lại bridge
            mrun.reset_mock()
            r2 = pm.list_scan_all()
            assert r2 == r1
            mrun.assert_not_called()

    def test_scan_no_bridge_does_not_cache_empty(self, tmp_path):
        """Bridge chưa tồn tại → trả [] nhưng KHÔNG cache → lần sau thử lại
        (trước đây cache [] vĩnh viễn → MASTERING PANEL trống)."""
        pm = self._pm(tmp_path)
        with patch("app.core.native_render.find_fx_bridge_exe", return_value=""):
            assert pm.list_scan_all() == []
        assert pm._fx_scan_all is None  # KHÔNG cache rỗng
        # Có bridge → scan thật chạy (plugin tồn tại trên đĩa)
        open(os.path.join(str(tmp_path), "Ozone Pro.vst3"), "w").close()
        with patch("app.core.native_render.find_fx_bridge_exe", return_value="fx_bridge.exe"), \
             patch("subprocess.run", return_value=MagicMock(
                 returncode=0, stdout=self._scan_json(str(tmp_path)), stderr="")):
            r = pm.list_scan_all()
            assert len(r) == 1

    def test_scan_ttl_expiry_rescans(self, tmp_path):
        pm = self._pm(tmp_path)
        with patch("app.core.native_render.find_fx_bridge_exe", return_value="fx_bridge.exe"), \
             patch("subprocess.run", return_value=MagicMock(
                 returncode=0, stdout=self.SCAN_JSON, stderr="")) as mrun:
            pm.list_scan_all()
            mrun.assert_called_once()
            # Giả TTL đã hết hạn → gọi lại phải scan
            pm._fx_scan_all_ts -= (pm.FX_SCAN_TTL + 1)
            pm.list_scan_all()
            assert mrun.call_count == 2

    def test_invalidate_fx_scan(self, tmp_path):
        pm = self._pm(tmp_path)
        with patch("app.core.native_render.find_fx_bridge_exe", return_value="fx_bridge.exe"), \
             patch("subprocess.run", return_value=MagicMock(
                 returncode=0, stdout=self.SCAN_JSON, stderr="")):
            pm.list_scan_all()
            assert pm._fx_scan_all is not None
        pm.invalidate_fx_scan()
        assert pm._fx_scan_all is None
        assert pm._fx_scan_all_ts == 0.0
        assert pm._fx_scan_cache is None


class TestMidiNoteToFreq:
    def test_a4_440(self):
        assert midi_note_to_freq(69) == 440.0

    def test_c4(self):
        # C4 = MIDI 60 = 261.63
        freq = midi_note_to_freq(60)
        assert abs(freq - 261.63) < 0.5

    def test_note_zero_to_freq(self):
        freq = midi_note_to_freq(0)
        assert freq > 0 and freq < 10


class TestRenderMidiToAudio:
    def test_render_single_note(self):
        events = [{"note": 60, "start_beat": 0, "duration_beats": 1, "velocity": 100}]
        audio = render_midi_events_to_audio(events, sr=44100, bpm=120)
        assert audio.shape[0] == 2  # Stereo
        assert audio.shape[1] > 0
        # Should have non-zero samples
        assert np.max(np.abs(audio)) > 0

    def test_render_empty_events(self):
        audio = render_midi_events_to_audio([], sr=44100, bpm=120)
        assert audio.shape[0] == 2  # Stereo
        # Empty events defaults to 2 seconds of silence (minimum length)
        assert audio.shape[1] >= 0

    def test_render_multiple_notes(self):
        events = [
            {"note": 60, "start_beat": 0, "duration_beats": 1, "velocity": 100},
            {"note": 64, "start_beat": 1, "duration_beats": 1, "velocity": 80},
            {"note": 67, "start_beat": 2, "duration_beats": 1, "velocity": 90},
        ]
        audio = render_midi_events_to_audio(events, sr=44100, bpm=120)
        assert audio.shape[0] == 2
        assert audio.shape[1] > 44100 * 1