2 # -*- coding: utf-8 -*-
4 Test suite highly borrowed^Wsteal from python-mpd2 [0] project.
6 [0] https://github.com/Mic92/python-mpd2
20 import unittest.mock as mock
25 print("Please install mock from PyPI to run tests!")
28 # show deprecation warnings
29 warnings.simplefilter('default')
32 TEST_MPD_HOST, TEST_MPD_PORT = ('example.com', 10000)
35 class testEnvVar(unittest.TestCase):
37 def test_envvar(self):
38 os.environ.pop('MPD_HOST', None)
39 os.environ.pop('MPD_PORT', None)
40 client = musicpd.MPDClient()
41 self.assertEqual(client.host, 'localhost')
42 self.assertEqual(client.port, '6600')
44 os.environ['MPD_HOST'] = 'pa55w04d@example.org'
45 client = musicpd.MPDClient()
46 self.assertEqual(client.pwd, 'pa55w04d')
47 self.assertEqual(client.host, 'example.org')
48 self.assertEqual(client.port, '6600')
50 os.environ.pop('MPD_HOST', None)
51 os.environ['MPD_PORT'] = '6666'
52 client = musicpd.MPDClient()
53 self.assertEqual(client.pwd, None)
54 self.assertEqual(client.host, 'localhost')
55 self.assertEqual(client.port, '6666')
57 # Test unix socket fallback
58 os.environ.pop('MPD_HOST', None)
59 os.environ.pop('MPD_PORT', None)
60 os.environ.pop('XDG_RUNTIME_DIR', None)
61 with mock.patch('os.path.exists', return_value=True):
62 client = musicpd.MPDClient()
63 self.assertEqual(client.host, '/run/mpd/socket')
65 os.environ.pop('MPD_HOST', None)
66 os.environ.pop('MPD_PORT', None)
67 os.environ['XDG_RUNTIME_DIR'] = '/run/user/1000/'
68 with mock.patch('os.path.exists', return_value=True):
69 client = musicpd.MPDClient()
70 self.assertEqual(client.host, '/run/user/1000/mpd/socket')
72 class TestMPDClient(unittest.TestCase):
75 # last sync: musicpd 0.4.2 unreleased / Mon Nov 17 21:45:22 CET 2014
78 'clearerror': 'nothing',
79 'currentsong': 'object',
84 # Playback Option Commands
86 'crossfade': 'nothing',
87 'mixrampdb': 'nothing',
88 'mixrampdelay': 'nothing',
93 'replay_gain_mode': 'nothing',
94 'replay_gain_status': 'item',
96 # Playback Control Commands
101 'previous': 'nothing',
104 'seekcur': 'nothing',
111 'deleteid': 'nothing',
114 'playlist': 'playlist',
115 'playlistfind': 'songs',
116 'playlistid': 'songs',
117 'playlistinfo': 'songs',
118 'playlistsearch': 'songs',
119 'plchanges': 'songs',
120 'plchangesposid': 'changes',
123 'rangeid': 'nothing',
124 'shuffle': 'nothing',
127 'addtagid': 'nothing',
128 'cleartagid': 'nothing',
129 # Stored Playlist Commands
130 'listplaylist': 'list',
131 'listplaylistinfo': 'songs',
132 'listplaylists': 'playlists',
134 'playlistadd': 'nothing',
135 'playlistclear': 'nothing',
136 'playlistdelete': 'nothing',
137 'playlistmove': 'nothing',
144 'findadd': 'nothing',
146 'listall': 'database',
147 'listallinfo': 'database',
148 'lsinfo': 'database',
150 'searchadd': 'nothing',
151 'searchaddpl': 'nothing',
154 'readcomments': 'object',
155 # Mounts and neighbors
157 'unmount': 'nothing',
158 'listmounts': 'mounts',
159 'listneighbors': 'neighbors',
161 'sticker get': 'item',
162 'sticker set': 'nothing',
163 'sticker delete': 'nothing',
164 'sticker list': 'list',
165 'sticker find': 'songs',
166 # Connection Commands
169 'password': 'nothing',
172 'partition': 'nothing',
173 'listpartitions': 'list',
174 'newpartition': 'nothing',
175 # Audio Output Commands
176 'disableoutput': 'nothing',
177 'enableoutput': 'nothing',
178 'toggleoutput': 'nothing',
179 'outputs': 'outputs',
180 # Reflection Commands
183 'notcommands': 'list',
185 'urlhandlers': 'list',
186 'decoders': 'plugins',
188 'subscribe': 'nothing',
189 'unsubscribe': 'nothing',
191 'readmessages': 'messages',
192 'sendmessage': 'nothing',
196 self.socket_patch = mock.patch('musicpd.socket')
197 self.socket_mock = self.socket_patch.start()
198 self.socket_mock.getaddrinfo.return_value = [range(5)]
200 self.socket_mock.socket.side_effect = (
202 # Create a new socket.socket() mock with default attributes,
203 # each time we are calling it back (otherwise, it keeps set
204 # attributes across calls).
205 # That's probably what we want, since reconnecting is like
206 # reinitializing the entire connection, and so, the mock.
207 mock.MagicMock(name='socket.socket'))
209 self.client = musicpd.MPDClient()
210 self.client.connect(TEST_MPD_HOST, TEST_MPD_PORT)
211 self.client._sock.reset_mock()
212 self.MPDWillReturn('ACK don\'t forget to setup your mock\n')
215 self.socket_patch.stop()
217 def MPDWillReturn(self, *lines):
218 # Return what the caller wants first, then do as if the socket was
220 self.client._rfile.readline.side_effect = itertools.chain(
221 lines, itertools.repeat(''))
223 def assertMPDReceived(self, *lines):
224 self.client._wfile.write.assert_called_with(*lines)
226 def test_metaclass_commands(self):
227 """Controls client has at least commands as last synchronized in
228 TestMPDClient.commands"""
229 for cmd, ret in TestMPDClient.commands.items():
230 self.assertTrue(hasattr(self.client, cmd), msg='cmd "{}" not available!'.format(cmd))
232 self.assertTrue(hasattr(self.client, cmd.replace(' ', '_')))
234 def test_fetch_nothing(self):
235 self.MPDWillReturn('OK\n')
236 self.assertIsNone(self.client.ping())
237 self.assertMPDReceived('ping\n')
239 def test_fetch_list(self):
240 self.MPDWillReturn('OK\n')
241 self.assertIsInstance(self.client.list('album'), list)
242 self.assertMPDReceived('list "album"\n')
244 def test_fetch_item(self):
245 self.MPDWillReturn('updating_db: 42\n', 'OK\n')
246 self.assertIsNotNone(self.client.update())
248 def test_fetch_object(self):
249 # XXX: _read_objects() doesn't wait for the final OK
250 self.MPDWillReturn('volume: 63\n', 'OK\n')
251 status = self.client.status()
252 self.assertMPDReceived('status\n')
253 self.assertIsInstance(status, dict)
255 # XXX: _read_objects() doesn't wait for the final OK
256 self.MPDWillReturn('OK\n')
257 stats = self.client.stats()
258 self.assertMPDReceived('stats\n')
259 self.assertIsInstance(stats, dict)
261 def test_fetch_songs(self):
262 self.MPDWillReturn('file: my-song.ogg\n', 'Pos: 0\n', 'Id: 66\n', 'OK\n')
263 playlist = self.client.playlistinfo()
265 self.assertMPDReceived('playlistinfo\n')
266 self.assertIsInstance(playlist, list)
267 self.assertEqual(1, len(playlist))
269 self.assertIsInstance(e, dict)
270 self.assertEqual('my-song.ogg', e['file'])
271 self.assertEqual('0', e['pos'])
272 self.assertEqual('66', e['id'])
274 def test_send_and_fetch(self):
275 self.MPDWillReturn('volume: 50\n', 'OK\n')
276 result = self.client.send_status()
277 self.assertEqual(None, result)
278 self.assertMPDReceived('status\n')
280 status = self.client.fetch_status()
281 self.assertEqual(1, self.client._wfile.write.call_count)
282 self.assertEqual({'volume': '50'}, status)
284 def test_iterating(self):
285 self.MPDWillReturn('file: my-song.ogg\n', 'Pos: 0\n', 'Id: 66\n',
286 'file: my-song.ogg\n', 'Pos: 0\n', 'Id: 66\n', 'OK\n')
287 self.client.iterate = True
288 playlist = self.client.playlistinfo()
289 self.assertMPDReceived('playlistinfo\n')
290 self.assertIsInstance(playlist, types.GeneratorType)
291 self.assertTrue(self.client._iterating)
292 for song in playlist:
293 self.assertRaises(musicpd.IteratingError, self.client.status)
294 self.assertIsInstance(song, dict)
295 self.assertEqual('my-song.ogg', song['file'])
296 self.assertEqual('0', song['pos'])
297 self.assertEqual('66', song['id'])
298 self.assertFalse(self.client._iterating)
300 def test_noidle(self):
301 self.MPDWillReturn('OK\n') # nothing changed after idle-ing
302 self.client.send_idle()
303 self.MPDWillReturn('OK\n') # nothing changed after noidle
304 self.assertEqual(self.client.noidle(), [])
305 self.assertMPDReceived('noidle\n')
306 self.MPDWillReturn('volume: 50\n', 'OK\n')
308 self.assertMPDReceived('status\n')
310 def test_noidle_while_idle_started_sending(self):
311 self.MPDWillReturn('OK\n') # nothing changed after idle
312 self.client.send_idle()
313 self.MPDWillReturn('CHANGED: player\n', 'OK\n') # noidle response
314 self.assertEqual(self.client.noidle(), ['player'])
315 self.MPDWillReturn('volume: 50\n', 'OK\n')
316 status = self.client.status()
317 self.assertMPDReceived('status\n')
318 self.assertEqual({'volume': '50'}, status)
320 def test_throw_when_calling_noidle_withoutidling(self):
321 self.assertRaises(musicpd.CommandError, self.client.noidle)
322 self.client.send_status()
323 self.assertRaises(musicpd.CommandError, self.client.noidle)
325 def test_client_to_client(self):
326 # client to client is at this time in beta!
328 self.MPDWillReturn('OK\n')
329 self.assertIsNone(self.client.subscribe("monty"))
330 self.assertMPDReceived('subscribe "monty"\n')
332 self.MPDWillReturn('channel: monty\n', 'OK\n')
333 channels = self.client.channels()
334 self.assertMPDReceived('channels\n')
335 self.assertEqual(['monty'], channels)
337 self.MPDWillReturn('OK\n')
338 self.assertIsNone(self.client.sendmessage('monty', 'SPAM'))
339 self.assertMPDReceived('sendmessage "monty" "SPAM"\n')
341 self.MPDWillReturn('channel: monty\n', 'message: SPAM\n', 'OK\n')
342 msg = self.client.readmessages()
343 self.assertMPDReceived('readmessages\n')
344 self.assertEqual(msg, [{'channel': 'monty', 'message': 'SPAM'}])
346 self.MPDWillReturn('OK\n')
347 self.assertIsNone(self.client.unsubscribe('monty'))
348 self.assertMPDReceived('unsubscribe "monty"\n')
350 self.MPDWillReturn('OK\n')
351 channels = self.client.channels()
352 self.assertMPDReceived('channels\n')
353 self.assertEqual([], channels)
355 def test_ranges_in_command_args(self):
356 self.MPDWillReturn('OK\n')
357 self.client.playlistinfo((10,))
358 self.assertMPDReceived('playlistinfo 10:\n')
360 self.MPDWillReturn('OK\n')
361 self.client.playlistinfo(('10',))
362 self.assertMPDReceived('playlistinfo 10:\n')
364 self.MPDWillReturn('OK\n')
365 self.client.playlistinfo((10, 12))
366 self.assertMPDReceived('playlistinfo 10:12\n')
368 self.MPDWillReturn('OK\n')
369 self.client.rangeid(())
370 self.assertMPDReceived('rangeid :\n')
372 for arg in [(10, 't'), (10, 1, 1), (None,1)]:
373 self.MPDWillReturn('OK\n')
374 with self.assertRaises(musicpd.CommandError):
375 self.client.playlistinfo(arg)
377 def test_numbers_as_command_args(self):
378 self.MPDWillReturn('OK\n')
379 self.client.find('file', 1)
380 self.assertMPDReceived('find "file" "1"\n')
382 def test_commands_without_callbacks(self):
383 self.MPDWillReturn('\n')
385 self.assertMPDReceived('close\n')
387 # XXX: what are we testing here?
389 self.client.connect(TEST_MPD_HOST, TEST_MPD_PORT)
391 def test_connection_lost(self):
392 # Simulate a connection lost: the socket returns empty strings
393 self.MPDWillReturn('')
395 with self.assertRaises(musicpd.ConnectionError):
398 # consistent behaviour
399 # solves <https://github.com/Mic92/python-mpd2/issues/11> (also present
401 with self.assertRaises(musicpd.ConnectionError):
404 self.assertIs(self.client._sock, None)
406 def test_parse_sticker_get_no_sticker(self):
407 self.MPDWillReturn('ACK [50@0] {sticker} no such sticker\n')
408 self.assertRaises(musicpd.CommandError,
409 self.client.sticker_get, 'song', 'baz', 'foo')
411 def test_parse_sticker_list(self):
412 self.MPDWillReturn('sticker: foo=bar\n', 'sticker: lom=bok\n', 'OK\n')
413 res = self.client.sticker_list('song', 'baz')
414 self.assertEqual(['foo=bar', 'lom=bok'], res)
416 # Even with only one sticker, we get a dict
417 self.MPDWillReturn('sticker: foo=bar\n', 'OK\n')
418 res = self.client.sticker_list('song', 'baz')
419 self.assertEqual(['foo=bar'], res)
421 def test_albumart(self):
422 data = bytes('\xff\xd8\xff\xe0\x00\x10JFIF\x00\x01\x01\x00\x00\x01'
423 '\x00\x01\x00\x00\xff\xdb\x00C\x00\x05\x03\x04',
425 data_str = data.decode(encoding='utf-8', errors='surrogateescape')
426 self.MPDWillReturn('size: 36474\n', 'binary: 8192\n',
427 data_str+'\n', 'OK\n')
428 res = self.client.albumart('muse/Raised Fist/2002-Dedication/', 0)
429 self.assertEqual(res.get('data'), data)
431 if __name__ == '__main__':