import os import json import pytest from fastapi.testclient import TestClient from unittest.mock import patch, MagicMock from app.main import app client = TestClient(app) def get_admin_token(): # test_auth_and_quota.py có thể đã rotate password; thử cả 2. for pwd in ("admin123", "admin_new_password_2026"): resp = client.post("/api/v1/auth/login", json={"username": "admin", "password": pwd}) if resp.status_code != 200: continue token = resp.json()["access_token"] # Admin is seeded with must_change_password=1; the app blocks music # processing until the first password change. Complete that flow here # (keeping the same password) so feature tests run unblocked. user = resp.json()["user"] if user.get("must_change_password"): r = client.post("/api/v1/auth/change-password", headers={"Authorization": f"Bearer {token}"}, json={"old_password": pwd, "new_password": pwd}) if r.status_code == 200: token = r.json()["access_token"] return token return None class TestPluginAPI: def test_fx_render_builtin_gain(self): """POST /api/v1/plugins/fx-render với builtin gain (bridge --render-fx), không cần VST — output WAV mới trong PROCESSED_DIR.""" import numpy as np import soundfile as sf from app.core import native_render from app.config import settings exe = native_render.find_bridge_exe() if not exe: pytest.skip("bridge exe missing") token = get_admin_token() if not token: pytest.skip("no admin token") h = {"Authorization": f"Bearer {token}"} sr = 44100 t = np.linspace(0, 0.5, int(sr * 0.5), endpoint=False) data = (0.05 * np.sin(2 * np.pi * 440 * t)).astype(np.float32) os.makedirs(settings.PROCESSED_DIR, exist_ok=True) inp = os.path.join(settings.PROCESSED_DIR, "fx_api_in_test.wav") sf.write(inp, data, sr, subtype="FLOAT") try: resp = client.post("/api/v1/plugins/fx-render", headers=h, json={ "input_file_id": "fx_api_in_test.wav", "fx_chain": [{"type": "builtin", "id": "gain", "params": {"db": 6.0}}], "sample_rate": sr, }) assert resp.status_code == 200, resp.text[-500:] body = resp.json() assert body["success"] is True assert os.path.isfile(body["path"]) d, _ = sf.read(body["path"], dtype="float32", always_2d=True) assert float(np.max(np.abs(d))) > 0.07 # ~2x 0.05 finally: client.cookies.clear() # không làm hỏng test requires_auth kế tiếp for f in (inp, os.path.join(settings.PROCESSED_DIR, "fx_api_out_test.wav")): if os.path.exists(f): try: os.remove(f) except Exception: pass def test_list_plugins_requires_auth(self): resp = client.get("/api/v1/plugins/available") assert resp.status_code in (401, 403) def test_list_plugins_authenticated(self): token = get_admin_token() if not token: pytest.skip("Cannot get admin token") resp = client.get("/api/v1/plugins/available", headers={"Authorization": f"Bearer {token}"}) assert resp.status_code == 200 data = resp.json() assert "vst_instruments" in data assert "soundfonts" in data def test_list_default_soundfonts(self): resp = client.get("/api/v1/plugins/default-soundfonts") assert resp.status_code == 200 data = resp.json() assert isinstance(data, list) def test_render_project_invalid_json(self): token = get_admin_token() if not token: pytest.skip("Cannot get admin token") resp = client.post("/api/v1/plugins/render", headers={"Authorization": f"Bearer {token}"}, json={"project_json": {}}) # Should fail because project is empty, but API should return 500 or error assert resp.status_code in (400, 422, 500) def test_upload_soundfont_requires_auth(self): resp = client.post("/api/v1/plugins/upload-soundfont") assert resp.status_code in (401, 403, 422) def test_upload_soundfont_invalid_magic(self): token = get_admin_token() if not token: pytest.skip("Cannot get admin token") # Upload a file with invalid magic bytes fake_content = b'XXXX\x00\x00\x00\x00YYYY' * 100 resp = client.post( "/api/v1/plugins/upload-soundfont", headers={"Authorization": f"Bearer {token}"}, files={"file": ("fake.sf2", fake_content, "application/octet-stream")} ) assert resp.status_code == 400 assert "Invalid SoundFont" in resp.json().get("detail", "") def test_upload_soundfont_valid_magic(self): token = get_admin_token() if not token: pytest.skip("Cannot get admin token") # Upload a file with valid RIFF + sfbk magic valid_content = b'RIFF\x00\x00\x00\x00sfbk' + b'\x00' * 200 resp = client.post( "/api/v1/plugins/upload-soundfont", headers={"Authorization": f"Bearer {token}"}, files={"file": ("test.sf2", valid_content, "application/octet-stream")} ) # Should succeed (200) unless auth/permission issues if resp.status_code == 200: data = resp.json() assert "id" in data assert "size_bytes" in data assert data["size_bytes"] == len(valid_content) elif resp.status_code == 403: pytest.skip("Permission denied for admin user") def test_plugin_dirs_save_list(self): """plugin_dirs (list) — save/get/effective + scan phân loại riêng rẽ.""" token = get_admin_token() if not token: pytest.skip("Cannot get admin token") h = {"Authorization": f"Bearer {token}"} import tempfile with tempfile.TemporaryDirectory() as td: # 2 dir giả: 1 chứa VST, 1 chứa SoundFont vst_dir = os.path.join(td, "vsts") sf_dir = os.path.join(td, "sfs") os.makedirs(vst_dir) os.makedirs(sf_dir) open(os.path.join(vst_dir, "Synth1.vst3"), "w").write("x") open(os.path.join(sf_dir, "piano.sf2"), "w").write("x") # Save list r = client.post("/api/v1/plugins/dirs", headers=h, json={"plugin_dirs": [vst_dir, sf_dir]}) assert r.status_code == 200 assert r.json()["plugin_dirs"] == [vst_dir, sf_dir] # Get lại g = client.get("/api/v1/plugins/dirs", headers=h) assert g.json()["plugin_dirs"] == [vst_dir, sf_dir] # Scan → phân loại riêng rẽ s = client.post("/api/v1/plugins/scan", headers=h) assert s.status_code == 200 data = s.json() assert any(v["name"] == "Synth1" for v in data["vst_found"]) assert any(x["name"] == "piano" for x in data["soundfonts"]) assert data["vst_count"] == 1 assert data["soundfont_count"] == 1 # Mỗi entry có dir gốc assert data["vst_found"][0]["dir"] == vst_dir assert data["soundfonts"][0]["dir"] == sf_dir def test_plugin_dirs_remove(self): """Xóa 1 dir khỏi list → save lại → không còn trong effective.""" token = get_admin_token() if not token: pytest.skip("Cannot get admin token") h = {"Authorization": f"Bearer {token}"} import tempfile with tempfile.TemporaryDirectory() as td: d1 = os.path.join(td, "d1") d2 = os.path.join(td, "d2") os.makedirs(d1) os.makedirs(d2) client.post("/api/v1/plugins/dirs", headers=h, json={"plugin_dirs": [d1, d2]}) client.post("/api/v1/plugins/dirs", headers=h, json={"plugin_dirs": [d1]}) g = client.get("/api/v1/plugins/dirs", headers=h) assert g.json()["plugin_dirs"] == [d1] class TestMidiRenderVSTi: """Preview/export MIDI notes với âm VSTi — feature Carla bridge → pedalboard.""" def test_midi_render_requires_auth(self): client.cookies.clear() # TestClient giữ cookie login từ test trước resp = client.post("/api/v1/plugins/midi-render", json={"instrument_id": "x", "notes": [{"pitch": 60}]}) assert resp.status_code == 401 def test_midi_render_no_notes(self): token = get_admin_token() if not token: pytest.skip("Cannot get admin token") resp = client.post("/api/v1/plugins/midi-render", headers={"Authorization": f"Bearer {token}"}, json={"instrument_id": "x", "notes": []}) assert resp.status_code == 400 def test_midi_render_unknown_instrument(self): token = get_admin_token() if not token: pytest.skip("Cannot get admin token") resp = client.post("/api/v1/plugins/midi-render", headers={"Authorization": f"Bearer {token}"}, json={"instrument_id": "NoSuchPluginXYZ", "notes": [{"pitch": 60, "start_beat": 0, "duration_beats": 1, "velocity": 0.8}]}) assert resp.status_code == 404 assert "VSTi" in resp.json()["detail"] def test_carla_play_notes_requires_auth(self): client.cookies.clear() # TestClient giữ cookie login từ test trước resp = client.post("/api/v1/plugins/carla-play-notes", json={"notes": [{"pitch": 60}], "bpm": 120}) assert resp.status_code == 401 def test_carla_play_notes_no_notes(self): token = get_admin_token() if not token: pytest.skip("Cannot get admin token") resp = client.post("/api/v1/plugins/carla-play-notes", headers={"Authorization": f"Bearer {token}"}, json={"notes": [], "bpm": 120}) assert resp.status_code == 400 def test_carla_play_notes_no_carla(self): # Máy test không có Carla local → 409 hướng dẫn định vị/mở Carla. token = get_admin_token() if not token: pytest.skip("Cannot get admin token") resp = client.post("/api/v1/plugins/carla-play-notes", headers={"Authorization": f"Bearer {token}"}, json={"notes": [{"pitch": 60, "start_beat": 0, "duration_beats": 1}], "bpm": 120}) assert resp.status_code == 409 # --- fx-gui embed (PLAN_MASTERBUS_VST_GUI_EMBED.md Phase 1) -------------- def test_fx_gui_open_embed_returns_embed_url(self): token = get_admin_token() if not token: pytest.skip("Cannot get admin token") fake_proc = MagicMock() fake_proc.stdout = iter(["some log\n", "SF_FXGUI_PORT=54321\n"]) with patch("app.core.native_render.find_bridge_exe", return_value="C:\\bridge\\daw_vst_bridge.exe"), \ patch("app.api.v1.plugins.subprocess.Popen", return_value=fake_proc): resp = client.post("/api/v1/plugins/fx-gui", headers={"Authorization": f"Bearer {token}"}, json={"path": "C:\\VST\\Ozone11.vst3", "name": "Ozone 11", "embed": True}) assert resp.status_code == 200 data = resp.json() assert data["embed_url"] == "http://127.0.0.1:54321" assert data["started"] is True # registry có process + path để close tìm thấy from app.api.v1 import plugins as plugins_mod assert any(getattr(p, "_sf_plugin_path", None) == "C:\\VST\\Ozone11.vst3" for p in plugins_mod._FX_GUI_PROCESSES) plugins_mod._FX_GUI_PROCESSES.clear() def test_fx_gui_open_no_port_is_error(self): token = get_admin_token() if not token: pytest.skip("Cannot get admin token") fake_proc = MagicMock() fake_proc.stdout = iter(["never prints port\n"]) with patch("app.core.native_render.find_bridge_exe", return_value="C:\\bridge\\daw_vst_bridge.exe"), \ patch("app.api.v1.plugins.subprocess.Popen", return_value=fake_proc): resp = client.post("/api/v1/plugins/fx-gui", headers={"Authorization": f"Bearer {token}"}, json={"path": "C:\\VST\\Ozone11.vst3", "embed": True}) assert resp.status_code == 500 assert "port" in resp.json()["detail"] fake_proc.terminate.assert_called_once() def test_fx_gui_close_kills_process(self): token = get_admin_token() if not token: pytest.skip("Cannot get admin token") from app.api.v1 import plugins as plugins_mod proc = MagicMock() proc.poll.return_value = None proc._sf_plugin_path = "C:\\VST\\Ozone11.vst3" plugins_mod._FX_GUI_PROCESSES.append(proc) try: resp = client.post("/api/v1/plugins/fx-gui/close", headers={"Authorization": f"Bearer {token}"}, json={"path": "C:\\VST\\Ozone11.vst3"}) finally: plugins_mod._FX_GUI_PROCESSES.clear() assert resp.status_code == 200 assert resp.json()["closed"] is True proc.terminate.assert_called_once() def test_fx_gui_close_unknown_plugin(self): token = get_admin_token() if not token: pytest.skip("Cannot get admin token") resp = client.post("/api/v1/plugins/fx-gui/close", headers={"Authorization": f"Bearer {token}"}, json={"path": "C:\\VST\\NoSuchPlugin.vst3"}) assert resp.status_code == 200 assert resp.json()["closed"] is False