admin
6dfe390bda
- GUI process peek ring SHM đầu vào (không consume, không tranh chấp RealtimeFxLoop) - gọi fx->processAudio() trên instance GUI để meter/UI plugin chuyển động theo âm thật - performEdit forward sang controller -> meter điều khiển bởi param cũng cập nhật - bridge /shm: frontend đẩy tên session live vào mỗi GUI đang mở (start/stop/open) - tests: 127 passed
341 lines
15 KiB
Python
341 lines
15 KiB
Python
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_shm_written_to_job(self):
|
|
"""shm (tên session realtime) phải vào job.json → bridge fx-gui mở SHM
|
|
để feed audio vào instance GUI (meter nhảy theo âm thật)."""
|
|
token = get_admin_token()
|
|
if not token:
|
|
pytest.skip("Cannot get admin token")
|
|
import tempfile as _tf
|
|
fd, known = _tf.mkstemp(prefix="sf_fx_gui_test_", suffix=".json")
|
|
fake_proc = MagicMock()
|
|
fake_proc.stdout = iter(["SF_FXGUI_PORT=54322\n"])
|
|
from app.api.v1 import plugins as plugins_mod
|
|
try:
|
|
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), \
|
|
patch("app.api.v1.plugins.tempfile.mkstemp", return_value=(fd, known)):
|
|
resp = client.post("/api/v1/plugins/fx-gui",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
json={"path": "C:\\VST\\Ozone11.vst3", "name": "Ozone 11",
|
|
"embed": True, "shm": "SonicForge_FXRT_abc123"})
|
|
assert resp.status_code == 200
|
|
with open(known, "r", encoding="utf-8") as f:
|
|
job = json.load(f)
|
|
assert job.get("shm") == "SonicForge_FXRT_abc123"
|
|
finally:
|
|
plugins_mod._FX_GUI_PROCESSES.clear()
|
|
if os.path.exists(known):
|
|
try:
|
|
os.remove(known)
|
|
except OSError:
|
|
pass
|
|
|
|
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
|