test: regresja arytmetyki kolejki wskaźnikowej
Kolejka jest nieusuwalna, a pozycję śledzi indeks, więc przeskoki, usuwanie i zmiana kolejności muszą ten indeks poprawnie przesuwać. To najbardziej podatna na błędy część przeniesionego kodu, więc dostaje testy przypinające zachowanie: - start odtwarzania przy dodaniu do pustej kolejki - limit odtworzeń usuwający utwór i cofający indeks - przechodzenie dalej oraz oba tryby zapętlania - koniec kolejki czyszczący aktualny utwór - przeskok, usunięcie utworu przed indeksem i usunięcie odtwarzanego - zmiana kolejności zachowująca indeks na odtwarzanym utworze - stosowanie głośności serwera i zatrzymanie przy niedziałających utworach 16 przypadków na atrapach Discorda, bez sieci i ffmpega. Uruchamiane przez python -m unittest discover -s tests -t . - bez nowych zależności. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
lgooseandClaude Bot committed
1 parent
c8a188647b
commit
0879120c85
3 files changed
+241
No files matched your search
@@ -0,0 +1 @@
|
||||
"""Testy regresyjne - uruchamiane przez ``python -m unittest``."""
|
||||
@@ -0,0 +1,81 @@
|
||||
"""Atrapy pozwalające testować ``MusicPlayer`` bez Discorda i ffmpeg."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from zyrarbot.music.music_player import MusicPlayer
|
||||
|
||||
|
||||
class FakeTrack:
|
||||
"""Odpowiednik ``YTDLSource`` o tym samym interfejsie, ale bez audio."""
|
||||
|
||||
def __init__(self, track_id: str):
|
||||
self.id = track_id
|
||||
self.title = track_id
|
||||
self.data: dict = {}
|
||||
self.url = None
|
||||
self.duration = 10
|
||||
self.elapsed = 0.0
|
||||
self.plays_left: int | None = None
|
||||
self.played = False
|
||||
self.volume = 0.5
|
||||
self.is_local = False
|
||||
self.filepath = None
|
||||
self.display_url = None
|
||||
self.has_web_link = False
|
||||
|
||||
|
||||
class FakeVoiceClient:
|
||||
"""Klient głosowy, który zapamiętuje odtwarzane źródło zamiast je grać.
|
||||
|
||||
Nie wywołuje zwrotnie ``after`` - w testach kolejny utwór uruchamiamy ręcznie,
|
||||
tak jak zrobiłaby to biblioteka discord.py po zakończeniu ścieżki.
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.playing = None
|
||||
self.after = None
|
||||
self.source = None
|
||||
|
||||
def is_playing(self) -> bool:
|
||||
return self.playing is not None
|
||||
|
||||
def is_paused(self) -> bool:
|
||||
return False
|
||||
|
||||
def play(self, source, after=None) -> None:
|
||||
self.playing = source
|
||||
self.source = source
|
||||
self.after = after
|
||||
|
||||
def stop(self) -> None:
|
||||
self.playing = None
|
||||
|
||||
|
||||
def build_player(default_plays: int = 5) -> tuple[MusicPlayer, FakeVoiceClient]:
|
||||
bot = MagicMock()
|
||||
bot.loop = asyncio.get_event_loop()
|
||||
voice_client = FakeVoiceClient()
|
||||
|
||||
async def recreate(track, loop=None):
|
||||
fresh = FakeTrack(track.id)
|
||||
fresh.plays_left = track.plays_left
|
||||
return fresh
|
||||
|
||||
tracks = MagicMock()
|
||||
tracks.recreate = recreate
|
||||
|
||||
history = MagicMock()
|
||||
history.list.return_value = []
|
||||
|
||||
config = MagicMock()
|
||||
config.get_default_plays.return_value = default_plays
|
||||
|
||||
channels = MagicMock()
|
||||
channels.resolve.return_value = None
|
||||
|
||||
player = MusicPlayer(bot, tracks, history, config, channels)
|
||||
player.voice_client = lambda guild_id: voice_client # type: ignore[method-assign]
|
||||
return player, voice_client
|
||||
@@ -0,0 +1,159 @@
|
||||
"""Sprawdza, że wskaźnikowa kolejka zachowuje się tak jak przed refaktorem."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
|
||||
from tests.fakes import FakeTrack, build_player
|
||||
from zyrarbot.music.loop_mode import LoopMode
|
||||
|
||||
GUILD = 1
|
||||
|
||||
|
||||
class MusicPlayerQueueTest(unittest.IsolatedAsyncioTestCase):
|
||||
async def _queue_abc(self, default_plays: int = 5):
|
||||
player, voice_client = build_player(default_plays=default_plays)
|
||||
for track_id in "abc":
|
||||
await player.enqueue(GUILD, FakeTrack(track_id))
|
||||
return player, voice_client, player.state(GUILD)
|
||||
|
||||
async def test_enqueue_on_idle_guild_starts_playback(self):
|
||||
player, voice_client = build_player()
|
||||
track = FakeTrack("a")
|
||||
await player.enqueue(GUILD, track)
|
||||
|
||||
self.assertIs(voice_client.playing, track)
|
||||
self.assertEqual(player.state(GUILD).index, 0)
|
||||
|
||||
async def test_single_play_budget_drops_track_from_queue(self):
|
||||
player, _ = build_player(default_plays=1)
|
||||
track = FakeTrack("a")
|
||||
await player.enqueue(GUILD, track)
|
||||
|
||||
state = player.state(GUILD)
|
||||
self.assertEqual(state.queue, [])
|
||||
self.assertEqual(state.index, -1)
|
||||
self.assertIs(state.current, track)
|
||||
|
||||
async def test_enqueue_while_playing_only_appends(self):
|
||||
_, _, state = await self._queue_abc()
|
||||
|
||||
self.assertEqual([track.id for track in state.queue], ["a", "b", "c"])
|
||||
self.assertEqual(state.index, 0)
|
||||
self.assertEqual(state.current.id, "a")
|
||||
|
||||
async def test_advance_moves_one_position_forward(self):
|
||||
player, _, state = await self._queue_abc()
|
||||
await player.play_next(GUILD, advance=True)
|
||||
|
||||
self.assertEqual(state.index, 1)
|
||||
self.assertEqual(state.current.id, "b")
|
||||
|
||||
async def test_loop_track_replays_same_index(self):
|
||||
player, _, state = await self._queue_abc()
|
||||
await player.play_next(GUILD, advance=True)
|
||||
state.loop_mode = LoopMode.TRACK
|
||||
await player.play_next(GUILD, advance=True)
|
||||
|
||||
self.assertEqual(state.index, 1)
|
||||
self.assertEqual(state.current.id, "b")
|
||||
|
||||
async def test_loop_queue_wraps_to_start(self):
|
||||
player, _, state = await self._queue_abc()
|
||||
state.loop_mode = LoopMode.QUEUE
|
||||
state.index = 2
|
||||
await player.play_next(GUILD, advance=True)
|
||||
|
||||
self.assertEqual(state.index, 0)
|
||||
|
||||
async def test_queue_end_clears_current_track(self):
|
||||
player, _, state = await self._queue_abc()
|
||||
state.index = 2
|
||||
await player.play_next(GUILD, advance=True)
|
||||
|
||||
self.assertIsNone(state.current)
|
||||
self.assertEqual(state.index, 3)
|
||||
|
||||
async def test_jump_selects_target_without_advancing_past_it(self):
|
||||
player, _, state = await self._queue_abc()
|
||||
await player.jump(GUILD, 2)
|
||||
await player.play_next(GUILD, advance=True) # discord.py: zwrotka `after`
|
||||
|
||||
self.assertEqual(state.index, 2)
|
||||
self.assertEqual(state.current.id, "c")
|
||||
|
||||
async def test_removing_earlier_track_shifts_index_back(self):
|
||||
player, _, state = await self._queue_abc()
|
||||
await player.jump(GUILD, 2)
|
||||
await player.play_next(GUILD, advance=True)
|
||||
|
||||
self.assertTrue(await player.remove(GUILD, "a"))
|
||||
self.assertEqual([track.id for track in state.queue], ["b", "c"])
|
||||
self.assertEqual(state.index, 1)
|
||||
|
||||
async def test_removing_playing_track_plays_the_next_one(self):
|
||||
player, _, state = await self._queue_abc()
|
||||
await player.jump(GUILD, 1)
|
||||
await player.play_next(GUILD, advance=True)
|
||||
|
||||
self.assertTrue(await player.remove(GUILD, "b"))
|
||||
await player.play_next(GUILD, advance=True)
|
||||
|
||||
self.assertEqual([track.id for track in state.queue], ["a", "c"])
|
||||
self.assertEqual(state.index, 1)
|
||||
self.assertEqual(state.current.id, "c")
|
||||
|
||||
async def test_removing_unknown_track_reports_failure(self):
|
||||
player, _, _ = await self._queue_abc()
|
||||
self.assertFalse(await player.remove(GUILD, "nope"))
|
||||
|
||||
async def test_reorder_keeps_index_on_playing_track(self):
|
||||
player, _, state = await self._queue_abc()
|
||||
await player.jump(GUILD, 0)
|
||||
await player.play_next(GUILD, advance=True)
|
||||
|
||||
player.reorder(GUILD, ["c", "b", "a"])
|
||||
|
||||
self.assertEqual([track.id for track in state.queue], ["c", "b", "a"])
|
||||
self.assertEqual(state.index, 2)
|
||||
self.assertIs(state.queue[state.index], state.current)
|
||||
|
||||
async def test_guild_volume_applies_to_new_track(self):
|
||||
player, _ = build_player()
|
||||
player.set_volume(GUILD, 0.25)
|
||||
|
||||
track = FakeTrack("v")
|
||||
await player.enqueue(GUILD, track)
|
||||
|
||||
self.assertEqual(track.volume, 0.25)
|
||||
|
||||
async def test_broken_tracks_stop_instead_of_looping_forever(self):
|
||||
player, _, state = await self._queue_abc()
|
||||
|
||||
async def always_fails(track, loop=None):
|
||||
raise RuntimeError("brak dostępu do źródła")
|
||||
|
||||
player._tracks.recreate = always_fails
|
||||
for track in state.queue:
|
||||
track.played = True
|
||||
|
||||
state.loop_mode = LoopMode.QUEUE
|
||||
await player.play_next(GUILD, advance=True)
|
||||
|
||||
self.assertIsNone(state.current)
|
||||
|
||||
|
||||
class LoopModeTest(unittest.TestCase):
|
||||
def test_cycle_order(self):
|
||||
self.assertIs(LoopMode.OFF.next(), LoopMode.QUEUE)
|
||||
self.assertIs(LoopMode.QUEUE.next(), LoopMode.TRACK)
|
||||
self.assertIs(LoopMode.TRACK.next(), LoopMode.OFF)
|
||||
|
||||
def test_api_values_match_panel_contract(self):
|
||||
self.assertEqual(LoopMode.OFF.value, "off")
|
||||
self.assertEqual(LoopMode.QUEUE.value, "playlist")
|
||||
self.assertEqual(LoopMode.TRACK.value, "single")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user