2 # -*- coding: utf-8 -*-
4 Test suite highly borrowed^Wsteal from python-mpd2 [0] project.
6 [0] https://github.com/Mic92/python-mpd2
19 import unittest.mock as mock
24 print("Please install mock from PyPI to run tests!")
27 # show deprecation warnings
28 warnings.simplefilter('default')
31 TEST_MPD_HOST, TEST_MPD_PORT = ('example.com', 10000)
34 class TestMPDClient(unittest.TestCase):
37 # last sync: musicpd 0.4.0
40 "clearerror": "nothing",
41 "currentsong": "object",
46 # Playback Option Commands
48 "crossfade": "nothing",
49 "mixrampdb": "nothing",
50 "mixrampdelay": "nothing",
55 "replay_gain_mode": "nothing",
56 "replay_gain_status": "item",
58 # Playback Control Commands
63 "previous": "nothing",
73 "deleteid": "nothing",
76 "playlist": "playlist",
77 "playlistfind": "songs",
78 "playlistid": "songs",
79 "playlistinfo": "songs",
80 "playlistsearch": "songs",
82 "plchangesposid": "changes",
86 # Stored Playlist Commands
87 "listplaylist": "list",
88 "listplaylistinfo": "songs",
89 "listplaylists": "playlists",
91 "playlistadd": "nothing",
92 "playlistclear": "nothing",
93 "playlistdelete": "nothing",
94 "playlistmove": "nothing",
101 "findadd": "nothing",
103 "listall": "database",
104 "listallinfo": "database",
105 "lsinfo": "database",
107 "searchadd": "nothing",
108 "searchaddpl": "nothing",
111 "readcomments": "object",
113 "sticker get": "item",
114 "sticker set": "nothing",
115 "sticker delete": "nothing",
116 "sticker list": "list",
117 "sticker find": "songs",
118 # Connection Commands
121 "password": "nothing",
123 # Audio Output Commands
124 "disableoutput": "nothing",
125 "enableoutput": "nothing",
126 "toggleoutput": "nothing",
127 "outputs": "outputs",
128 # Reflection Commands
130 "notcommands": "list",
132 "urlhandlers": "list",
133 "decoders": "plugins",
135 "subscribe": "nothing",
136 "unsubscribe": "nothing",
138 "readmessages": "messages",
139 "sendmessage": "nothing",
143 self.socket_patch = mock.patch("musicpd.socket")
144 self.socket_mock = self.socket_patch.start()
145 self.socket_mock.getaddrinfo.return_value = [range(5)]
147 self.socket_mock.socket.side_effect = (
149 # Create a new socket.socket() mock with default attributes,
150 # each time we are calling it back (otherwise, it keeps set
151 # attributes across calls).
152 # That's probably what we want, since reconnecting is like
153 # reinitializing the entire connection, and so, the mock.
154 mock.MagicMock(name="socket.socket"))
156 self.client = musicpd.MPDClient()
157 self.client.connect(TEST_MPD_HOST, TEST_MPD_PORT)
158 self.client._sock.reset_mock()
159 self.MPDWillReturn("ACK don't forget to setup your mock\n")
162 self.socket_patch.stop()
164 def MPDWillReturn(self, *lines):
165 # Return what the caller wants first, then do as if the socket was
167 self.client._rfile.readline.side_effect = itertools.chain(
168 lines, itertools.repeat(''))
170 def assertMPDReceived(self, *lines):
171 self.client._wfile.write.assert_called_with(*lines)
173 def test_metaclass_commands(self):
174 for cmd, ret in TestMPDClient.commands.items():
175 self.assertTrue(hasattr(self.client, cmd), msg='fails for{}'.format(cmd))
177 self.assertTrue(hasattr(self.client, cmd.replace(' ', '_')))
179 def test_fetch_nothing(self):
180 self.MPDWillReturn('OK\n')
181 self.assertIsNone(self.client.ping())
182 self.assertMPDReceived('ping\n')
184 def test_fetch_list(self):
185 self.MPDWillReturn('OK\n')
186 self.assertIsInstance(self.client.list('album'), list)
187 self.assertMPDReceived('list "album"\n')
189 def test_fetch_item(self):
190 self.MPDWillReturn('updating_db: 42\n', 'OK\n')
191 self.assertIsNotNone(self.client.update())
193 def test_fetch_object(self):
194 # XXX: _read_objects() doesn't wait for the final OK
195 self.MPDWillReturn('volume: 63\n', 'OK\n')
196 status = self.client.status()
197 self.assertMPDReceived('status\n')
198 self.assertIsInstance(status, dict)
200 # XXX: _read_objects() doesn't wait for the final OK
201 self.MPDWillReturn('OK\n')
202 stats = self.client.stats()
203 self.assertMPDReceived('stats\n')
204 self.assertIsInstance(stats, dict)
206 def test_fetch_songs(self):
207 self.MPDWillReturn("file: my-song.ogg\n", "Pos: 0\n", "Id: 66\n", "OK\n")
208 playlist = self.client.playlistinfo()
210 self.assertMPDReceived('playlistinfo\n')
211 self.assertIsInstance(playlist, list)
212 self.assertEqual(1, len(playlist))
214 self.assertIsInstance(e, dict)
215 self.assertEqual('my-song.ogg', e['file'])
216 self.assertEqual('0', e['pos'])
217 self.assertEqual('66', e['id'])
219 def test_send_and_fetch(self):
220 self.MPDWillReturn("volume: 50\n", "OK\n")
221 result = self.client.send_status()
222 self.assertEqual(None, result)
223 self.assertMPDReceived('status\n')
225 status = self.client.fetch_status()
226 self.assertEqual(1, self.client._wfile.write.call_count)
227 self.assertEqual({'volume': '50'}, status)
229 def test_iterating(self):
230 self.MPDWillReturn("file: my-song.ogg\n", "Pos: 0\n", "Id: 66\n", "OK\n")
231 self.client.iterate = True
232 playlist = self.client.playlistinfo()
233 self.assertMPDReceived('playlistinfo\n')
234 self.assertIsInstance(playlist, types.GeneratorType)
235 for song in playlist:
236 self.assertIsInstance(song, dict)
237 self.assertEqual('my-song.ogg', song['file'])
238 self.assertEqual('0', song['pos'])
239 self.assertEqual('66', song['id'])
241 def test_noidle(self):
242 self.MPDWillReturn('OK\n') # nothing changed after idle-ing
243 self.client.send_idle()
244 self.MPDWillReturn('OK\n') # nothing changed after noidle
245 self.assertEqual(self.client.noidle(), [])
246 self.assertMPDReceived('noidle\n')
247 self.MPDWillReturn("volume: 50\n", "OK\n")
249 self.assertMPDReceived('status\n')
251 def test_noidle_while_idle_started_sending(self):
252 self.MPDWillReturn('OK\n') # nothing changed after idle
253 self.client.send_idle()
254 self.MPDWillReturn('CHANGED: player\n', 'OK\n') # noidle response
255 self.assertEqual(self.client.noidle(), ['player'])
256 self.MPDWillReturn("volume: 50\n", "OK\n")
257 status = self.client.status()
258 self.assertMPDReceived('status\n')
259 self.assertEqual({'volume': '50'}, status)
261 def test_throw_when_calling_noidle_withoutidling(self):
262 self.assertRaises(musicpd.CommandError, self.client.noidle)
263 self.client.send_status()
264 self.assertRaises(musicpd.CommandError, self.client.noidle)
266 def test_client_to_client(self):
267 # client to client is at this time in beta!
269 self.MPDWillReturn('OK\n')
270 self.assertIsNone(self.client.subscribe("monty"))
271 self.assertMPDReceived('subscribe "monty"\n')
273 self.MPDWillReturn('channel: monty\n', 'OK\n')
274 channels = self.client.channels()
275 self.assertMPDReceived('channels\n')
276 self.assertEqual(["monty"], channels)
278 self.MPDWillReturn('OK\n')
279 self.assertIsNone(self.client.sendmessage("monty", "SPAM"))
280 self.assertMPDReceived('sendmessage "monty" "SPAM"\n')
282 self.MPDWillReturn('channel: monty\n', 'message: SPAM\n', 'OK\n')
283 msg = self.client.readmessages()
284 self.assertMPDReceived('readmessages\n')
285 self.assertEqual(msg, [{"channel":"monty", "message": "SPAM"}])
287 self.MPDWillReturn('OK\n')
288 self.assertIsNone(self.client.unsubscribe("monty"))
289 self.assertMPDReceived('unsubscribe "monty"\n')
291 self.MPDWillReturn('OK\n')
292 channels = self.client.channels()
293 self.assertMPDReceived('channels\n')
294 self.assertEqual([], channels)
296 def test_ranges_in_command_args(self):
297 self.MPDWillReturn("OK\n")
298 self.client.playlistinfo((10,))
299 self.assertMPDReceived('playlistinfo 10:\n')
301 self.MPDWillReturn("OK\n")
302 self.client.playlistinfo(("10",))
303 self.assertMPDReceived('playlistinfo 10:\n')
305 self.MPDWillReturn("OK\n")
306 self.client.playlistinfo((10, 12))
307 self.assertMPDReceived('playlistinfo 10:12\n')
309 for arg in [(10, "t"), (10, 1, 1)]:
310 self.MPDWillReturn("OK\n")
311 with self.assertRaises(musicpd.CommandError):
312 self.client.playlistinfo(arg)
314 def test_numbers_as_command_args(self):
315 self.MPDWillReturn("OK\n")
316 self.client.find("file", 1)
317 self.assertMPDReceived('find "file" "1"\n')
319 def test_commands_without_callbacks(self):
320 self.MPDWillReturn("\n")
322 self.assertMPDReceived('close\n')
324 # XXX: what are we testing here?
326 self.client.connect(TEST_MPD_HOST, TEST_MPD_PORT)
328 def test_connection_lost(self):
329 # Simulate a connection lost: the socket returns empty strings
330 self.MPDWillReturn('')
332 with self.assertRaises(musicpd.ConnectionError):
335 # consistent behaviour
336 # solves <https://github.com/Mic92/python-mpd2/issues/11> (also present
338 with self.assertRaises(musicpd.ConnectionError):
341 self.assertIs(self.client._sock, None)
343 def test_parse_sticker_get_no_sticker(self):
344 self.MPDWillReturn("ACK [50@0] {sticker} no such sticker\n")
345 self.assertRaises(musicpd.CommandError,
346 self.client.sticker_get, 'song', 'baz', 'foo')
348 def test_parse_sticker_list(self):
349 self.MPDWillReturn("sticker: foo=bar\n", "sticker: lom=bok\n", "OK\n")
350 res = self.client.sticker_list('song', 'baz')
351 self.assertEqual(['foo=bar', 'lom=bok'], res)
353 # Even with only one sticker, we get a dict
354 self.MPDWillReturn("sticker: foo=bar\n", "OK\n")
355 res = self.client.sticker_list('song', 'baz')
356 self.assertEqual(['foo=bar'], res)
359 if __name__ == '__main__':