[Pkg-mpd-commits] [python-mpd] 35/91: Add tests for twisted client

Simon McVittie smcv at debian.org
Sat Feb 24 14:55:31 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 5bf0e0643d80a5c1431d19fdcb05b25a0f930e24
Author: Robert Niederreiter <office at squarewave.at>
Date:   Mon Sep 12 11:21:32 2016 +0200

    Add tests for twisted client
---
 .travis.yml    |   3 +
 mpd/tests.py   | 292 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++
 mpd/twisted.py |  11 ++-
 setup.py       |  14 ++-
 tox.ini        |   6 +-
 5 files changed, 318 insertions(+), 8 deletions(-)

diff --git a/.travis.yml b/.travis.yml
index 420e7a2..8aec8db 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -9,4 +9,7 @@ python:
   - "3.5"
   - "pypy"
 
+install:
+  - if [[ $TRAVIS_PYTHON_VERSION != '3.2' ]]; then pip install Twisted; fi
+
 script: python -m unittest mpd.tests
diff --git a/mpd/tests.py b/mpd/tests.py
index 873e688..5116841 100755
--- a/mpd/tests.py
+++ b/mpd/tests.py
@@ -1,6 +1,7 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 
+from __future__ import absolute_import
 import itertools
 import mpd
 import sys
@@ -21,6 +22,14 @@ except ImportError:
         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
+
+try:
     import mock
 except ImportError:
     print("Please install mock from PyPI to run tests!")
@@ -599,5 +608,288 @@ class TestMPDClient(unittest.TestCase):
             ('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)
+
+
+ at 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_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 at 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 at 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 at 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 at 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 at 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 at 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)
+
+
 if __name__ == '__main__':
     unittest.main()
diff --git a/mpd/twisted.py b/mpd/twisted.py
index 3793d81..4b65c6f 100644
--- a/mpd/twisted.py
+++ b/mpd/twisted.py
@@ -44,6 +44,7 @@ def _create_command(wrapper, name, callback):
     def mpd_command(self, *args):
         def bound_callback(lines):
             return callback(self, lines)
+        bound_callback.callback = callback
         return wrapper(self, name, args, bound_callback)
     return mpd_command
 
@@ -99,7 +100,6 @@ class MPDProtocol(basic.LineReceiver, MPDClientBase):
                 del self._command_list_results[0]
             else:
                 state_list.pop(0).errback(CommandError(error))
-            # XXX: reset received lines here?
             self._continue_idle()
         elif line == SUCCESS or (command_list and line == NEXT):
             state_list.pop(0).callback(self._rcvd_lines[:])
@@ -110,9 +110,14 @@ class MPDProtocol(basic.LineReceiver, MPDClientBase):
         else:
             self._rcvd_lines.append(line)
 
+    def _lookup_callback(self, parser):
+        if hasattr(parser, 'callback'):
+            return parser.callback
+        return parser
+
     def _execute(self, command, args, parser):
         # close or kill command in command list not allowed
-        if self._command_list and not callable(parser):
+        if self._command_list and self._lookup_callback(parser) is self.NOOP:
             msg = '{} not allowed in command list'.format(command)
             raise CommandListError(msg)
         # default state idle and currently in idle state, trigger noidle
@@ -126,7 +131,7 @@ class MPDProtocol(basic.LineReceiver, MPDClientBase):
         state = self._state[-1] if self._command_list else self._state
         state.append(deferred)
         # NOOP is for close and kill commands
-        if parser is not self.NOOP:
+        if self._lookup_callback(parser) is not self.NOOP:
             # attach command related result parser
             deferred.addCallback(parser)
             # command list, attach handler for collecting command list results
diff --git a/setup.py b/setup.py
index 2d21944..09f1175 100644
--- a/setup.py
+++ b/setup.py
@@ -77,9 +77,17 @@ setup(
     zip_safe=True,
     keywords=["mpd"],
     test_suite="mpd.tests",
-    tests_require=['tox'],
-    cmdclass={'test': Tox},
-    extras_require={'twisted':  ["twisted"]}
+    tests_require=[
+        'tox',
+        'mock',
+        'Twisted'
+    ],
+    cmdclass={
+        'test': Tox
+    },
+    extras_require={
+        'twisted': ['Twisted']
+    }
 )
 
 # vim: set expandtab shiftwidth=4 softtabstop=4 textwidth=79:
diff --git a/tox.ini b/tox.ini
index 94666aa..28d12b9 100644
--- a/tox.ini
+++ b/tox.ini
@@ -4,8 +4,9 @@ envlist = py26,py27,py32,py33,py34,py35,pypy
 [testenv]
 deps = mock
        coverage
+       Twisted
 commands = coverage erase
-           coverage run mpd/tests.py
+           coverage run -m unittest mpd.tests
            coverage report
            coverage html -d coverage_html/{envname}
 
@@ -13,7 +14,8 @@ commands = coverage erase
 deps = mock
        unittest2
        coverage
+       Twisted
 commands = coverage erase
-           coverage run mpd/tests.py
+           coverage run -m unittest mpd.tests
            coverage report
            coverage html -d coverage_html/{envname}

-- 
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