73 lines
2.6 KiB
Python
73 lines
2.6 KiB
Python
"""Phase 1 chain hợp nhất + control protocol (PLAN_DAW_A.md).
|
|
|
|
Unit test không spawn bridge: dùng object.__new__ để bỏ qua __init__/subprocess,
|
|
chỉ test logic thuần của FxRealtimeSession.
|
|
"""
|
|
import threading
|
|
|
|
from app.core import fx_realtime
|
|
|
|
|
|
def make_session(chain):
|
|
s = object.__new__(fx_realtime.FxRealtimeSession)
|
|
s.chain = list(chain)
|
|
s._lock = threading.Lock()
|
|
s.pending_params = {}
|
|
s.latencies = {}
|
|
return s
|
|
|
|
|
|
CHAIN = [
|
|
{"type": "vst3", "path": "C:\\Plugins\\Eq.vst3",
|
|
"preset_b64": "AAA=", "bypass": False, "active": True},
|
|
{"type": "eq", "params": {"g1": 0.5}, "bypass": False, "active": True},
|
|
{"type": "compressor", "params": {"threshold": -12.0, "ratio": 4.0},
|
|
"bypass": True, "active": True},
|
|
{"type": "vst3", "path": "C:\\Plugins\\Disabled.vst3", "active": False},
|
|
{"type": "limiter", "params": {}, "bypass": False, "active": True},
|
|
]
|
|
|
|
|
|
def test_chain_json_keeps_order_and_skips_inactive():
|
|
"""_chain_json_for_bridge: giữ NGUYÊN thứ tự UI (vst3+builtin xen kẽ),
|
|
loại slot active=false, giữ type/path/preset_b64/params/bypass."""
|
|
s = make_session(CHAIN)
|
|
out = s._chain_json_for_bridge()
|
|
assert [o["type"] for o in out] == ["vst3", "eq", "compressor", "limiter"]
|
|
assert out[0]["path"] == "C:\\Plugins\\Eq.vst3"
|
|
assert out[0]["preset_b64"] == "AAA="
|
|
assert out[1]["params"] == {"g1": 0.5}
|
|
assert out[2]["bypass"] is True
|
|
assert out[2]["params"] == {"threshold": -12.0, "ratio": 4.0}
|
|
# params rỗng → không đưa key params (khớp job schema)
|
|
assert out[3]["type"] == "limiter"
|
|
assert "params" not in out[3]
|
|
|
|
|
|
def test_chain_json_handles_malformed_slots():
|
|
s = make_session([None, "x", {"active": False}, {}, {"type": "eq"}])
|
|
out = s._chain_json_for_bridge()
|
|
assert [o["type"] for o in out] == ["eq"]
|
|
|
|
|
|
def test_set_param_report_latency_drain_roundtrip():
|
|
"""Control protocol Phase 1: set_param/report_latency/drain hoạt động
|
|
đúng (engine nhận từ WS text, giữ pending_params + latencies)."""
|
|
s = make_session([])
|
|
s.set_param(1, "g1", 0.75)
|
|
s.set_param(1, "g2", -1.5)
|
|
s.set_param(3, "threshold", -20.0)
|
|
s.report_latency(1, 128)
|
|
s.report_latency(3, 256)
|
|
assert s.get_latencies() == {1: 128, 3: 256}
|
|
assert s.drain_params() == {1: {"g1": 0.75, "g2": -1.5},
|
|
3: {"threshold": -20.0}}
|
|
assert s.drain_params() == {}
|
|
|
|
|
|
def test_set_param_overwrites_same_key():
|
|
s = make_session([])
|
|
s.set_param(0, "wet", 0.5)
|
|
s.set_param(0, "wet", 0.9)
|
|
assert s.drain_params() == {0: {"wet": 0.9}}
|