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).
354 lines
11 KiB
Python
354 lines
11 KiB
Python
"""Tests for the music playback plugin."""
|
|
|
|
import asyncio
|
|
import importlib.util
|
|
import sys
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
# -- Load plugin module directly ---------------------------------------------
|
|
|
|
_spec = importlib.util.spec_from_file_location("music", "plugins/music.py")
|
|
_mod = importlib.util.module_from_spec(_spec)
|
|
sys.modules["music"] = _mod
|
|
_spec.loader.exec_module(_mod)
|
|
|
|
|
|
# -- Fakes -------------------------------------------------------------------
|
|
|
|
|
|
class _FakeState:
|
|
def __init__(self):
|
|
self._store: dict[str, dict[str, str]] = {}
|
|
|
|
def get(self, ns: str, key: str) -> str | None:
|
|
return self._store.get(ns, {}).get(key)
|
|
|
|
def set(self, ns: str, key: str, value: str) -> None:
|
|
self._store.setdefault(ns, {})[key] = value
|
|
|
|
def delete(self, ns: str, key: str) -> None:
|
|
self._store.get(ns, {}).pop(key, None)
|
|
|
|
def keys(self, ns: str) -> list[str]:
|
|
return list(self._store.get(ns, {}).keys())
|
|
|
|
|
|
class _FakeBot:
|
|
"""Minimal bot for music plugin testing."""
|
|
|
|
def __init__(self, *, mumble: bool = True):
|
|
self.sent: list[tuple[str, str]] = []
|
|
self.replied: list[str] = []
|
|
self.state = _FakeState()
|
|
self._pstate: dict = {}
|
|
self._tasks: set[asyncio.Task] = set()
|
|
if mumble:
|
|
self.stream_audio = AsyncMock()
|
|
|
|
async def send(self, target: str, text: str) -> None:
|
|
self.sent.append((target, text))
|
|
|
|
async def reply(self, message, text: str) -> None:
|
|
self.replied.append(text)
|
|
|
|
def _is_admin(self, message) -> bool:
|
|
return False
|
|
|
|
def _spawn(self, coro, *, name=None):
|
|
task = asyncio.ensure_future(coro)
|
|
self._tasks.add(task)
|
|
task.add_done_callback(self._tasks.discard)
|
|
return task
|
|
|
|
|
|
class _Msg:
|
|
"""Minimal message object."""
|
|
|
|
def __init__(self, text="!play url", nick="Alice", target="0",
|
|
is_channel=True):
|
|
self.text = text
|
|
self.nick = nick
|
|
self.target = target
|
|
self.is_channel = is_channel
|
|
self.prefix = nick
|
|
self.command = "PRIVMSG"
|
|
self.params = [target, text]
|
|
self.tags = {}
|
|
self.raw = {}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestMumbleGuard
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestMumbleGuard:
|
|
def test_is_mumble_true(self):
|
|
bot = _FakeBot(mumble=True)
|
|
assert _mod._is_mumble(bot) is True
|
|
|
|
def test_is_mumble_false(self):
|
|
bot = _FakeBot(mumble=False)
|
|
assert _mod._is_mumble(bot) is False
|
|
|
|
def test_play_non_mumble(self):
|
|
bot = _FakeBot(mumble=False)
|
|
msg = _Msg(text="!play https://example.com")
|
|
asyncio.run(_mod.cmd_play(bot, msg))
|
|
assert any("Mumble-only" in r for r in bot.replied)
|
|
|
|
def test_stop_non_mumble_silent(self):
|
|
bot = _FakeBot(mumble=False)
|
|
msg = _Msg(text="!stop")
|
|
asyncio.run(_mod.cmd_stop(bot, msg))
|
|
assert bot.replied == []
|
|
|
|
def test_skip_non_mumble_silent(self):
|
|
bot = _FakeBot(mumble=False)
|
|
msg = _Msg(text="!skip")
|
|
asyncio.run(_mod.cmd_skip(bot, msg))
|
|
assert bot.replied == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestPlayCommand
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestPlayCommand:
|
|
def test_play_no_url(self):
|
|
bot = _FakeBot()
|
|
msg = _Msg(text="!play")
|
|
asyncio.run(_mod.cmd_play(bot, msg))
|
|
assert any("Usage" in r for r in bot.replied)
|
|
|
|
def test_play_queues_track(self):
|
|
bot = _FakeBot()
|
|
msg = _Msg(text="!play https://example.com/track")
|
|
with patch.object(_mod, "_resolve_title", return_value="Test Track"):
|
|
with patch.object(_mod, "_ensure_loop"):
|
|
asyncio.run(_mod.cmd_play(bot, msg))
|
|
assert any("Playing" in r for r in bot.replied)
|
|
ps = _mod._ps(bot)
|
|
assert len(ps["queue"]) == 1
|
|
assert ps["queue"][0].title == "Test Track"
|
|
|
|
def test_play_shows_queued_when_busy(self):
|
|
bot = _FakeBot()
|
|
ps = _mod._ps(bot)
|
|
ps["current"] = _mod._Track(
|
|
url="x", title="Current", requester="Bob",
|
|
)
|
|
msg = _Msg(text="!play https://example.com/next")
|
|
with patch.object(_mod, "_resolve_title", return_value="Next Track"):
|
|
asyncio.run(_mod.cmd_play(bot, msg))
|
|
assert any("Queued" in r for r in bot.replied)
|
|
|
|
def test_play_queue_full(self):
|
|
bot = _FakeBot()
|
|
ps = _mod._ps(bot)
|
|
ps["queue"] = [
|
|
_mod._Track(url="x", title="t", requester="a")
|
|
for _ in range(_mod._MAX_QUEUE)
|
|
]
|
|
msg = _Msg(text="!play https://example.com/overflow")
|
|
with patch.object(_mod, "_resolve_title", return_value="Overflow"):
|
|
asyncio.run(_mod.cmd_play(bot, msg))
|
|
assert any("full" in r.lower() for r in bot.replied)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestStopCommand
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestStopCommand:
|
|
def test_stop_clears_queue(self):
|
|
bot = _FakeBot()
|
|
ps = _mod._ps(bot)
|
|
ps["queue"] = [_mod._Track(url="x", title="t", requester="a")]
|
|
ps["current"] = _mod._Track(url="y", title="s", requester="b")
|
|
msg = _Msg(text="!stop")
|
|
asyncio.run(_mod.cmd_stop(bot, msg))
|
|
assert ps["queue"] == []
|
|
assert ps["current"] is None
|
|
assert any("Stopped" in r for r in bot.replied)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestSkipCommand
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSkipCommand:
|
|
def test_skip_nothing_playing(self):
|
|
bot = _FakeBot()
|
|
msg = _Msg(text="!skip")
|
|
asyncio.run(_mod.cmd_skip(bot, msg))
|
|
assert any("Nothing" in r for r in bot.replied)
|
|
|
|
def test_skip_with_queue(self):
|
|
bot = _FakeBot()
|
|
ps = _mod._ps(bot)
|
|
ps["current"] = _mod._Track(url="a", title="First", requester="x")
|
|
ps["queue"] = [_mod._Track(url="b", title="Second", requester="y")]
|
|
# We need to mock the task
|
|
mock_task = MagicMock()
|
|
mock_task.done.return_value = False
|
|
ps["task"] = mock_task
|
|
msg = _Msg(text="!skip")
|
|
with patch.object(_mod, "_ensure_loop"):
|
|
asyncio.run(_mod.cmd_skip(bot, msg))
|
|
assert any("Skipped" in r for r in bot.replied)
|
|
mock_task.cancel.assert_called_once()
|
|
|
|
def test_skip_empty_queue(self):
|
|
bot = _FakeBot()
|
|
ps = _mod._ps(bot)
|
|
ps["current"] = _mod._Track(url="a", title="Only", requester="x")
|
|
mock_task = MagicMock()
|
|
mock_task.done.return_value = False
|
|
ps["task"] = mock_task
|
|
msg = _Msg(text="!skip")
|
|
asyncio.run(_mod.cmd_skip(bot, msg))
|
|
assert any("empty" in r.lower() for r in bot.replied)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestQueueCommand
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestQueueCommand:
|
|
def test_queue_empty(self):
|
|
bot = _FakeBot()
|
|
msg = _Msg(text="!queue")
|
|
asyncio.run(_mod.cmd_queue(bot, msg))
|
|
assert any("empty" in r.lower() for r in bot.replied)
|
|
|
|
def test_queue_with_tracks(self):
|
|
bot = _FakeBot()
|
|
ps = _mod._ps(bot)
|
|
ps["current"] = _mod._Track(url="a", title="Now", requester="x")
|
|
ps["queue"] = [
|
|
_mod._Track(url="b", title="Next", requester="y"),
|
|
]
|
|
msg = _Msg(text="!queue")
|
|
asyncio.run(_mod.cmd_queue(bot, msg))
|
|
assert any("Now" in r for r in bot.replied)
|
|
assert any("Next" in r for r in bot.replied)
|
|
|
|
def test_queue_with_url_delegates(self):
|
|
bot = _FakeBot()
|
|
msg = _Msg(text="!queue https://example.com/track")
|
|
with patch.object(_mod, "_resolve_title", return_value="Title"):
|
|
with patch.object(_mod, "_ensure_loop"):
|
|
asyncio.run(_mod.cmd_queue(bot, msg))
|
|
# Should have called cmd_play logic
|
|
assert any("Playing" in r or "Queued" in r for r in bot.replied)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestNpCommand
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestNpCommand:
|
|
def test_np_nothing(self):
|
|
bot = _FakeBot()
|
|
msg = _Msg(text="!np")
|
|
asyncio.run(_mod.cmd_np(bot, msg))
|
|
assert any("Nothing" in r for r in bot.replied)
|
|
|
|
def test_np_playing(self):
|
|
bot = _FakeBot()
|
|
ps = _mod._ps(bot)
|
|
ps["current"] = _mod._Track(
|
|
url="x", title="Cool Song", requester="DJ",
|
|
)
|
|
msg = _Msg(text="!np")
|
|
asyncio.run(_mod.cmd_np(bot, msg))
|
|
assert any("Cool Song" in r for r in bot.replied)
|
|
assert any("DJ" in r for r in bot.replied)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestVolumeCommand
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestVolumeCommand:
|
|
def test_volume_show(self):
|
|
bot = _FakeBot()
|
|
msg = _Msg(text="!volume")
|
|
asyncio.run(_mod.cmd_volume(bot, msg))
|
|
assert any("50%" in r for r in bot.replied)
|
|
|
|
def test_volume_set(self):
|
|
bot = _FakeBot()
|
|
msg = _Msg(text="!volume 75")
|
|
asyncio.run(_mod.cmd_volume(bot, msg))
|
|
ps = _mod._ps(bot)
|
|
assert ps["volume"] == 75
|
|
assert any("75%" in r for r in bot.replied)
|
|
|
|
def test_volume_out_of_range(self):
|
|
bot = _FakeBot()
|
|
msg = _Msg(text="!volume 150")
|
|
asyncio.run(_mod.cmd_volume(bot, msg))
|
|
assert any("0-100" in r for r in bot.replied)
|
|
|
|
def test_volume_negative(self):
|
|
bot = _FakeBot()
|
|
msg = _Msg(text="!volume -10")
|
|
asyncio.run(_mod.cmd_volume(bot, msg))
|
|
assert any("0-100" in r for r in bot.replied)
|
|
|
|
def test_volume_invalid(self):
|
|
bot = _FakeBot()
|
|
msg = _Msg(text="!volume abc")
|
|
asyncio.run(_mod.cmd_volume(bot, msg))
|
|
assert any("Usage" in r for r in bot.replied)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestPerBotState
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestPerBotState:
|
|
def test_ps_initializes(self):
|
|
bot = _FakeBot()
|
|
ps = _mod._ps(bot)
|
|
assert ps["queue"] == []
|
|
assert ps["current"] is None
|
|
assert ps["volume"] == 50
|
|
|
|
def test_ps_stable_reference(self):
|
|
bot = _FakeBot()
|
|
ps1 = _mod._ps(bot)
|
|
ps2 = _mod._ps(bot)
|
|
assert ps1 is ps2
|
|
|
|
def test_ps_isolated_per_bot(self):
|
|
bot1 = _FakeBot()
|
|
bot2 = _FakeBot()
|
|
_mod._ps(bot1)["volume"] = 80
|
|
assert _mod._ps(bot2)["volume"] == 50
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestHelpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestMusicHelpers:
|
|
def test_truncate_short(self):
|
|
assert _mod._truncate("short") == "short"
|
|
|
|
def test_truncate_long(self):
|
|
long = "x" * 100
|
|
result = _mod._truncate(long)
|
|
assert len(result) == 80
|
|
assert result.endswith("...")
|