[Pkg-mpd-commits] [python-mpd] 06/91: Command registration is now done via ``mpd_commands`` decorator. Commands are registered on ``MPDClientBase`` class now with the corresponding result parser callbacks. Adopt ``MPDClient`` class. Add more tests
Simon McVittie
smcv at debian.org
Sat Feb 24 14:55:26 UTC 2018
This is an automated email from the git hooks/post-receive script.
smcv pushed a commit to branch debian/master
in repository python-mpd.
commit 7a80317a22de7723566ce4d24a09887118f6bc3c
Author: Robert Niederreiter <office at squarewave.at>
Date: Thu Aug 25 09:21:16 2016 +0200
Command registration is now done via ``mpd_commands`` decorator. Commands are registered on ``MPDClientBase`` class now with the corresponding result parser callbacks. Adopt ``MPDClient`` class. Add more tests
---
mpd/base.py | 600 +++++++++++++++++++++++++++++------------------------------
mpd/tests.py | 228 ++++++++++++++++++++---
2 files changed, 502 insertions(+), 326 deletions(-)
diff --git a/mpd/base.py b/mpd/base.py
index 64c1082..d8219d3 100644
--- a/mpd/base.py
+++ b/mpd/base.py
@@ -17,11 +17,16 @@
# You should have received a copy of the GNU Lesser General Public License
# along with python-mpd2. If not, see <http://www.gnu.org/licenses/>.
+from collections import Callable
import logging
-import sys
import socket
+import sys
import warnings
-from collections import Callable
+
+
+###############################################################################
+# constants
+###############################################################################
VERSION = (0, 6, 0)
HELLO_PREFIX = "OK MPD "
@@ -29,6 +34,11 @@ ERROR_PREFIX = "ACK "
SUCCESS = "OK"
NEXT = "list_OK"
+
+###############################################################################
+# utils
+###############################################################################
+
IS_PYTHON2 = sys.version_info < (3, 0)
if IS_PYTHON2:
def decode_str(s):
@@ -40,11 +50,20 @@ if IS_PYTHON2:
else:
return (unicode(s)).encode("utf-8")
else:
-
def decode_str(s):
return s
+
encode_str = str
+
+def escape(text):
+ return text.replace("\\", "\\\\").replace('"', '\\"')
+
+
+###############################################################################
+# logging
+###############################################################################
+
try:
from logging import NullHandler
except ImportError: # NullHandler was introduced in python2.7
@@ -52,10 +71,15 @@ except ImportError: # NullHandler was introduced in python2.7
def emit(self, record):
pass
+
logger = logging.getLogger(__name__)
logger.addHandler(NullHandler())
+###############################################################################
+# exceptions
+###############################################################################
+
class MPDError(Exception):
pass
@@ -84,163 +108,100 @@ class IteratingError(MPDError):
pass
-class _NotConnected(object):
- def __getattr__(self, attr):
- return self._dummy
+###############################################################################
+# command registration
+###############################################################################
- def _dummy(*args):
- raise ConnectionError("Not connected")
+class mpd_commands(object):
+ """Decorator for registering MPD commands with it's corresponding result
+ callback.
+ """
+
+ def __init__(self, *commands):
+ self.commands = commands
-_commands = {
- # Status Commands
- "clearerror": "_fetch_nothing",
- "currentsong": "_fetch_object",
- "idle": "_fetch_idle",
- "status": "_fetch_object",
- "stats": "_fetch_object",
- # Playback Option Commands
- "consume": "_fetch_nothing",
- "crossfade": "_fetch_nothing",
- "mixrampdb": "_fetch_nothing",
- "mixrampdelay": "_fetch_nothing",
- "random": "_fetch_nothing",
- "repeat": "_fetch_nothing",
- "setvol": "_fetch_nothing",
- "single": "_fetch_nothing",
- "replay_gain_mode": "_fetch_nothing",
- "replay_gain_status": "_fetch_item",
- # Playback Control Commands
- "next": "_fetch_nothing",
- "pause": "_fetch_nothing",
- "play": "_fetch_nothing",
- "playid": "_fetch_nothing",
- "previous": "_fetch_nothing",
- "seek": "_fetch_nothing",
- "seekid": "_fetch_nothing",
- "seekcur": "_fetch_nothing",
- "stop": "_fetch_nothing",
- # Playlist Commands
- "add": "_fetch_nothing",
- "addid": "_fetch_item",
- "addtagid": "_fetch_nothing",
- "cleartagid": "_fetch_nothing",
- "clear": "_fetch_nothing",
- "delete": "_fetch_nothing",
- "deleteid": "_fetch_nothing",
- "move": "_fetch_nothing",
- "moveid": "_fetch_nothing",
- "playlist": "_fetch_playlist",
- "playlistfind": "_fetch_songs",
- "playlistid": "_fetch_songs",
- "playlistinfo": "_fetch_songs",
- "playlistsearch": "_fetch_songs",
- "plchanges": "_fetch_songs",
- "plchangesposid": "_fetch_changes",
- "prio": "_fetch_nothing",
- "prioid": "_fetch_nothing",
- "rangeid": "_fetch_nothing",
- "shuffle": "_fetch_nothing",
- "swap": "_fetch_nothing",
- "swapid": "_fetch_nothing",
- # Stored Playlist Commands
- "listplaylist": "_fetch_list",
- "listplaylistinfo": "_fetch_songs",
- "listplaylists": "_fetch_playlists",
- "load": "_fetch_nothing",
- "playlistadd": "_fetch_nothing",
- "playlistclear": "_fetch_nothing",
- "playlistdelete": "_fetch_nothing",
- "playlistmove": "_fetch_nothing",
- "rename": "_fetch_nothing",
- "rm": "_fetch_nothing",
- "save": "_fetch_nothing",
- # Database Commands
- "count": "_fetch_object",
- "find": "_fetch_songs",
- "findadd": "_fetch_nothing",
- "list": "_fetch_list",
- "listall": "_fetch_database",
- "listallinfo": "_fetch_database",
- "listfiles": "_fetch_database",
- "lsinfo": "_fetch_database",
- "readcomments": "_fetch_object",
- "search": "_fetch_songs",
- "searchadd": "_fetch_nothing",
- "searchaddpl": "_fetch_nothing",
- "update": "_fetch_item",
- "rescan": "_fetch_item",
- # Mounts and neighbors
- "mount": "_fetch_nothing",
- "umount": "_fetch_nothing",
- "listmounts": "_fetch_mounts",
- "listneighbors": "_fetch_neighbors",
- # Sticker Commands
- "sticker get": "_fetch_sticker",
- "sticker set": "_fetch_nothing",
- "sticker delete": "_fetch_nothing",
- "sticker list": "_fetch_stickers",
- "sticker find": "_fetch_songs",
- # Connection Commands
- "close": None,
- "kill": None,
- "password": "_fetch_nothing",
- "ping": "_fetch_nothing",
- # Audio Output Commands
- "disableoutput": "_fetch_nothing",
- "enableoutput": "_fetch_nothing",
- "toggleoutput": "_fetch_nothing",
- "outputs": "_fetch_outputs",
- # Reflection Commands
- "config": "_fetch_item",
- "commands": "_fetch_list",
- "notcommands": "_fetch_list",
- "tagtypes": "_fetch_list",
- "urlhandlers": "_fetch_list",
- "decoders": "_fetch_plugins",
- # Client To Client
- "subscribe": "_fetch_nothing",
- "unsubscribe": "_fetch_nothing",
- "channels": "_fetch_list",
- "readmessages": "_fetch_messages",
- "sendmessage": "_fetch_nothing",
-}
+ def __call__(self, ob):
+ ob.mpd_commands = self.commands
+ return ob
+def mpd_command_provider(cls):
+ """Decorator hooking up registered MPD commands to concrete client
+ implementation.
+
+ A class using this decorator must inherit from ``MPDClientBase`` and
+ implement it's ``add_command`` function.
+ """
+ def collect(cls, callbacks=dict()):
+ """Collect MPD command callbacks from given class.
+
+ Searches class __dict__ on given class and all it's bases for functions
+ which have been decorated with @mpd_commands and returns a dict
+ containing callback name as keys and
+ (callback, callback implementing class) tuples as values.
+ """
+ for name, ob in cls.__dict__.items():
+ if hasattr(ob, "mpd_commands") and name not in callbacks:
+ callbacks[name] = (ob, cls)
+ for base in cls.__bases__:
+ callbacks = collect(base, callbacks)
+ return callbacks
+
+ for name, value in collect(cls).items():
+ callback, from_ = value
+ for command in callback.mpd_commands:
+ cls.add_command(command, callback)
+ return cls
+
+
+class Noop(object):
+ """An instance of this class represents a MPD command callback which
+ does nothing.
+ """
+ mpd_commands = None
+
+
+###############################################################################
+# abstract base client
+###############################################################################
+
class MPDClientBase(object):
+ """Abstract MPD client.
+
+ This class defines a general client contract, provides MPD protocol parsers
+ and defines all available MPD commands and it's corresponding result
+ parsing callbacks. There might be the need of overriding some callbacks on
+ subclasses.
+ """
def __init__(self, use_unicode=False):
self.iterate = False
self.use_unicode = use_unicode
self._reset()
- def disconnect(self):
+ @classmethod
+ def add_command(cls, name, callback):
raise NotImplementedError(
- "MPDClientBase does not implement disconnect")
+ 'Abstract ``MPDClientBase`` does not implement ``add_command``')
+
+ def noidle(self):
+ raise NotImplementedError(
+ 'Abstract ``MPDClientBase`` does not implement ``noidle``')
+
+ def command_list_ok_begin(self):
+ raise NotImplementedError(
+ 'Abstract ``MPDClientBase`` does not implement '
+ '``command_list_ok_begin``')
+
+ def command_list_end(self):
+ raise NotImplementedError(
+ 'Abstract ``MPDClientBase`` does not implement '
+ '``command_list_end``')
def _reset(self):
self.mpd_version = None
- self._pending = []
self._command_list = None
- def _parse_line(self, line):
- if self.use_unicode:
- line = decode_str(line)
- if not line.endswith("\n"):
- self.disconnect()
- raise ConnectionError("Connection lost while reading line")
- line = line.rstrip("\n")
- if line.startswith(ERROR_PREFIX):
- error = line[len(ERROR_PREFIX):].strip()
- raise CommandError(error)
- if self._command_list is not None:
- if line == NEXT:
- return
- if line == SUCCESS:
- raise ProtocolError("Got unexpected '{}'".format(SUCCESS))
- elif line == SUCCESS:
- return
- return line
-
def _parse_pair(self, line, separator):
if line is None:
return
@@ -253,20 +214,6 @@ class MPDClientBase(object):
for line in lines:
yield self._parse_pair(line, separator)
- def _parse_list(self, lines):
- seen = None
- for key, value in self._parse_pairs(lines):
- if key != seen:
- if seen is not None:
- raise ProtocolError(
- "Expected key '{}', got '{}'".format(seen, key))
- seen = key
- yield value
-
- def _parse_playlist(self, lines):
- for key, value in self._parse_pairs(lines, ":"):
- yield value
-
def _parse_objects(self, lines, delimiters=[]):
obj = {}
for key, value in self._parse_pairs(lines):
@@ -285,7 +232,7 @@ class MPDClientBase(object):
if obj:
yield obj
- def _parse_stickers(self, lines):
+ def _parse_raw_stickers(self, lines):
for key, sticker in self._parse_pairs(lines):
value = sticker.split('=', 1)
if len(value) < 2:
@@ -293,16 +240,171 @@ class MPDClientBase(object):
"Could not parse sticker: {}".format(repr(sticker)))
yield tuple(value)
+ NOOP = mpd_commands('close', 'kill')(Noop())
+
+ @mpd_commands('plchangesposid')
+ def _parse_changes(self, lines):
+ return self._parse_objects(lines, ["cpos"])
+
+ @mpd_commands('listall', 'listallinfo', 'listfiles', 'lsinfo')
+ def _parse_database(self, lines):
+ return self._parse_objects(lines, ["file", "directory", "playlist"])
+
+ @mpd_commands('idle')
+ def _parse_idle(self, lines):
+ return self._parse_list(lines)
+
+ @mpd_commands('addid', 'config', 'replay_gain_status', 'rescan', 'update')
+ def _parse_item(self, lines):
+ pairs = list(self._parse_pairs(lines))
+ if len(pairs) != 1:
+ return
+ return pairs[0][1]
+ @mpd_commands(
+ 'channels', 'commands', 'list', 'listplaylist', 'notcommands',
+ 'tagtypes', 'urlhandlers')
+ def _parse_list(self, lines):
+ seen = None
+ for key, value in self._parse_pairs(lines):
+ if key != seen:
+ if seen is not None:
+ raise ProtocolError(
+ "Expected key '{}', got '{}'".format(seen, key))
+ seen = key
+ yield value
+
+ @mpd_commands('readmessages')
+ def _parse_messages(self, lines):
+ return self._parse_objects(lines, ["channel"])
+
+ @mpd_commands('listmounts')
+ def _parse_mounts(self, lines):
+ return self._parse_objects(lines, ["mount"])
+
+ @mpd_commands('listneighbors')
+ def _parse_neighbors(self, lines):
+ return self._parse_objects(lines, ["neighbor"])
+
+ @mpd_commands(
+ 'add', 'addtagid', 'clear', 'clearerror', 'cleartagid', 'consume',
+ 'crossfade', 'delete', 'deleteid', 'disableoutput', 'enableoutput',
+ 'findadd', 'load', 'mixrampdb', 'mixrampdelay', 'mount', 'move',
+ 'moveid', 'next', 'password', 'pause', 'ping', 'play', 'playid',
+ 'playlistadd', 'playlistclear', 'playlistdelete', 'playlistmove',
+ 'previous', 'prio', 'prioid', 'random', 'rangeid', 'rename', 'repeat',
+ 'replay_gain_mode', 'rm', 'save', 'searchadd', 'searchaddpl', 'seek',
+ 'seekcur', 'seekid', 'sendmessage', 'setvol', 'shuffle', 'single',
+ 'sticker delete', 'sticker set', 'stop', 'subscribe', 'swap', 'swapid',
+ 'toggleoutput', 'umount', 'unsubscribe')
+ def _parse_nothing(self, lines):
+ for line in lines:
+ raise ProtocolError(
+ "Got unexpected return value: '{}'".format(', '.join(lines)))
+
+ @mpd_commands('count', 'currentsong', 'readcomments', 'stats', 'status')
+ def _parse_object(self, lines):
+ objs = list(self._parse_objects(lines))
+ if not objs:
+ return {}
+ return objs[0]
+
+ @mpd_commands('outputs')
+ def _parse_outputs(self, lines):
+ return self._parse_objects(lines, ["outputid"])
+
+ @mpd_commands('playlist')
+ def _parse_playlist(self, lines):
+ for key, value in self._parse_pairs(lines, ":"):
+ yield value
+
+ @mpd_commands('listplaylists')
+ def _parse_playlists(self, lines):
+ return self._parse_objects(lines, ["playlist"])
+
+ @mpd_commands('decoders')
+ def _parse_plugins(self, lines):
+ return self._parse_objects(lines, ["plugin"])
+
+ @mpd_commands(
+ 'find', 'listplaylistinfo', 'playlistfind', 'playlistid',
+ 'playlistinfo', 'playlistsearch', 'plchanges', 'search', 'sticker find')
+ def _parse_songs(self, lines):
+ return self._parse_objects(lines, ["file"])
+
+ @mpd_commands('sticker get')
+ def _parse_sticker(self, lines):
+ key, value = list(self._parse_raw_stickers(lines))[0]
+ return value
+
+ @mpd_commands('sticker list')
+ def _parse_stickers(self, lines):
+ return dict(self._parse_raw_stickers(lines))
+
+
+###############################################################################
+# sync client
+###############################################################################
+
+def _create_callback(self, function, wrap_result):
+ """Create MPD command related response callback.
+ """
+ if not isinstance(function, Callable):
+ return None
+ def command_callback():
+ # command result callback expects response from MPD as iterable lines,
+ # thus read available lines from socket
+ res = function(self, self._read_lines())
+ # wrap result in iterator helper if desired
+ if wrap_result:
+ res = self._wrap_iterator(res)
+ return res
+ return command_callback
+
+
+def _create_command(wrapper, name, return_value, wrap_result):
+ """Create MPD command related function.
+ """
+ def mpd_command(self, *args):
+ callback = _create_callback(self, return_value, wrap_result)
+ return wrapper(self, name, args, callback)
+ return mpd_command
+
+
+class _NotConnected(object):
+ def __getattr__(self, attr):
+ return self._dummy
+
+ def _dummy(*args):
+ raise ConnectionError("Not connected")
+
+
+ at mpd_command_provider
class MPDClient(MPDClientBase):
idletimeout = None
_timeout = None
+ _wrap_iterator_parsers = [
+ MPDClientBase._parse_list,
+ MPDClientBase._parse_playlist,
+ MPDClientBase._parse_changes,
+ MPDClientBase._parse_songs,
+ MPDClientBase._parse_mounts,
+ MPDClientBase._parse_neighbors,
+ MPDClientBase._parse_playlists,
+ MPDClientBase._parse_database,
+ MPDClientBase._parse_messages,
+ MPDClientBase._parse_outputs,
+ MPDClientBase._parse_plugins
+ ]
+ if IS_PYTHON2:
+ _wrap_iterator_parsers = [f.__func__ for f in _wrap_iterator_parsers]
def __init__(self, use_unicode=False):
super(MPDClient, self).__init__(use_unicode=use_unicode)
def _reset(self):
super(MPDClient, self)._reset()
+ self._pending = []
self._iterating = False
self._sock = None
self._rfile = _NotConnected()
@@ -374,13 +476,28 @@ class MPDClient(MPDClientBase):
logger.debug("Calling MPD password(******)")
else:
logger.debug("Calling MPD %s%r", command, args)
- self._write_line(" ".join(parts))
-
- ##################
- # response helpers
+ cmd = " ".join(parts)
+ self._write_line(cmd)
def _read_line(self):
- return self._parse_line(self._rfile.readline())
+ line = self._rfile.readline()
+ if self.use_unicode:
+ line = decode_str(line)
+ if not line.endswith("\n"):
+ self.disconnect()
+ raise ConnectionError("Connection lost while reading line")
+ line = line.rstrip("\n")
+ if line.startswith(ERROR_PREFIX):
+ error = line[len(ERROR_PREFIX):].strip()
+ raise CommandError(error)
+ if self._command_list is not None:
+ if line == NEXT:
+ return
+ if line == SUCCESS:
+ raise ProtocolError("Got unexpected '{}'".format(SUCCESS))
+ elif line == SUCCESS:
+ return
+ return line
def _read_lines(self):
line = self._read_line()
@@ -388,31 +505,13 @@ class MPDClient(MPDClientBase):
yield line
line = self._read_line()
- def _read_pair(self, separator):
- return self._parse_pair(self._read_line(), separator)
-
- def _read_pairs(self, separator=": "):
- return self._parse_pairs(self._read_lines(), separator)
-
- def _read_list(self):
- return self._parse_list(self._read_lines())
-
- def _read_playlist(self):
- return self._parse_playlist(self._read_lines())
-
- def _read_objects(self, delimiters=[]):
- return self._parse_objects(self._read_lines(), delimiters)
-
- def _read_stickers(self):
- return self._parse_stickers(self._read_lines())
-
def _read_command_list(self):
try:
for retval in self._command_list:
yield retval()
finally:
self._command_list = None
- self._fetch_nothing()
+ self._parse_nothing(self._read_lines())
def _iterator_wrapper(self, iterator):
try:
@@ -427,83 +526,6 @@ class MPDClient(MPDClientBase):
self._iterating = True
return self._iterator_wrapper(iterator)
- ####################
- # response callbacks
-
- def _fetch_nothing(self):
- line = self._read_line()
- if line is not None:
- raise ProtocolError(
- "Got unexpected return value: '{}'".format(line))
-
- def _fetch_item(self):
- pairs = list(self._read_pairs())
- if len(pairs) != 1:
- return
- return pairs[0][1]
-
- def _fetch_sticker(self):
- # Either we get one or we get an error while reading the line
- key, value = list(self._read_stickers())[0]
- return value
-
- def _fetch_stickers(self):
- return dict(self._read_stickers())
-
- def _fetch_list(self):
- return self._wrap_iterator(self._read_list())
-
- def _fetch_playlist(self):
- return self._wrap_iterator(self._read_playlist())
-
- def _fetch_object(self):
- objs = list(self._read_objects())
- if not objs:
- return {}
- return objs[0]
-
- def _fetch_objects(self, delimiters):
- return self._wrap_iterator(self._read_objects(delimiters))
-
- def _fetch_changes(self):
- return self._fetch_objects(["cpos"])
-
- def _fetch_idle(self):
- self._sock.settimeout(self.idletimeout)
- ret = self._fetch_list()
- self._sock.settimeout(self._timeout)
- return ret
-
- def _fetch_songs(self):
- return self._fetch_objects(["file"])
-
- def _fetch_mounts(self):
- return self._fetch_objects(["mount"])
-
- def _fetch_neighbors(self):
- return self._fetch_objects(["neighbor"])
-
- def _fetch_playlists(self):
- return self._fetch_objects(["playlist"])
-
- def _fetch_database(self):
- return self._fetch_objects(["file", "directory", "playlist"])
-
- def _fetch_messages(self):
- return self._fetch_objects(["channel"])
-
- def _fetch_outputs(self):
- return self._fetch_objects(["outputid"])
-
- def _fetch_plugins(self):
- return self._fetch_objects(["plugin"])
-
- def _fetch_command_list(self):
- return self._wrap_iterator(self._read_command_list())
-
- # end response callbacks
- ########################
-
def _hello(self):
line = self._rfile.readline()
if not line.endswith("\n"):
@@ -549,6 +571,13 @@ class MPDClient(MPDClientBase):
else:
raise ConnectionError("getaddrinfo returns an empty list")
+ @mpd_commands('idle')
+ def _parse_idle(self, lines):
+ self._sock.settimeout(self.idletimeout)
+ ret = self._wrap_iterator(self._parse_list(lines))
+ self._sock.settimeout(self._timeout)
+ return ret
+
@property
def timeout(self):
return self._timeout
@@ -619,7 +648,7 @@ class MPDClient(MPDClientBase):
raise CommandError(msg)
del self._pending[0]
self._write_command("noidle")
- return self._fetch_list()
+ return self._wrap_iterator(self._parse_list(self._read_lines()))
def command_list_ok_begin(self):
if self._command_list is not None:
@@ -638,19 +667,20 @@ class MPDClient(MPDClientBase):
if self._iterating:
raise IteratingError("Already iterating over a command list")
self._write_command("command_list_end")
- return self._fetch_command_list()
+ return self._wrap_iterator(self._read_command_list())
@classmethod
def add_command(cls, name, callback):
- method = new_function(cls._execute, name, callback)
- send_method = new_function(cls._send, name, callback)
- fetch_method = new_function(cls._fetch, name, callback)
+ wrap_result = callback in cls._wrap_iterator_parsers
+ method = _create_command(cls._execute, name, callback, wrap_result)
+ send_method = _create_command(cls._send, name, callback, wrap_result)
+ fetch_method = _create_command(cls._fetch, name, callback, wrap_result)
# create new mpd commands as function in three flavors:
# normal, with "send_"-prefix and with "fetch_"-prefix
escaped_name = name.replace(" ", "_")
setattr(cls, escaped_name, method)
- setattr(cls, "send_"+escaped_name, send_method)
- setattr(cls, "fetch_"+escaped_name, fetch_method)
+ setattr(cls, "send_" + escaped_name, send_method)
+ setattr(cls, "fetch_" + escaped_name, fetch_method)
@classmethod
def remove_command(cls, name):
@@ -662,42 +692,4 @@ class MPDClient(MPDClientBase):
delattr(cls, str("send_" + name))
delattr(cls, str("fetch_" + name))
-
-def bound_decorator(self, function):
- """Bind decorator to self.
- """
- if not isinstance(function, Callable):
- return None
- def decorator(*args, **kwargs):
- return function(self, *args, **kwargs)
- return decorator
-
-
-def new_function(wrapper, name, return_value):
- def decorator(self, *args):
- return wrapper(self, name, args, bound_decorator(self, return_value))
- return decorator
-
-
-def lookup_func(cls, name):
- func = None
- if name in cls.__dict__:
- func = cls.__dict__[name]
- else:
- for base in cls.__bases__:
- func = lookup_func(base, name)
- if func is not None:
- break
- return func
-
-
-for key, value in _commands.items():
- return_value = lookup_func(MPDClient, value)
- MPDClient.add_command(key, return_value)
-
-
-def escape(text):
- return text.replace("\\", "\\\\").replace('"', '\\"')
-
-
# vim: set expandtab shiftwidth=4 softtabstop=4 textwidth=79:
diff --git a/mpd/tests.py b/mpd/tests.py
index 908d72c..873e688 100755
--- a/mpd/tests.py
+++ b/mpd/tests.py
@@ -2,12 +2,11 @@
# -*- coding: utf-8 -*-
import itertools
+import mpd
import sys
import types
import warnings
-import mpd
-
try:
# is required for python2.6
# python2.7 works with this module too
@@ -69,6 +68,20 @@ class TestMPDClient(unittest.TestCase):
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"))
@@ -89,7 +102,7 @@ class TestMPDClient(unittest.TestCase):
self.assertIsInstance(song["track"], list)
self.assertMPDReceived('currentsong\n')
- def test_fetch_nothing(self):
+ def test_parse_nothing(self):
self.MPDWillReturn('OK\n', 'OK\n')
self.assertIsNone(self.client.ping())
@@ -98,17 +111,17 @@ class TestMPDClient(unittest.TestCase):
self.assertIsNone(self.client.clearerror())
self.assertMPDReceived('clearerror\n')
- def test_fetch_list(self):
+ def test_parse_list(self):
self.MPDWillReturn('OK\n')
self.assertIsInstance(self.client.list("album"), list)
self.assertMPDReceived('list "album"\n')
- def test_fetch_item(self):
+ def test_parse_item(self):
self.MPDWillReturn('updating_db: 42\n', 'OK\n')
self.assertIsNotNone(self.client.update())
- def test_fetch_object(self):
+ 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()
@@ -121,7 +134,7 @@ class TestMPDClient(unittest.TestCase):
self.assertMPDReceived('stats\n')
self.assertIsInstance(stats, dict)
- def test_fetch_songs(self):
+ def test_parse_songs(self):
self.MPDWillReturn("file: my-song.ogg\n",
"Pos: 0\n",
"Id: 66\n",
@@ -217,7 +230,7 @@ class TestMPDClient(unittest.TestCase):
self.MPDWillReturn("ACK awesome command\n")
self.client.add_command("awesome command",
- mpd.MPDClient._fetch_nothing)
+ 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"))
@@ -302,6 +315,7 @@ class TestMPDClient(unittest.TestCase):
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)
@@ -372,27 +386,139 @@ class TestMPDClient(unittest.TestCase):
self.client.move((1, "garbage"), 2)
self.assertMPDReceived('move "1:" "2"\n')
- def test_read_stickers(self):
+ 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._read_stickers()
+ 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._read_stickers()
+ res = self.client._parse_raw_stickers(self.client._read_lines())
self.assertEqual([('foo', 'bar'), ('l', 'b')], list(res))
- def test_fetch_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_read_sticker_with_special_value(self):
+ def test_parse_raw_sticker_with_special_value(self):
self.MPDWillReturn("sticker: foo==uv=vu\n", "OK\n")
- res = self.client._read_stickers()
+ res = self.client._parse_raw_stickers(self.client._read_lines())
self.assertEqual([('foo', '=uv=vu')], list(res))
def test_parse_sticket_get_one(self):
@@ -415,5 +541,63 @@ class TestMPDClient(unittest.TestCase):
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()))
+
if __name__ == '__main__':
unittest.main()
--
Alioth's /usr/local/bin/git-commit-notice on /srv/git.debian.org/git/pkg-mpd/python-mpd.git
More information about the Pkg-mpd-commits
mailing list