243 lines
8.7 KiB
Python
243 lines
8.7 KiB
Python
"""Sprawdza, że wskaźnikowa kolejka zachowuje się tak jak przed refaktorem."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import tempfile
|
|
import unittest
|
|
|
|
from tests.fakes import FakeTrack, build_player
|
|
from zyrarbot.music.loop_mode import LoopMode
|
|
from zyrarbot.storage.queue_repository import QueueRepository
|
|
|
|
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)
|
|
self.assertEqual(track.plays_left, 5)
|
|
|
|
async def test_finish_current_decrements_plays_without_going_below_zero(self):
|
|
player, _ = build_player(default_plays=1)
|
|
track = FakeTrack("a")
|
|
await player.enqueue(GUILD, track)
|
|
|
|
player.finish_current(GUILD)
|
|
player.finish_current(GUILD)
|
|
|
|
self.assertEqual(track.plays_left, 0)
|
|
|
|
async def test_skip_does_not_decrement_current_track(self):
|
|
player, _, state = await self._queue_abc(default_plays=2)
|
|
|
|
player.skip(GUILD)
|
|
await player.play_next(GUILD, advance=True, consume_current=False)
|
|
|
|
self.assertEqual(state.queue[0].plays_left, 2)
|
|
self.assertEqual(state.current.id, "b")
|
|
|
|
async def test_last_play_remains_pointed_to_until_playback_finishes(self):
|
|
player, _ = build_player(default_plays=1)
|
|
track = FakeTrack("a")
|
|
await player.enqueue(GUILD, track)
|
|
|
|
state = player.state(GUILD)
|
|
self.assertEqual(state.queue, [track])
|
|
self.assertEqual(state.index, 0)
|
|
self.assertIs(state.current, track)
|
|
|
|
await player.play_next(GUILD, advance=True)
|
|
|
|
self.assertEqual(state.queue, [])
|
|
self.assertEqual(state.index, -1)
|
|
self.assertIsNone(state.current)
|
|
|
|
async def test_last_play_of_recreated_track_is_removed_after_callback(self):
|
|
player, _, state = await self._queue_abc(default_plays=2)
|
|
state.loop_mode = LoopMode.TRACK
|
|
|
|
await player.play_next(GUILD, advance=True)
|
|
self.assertEqual([track.id for track in state.queue], ["a", "b", "c"])
|
|
self.assertEqual(state.index, 0)
|
|
self.assertEqual(state.current.plays_left, 1)
|
|
|
|
await player.play_next(GUILD, advance=True)
|
|
|
|
self.assertEqual([track.id for track in state.queue], ["b", "c"])
|
|
self.assertEqual(state.index, 0)
|
|
self.assertEqual(state.current.id, "b")
|
|
|
|
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_default_plays_are_applied_per_guild(self):
|
|
player, _ = build_player()
|
|
player._config.get_default_plays.side_effect = lambda guild_id: {1: 2, 2: 7}.get(
|
|
guild_id, 5
|
|
)
|
|
|
|
first = FakeTrack("first")
|
|
second = FakeTrack("second")
|
|
await player.enqueue(1, first)
|
|
await player.enqueue(2, second)
|
|
|
|
self.assertEqual(first.plays_left, 2)
|
|
self.assertEqual(second.plays_left, 7)
|
|
|
|
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)
|
|
|
|
async def test_restore_replaces_active_queue_without_old_callback_interference(self):
|
|
player, voice_client = build_player()
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
player._queues = QueueRepository(directory)
|
|
player._queues.save(
|
|
GUILD,
|
|
[{"id": "restored", "title": "Restored", "plays_left": 3}],
|
|
)
|
|
|
|
async def from_url(url, loop=None):
|
|
return FakeTrack("restored")
|
|
|
|
player._tracks.from_url = from_url
|
|
await player.enqueue(GUILD, FakeTrack("old"))
|
|
old_callback = voice_client.after
|
|
|
|
count = await player.restore_queue(GUILD)
|
|
|
|
self.assertEqual(count, 1)
|
|
self.assertEqual([track.id for track in player.state(GUILD).queue], ["restored"])
|
|
self.assertIs(player.state(GUILD).current, player.state(GUILD).queue[0])
|
|
old_callback(None)
|
|
await asyncio.sleep(0)
|
|
self.assertEqual([track.id for track in player.state(GUILD).queue], ["restored"])
|
|
|
|
|
|
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()
|