47b13c3f1f
ctypes libopus encoder (src/derp/opus.py), voice varint/packet builder and stream_audio method on MumbleBot (src/derp/mumble.py), music plugin with play/stop/skip/queue/np/volume commands (plugins/music.py). Audio pipeline: yt-dlp|ffmpeg subprocess -> PCM -> Opus -> UDPTunnel. 67 new tests (1561 total).
155 lines
4.9 KiB
Python
155 lines
4.9 KiB
Python
"""Tests for the Opus ctypes wrapper."""
|
|
|
|
import math
|
|
import struct
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from derp.opus import CHANNELS, FRAME_BYTES, FRAME_SIZE, SAMPLE_RATE
|
|
|
|
# -- Helpers -----------------------------------------------------------------
|
|
|
|
|
|
def _silence() -> bytes:
|
|
"""Generate one frame of silence (1920 bytes of zeros)."""
|
|
return b"\x00" * FRAME_BYTES
|
|
|
|
|
|
def _sine_frame(freq: float = 440.0) -> bytes:
|
|
"""Generate one 20ms frame of a sine wave at the given frequency."""
|
|
samples = []
|
|
for i in range(FRAME_SIZE):
|
|
t = i / SAMPLE_RATE
|
|
val = int(16000 * math.sin(2 * math.pi * freq * t))
|
|
samples.append(struct.pack("<h", max(-32768, min(32767, val))))
|
|
return b"".join(samples)
|
|
|
|
|
|
# -- Mock libopus for unit testing without system library --------------------
|
|
|
|
|
|
def _mock_lib():
|
|
"""Build a mock ctypes CDLL that simulates libopus calls."""
|
|
lib = MagicMock()
|
|
lib.opus_encoder_get_size.return_value = 256
|
|
lib.opus_encoder_init.return_value = 0
|
|
lib.opus_encoder_ctl.return_value = 0
|
|
lib.opus_encode.return_value = 10 # 10 bytes output
|
|
return lib
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_lib_cache():
|
|
"""Reset the cached _lib before each test."""
|
|
import derp.opus as _mod
|
|
old = _mod._lib
|
|
_mod._lib = None
|
|
yield
|
|
_mod._lib = old
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestOpusConstants
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestOpusConstants:
|
|
def test_sample_rate(self):
|
|
assert SAMPLE_RATE == 48000
|
|
|
|
def test_channels(self):
|
|
assert CHANNELS == 1
|
|
|
|
def test_frame_size(self):
|
|
assert FRAME_SIZE == 960
|
|
|
|
def test_frame_bytes(self):
|
|
assert FRAME_BYTES == FRAME_SIZE * CHANNELS * 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestOpusEncoder (mocked libopus)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestOpusEncoder:
|
|
def test_encode_silence(self):
|
|
"""Encoding silence produces bytes output."""
|
|
lib = _mock_lib()
|
|
with patch("derp.opus._load_lib", return_value=lib):
|
|
from derp.opus import OpusEncoder
|
|
enc = OpusEncoder()
|
|
result = enc.encode(_silence())
|
|
assert isinstance(result, bytes)
|
|
assert len(result) > 0
|
|
enc.close()
|
|
|
|
def test_encode_sine(self):
|
|
"""Encoding a sine wave produces bytes output."""
|
|
lib = _mock_lib()
|
|
with patch("derp.opus._load_lib", return_value=lib):
|
|
from derp.opus import OpusEncoder
|
|
enc = OpusEncoder()
|
|
result = enc.encode(_sine_frame())
|
|
assert isinstance(result, bytes)
|
|
enc.close()
|
|
|
|
def test_encode_wrong_size(self):
|
|
"""Passing wrong buffer size raises ValueError."""
|
|
lib = _mock_lib()
|
|
with patch("derp.opus._load_lib", return_value=lib):
|
|
from derp.opus import OpusEncoder
|
|
enc = OpusEncoder()
|
|
with pytest.raises(ValueError, match="expected 1920"):
|
|
enc.encode(b"\x00" * 100)
|
|
enc.close()
|
|
|
|
def test_encode_multi_frame(self):
|
|
"""Multiple sequential encodes work."""
|
|
lib = _mock_lib()
|
|
with patch("derp.opus._load_lib", return_value=lib):
|
|
from derp.opus import OpusEncoder
|
|
enc = OpusEncoder()
|
|
for _ in range(5):
|
|
result = enc.encode(_silence())
|
|
assert isinstance(result, bytes)
|
|
enc.close()
|
|
|
|
def test_custom_bitrate(self):
|
|
"""Custom bitrate is passed to opus_encoder_ctl."""
|
|
lib = _mock_lib()
|
|
with patch("derp.opus._load_lib", return_value=lib):
|
|
from derp.opus import OpusEncoder
|
|
enc = OpusEncoder(bitrate=96000)
|
|
assert lib.opus_encoder_ctl.called
|
|
enc.close()
|
|
|
|
def test_init_failure(self):
|
|
"""RuntimeError on encoder init failure."""
|
|
lib = _mock_lib()
|
|
lib.opus_encoder_init.return_value = -1
|
|
with patch("derp.opus._load_lib", return_value=lib):
|
|
from derp.opus import OpusEncoder
|
|
with pytest.raises(RuntimeError, match="opus_encoder_init"):
|
|
OpusEncoder()
|
|
|
|
def test_encode_failure(self):
|
|
"""RuntimeError on encode failure."""
|
|
lib = _mock_lib()
|
|
lib.opus_encode.return_value = -1
|
|
with patch("derp.opus._load_lib", return_value=lib):
|
|
from derp.opus import OpusEncoder
|
|
enc = OpusEncoder()
|
|
with pytest.raises(RuntimeError, match="opus_encode"):
|
|
enc.encode(_silence())
|
|
|
|
def test_close_clears_state(self):
|
|
"""close() sets internal state to None."""
|
|
lib = _mock_lib()
|
|
with patch("derp.opus._load_lib", return_value=lib):
|
|
from derp.opus import OpusEncoder
|
|
enc = OpusEncoder()
|
|
enc.close()
|
|
assert enc._state is None
|