2 # -*- coding: utf-8 -*-
3 # pylint: disable=missing-docstring
5 Test suite highly borrowed^Wsteal from python-mpd2 [0] project.
7 [0] https://github.com/Mic92/python-mpd2
21 import unittest.mock as mock
26 print("Please install mock from PyPI to run tests!")
29 # show deprecation warnings
30 warnings.simplefilter('default')
33 TEST_MPD_HOST, TEST_MPD_PORT = ('example.com', 10000)
36 class testEnvVar(unittest.TestCase):
38 def test_envvar(self):
39 os.environ.pop('MPD_HOST', None)
40 os.environ.pop('MPD_PORT', None)
41 client = musicpd.MPDClient()
42 self.assertEqual(client.host, 'localhost')
43 self.assertEqual(client.port, '6600')
45 os.environ['MPD_HOST'] = 'pa55w04d@example.org'
46 client = musicpd.MPDClient()
47 self.assertEqual(client.pwd, 'pa55w04d')
48 self.assertEqual(client.host, 'example.org')
49 self.assertEqual(client.port, '6600')
51 os.environ.pop('MPD_HOST', None)
52 os.environ['MPD_PORT'] = '6666'
53 client = musicpd.MPDClient()
54 self.assertEqual(client.pwd, None)
55 self.assertEqual(client.host, 'localhost')
56 self.assertEqual(client.port, '6666')
58 # Test unix socket fallback
59 os.environ.pop('MPD_HOST', None)
60 os.environ.pop('MPD_PORT', None)
61 os.environ.pop('XDG_RUNTIME_DIR', None)
62 with mock.patch('os.path.exists', return_value=True):
63 client = musicpd.MPDClient()
64 self.assertEqual(client.host, '/run/mpd/socket')
66 os.environ.pop('MPD_HOST', None)
67 os.environ.pop('MPD_PORT', None)
68 os.environ['XDG_RUNTIME_DIR'] = '/run/user/1000/'
69 with mock.patch('os.path.exists', return_value=True):
70 client = musicpd.MPDClient()
71 self.assertEqual(client.host, '/run/user/1000/mpd/socket')
73 os.environ.pop('MPD_TIMEOUT', None)
74 client = musicpd.MPDClient()
75 self.assertEqual(client.mpd_timeout, 30)
76 os.environ['MPD_TIMEOUT'] = 'garbage'
77 client = musicpd.MPDClient()
78 self.assertEqual(client.mpd_timeout, 30)
79 os.environ['MPD_TIMEOUT'] = '42'
80 client = musicpd.MPDClient()
81 self.assertEqual(client.mpd_timeout, 42)
84 class TestMPDClient(unittest.TestCase):
87 # last sync: musicpd 0.4.2 unreleased / Mon Nov 17 21:45:22 CET 2014
90 'clearerror': 'nothing',
91 'currentsong': 'object',
96 # Playback Option Commands
98 'crossfade': 'nothing',
99 'mixrampdb': 'nothing',
100 'mixrampdelay': 'nothing',
105 'replay_gain_mode': 'nothing',
106 'replay_gain_status': 'item',
108 # Playback Control Commands
113 'previous': 'nothing',
116 'seekcur': 'nothing',
123 'deleteid': 'nothing',
126 'playlist': 'playlist',
127 'playlistfind': 'songs',
128 'playlistid': 'songs',
129 'playlistinfo': 'songs',
130 'playlistsearch': 'songs',
131 'plchanges': 'songs',
132 'plchangesposid': 'changes',
135 'rangeid': 'nothing',
136 'shuffle': 'nothing',
139 'addtagid': 'nothing',
140 'cleartagid': 'nothing',
141 # Stored Playlist Commands
142 'listplaylist': 'list',
143 'listplaylistinfo': 'songs',
144 'listplaylists': 'playlists',
146 'playlistadd': 'nothing',
147 'playlistclear': 'nothing',
148 'playlistdelete': 'nothing',
149 'playlistmove': 'nothing',
156 'findadd': 'nothing',
158 'listall': 'database',
159 'listallinfo': 'database',
160 'lsinfo': 'database',
162 'searchadd': 'nothing',
163 'searchaddpl': 'nothing',
166 'readcomments': 'object',
167 # Mounts and neighbors
169 'unmount': 'nothing',
170 'listmounts': 'mounts',
171 'listneighbors': 'neighbors',
173 'sticker get': 'item',
174 'sticker set': 'nothing',
175 'sticker delete': 'nothing',
176 'sticker list': 'list',
177 'sticker find': 'songs',
178 # Connection Commands
181 'password': 'nothing',
184 'partition': 'nothing',
185 'listpartitions': 'list',
186 'newpartition': 'nothing',
187 # Audio Output Commands
188 'disableoutput': 'nothing',
189 'enableoutput': 'nothing',
190 'toggleoutput': 'nothing',
191 'outputs': 'outputs',
192 # Reflection Commands
195 'notcommands': 'list',
197 'urlhandlers': 'list',
198 'decoders': 'plugins',
200 'subscribe': 'nothing',
201 'unsubscribe': 'nothing',
203 'readmessages': 'messages',
204 'sendmessage': 'nothing',
208 self.socket_patch = mock.patch('musicpd.socket')
209 self.socket_mock = self.socket_patch.start()
210 self.socket_mock.getaddrinfo.return_value = [range(5)]
212 self.socket_mock.socket.side_effect = (
214 # Create a new socket.socket() mock with default attributes,
215 # each time we are calling it back (otherwise, it keeps set
216 # attributes across calls).
217 # That's probably what we want, since reconnecting is like
218 # reinitializing the entire connection, and so, the mock.
219 mock.MagicMock(name='socket.socket'))
221 self.client = musicpd.MPDClient()
222 self.client.connect(TEST_MPD_HOST, TEST_MPD_PORT)
223 self.client._sock.reset_mock()
224 self.MPDWillReturn('ACK don\'t forget to setup your mock\n')
227 self.socket_patch.stop()
229 def MPDWillReturn(self, *lines):
230 # Return what the caller wants first, then do as if the socket was
232 self.client._rfile.readline.side_effect = itertools.chain(
233 lines, itertools.repeat(''))
235 def MPDWillReturnBinary(self, lines):
236 data = bytearray(b''.join(lines))
244 val.append(data.pop(0))
249 while not val.endswith(b'\x0a'):
250 val.append(data.pop(0))
252 self.client._rbfile.readline.side_effect = readline
253 self.client._rbfile.read.side_effect = read
255 def assertMPDReceived(self, *lines):
256 self.client._wfile.write.assert_called_with(*lines)
258 def test_metaclass_commands(self):
259 """Controls client has at least commands as last synchronized in
260 TestMPDClient.commands"""
261 for cmd, ret in TestMPDClient.commands.items():
262 self.assertTrue(hasattr(self.client, cmd), msg='cmd "{}" not available!'.format(cmd))
264 self.assertTrue(hasattr(self.client, cmd.replace(' ', '_')))
266 def test_fetch_nothing(self):
267 self.MPDWillReturn('OK\n')
268 self.assertIsNone(self.client.ping())
269 self.assertMPDReceived('ping\n')
271 def test_fetch_list(self):
272 self.MPDWillReturn('OK\n')
273 self.assertIsInstance(self.client.list('album'), list)
274 self.assertMPDReceived('list "album"\n')
276 def test_fetch_item(self):
277 self.MPDWillReturn('updating_db: 42\n', 'OK\n')
278 self.assertIsNotNone(self.client.update())
280 def test_fetch_object(self):
281 # XXX: _read_objects() doesn't wait for the final OK
282 self.MPDWillReturn('volume: 63\n', 'OK\n')
283 status = self.client.status()
284 self.assertMPDReceived('status\n')
285 self.assertIsInstance(status, dict)
287 # XXX: _read_objects() doesn't wait for the final OK
288 self.MPDWillReturn('OK\n')
289 stats = self.client.stats()
290 self.assertMPDReceived('stats\n')
291 self.assertIsInstance(stats, dict)
293 def test_fetch_songs(self):
294 self.MPDWillReturn('file: my-song.ogg\n', 'Pos: 0\n', 'Id: 66\n', 'OK\n')
295 playlist = self.client.playlistinfo()
297 self.assertMPDReceived('playlistinfo\n')
298 self.assertIsInstance(playlist, list)
299 self.assertEqual(1, len(playlist))
301 self.assertIsInstance(e, dict)
302 self.assertEqual('my-song.ogg', e['file'])
303 self.assertEqual('0', e['pos'])
304 self.assertEqual('66', e['id'])
306 def test_send_and_fetch(self):
307 self.MPDWillReturn('volume: 50\n', 'OK\n')
308 result = self.client.send_status()
309 self.assertEqual(None, result)
310 self.assertMPDReceived('status\n')
312 status = self.client.fetch_status()
313 self.assertEqual(1, self.client._wfile.write.call_count)
314 self.assertEqual({'volume': '50'}, status)
316 def test_iterating(self):
317 self.MPDWillReturn('file: my-song.ogg\n', 'Pos: 0\n', 'Id: 66\n',
318 'file: my-song.ogg\n', 'Pos: 0\n', 'Id: 66\n', 'OK\n')
319 self.client.iterate = True
320 playlist = self.client.playlistinfo()
321 self.assertMPDReceived('playlistinfo\n')
322 self.assertIsInstance(playlist, types.GeneratorType)
323 self.assertTrue(self.client._iterating)
324 for song in playlist:
325 self.assertRaises(musicpd.IteratingError, self.client.status)
326 self.assertIsInstance(song, dict)
327 self.assertEqual('my-song.ogg', song['file'])
328 self.assertEqual('0', song['pos'])
329 self.assertEqual('66', song['id'])
330 self.assertFalse(self.client._iterating)
332 def test_noidle(self):
333 self.MPDWillReturn('OK\n') # nothing changed after idle-ing
334 self.client.send_idle()
335 self.MPDWillReturn('OK\n') # nothing changed after noidle
336 self.assertEqual(self.client.noidle(), [])
337 self.assertMPDReceived('noidle\n')
338 self.MPDWillReturn('volume: 50\n', 'OK\n')
340 self.assertMPDReceived('status\n')
342 def test_noidle_while_idle_started_sending(self):
343 self.MPDWillReturn('OK\n') # nothing changed after idle
344 self.client.send_idle()
345 self.MPDWillReturn('CHANGED: player\n', 'OK\n') # noidle response
346 self.assertEqual(self.client.noidle(), ['player'])
347 self.MPDWillReturn('volume: 50\n', 'OK\n')
348 status = self.client.status()
349 self.assertMPDReceived('status\n')
350 self.assertEqual({'volume': '50'}, status)
352 def test_throw_when_calling_noidle_withoutidling(self):
353 self.assertRaises(musicpd.CommandError, self.client.noidle)
354 self.client.send_status()
355 self.assertRaises(musicpd.CommandError, self.client.noidle)
357 def test_client_to_client(self):
358 # client to client is at this time in beta!
360 self.MPDWillReturn('OK\n')
361 self.assertIsNone(self.client.subscribe("monty"))
362 self.assertMPDReceived('subscribe "monty"\n')
364 self.MPDWillReturn('channel: monty\n', 'OK\n')
365 channels = self.client.channels()
366 self.assertMPDReceived('channels\n')
367 self.assertEqual(['monty'], channels)
369 self.MPDWillReturn('OK\n')
370 self.assertIsNone(self.client.sendmessage('monty', 'SPAM'))
371 self.assertMPDReceived('sendmessage "monty" "SPAM"\n')
373 self.MPDWillReturn('channel: monty\n', 'message: SPAM\n', 'OK\n')
374 msg = self.client.readmessages()
375 self.assertMPDReceived('readmessages\n')
376 self.assertEqual(msg, [{'channel': 'monty', 'message': 'SPAM'}])
378 self.MPDWillReturn('OK\n')
379 self.assertIsNone(self.client.unsubscribe('monty'))
380 self.assertMPDReceived('unsubscribe "monty"\n')
382 self.MPDWillReturn('OK\n')
383 channels = self.client.channels()
384 self.assertMPDReceived('channels\n')
385 self.assertEqual([], channels)
387 def test_ranges_in_command_args(self):
388 self.MPDWillReturn('OK\n')
389 self.client.playlistinfo((10,))
390 self.assertMPDReceived('playlistinfo 10:\n')
392 self.MPDWillReturn('OK\n')
393 self.client.playlistinfo(('10',))
394 self.assertMPDReceived('playlistinfo 10:\n')
396 self.MPDWillReturn('OK\n')
397 self.client.playlistinfo((10, 12))
398 self.assertMPDReceived('playlistinfo 10:12\n')
400 self.MPDWillReturn('OK\n')
401 self.client.rangeid(())
402 self.assertMPDReceived('rangeid :\n')
404 for arg in [(10, 't'), (10, 1, 1), (None,1)]:
405 self.MPDWillReturn('OK\n')
406 with self.assertRaises(musicpd.CommandError):
407 self.client.playlistinfo(arg)
409 def test_numbers_as_command_args(self):
410 self.MPDWillReturn('OK\n')
411 self.client.find('file', 1)
412 self.assertMPDReceived('find "file" "1"\n')
414 def test_commands_without_callbacks(self):
415 self.MPDWillReturn('\n')
417 self.assertMPDReceived('close\n')
419 # XXX: what are we testing here?
421 self.client.connect(TEST_MPD_HOST, TEST_MPD_PORT)
423 def test_connection_lost(self):
424 # Simulate a connection lost: the socket returns empty strings
425 self.MPDWillReturn('')
427 with self.assertRaises(musicpd.ConnectionError):
430 # consistent behaviour
431 # solves <https://github.com/Mic92/python-mpd2/issues/11> (also present
433 with self.assertRaises(musicpd.ConnectionError):
436 self.assertIs(self.client._sock, None)
438 def test_parse_sticker_get_no_sticker(self):
439 self.MPDWillReturn('ACK [50@0] {sticker} no such sticker\n')
440 self.assertRaises(musicpd.CommandError,
441 self.client.sticker_get, 'song', 'baz', 'foo')
443 def test_parse_sticker_list(self):
444 self.MPDWillReturn('sticker: foo=bar\n', 'sticker: lom=bok\n', 'OK\n')
445 res = self.client.sticker_list('song', 'baz')
446 self.assertEqual(['foo=bar', 'lom=bok'], res)
448 # Even with only one sticker, we get a dict
449 self.MPDWillReturn('sticker: foo=bar\n', 'OK\n')
450 res = self.client.sticker_list('song', 'baz')
451 self.assertEqual(['foo=bar'], res)
453 def test_albumart(self):
454 # here is a 34 bytes long data
455 data = bytes('\xff\xd8\xff\xe0\x00\x10JFIF\x00\x01\x01\x00\x00\x01'
456 '\x00\x01\x00\x00\xff\xdb\x00C\x00\x05\x03\x04',
458 read_lines = [b'size: 42\nbinary: 34\n', data, b'\nOK\n']
459 self.MPDWillReturnBinary(read_lines)
460 # Reading albumart / offset 0 should return the data
461 res = self.client.albumart('muse/Raised Fist/2002-Dedication/', 0)
462 self.assertEqual(res.get('data'), data)
465 if __name__ == '__main__':