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 class TestMPDClient(unittest.TestCase):
76 # last sync: musicpd 0.4.2 unreleased / Mon Nov 17 21:45:22 CET 2014
79 'clearerror': 'nothing',
80 'currentsong': 'object',
85 # Playback Option Commands
87 'crossfade': 'nothing',
88 'mixrampdb': 'nothing',
89 'mixrampdelay': 'nothing',
94 'replay_gain_mode': 'nothing',
95 'replay_gain_status': 'item',
97 # Playback Control Commands
102 'previous': 'nothing',
105 'seekcur': 'nothing',
112 'deleteid': 'nothing',
115 'playlist': 'playlist',
116 'playlistfind': 'songs',
117 'playlistid': 'songs',
118 'playlistinfo': 'songs',
119 'playlistsearch': 'songs',
120 'plchanges': 'songs',
121 'plchangesposid': 'changes',
124 'rangeid': 'nothing',
125 'shuffle': 'nothing',
128 'addtagid': 'nothing',
129 'cleartagid': 'nothing',
130 # Stored Playlist Commands
131 'listplaylist': 'list',
132 'listplaylistinfo': 'songs',
133 'listplaylists': 'playlists',
135 'playlistadd': 'nothing',
136 'playlistclear': 'nothing',
137 'playlistdelete': 'nothing',
138 'playlistmove': 'nothing',
145 'findadd': 'nothing',
147 'listall': 'database',
148 'listallinfo': 'database',
149 'lsinfo': 'database',
151 'searchadd': 'nothing',
152 'searchaddpl': 'nothing',
155 'readcomments': 'object',
156 # Mounts and neighbors
158 'unmount': 'nothing',
159 'listmounts': 'mounts',
160 'listneighbors': 'neighbors',
162 'sticker get': 'item',
163 'sticker set': 'nothing',
164 'sticker delete': 'nothing',
165 'sticker list': 'list',
166 'sticker find': 'songs',
167 # Connection Commands
170 'password': 'nothing',
173 'partition': 'nothing',
174 'listpartitions': 'list',
175 'newpartition': 'nothing',
176 # Audio Output Commands
177 'disableoutput': 'nothing',
178 'enableoutput': 'nothing',
179 'toggleoutput': 'nothing',
180 'outputs': 'outputs',
181 # Reflection Commands
184 'notcommands': 'list',
186 'urlhandlers': 'list',
187 'decoders': 'plugins',
189 'subscribe': 'nothing',
190 'unsubscribe': 'nothing',
192 'readmessages': 'messages',
193 'sendmessage': 'nothing',
197 self.socket_patch = mock.patch('musicpd.socket')
198 self.socket_mock = self.socket_patch.start()
199 self.socket_mock.getaddrinfo.return_value = [range(5)]
201 self.socket_mock.socket.side_effect = (
203 # Create a new socket.socket() mock with default attributes,
204 # each time we are calling it back (otherwise, it keeps set
205 # attributes across calls).
206 # That's probably what we want, since reconnecting is like
207 # reinitializing the entire connection, and so, the mock.
208 mock.MagicMock(name='socket.socket'))
210 self.client = musicpd.MPDClient()
211 self.client.connect(TEST_MPD_HOST, TEST_MPD_PORT)
212 self.client._sock.reset_mock()
213 self.MPDWillReturn('ACK don\'t forget to setup your mock\n')
216 self.socket_patch.stop()
218 def MPDWillReturn(self, *lines):
219 # Return what the caller wants first, then do as if the socket was
221 self.client._rfile.readline.side_effect = itertools.chain(
222 lines, itertools.repeat(''))
224 def MPDWillReturnBinary(self, lines):
225 data = bytearray(b''.join(lines))
233 val.append(data.pop(0))
238 while not val.endswith(b'\x0a'):
239 val.append(data.pop(0))
241 self.client._rbfile.readline.side_effect = readline
242 self.client._rbfile.read.side_effect = read
244 def assertMPDReceived(self, *lines):
245 self.client._wfile.write.assert_called_with(*lines)
247 def test_metaclass_commands(self):
248 """Controls client has at least commands as last synchronized in
249 TestMPDClient.commands"""
250 for cmd, ret in TestMPDClient.commands.items():
251 self.assertTrue(hasattr(self.client, cmd), msg='cmd "{}" not available!'.format(cmd))
253 self.assertTrue(hasattr(self.client, cmd.replace(' ', '_')))
255 def test_fetch_nothing(self):
256 self.MPDWillReturn('OK\n')
257 self.assertIsNone(self.client.ping())
258 self.assertMPDReceived('ping\n')
260 def test_fetch_list(self):
261 self.MPDWillReturn('OK\n')
262 self.assertIsInstance(self.client.list('album'), list)
263 self.assertMPDReceived('list "album"\n')
265 def test_fetch_item(self):
266 self.MPDWillReturn('updating_db: 42\n', 'OK\n')
267 self.assertIsNotNone(self.client.update())
269 def test_fetch_object(self):
270 # XXX: _read_objects() doesn't wait for the final OK
271 self.MPDWillReturn('volume: 63\n', 'OK\n')
272 status = self.client.status()
273 self.assertMPDReceived('status\n')
274 self.assertIsInstance(status, dict)
276 # XXX: _read_objects() doesn't wait for the final OK
277 self.MPDWillReturn('OK\n')
278 stats = self.client.stats()
279 self.assertMPDReceived('stats\n')
280 self.assertIsInstance(stats, dict)
282 def test_fetch_songs(self):
283 self.MPDWillReturn('file: my-song.ogg\n', 'Pos: 0\n', 'Id: 66\n', 'OK\n')
284 playlist = self.client.playlistinfo()
286 self.assertMPDReceived('playlistinfo\n')
287 self.assertIsInstance(playlist, list)
288 self.assertEqual(1, len(playlist))
290 self.assertIsInstance(e, dict)
291 self.assertEqual('my-song.ogg', e['file'])
292 self.assertEqual('0', e['pos'])
293 self.assertEqual('66', e['id'])
295 def test_send_and_fetch(self):
296 self.MPDWillReturn('volume: 50\n', 'OK\n')
297 result = self.client.send_status()
298 self.assertEqual(None, result)
299 self.assertMPDReceived('status\n')
301 status = self.client.fetch_status()
302 self.assertEqual(1, self.client._wfile.write.call_count)
303 self.assertEqual({'volume': '50'}, status)
305 def test_iterating(self):
306 self.MPDWillReturn('file: my-song.ogg\n', 'Pos: 0\n', 'Id: 66\n',
307 'file: my-song.ogg\n', 'Pos: 0\n', 'Id: 66\n', 'OK\n')
308 self.client.iterate = True
309 playlist = self.client.playlistinfo()
310 self.assertMPDReceived('playlistinfo\n')
311 self.assertIsInstance(playlist, types.GeneratorType)
312 self.assertTrue(self.client._iterating)
313 for song in playlist:
314 self.assertRaises(musicpd.IteratingError, self.client.status)
315 self.assertIsInstance(song, dict)
316 self.assertEqual('my-song.ogg', song['file'])
317 self.assertEqual('0', song['pos'])
318 self.assertEqual('66', song['id'])
319 self.assertFalse(self.client._iterating)
321 def test_noidle(self):
322 self.MPDWillReturn('OK\n') # nothing changed after idle-ing
323 self.client.send_idle()
324 self.MPDWillReturn('OK\n') # nothing changed after noidle
325 self.assertEqual(self.client.noidle(), [])
326 self.assertMPDReceived('noidle\n')
327 self.MPDWillReturn('volume: 50\n', 'OK\n')
329 self.assertMPDReceived('status\n')
331 def test_noidle_while_idle_started_sending(self):
332 self.MPDWillReturn('OK\n') # nothing changed after idle
333 self.client.send_idle()
334 self.MPDWillReturn('CHANGED: player\n', 'OK\n') # noidle response
335 self.assertEqual(self.client.noidle(), ['player'])
336 self.MPDWillReturn('volume: 50\n', 'OK\n')
337 status = self.client.status()
338 self.assertMPDReceived('status\n')
339 self.assertEqual({'volume': '50'}, status)
341 def test_throw_when_calling_noidle_withoutidling(self):
342 self.assertRaises(musicpd.CommandError, self.client.noidle)
343 self.client.send_status()
344 self.assertRaises(musicpd.CommandError, self.client.noidle)
346 def test_client_to_client(self):
347 # client to client is at this time in beta!
349 self.MPDWillReturn('OK\n')
350 self.assertIsNone(self.client.subscribe("monty"))
351 self.assertMPDReceived('subscribe "monty"\n')
353 self.MPDWillReturn('channel: monty\n', 'OK\n')
354 channels = self.client.channels()
355 self.assertMPDReceived('channels\n')
356 self.assertEqual(['monty'], channels)
358 self.MPDWillReturn('OK\n')
359 self.assertIsNone(self.client.sendmessage('monty', 'SPAM'))
360 self.assertMPDReceived('sendmessage "monty" "SPAM"\n')
362 self.MPDWillReturn('channel: monty\n', 'message: SPAM\n', 'OK\n')
363 msg = self.client.readmessages()
364 self.assertMPDReceived('readmessages\n')
365 self.assertEqual(msg, [{'channel': 'monty', 'message': 'SPAM'}])
367 self.MPDWillReturn('OK\n')
368 self.assertIsNone(self.client.unsubscribe('monty'))
369 self.assertMPDReceived('unsubscribe "monty"\n')
371 self.MPDWillReturn('OK\n')
372 channels = self.client.channels()
373 self.assertMPDReceived('channels\n')
374 self.assertEqual([], channels)
376 def test_ranges_in_command_args(self):
377 self.MPDWillReturn('OK\n')
378 self.client.playlistinfo((10,))
379 self.assertMPDReceived('playlistinfo 10:\n')
381 self.MPDWillReturn('OK\n')
382 self.client.playlistinfo(('10',))
383 self.assertMPDReceived('playlistinfo 10:\n')
385 self.MPDWillReturn('OK\n')
386 self.client.playlistinfo((10, 12))
387 self.assertMPDReceived('playlistinfo 10:12\n')
389 self.MPDWillReturn('OK\n')
390 self.client.rangeid(())
391 self.assertMPDReceived('rangeid :\n')
393 for arg in [(10, 't'), (10, 1, 1), (None,1)]:
394 self.MPDWillReturn('OK\n')
395 with self.assertRaises(musicpd.CommandError):
396 self.client.playlistinfo(arg)
398 def test_numbers_as_command_args(self):
399 self.MPDWillReturn('OK\n')
400 self.client.find('file', 1)
401 self.assertMPDReceived('find "file" "1"\n')
403 def test_commands_without_callbacks(self):
404 self.MPDWillReturn('\n')
406 self.assertMPDReceived('close\n')
408 # XXX: what are we testing here?
410 self.client.connect(TEST_MPD_HOST, TEST_MPD_PORT)
412 def test_connection_lost(self):
413 # Simulate a connection lost: the socket returns empty strings
414 self.MPDWillReturn('')
416 with self.assertRaises(musicpd.ConnectionError):
419 # consistent behaviour
420 # solves <https://github.com/Mic92/python-mpd2/issues/11> (also present
422 with self.assertRaises(musicpd.ConnectionError):
425 self.assertIs(self.client._sock, None)
427 def test_parse_sticker_get_no_sticker(self):
428 self.MPDWillReturn('ACK [50@0] {sticker} no such sticker\n')
429 self.assertRaises(musicpd.CommandError,
430 self.client.sticker_get, 'song', 'baz', 'foo')
432 def test_parse_sticker_list(self):
433 self.MPDWillReturn('sticker: foo=bar\n', 'sticker: lom=bok\n', 'OK\n')
434 res = self.client.sticker_list('song', 'baz')
435 self.assertEqual(['foo=bar', 'lom=bok'], res)
437 # Even with only one sticker, we get a dict
438 self.MPDWillReturn('sticker: foo=bar\n', 'OK\n')
439 res = self.client.sticker_list('song', 'baz')
440 self.assertEqual(['foo=bar'], res)
442 def test_albumart(self):
443 # here is a 34 bytes long data
444 data = bytes('\xff\xd8\xff\xe0\x00\x10JFIF\x00\x01\x01\x00\x00\x01'
445 '\x00\x01\x00\x00\xff\xdb\x00C\x00\x05\x03\x04',
447 read_lines = [b'size: 42\nbinary: 34\n', data, b'\nOK\n']
448 self.MPDWillReturnBinary(read_lines)
449 # Reading albumart / offset 0 should return the data
450 res = self.client.albumart('muse/Raised Fist/2002-Dedication/', 0)
451 self.assertEqual(res.get('data'), data)
454 if __name__ == '__main__':