import asyncio from unittest.mock import Mock, AsyncMock, patch import pytest from discord.ext import commands @pytest.mark.asyncio async def test_bot_ensure_voice(mbot, ctx): # Connects to voice channel of author if possible ctx.voice_client = None ctx.author.voice = AsyncMock() await mbot.ensure_voice(ctx) assert ctx.author.voice.channel.connect.call_count == 1, "Did not connect to voice channel of author" ctx.reset_mock(return_value=True) # Error if author not inside a channel ctx.voice_client = None ctx.author.voice = None with pytest.raises(commands.CommandError): await mbot.ensure_voice(ctx) def mock_ytdl_extract_info(ytdl, url, title): ytdl.extract_info.return_value = {"entries": [{"url": url, "title": title}]} def mock_ffmpeg_pcm_audio(ffmpeg_pcm_audio): deliver_us_audio = Mock() ffmpeg_pcm_audio.return_value = deliver_us_audio return deliver_us_audio @pytest.mark.asyncio async def test_bot_playback(mbot, ctx): with patch.object(mbot, '_ytdl') as ytdl: with patch('discord.FFmpegPCMAudio') as ffmpeg_pcm_audio: ctx.voice_client.is_playing.return_value = False url = "https://www.youtube.com/watch?v=Wr9LZ1hAFpQ" title = "In Flames - Deliver Us (Official Video)" mock_ytdl_extract_info(ytdl, url, title) deliver_us_audio = mock_ffmpeg_pcm_audio(ffmpeg_pcm_audio) query = 'in flames deliver us' # pylint: disable=too-many-function-args await mbot.play(mbot, ctx, query=query) assert \ ytdl.extract_info.call_args.args == (query,) and ytdl.extract_info.call_args.kwargs == {"download": False}, \ f"ytdl.extract_info was not called with {query}, {{ download: False }}" assert \ ffmpeg_pcm_audio.call_args is None, \ f"FFmpegPCMAudio was immediately called with {url} instead of being queued" assert \ ctx.voice_client.play.call_args is None, \ "Did immediately playback audio instead of being queued" assert \ ctx.send.call_args is None, \ "Did immediately send 'Now playing:' message of being queued" await asyncio.sleep(0) assert \ ffmpeg_pcm_audio.call_args.args == (url,), \ f"FFmpegPCMAudio was not called with {url}" assert \ ctx.voice_client.play.call_args.args == (deliver_us_audio,), \ "Did not playback correct audio" assert \ ctx.send.call_args.args == (f"Now playing: {title}",), \ "Did not send 'Now playing:' message" ctx.voice_client.is_playing.return_value = True url = "https://www.youtube.com/watch?v=pMDcYX2wRSg" title = "Three Days Grace - Time of Dying (lyrics)" mock_ytdl_extract_info(ytdl, url, title) time_of_dying_audio = mock_ffmpeg_pcm_audio(ffmpeg_pcm_audio) # pylint: disable=too-many-function-args query = "three days grace time of dying" await mbot.play(mbot, ctx, query=query) assert \ ytdl.extract_info.call_args.args == (query,) and ytdl.extract_info.call_args.kwargs == {"download": False}, \ f"ytdl.extract_info was not called with {query}, {{ download: False }}" assert \ not ffmpeg_pcm_audio.call_args.args == (url,), \ f"FFmpegPCMAudio was immediately called with {url} instead of being queued" assert \ not ctx.voice_client.play.call_args.args == (time_of_dying_audio,), \ "Did immediately playback audio instead of being queued" assert \ ctx.send.call_args.args == (f"Queued: {title}",), \ "Did not send 'Queued:' message" await asyncio.sleep(0) # Still no playback because previous song not finished assert \ not ffmpeg_pcm_audio.call_args.args == (url,), \ f"FFmpegPCMAudio was called with {url} before previous song finished" ctx.voice_client.play.call_args.kwargs["after"](None) # Execute callback for song finish event await asyncio.sleep(0) assert \ ctx.voice_client.play.call_args.args == (time_of_dying_audio,), \ "Did not queue next song"