musicube/test/test_bot.py

import asyncio
from unittest.mock import AsyncMock, Mock, patch

import pytest
from discord.ext import commands


@pytest.mark.asyncio
async def test_bot_ensure_voice(mbot, ctx):
  # TEST: 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)

  # TEST: 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, *, title, webpage_url, audio_url):
  ytdl.extract_info.return_value = {'entries': [{'title': title, 'webpage_url': webpage_url, 'url': audio_url}]}


def mock_ffmpeg_pcm_audio(ffmpeg_pcm_audio):
  audio = Mock()
  ffmpeg_pcm_audio.return_value = audio
  return 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:

      # TEST: First song queued is immediately played
      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, title=title, webpage_url=url, audio_url=url)
      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'
      content = ctx.send.call_args.kwargs['content']
      assert content == 'Now playing:', "Did not send 'Now playing:' message"
      embed = ctx.send.call_args.kwargs['embed']
      assert embed.title == title, "Did not send 'Now playing:' embed"

      # TEST: Following songs are put inside a queue
      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, title=title, webpage_url=url, audio_url=url)
      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'
      content = ctx.send.call_args.kwargs['content']
      assert content == 'Queued:', "Did not send 'Queued:' message"
      embed = ctx.send.call_args.kwargs['embed']
      assert embed.title == title, "Did not send 'Queued:' embed"
      await asyncio.sleep(0)
      # Assert that there is still no playback because previous song is not finished yet
      assert \
          not ffmpeg_pcm_audio.call_args.args == (url,), \
          f'FFmpegPCMAudio was called with {url} before previous song finished'
      # Execute callback for song finish event
      ctx.voice_client.play.call_args.kwargs['after'](None)
      await asyncio.sleep(0)
      assert \
          ctx.voice_client.play.call_args.args == (time_of_dying_audio,), \
          'Did not queue next song'


@pytest.mark.asyncio
async def test_bot_skip(mbot, ctx):
  with patch.object(mbot, '_ytdl') as ytdl:
    with patch('discord.FFmpegPCMAudio') as ffmpeg_pcm_audio:

      # TEST: One can skip songs to immediately play the next song in queue

      # Queue first song
      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, title=title, webpage_url=url, audio_url=url)
      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)
      await asyncio.sleep(0)
      assert \
          ctx.voice_client.play.call_args.args == (deliver_us_audio,), \
          'Did not playback correct audio'

      # Queue second song
      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, title=title, webpage_url=url, audio_url=url)
      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)
      await asyncio.sleep(0)
      assert \
          not ctx.voice_client.play.call_args.args == (time_of_dying_audio,), \
          'Did immediately playback audio instead of being queued'

      # Now skip first song
      await mbot.skip(mbot, ctx)
      await asyncio.sleep(0)
      assert \
          ctx.voice_client.play.call_args.args == (time_of_dying_audio,), \
          'Did not skip song'

      # TEST: Error if no song playing
      ctx.voice_client.is_playing.return_value = False
      with pytest.raises(commands.CommandError):
        await mbot.skip(mbot, ctx)

      # TEST: Error if no voice client
      ctx.voice_client = None
      with pytest.raises(commands.CommandError):
        await mbot.skip(mbot, ctx)