[Pkg-mpd-commits] [python-mpd] 159/262: Implement tests which don't rely on a running MPD server
Simon McVittie
smcv at debian.org
Sun May 22 18:16:41 UTC 2016
This is an automated email from the git hooks/post-receive script.
smcv pushed a commit to branch upstream
in repository python-mpd.
commit adaf615a164a36f4cb0957e4d5951c73d2474af3
Author: Jonathan Ballet <jon at multani.info>
Date: Wed Dec 12 17:21:19 2012 +0800
Implement tests which don't rely on a running MPD server
They use instead a mock-ed socket to "communicate" with the rest of the
tests which allow to control and test precisely what we want.
No tests communicate with a real MPD server anymore. If needed, we can still
get a way to add "integration" tests, or to allow some of those unittests
to communicate with a server (so-called "safe" tests).
Removed the tests which compare the commands allowed by the server and
those implemented by the library, I don't think this is the goal of the
unittest to do this, and should be done somewhere else (a separate,
dedicated script?)
---
mpd.py | 2 +-
test.py | 284 +++++++++++++++++++++++++++++++++++++++-------------------------
2 files changed, 173 insertions(+), 113 deletions(-)
diff --git a/mpd.py b/mpd.py
index ecd004b..3a43ef2 100644
--- a/mpd.py
+++ b/mpd.py
@@ -513,7 +513,7 @@ class MPDClient(object):
send_method = newFunction(cls._send, key, callback)
fetch_method = newFunction(cls._fetch, key, callback)
- # create new mpd commands as function in the tree flavors:
+ # 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)
diff --git a/test.py b/test.py
index 4e3b088..7810971 100755
--- a/test.py
+++ b/test.py
@@ -1,13 +1,15 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
+import itertools
import os
-import types
-import sys
from socket import error as SocketError
-import mpd
+import sys
+import types
import warnings
+import mpd
+
try:
# is required for python2.6
# python2.7 works with this module too
@@ -18,55 +20,56 @@ except ImportError:
if sys.version_info >= (2, 7):
import unittest
else:
- print("Please install unittest2 from pypi to run tests!")
+ print("Please install unittest2 from PyPI to run tests!")
sys.exit(1)
+try:
+ import mock
+except ImportError:
+ print("Please install mock from PyPI to run tests!")
+ sys.exit(1)
+
# show deprecation warnings
warnings.simplefilter('default')
-def setup_environment():
- # Alternate this to your setup
- # Make sure you have at least one song on your playlist
- global TEST_MPD_HOST, TEST_MPD_PORT, TEST_MPD_PASSWORD
-
- if 'TEST_MPD_PORT' not in os.environ:
- sys.stderr.write(
- "You should set the TEST_MPD_PORT environment variable to point "
- "to your test MPD running instance.\n")
- sys.exit(255)
-
- TEST_MPD_HOST = os.environ.get('TEST_MPD_HOST', "localhost")
- TEST_MPD_PORT = int(os.environ['TEST_MPD_PORT'])
- TEST_MPD_PASSWORD = os.environ.get('TEST_MPD_PASSWORD')
-
-setup_environment()
-
-def createMpdClient():
- global TEST_MPD_HOST, TEST_MPD_PORT, TEST_MPD_PASSWORD
- client = mpd.MPDClient()
- try:
- client.connect(TEST_MPD_HOST, TEST_MPD_PORT)
- commands = client.commands()
- except SocketError as e:
- raise Exception("Can't connect mpd! Start it or check the configuration: %s" % e)
- if TEST_MPD_PASSWORD != None:
- try:
- client.password(TEST_MPD_PASSWORD)
- except mpd.CommandError as e:
- raise Exception("Fail to authenticate to mpd.")
- return client
+
+TEST_MPD_HOST, TEST_MPD_PORT = ('example.com', 10000)
+
class TestMPDClient(unittest.TestCase):
longMessage = True
- @classmethod
- def setUpClass(self):
- self.client = createMpdClient()
+ def setUp(self):
+ self.socket_patch = mock.patch("mpd.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")
- @classmethod
- def tearDownClass(self):
- self.client.disconnect()
+ 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_metaclass_commands(self):
# just some random functions
@@ -82,164 +85,221 @@ class TestMPDClient(unittest.TestCase):
self.assertTrue(hasattr(self.client, "send_close"))
def test_fetch_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_fetch_list(self):
+ self.MPDWillReturn('OK\n')
+
self.assertIsInstance(self.client.list("album"), list)
+ self.assertMPDReceived('list "album"\n')
def test_fetch_item(self):
+ self.MPDWillReturn('updating_db: 42\n', 'OK\n')
self.assertIsNotNone(self.client.update())
def test_fetch_object(self):
+ # XXX: _read_objects() doesn't wait for the final OK
+ self.MPDWillReturn('volume: 63\n', 'OK\n')
status = self.client.status()
- stats = self.client.stats()
+ self.assertMPDReceived('status\n')
self.assertIsInstance(status, dict)
- # some keys should be there
- self.assertIn("volume", status)
- self.assertIn("song", status)
+
+ # 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)
- self.assertIn("artists", stats)
- self.assertIn("uptime", stats)
def test_fetch_songs(self):
+ self.MPDWillReturn("file: my-song.ogg\n", "Pos: 0\n", "Id: 66\n", "OK\n")
playlist = self.client.playlistinfo()
- self.assertTrue(type(playlist) is list)
- if len(playlist) > 0:
- self.assertIsInstance(playlist[0], dict)
+
+ 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.client.send_status()
- self.client.fetch_status()
+ 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_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.client.iterate = False
+ 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):
- idleclient = createMpdClient()
- # clean event mask
- idleclient.idle()
- idleclient.send_idle()
+ 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.client.update()
- event = idleclient.fetch_idle()
+ self.MPDWillReturn('changed: update\n', 'OK\n')
+
+ event = self.client.fetch_idle()
self.assertEqual(event, ['update'])
def test_add_and_remove_command(self):
+ self.MPDWillReturn("ACK awesome command\n")
+
self.client.add_command("awesome command", mpd.MPDClient._fetch_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
- with self.assertRaises(mpd.CommandError):
- self.client.awesome_command()
+ 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!
- if not "channels" in self.client.commands():
- return
+
+ 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.assertIn("monty", 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.assertNotIn("monty", channels)
-
- def test_commands_list(self):
- """
- Test if all implemented commands are valid
- and all avaible commands are implemented.
- This test may fail, if a implement command isn't
- avaible on older versions of mpd
- """
- avaible_cmds = set(self.client.commands() + self.client.notcommands())
- imple_cmds = set(mpd._commands.keys())
- sticker_cmds = set(["sticker get", "sticker set", "sticker delete",
- "sticker list", "sticker find"])
- imple_cmds = (imple_cmds - sticker_cmds)
- imple_cmds.add("sticker")
- imple_cmds.remove("noidle")
-
- self.assertEqual(set(), avaible_cmds - imple_cmds,
- "Not all commands supported by mpd are implemented!")
-
- long_desc = (
- "Not all commands implemented by this library are supported by "
- "the current mpd.\n" +
- "This either means the command list is wrong or mpd is not "
- "up-to-date.")
-
- self.assertEqual(set(), imple_cmds - avaible_cmds, long_desc)
+ self.assertMPDReceived('channels\n')
+ self.assertEqual([], channels)
def test_unicode_as_command_args(self):
if sys.version_info < (3, 0):
- raw_unicode = "☯☾☝♖✽".decode("utf-8")
- res = self.client.find("file", raw_unicode)
+ self.MPDWillReturn("OK\n")
+ res = self.client.find("file", unicode("☯☾☝♖✽", 'utf-8'))
self.assertIsInstance(res, list)
+ self.assertMPDReceived('find "file" "☯☾☝♖✽"\n')
- encoded_str = "☯☾☝♖✽"
- res2 = self.client.find("file", encoded_str)
+ 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):
- res = self.client.find("file", 1)
+ 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?
self.client._reset()
self.client.connect(TEST_MPD_HOST, TEST_MPD_PORT)
- def test_timeout(self):
- self.client.disconnect()
+ 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)
- with warnings.catch_warnings(record=True) as w:
- self.client.connect(TEST_MPD_HOST, TEST_MPD_PORT, timeout=5)
- self.assertEqual(self.client._sock.gettimeout(), 5)
- self.assertEqual(len(w), 1)
self.client.timeout = None
- self.assertEqual(self.client._sock.gettimeout(), 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))
def test_connection_lost(self):
- client = mpd.MPDClient()
- client.connect(TEST_MPD_HOST, TEST_MPD_PORT)
- # simulate a disconnect
- client._sock.send(b"close\n")
- client._sock.recv(4096)
+ # Simulate a connection lost: the socket returns empty strings
+ self.MPDWillReturn('')
+
with self.assertRaises(mpd.ConnectionError):
- client.status()
+ self.client.status()
+
# consistent behaviour, solves bug #11 (github)
with self.assertRaises(mpd.ConnectionError):
- client.status()
- self.assertIs(client._sock, None)
+ self.client.status()
+
+ 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"),
+ mock.call('w', encoding="utf-8")],
+ # 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])
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