python-mpd 1.0.0-1 / usr / lib / python2.7 / dist-packages / mpd / tests.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from __future__ import absolute_import
import itertools
import mpd
import sys
import types
import warnings

try:
    # is required for python2.6
    # python2.7 works with this module too
    import unittest2 as unittest
except ImportError:
    # required for python3
    # python2.7 works with this module too!
    if sys.version_info >= (2, 7):
        import unittest
    else:
        print("Please install unittest2 from PyPI to run tests!")
        sys.exit(1)

try:
    from twisted.python.failure import Failure
    TWISTED_MISSING = False
except ImportError:
    warnings.warn("No twisted installed: skip twisted related tests! " +
                  "(twisted is not available for python >= 3.0 && python < 3.3)")
    TWISTED_MISSING = True

if sys.version_info >= (3, 5):
    # asyncio would be available in 3.4, but it's not supported by mpd.asyncio
    import asyncio
else:
    asyncio = None

try:
    import mock
except ImportError:
    print("Please install mock from PyPI to run tests!")
    sys.exit(1)

# show deprecation warnings
warnings.simplefilter('default')


TEST_MPD_HOST, TEST_MPD_PORT = ('example.com', 10000)


class TestMPDClient(unittest.TestCase):

    longMessage = True

    def setUp(self):
        self.socket_patch = mock.patch("mpd.base.socket")
        self.socket_mock = self.socket_patch.start()
        self.socket_mock.getaddrinfo.return_value = [range(5)]

        self.socket_mock.socket.side_effect = (
            lambda *a, **kw:
            # Create a new socket.socket() mock with default attributes,
            # each time we are calling it back (otherwise, it keeps set
            # attributes across calls).
            # That's probablyy what we want, since reconnecting is like
            # reinitializing the entire connection, and so, the mock.
            mock.MagicMock(name="socket.socket"))

        self.client = mpd.MPDClient()
        self.client.connect(TEST_MPD_HOST, TEST_MPD_PORT)
        self.client._sock.reset_mock()
        self.MPDWillReturn("ACK don't forget to setup your mock\n")

    def tearDown(self):
        self.socket_patch.stop()

    def MPDWillReturn(self, *lines):
        # Return what the caller wants first, then do as if the socket was
        # disconnected.
        self.client._rfile.readline.side_effect = itertools.chain(
            lines, itertools.repeat(''))

    def assertMPDReceived(self, *lines):
        self.client._wfile.write.assert_called_with(*lines)

    def test_abstract_functions(self):
        MPDClientBase = mpd.base.MPDClientBase
        self.assertRaises(
            NotImplementedError,
            lambda: MPDClientBase.add_command('command_name', lambda x: x))
        client = MPDClientBase()
        self.assertRaises(NotImplementedError, lambda: client.noidle())
        self.assertRaises(
            NotImplementedError,
            lambda: client.command_list_ok_begin())
        self.assertRaises(
            NotImplementedError,
            lambda: client.command_list_end())

    def test_metaclass_commands(self):
        # just some random functions
        self.assertTrue(hasattr(self.client, "commands"))
        self.assertTrue(hasattr(self.client, "save"))
        self.assertTrue(hasattr(self.client, "random"))
        # space should be replaced
        self.assertFalse(hasattr(self.client, "sticker get"))
        self.assertTrue(hasattr(self.client, "sticker_get"))
        # each command should have prefixe
        self.assertTrue(hasattr(self.client, "close"))
        self.assertTrue(hasattr(self.client, "fetch_close"))
        self.assertTrue(hasattr(self.client, "send_close"))

    def test_duplicate_tags(self):
        self.MPDWillReturn('Track: file1\n', 'Track: file2\n', 'OK\n')
        song = self.client.currentsong()
        self.assertIsInstance(song, dict)
        self.assertIsInstance(song["track"], list)
        self.assertMPDReceived('currentsong\n')

    def test_parse_nothing(self):
        self.MPDWillReturn('OK\n', 'OK\n')

        self.assertIsNone(self.client.ping())
        self.assertMPDReceived('ping\n')

        self.assertIsNone(self.client.clearerror())
        self.assertMPDReceived('clearerror\n')

    def test_parse_list(self):
        self.MPDWillReturn('OK\n')

        self.assertIsInstance(self.client.list("album"), list)
        self.assertMPDReceived('list "album"\n')

    def test_parse_item(self):
        self.MPDWillReturn('updating_db: 42\n', 'OK\n')
        self.assertIsNotNone(self.client.update())

    def test_parse_object(self):
        # XXX: _read_objects() doesn't wait for the final OK
        self.MPDWillReturn('volume: 63\n', 'OK\n')
        status = self.client.status()
        self.assertMPDReceived('status\n')
        self.assertIsInstance(status, dict)

        # XXX: _read_objects() doesn't wait for the final OK
        self.MPDWillReturn('OK\n')
        stats = self.client.stats()
        self.assertMPDReceived('stats\n')
        self.assertIsInstance(stats, dict)

    def test_parse_songs(self):
        self.MPDWillReturn("file: my-song.ogg\n",
                           "Pos: 0\n",
                           "Id: 66\n",
                           "OK\n")
        playlist = self.client.playlistinfo()

        self.assertMPDReceived('playlistinfo\n')
        self.assertIsInstance(playlist, list)
        self.assertEqual(1, len(playlist))
        e = playlist[0]
        self.assertIsInstance(e, dict)
        self.assertEqual('my-song.ogg', e['file'])
        self.assertEqual('0', e['pos'])
        self.assertEqual('66', e['id'])

    def test_send_and_fetch(self):
        self.MPDWillReturn("volume: 50\n", "OK\n")
        result = self.client.send_status()
        self.assertEqual(None, result)
        self.assertMPDReceived('status\n')

        status = self.client.fetch_status()
        self.assertEqual(1, self.client._wfile.write.call_count)
        self.assertEqual({'volume': '50'}, status)

    def test_readcomments(self):
        self.MPDWillReturn("major_brand: M4V\n",
                           "minor_version: 1\n",
                           "lyrics: Lalala\n",
                           "OK\n")
        comments = self.client.readcomments()
        self.assertMPDReceived('readcomments\n')
        self.assertEqual(comments['major_brand'], "M4V")
        self.assertEqual(comments['minor_version'], "1")
        self.assertEqual(comments['lyrics'], "Lalala")

    def test_iterating(self):
        self.MPDWillReturn("file: my-song.ogg\n",
                           "Pos: 0\n",
                           "Id: 66\n",
                           "OK\n")
        self.client.iterate = True
        playlist = self.client.playlistinfo()
        self.assertMPDReceived('playlistinfo\n')
        self.assertIsInstance(playlist, types.GeneratorType)
        for song in playlist:
            self.assertIsInstance(song, dict)
            self.assertEqual('my-song.ogg', song['file'])
            self.assertEqual('0', song['pos'])
            self.assertEqual('66', song['id'])

    def test_idle(self):
        self.MPDWillReturn('OK\n')  # nothing changed after idle-ing
        self.client.idletimeout = 456
        res = self.client.idle()
        self.assertMPDReceived('idle\n')
        self.client._sock.settimeout.assert_has_calls([mock.call(456),
                                                       mock.call(None)])
        self.assertEqual([], res)

        self.client.send_idle()
        # new event
        self.MPDWillReturn('changed: update\n', 'OK\n')

        event = self.client.fetch_idle()
        self.assertEqual(event, ['update'])

    def test_noidle(self):
        self.MPDWillReturn('OK\n')  # nothing changed after idle-ing
        self.client.send_idle()
        self.MPDWillReturn('OK\n')  # nothing changed after noidle
        self.assertEqual(self.client.noidle(), [])
        self.assertMPDReceived('noidle\n')
        self.MPDWillReturn("volume: 50\n", "OK\n")
        self.client.status()
        self.assertMPDReceived('status\n')

    def test_noidle_while_idle_started_sending(self):
        self.MPDWillReturn('OK\n')  # nothing changed after idle-ing
        self.client.send_idle()
        self.MPDWillReturn('changed: player\n', 'OK\n')  # noidle response
        self.assertEqual(self.client.noidle(), ['player'])
        self.MPDWillReturn("volume: 50\n", "OK\n")
        status = self.client.status()
        self.assertEqual({'volume': '50'}, status)

    def test_throw_when_calling_noidle_withoutidling(self):
        self.assertRaises(mpd.CommandError, self.client.noidle)
        self.client.send_status()
        self.assertRaises(mpd.CommandError, self.client.noidle)

    def test_add_and_remove_command(self):
        self.MPDWillReturn("ACK awesome command\n")

        self.client.add_command("awesome command",
                                mpd.MPDClient._parse_nothing)
        self.assertTrue(hasattr(self.client, "awesome_command"))
        self.assertTrue(hasattr(self.client, "send_awesome_command"))
        self.assertTrue(hasattr(self.client, "fetch_awesome_command"))
        # should be unknown by mpd
        self.assertRaises(mpd.CommandError, self.client.awesome_command)

        self.client.remove_command("awesome_command")
        self.assertFalse(hasattr(self.client, "awesome_command"))
        self.assertFalse(hasattr(self.client, "send_awesome_command"))
        self.assertFalse(hasattr(self.client, "fetch_awesome_command"))

        # remove non existing command
        self.assertRaises(ValueError, self.client.remove_command,
                          "awesome_command")

    def test_client_to_client(self):
        # client to client is at this time in beta!

        self.MPDWillReturn('OK\n')
        self.assertIsNone(self.client.subscribe("monty"))
        self.assertMPDReceived('subscribe "monty"\n')

        self.MPDWillReturn('channel: monty\n', 'OK\n')
        channels = self.client.channels()
        self.assertMPDReceived('channels\n')
        self.assertEqual(["monty"], channels)

        self.MPDWillReturn('OK\n')
        self.assertIsNone(self.client.sendmessage("monty", "SPAM"))
        self.assertMPDReceived('sendmessage "monty" "SPAM"\n')

        self.MPDWillReturn('channel: monty\n', 'message: SPAM\n', 'OK\n')
        msg = self.client.readmessages()
        self.assertMPDReceived('readmessages\n')
        self.assertEqual(msg, [{"channel": "monty", "message": "SPAM"}])

        self.MPDWillReturn('OK\n')
        self.assertIsNone(self.client.unsubscribe("monty"))
        self.assertMPDReceived('unsubscribe "monty"\n')

        self.MPDWillReturn('OK\n')
        channels = self.client.channels()
        self.assertMPDReceived('channels\n')
        self.assertEqual([], channels)

    def test_unicode_as_command_args(self):
        if sys.version_info < (3, 0):
            self.MPDWillReturn("OK\n")
            res = self.client.find("file", unicode("☯☾☝♖✽", 'utf-8'))
            self.assertIsInstance(res, list)
            self.assertMPDReceived('find "file" "☯☾☝♖✽"\n')

            self.MPDWillReturn("OK\n")
            res2 = self.client.find("file", "☯☾☝♖✽")
            self.assertIsInstance(res2, list)
            self.assertMPDReceived('find "file" "☯☾☝♖✽"\n')
        else:
            self.MPDWillReturn("OK\n")
            res = self.client.find("file", "☯☾☝♖✽")
            self.assertIsInstance(res, list)
            self.assertMPDReceived('find "file" "☯☾☝♖✽"\n')

    @unittest.skipIf(sys.version_info >= (3, 0),
                     "Test special unicode handling only if python2")
    def test_unicode_as_reponse(self):
        self.MPDWillReturn("handler: http://\n", "OK\n")
        self.client.use_unicode = True
        self.assertIsInstance(self.client.urlhandlers()[0], unicode)

        self.MPDWillReturn("handler: http://\n", "OK\n")
        self.client.use_unicode = False
        self.assertIsInstance(self.client.urlhandlers()[0], str)

    def test_numbers_as_command_args(self):
        self.MPDWillReturn("OK\n")
        self.client.find("file", 1)
        self.assertMPDReceived('find "file" "1"\n')

    def test_commands_without_callbacks(self):
        self.MPDWillReturn("\n")
        self.client.close()
        self.assertMPDReceived('close\n')

        # XXX: what are we testing here?
        #      looks like reconnection test?
        self.client._reset()
        self.client.connect(TEST_MPD_HOST, TEST_MPD_PORT)

    def test_set_timeout_on_client(self):
        self.client.timeout = 1
        self.client._sock.settimeout.assert_called_with(1)
        self.assertEqual(self.client.timeout, 1)

        self.client.timeout = None
        self.client._sock.settimeout.assert_called_with(None)
        self.assertEqual(self.client.timeout, None)

    def test_set_timeout_from_connect(self):
        self.client.disconnect()
        with warnings.catch_warnings(record=True) as w:
            self.client.connect("example.com", 10000, timeout=5)
            self.client._sock.settimeout.assert_called_with(5)
            self.assertEqual(len(w), 1)
            self.assertIn('Use MPDClient.timeout', str(w[0].message))

    @unittest.skipIf(sys.version_info < (3, 3), "BrokenPipeError was introduced in python 3.3")
    def test_broken_pipe_error(self):
        self.MPDWillReturn('volume: 63\n', 'OK\n')
        self.client._wfile.write.side_effect = BrokenPipeError
        self.socket_mock.error = Exception

        with self.assertRaises(mpd.ConnectionError):
            self.client.status()

    def test_connection_lost(self):
        # Simulate a connection lost: the socket returns empty strings
        self.MPDWillReturn('')
        self.socket_mock.error = Exception

        with self.assertRaises(mpd.ConnectionError):
            self.client.status()
            self.socket_mock.unpack.assert_called()

        # consistent behaviour, solves bug #11 (github)
        with self.assertRaises(mpd.ConnectionError):
            self.client.status()
            self.socket_mock.unpack.assert_called()

        self.assertIs(self.client._sock, None)

    @unittest.skipIf(sys.version_info < (3, 0),
                     "Automatic decoding/encoding from the socket is only "
                     "available in Python 3")
    def test_force_socket_encoding_to_utf8(self):
        # Force the reconnection to refill the mock
        self.client.disconnect()
        self.client.connect(TEST_MPD_HOST, TEST_MPD_PORT)
        self.assertEqual([mock.call('r', encoding="utf-8", newline="\n"),
                          mock.call('w', encoding="utf-8", newline="\n")],
                         # We are onlyy interested into the 2 first entries,
                         # otherwise we get all the readline() & co...
                         self.client._sock.makefile.call_args_list[0:2])

    def test_ranges_as_argument(self):
        self.MPDWillReturn('OK\n')
        self.client.move((1, 2), 2)
        self.assertMPDReceived('move "1:2" "2"\n')

        self.MPDWillReturn('OK\n')
        self.client.move((1,), 2)
        self.assertMPDReceived('move "1:" "2"\n')

        # old code still works!
        self.MPDWillReturn('OK\n')
        self.client.move("1:2", 2)
        self.assertMPDReceived('move "1:2" "2"\n')

        # empty ranges
        self.MPDWillReturn('OK\n')
        self.client.rangeid(1, ())
        self.assertMPDReceived('rangeid "1" ":"\n')

        with self.assertRaises(ValueError):
            self.MPDWillReturn('OK\n')
            self.client.move((1, "garbage"), 2)
            self.assertMPDReceived('move "1:" "2"\n')

    def test_parse_changes(self):
        self.MPDWillReturn(
            'cpos: 0\n',
            'Id: 66\n',
            'cpos: 1\n',
            'Id: 67\n',
            'cpos: 2\n',
            'Id: 68\n',
            'cpos: 3\n',
            'Id: 69\n',
            'cpos: 4\n',
            'Id: 70\n',
            'OK\n')
        res = self.client.plchangesposid(0)
        self.assertEqual([
            {'cpos': '0', 'id': '66'},
            {'cpos': '1', 'id': '67'},
            {'cpos': '2', 'id': '68'},
            {'cpos': '3', 'id': '69'},
            {'cpos': '4', 'id': '70'}], res)

    def test_parse_database(self):
        self.MPDWillReturn(
            'directory: foo\n',
            'Last-Modified: 2014-01-23T16:42:46Z\n',
            'file: bar.mp3\n',
            'size: 59618802\n',
            'Last-Modified: 2014-11-02T19:57:00Z\n',
            'OK\n')
        self.client.listfiles("/")

    def test_parse_mounts(self):
        self.MPDWillReturn(
            'mount: \n',
            'storage: /home/foo/music\n',
            'mount: foo\n',
            'storage: nfs://192.168.1.4/export/mp3\n',
            'OK\n')
        res = self.client.listmounts()
        self.assertEqual([
            {'mount': '', 'storage': '/home/foo/music'},
            {'mount': 'foo', 'storage': 'nfs://192.168.1.4/export/mp3'}], res)

    def test_parse_neighbors(self):
        self.MPDWillReturn(
            'neighbor: smb://FOO\n',
            'name: FOO (Samba 4.1.11-Debian)\n',
            'OK\n')
        res = self.client.listneighbors()
        self.assertEqual(
            [{'name': 'FOO (Samba 4.1.11-Debian)', 'neighbor': 'smb://FOO'}],
            res)

    def test_parse_outputs(self):
        self.MPDWillReturn(
            'outputid: 0\n',
            'outputname: My ALSA Device\n',
            'outputenabled: 0\n',
            'OK\n')
        res = self.client.outputs()
        self.assertEqual([{
            'outputenabled': '0',
            'outputid': '0',
            'outputname': 'My ALSA Device'}], res)

    def test_parse_playlist(self):
        self.MPDWillReturn(
            '0:file: Weezer - Say It Ain\'t So.mp3\n',
            '1:file: Dire Straits - Walk of Life.mp3\n',
            '2:file: 01 - Love Delicatessen.mp3\n',
            '3:file: Guns N\' Roses - Paradise City.mp3\n',
            '4:file: Nirvana - Lithium.mp3\n',
            'OK\n')
        res = self.client.playlist()
        self.assertEqual([
            "file: Weezer - Say It Ain't So.mp3",
            'file: Dire Straits - Walk of Life.mp3',
            'file: 01 - Love Delicatessen.mp3',
            "file: Guns N' Roses - Paradise City.mp3",
            'file: Nirvana - Lithium.mp3'], res)

    def test_parse_playlists(self):
        self.MPDWillReturn(
            'playlist: Playlist\n',
            'Last-Modified: 2016-08-13T10:55:56Z\n',
            'OK\n')
        res = self.client.listplaylists()
        self.assertEqual([
            {'last-modified': '2016-08-13T10:55:56Z', 'playlist': 'Playlist'}
        ], res)

    def test_parse_plugins(self):
        self.MPDWillReturn(
            'plugin: vorbis\n',
            'suffix: ogg\n',
            'suffix: oga\n',
            'mime_type: application/ogg\n',
            'mime_type: application/x-ogg\n',
            'mime_type: audio/ogg\n',
            'mime_type: audio/vorbis\n',
            'mime_type: audio/vorbis+ogg\n',
            'mime_type: audio/x-ogg\n',
            'mime_type: audio/x-vorbis\n',
            'mime_type: audio/x-vorbis+ogg\n',
            'OK\n')
        res = self.client.decoders()
        self.assertEqual([{
            'mime_type': [
                'application/ogg',
                'application/x-ogg',
                'audio/ogg',
                'audio/vorbis',
                'audio/vorbis+ogg',
                'audio/x-ogg',
                'audio/x-vorbis',
                'audio/x-vorbis+ogg'],
            'plugin': 'vorbis',
            'suffix': [
                'ogg',
                'oga']}], list(res))

    def test_parse_raw_stickers(self):
        self.MPDWillReturn("sticker: foo=bar\n", "OK\n")
        res = self.client._parse_raw_stickers(self.client._read_lines())
        self.assertEqual([('foo', 'bar')], list(res))

        self.MPDWillReturn("sticker: foo=bar\n", "sticker: l=b\n", "OK\n")
        res = self.client._parse_raw_stickers(self.client._read_lines())
        self.assertEqual([('foo', 'bar'), ('l', 'b')], list(res))

    def test_parse_raw_sticker_with_special_value(self):
        self.MPDWillReturn("sticker: foo==uv=vu\n", "OK\n")
        res = self.client._parse_raw_stickers(self.client._read_lines())
        self.assertEqual([('foo', '=uv=vu')], list(res))

    def test_parse_sticket_get_one(self):
        self.MPDWillReturn("sticker: foo=bar\n", "OK\n")
        res = self.client.sticker_get('song', 'baz', 'foo')
        self.assertEqual('bar', res)

    def test_parse_sticket_get_no_sticker(self):
        self.MPDWillReturn("ACK [50@0] {sticker} no such sticker\n")
        self.assertRaises(mpd.CommandError,
                          self.client.sticker_get, 'song', 'baz', 'foo')

    def test_parse_sticker_list(self):
        self.MPDWillReturn("sticker: foo=bar\n", "sticker: lom=bok\n", "OK\n")
        res = self.client.sticker_list('song', 'baz')
        self.assertEqual({'foo': 'bar', 'lom': 'bok'}, res)

        # Even with only one sticker, we get a dict
        self.MPDWillReturn("sticker: foo=bar\n", "OK\n")
        res = self.client.sticker_list('song', 'baz')
        self.assertEqual({'foo': 'bar'}, res)

    def test_command_list(self):
        self.MPDWillReturn(
            'list_OK\n',
            'list_OK\n',
            'list_OK\n',
            'list_OK\n',
            'list_OK\n',
            'volume: 100\n',
            'repeat: 1\n',
            'random: 1\n',
            'single: 0\n',
            'consume: 0\n',
            'playlist: 68\n',
            'playlistlength: 5\n',
            'mixrampdb: 0.000000\n',
            'state: play\n',
            'xfade: 5\n',
            'song: 0\n',
            'songid: 56\n',
            'time: 0:259\n',
            'elapsed: 0.000\n',
            'bitrate: 0\n',
            'nextsong: 2\n',
            'nextsongid: 58\n',
            'list_OK\n',
            'OK\n')
        self.client.command_list_ok_begin()
        self.client.clear()
        self.client.load('Playlist')
        self.client.random(1)
        self.client.repeat(1)
        self.client.play(0)
        self.client.status()
        res = self.client.command_list_end()
        self.assertEqual(None, res[0])
        self.assertEqual(None, res[1])
        self.assertEqual(None, res[2])
        self.assertEqual(None, res[3])
        self.assertEqual(None, res[4])
        self.assertEqual([
            ('bitrate', '0'),
            ('consume', '0'),
            ('elapsed', '0.000'),
            ('mixrampdb', '0.000000'),
            ('nextsong', '2'),
            ('nextsongid', '58'),
            ('playlist', '68'),
            ('playlistlength', '5'),
            ('random', '1'),
            ('repeat', '1'),
            ('single', '0'),
            ('song', '0'),
            ('songid', '56'),
            ('state', 'play'),
            ('time', '0:259'),
            ('volume', '100'),
            ('xfade', '5')], sorted(res[5].items()))


class MockTransport(object):

    def __init__(self):
        self.written = list()

    def clear(self):
        self.written = list()

    def write(self, data):
        self.written.append(data)


@unittest.skipIf(TWISTED_MISSING, "requires twisted to be installed")
class TestMPDProtocol(unittest.TestCase):

    def init_protocol(self, default_idle=True, idle_result=None):
        self.protocol = mpd.MPDProtocol(
            default_idle=default_idle,
            idle_result=idle_result
        )
        self.protocol.transport = MockTransport()

    def test_create_command(self):
        self.init_protocol(default_idle=False)
        self.assertEqual(
            self.protocol._create_command('play'),
            b'play')
        self.assertEqual(
            self.protocol._create_command('rangeid', args=['1', ()]),
            b'rangeid "1" ":"')
        self.assertEqual(
            self.protocol._create_command('rangeid', args=['1', (1,)]),
            b'rangeid "1" "1:"')
        self.assertEqual(
            self.protocol._create_command('rangeid', args=['1', (1, 2)]),
            b'rangeid "1" "1:2"')

    def test_success(self):
        self.init_protocol(default_idle=False)

        def success(result):
            expected = {
                'file': 'Dire Straits - Walk of Life.mp3',
                'artist': 'Dire Straits',
                'title': 'Walk of Life',
                'genre': 'Rock/Pop',
                'track': '3',
                'album': 'Brothers in Arms',
                'id': '13',
                'last-modified': '2016-08-11T10:57:03Z',
                'pos': '4',
                'time': '253'
            }
            self.assertEqual(expected, result)

        self.protocol.currentsong().addCallback(success)
        self.assertEqual([b'currentsong\n'], self.protocol.transport.written)

        for line in [b'file: Dire Straits - Walk of Life.mp3',
                     b'Last-Modified: 2016-08-11T10:57:03Z',
                     b'Time: 253',
                     b'Artist: Dire Straits',
                     b'Title: Walk of Life',
                     b'Album: Brothers in Arms',
                     b'Track: 3',
                     b'Genre: Rock/Pop',
                     b'Pos: 4',
                     b'Id: 13',
                     b'OK']:
            self.protocol.lineReceived(line)

    def test_failure(self):
        self.init_protocol(default_idle=False)

        def error(result):
            self.assertIsInstance(result, Failure)
            self.assertEqual(
                result.getErrorMessage(),
                '[50@0] {load} No such playlist'
            )

        self.protocol.load('Foo').addErrback(error)
        self.assertEqual([b'load "Foo"\n'], self.protocol.transport.written)
        self.protocol.lineReceived(b'ACK [50@0] {load} No such playlist')

    def test_default_idle(self):

        def idle_result(result):
            self.assertEqual(list(result), ['player'])

        self.init_protocol(idle_result=idle_result)
        self.protocol.lineReceived(b'OK MPD 0.18.0')
        self.assertEqual([b'idle\n'], self.protocol.transport.written)
        self.protocol.transport.clear()
        self.protocol.lineReceived(b'changed: player')
        self.protocol.lineReceived(b'OK')
        self.assertEqual(
            [b'idle\n'],
            self.protocol.transport.written
        )

    def test_noidle_when_default_idle(self):
        self.init_protocol()
        self.protocol.lineReceived(b'OK MPD 0.18.0')
        self.protocol.pause()
        self.protocol.lineReceived(b'OK')
        self.protocol.lineReceived(b'OK')
        self.assertEqual(
            [b'idle\n', b'noidle\n', b'pause\n', b'idle\n'],
            self.protocol.transport.written
        )

    def test_already_idle(self):
        self.init_protocol(default_idle=False)
        self.protocol.idle()
        self.assertRaises(mpd.CommandError, lambda: self.protocol.idle())

    def test_already_noidle(self):
        self.init_protocol(default_idle=False)
        self.assertRaises(mpd.CommandError, lambda: self.protocol.noidle())

    def test_command_list(self):
        self.init_protocol(default_idle=False)

        def success(result):
            self.assertEqual([None, None], result)

        self.protocol.command_list_ok_begin()
        self.protocol.play()
        self.protocol.stop()
        self.protocol.command_list_end().addCallback(success)
        self.assertEqual(
            [
                b'command_list_ok_begin\n',
                b'play\n',
                b'stop\n',
                b'command_list_end\n',
            ],
            self.protocol.transport.written
        )
        self.protocol.transport.clear()
        self.protocol.lineReceived(b'list_OK')
        self.protocol.lineReceived(b'list_OK')
        self.protocol.lineReceived(b'OK')

    def test_command_list_failure(self):
        self.init_protocol(default_idle=False)

        def load_command_error(result):
            self.assertIsInstance(result, Failure)
            self.assertEqual(
                result.getErrorMessage(),
                '[50@0] {load} No such playlist'
            )

        def command_list_general_error(result):
            self.assertIsInstance(result, Failure)
            self.assertEqual(
                result.getErrorMessage(),
                'An earlier command failed.'
            )

        self.protocol.command_list_ok_begin()
        self.protocol.load('Foo').addErrback(load_command_error)
        self.protocol.play().addErrback(command_list_general_error)
        self.protocol.command_list_end().addErrback(load_command_error)
        self.assertEqual(
            [
                b'command_list_ok_begin\n',
                b'load "Foo"\n',
                b'play\n',
                b'command_list_end\n',
            ],
            self.protocol.transport.written
        )
        self.protocol.lineReceived(b'ACK [50@0] {load} No such playlist')

    def test_command_list_when_default_idle(self):
        self.init_protocol()
        self.protocol.lineReceived(b'OK MPD 0.18.0')

        def success(result):
            self.assertEqual([None, None], result)

        self.protocol.command_list_ok_begin()
        self.protocol.play()
        self.protocol.stop()
        self.protocol.command_list_end().addCallback(success)
        self.assertEqual(
            [
                b'idle\n',
                b'noidle\n',
                b'command_list_ok_begin\n',
                b'play\n',
                b'stop\n',
                b'command_list_end\n',
            ],
            self.protocol.transport.written
        )
        self.protocol.transport.clear()
        self.protocol.lineReceived(b'OK')
        self.protocol.lineReceived(b'list_OK')
        self.protocol.lineReceived(b'list_OK')
        self.protocol.lineReceived(b'OK')
        self.assertEqual([b'idle\n'], self.protocol.transport.written)

    def test_command_list_failure_when_default_idle(self):
        self.init_protocol()
        self.protocol.lineReceived(b'OK MPD 0.18.0')

        def load_command_error(result):
            self.assertIsInstance(result, Failure)
            self.assertEqual(
                result.getErrorMessage(),
                '[50@0] {load} No such playlist'
            )

        def command_list_general_error(result):
            self.assertIsInstance(result, Failure)
            self.assertEqual(
                result.getErrorMessage(),
                'An earlier command failed.'
            )

        self.protocol.command_list_ok_begin()
        self.protocol.load('Foo').addErrback(load_command_error)
        self.protocol.play().addErrback(command_list_general_error)
        self.protocol.command_list_end().addErrback(load_command_error)
        self.assertEqual(
            [
                b'idle\n',
                b'noidle\n',
                b'command_list_ok_begin\n',
                b'load "Foo"\n',
                b'play\n',
                b'command_list_end\n',
            ],
            self.protocol.transport.written
        )
        self.protocol.transport.clear()
        self.protocol.lineReceived(b'OK')
        self.protocol.lineReceived(b'ACK [50@0] {load} No such playlist')
        self.assertEqual([b'idle\n'], self.protocol.transport.written)

    def test_command_list_item_is_generator(self):
        self.init_protocol(default_idle=False)

        def success(result):
            self.assertEqual(result, [[
                "Weezer - Say It Ain't So.mp3",
                'Dire Straits - Walk of Life.mp3',
                '01 - Love Delicatessen.mp3',
                "Guns N' Roses - Paradise City.mp3"
            ]])

        self.protocol.command_list_ok_begin()
        self.protocol.listplaylist('Foo')
        self.protocol.command_list_end().addCallback(success)
        self.protocol.lineReceived(b'file: Weezer - Say It Ain\'t So.mp3')
        self.protocol.lineReceived(b'file: Dire Straits - Walk of Life.mp3')
        self.protocol.lineReceived(b'file: 01 - Love Delicatessen.mp3')
        self.protocol.lineReceived(b'file: Guns N\' Roses - Paradise City.mp3')
        self.protocol.lineReceived(b'list_OK')
        self.protocol.lineReceived(b'OK')

    def test_already_in_command_list(self):
        self.init_protocol(default_idle=False)
        self.protocol.command_list_ok_begin()
        self.assertRaises(
            mpd.CommandListError,
            lambda: self.protocol.command_list_ok_begin()
        )

    def test_not_in_command_list(self):
        self.init_protocol(default_idle=False)
        self.assertRaises(
            mpd.CommandListError,
            lambda: self.protocol.command_list_end()
        )

    def test_invalid_command_in_command_list(self):
        self.init_protocol(default_idle=False)
        self.protocol.command_list_ok_begin()
        self.assertRaises(
            mpd.CommandListError,
            lambda: self.protocol.kill()
        )

    def test_close(self):
        self.init_protocol(default_idle=False)

        def success(result):
            self.assertEqual(result, None)

        self.protocol.close().addCallback(success)

class AsyncMockServer:
    def __init__(self):
        self._output = asyncio.Queue()
        self._expectations = []

    def get_streams(self):
        result = asyncio.Future()
        result.set_result((self, self))
        return result

    def readline(self):
        # directly passing around the awaitable
        return self._output.get()

    def write(self, data):
        try:
            next_write = self._expectations[0][0][0]
        except IndexError:
            self.error("Data written to mock even though none expected: %r" % data)
        if next_write == data:
            self._expectations[0][0].pop(0)
            self._feed()
        else:
            self.error("Mock got %r, expected %r" % (data, next_write))

    def error(self, message):
        raise AssertionError(message)

    def _feed(self):
        if len(self._expectations[0][0]) == 0:
            _, response_lines = self._expectations.pop(0)
            for l in response_lines:
                self._output.put_nowait(l)

    def expect_exchange(self, request_lines, response_lines):
        self._expectations.append((request_lines, response_lines))
        self._feed()

@unittest.skipIf(asyncio is None, "requires asyncio to be available")
class TestAsyncioMPD(unittest.TestCase):
    def init_client(self, odd_hello=None):
        import mpd.asyncio

        self.loop = asyncio.get_event_loop()

        self.mockserver = AsyncMockServer()
        asyncio.open_connection = mock.MagicMock(return_value=self.mockserver.get_streams())

        if odd_hello is None:
            hello_lines = [b'OK MPD mocker\n']
        else:
            hello_lines = odd_hello

        self.mockserver.expect_exchange([], hello_lines)

        self.client = mpd.asyncio.MPDClient()
        self._await(self.client.connect(TEST_MPD_HOST, TEST_MPD_PORT, loop=self.loop))

        asyncio.open_connection.assert_called_with(TEST_MPD_HOST, TEST_MPD_PORT, loop=self.loop)

    def _await(self, future):
        return self.loop.run_until_complete(future)

    def test_oddhello(self):
        self.assertRaises(mpd.base.ProtocolError, self.init_client, odd_hello=[b'NOT OK\n'])

    @unittest.skip("This test would add 5 seconds of idling to the run")
    def test_noresponse(self):
        self.assertRaises(mpd.base.ConnectionError, self.init_client, odd_hello=[])

    def test_status(self):
        self.init_client()

        self.mockserver.expect_exchange([b"status\n"], [
            b"volume: 70\n",
            b"repeat: 0\n",
            b"random: 1\n",
            b"single: 0\n",
            b"consume: 0\n",
            b"playlist: 416\n",
            b"playlistlength: 7\n",
            b"mixrampdb: 0.000000\n",
            b"state: play\n",
            b"song: 4\n",
            b"songid: 19\n",
            b"time: 28:403\n",
            b"elapsed: 28.003\n",
            b"bitrate: 465\n",
            b"duration: 403.066\n",
            b"audio: 44100:16:2\n",
            b"OK\n",
            ])

        status = self._await(self.client.status())
        self.assertEqual(status, {
            'audio': '44100:16:2',
            'bitrate': '465',
            'consume': '0',
            'duration': '403.066',
            'elapsed': '28.003',
            'mixrampdb': '0.000000',
            'playlist': '416',
            'playlistlength': '7',
            'random': '1',
            'repeat': '0',
            'single': '0',
            'song': '4',
            'songid': '19',
            'state': 'play',
            'time': '28:403',
            'volume': '70',
            })

    def test_mocker(self):
        """Does the mock server refuse unexpected writes?"""
        self.init_client()

        self.mockserver.expect_exchange([b"expecting odd things\n"], [b""])
        self.assertRaises(AssertionError, self._await, self.client.status())

if __name__ == '__main__':
    unittest.main()