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 # mock "os.path.exists" here to ensure there are no socket in
40 # XDG_RUNTIME_DIR/mpd or /run/mpd since with test defaults fallbacks
42 # * neither MPD_HOST nor XDG_RUNTIME_DIR are not set
43 # * /run/mpd does not expose a socket
44 with mock.patch('os.path.exists', return_value=False):
45 os.environ.pop('MPD_HOST', None)
46 os.environ.pop('MPD_PORT', None)
47 client = musicpd.MPDClient()
48 self.assertEqual(client.host, 'localhost')
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 password extraction
59 os.environ['MPD_HOST'] = 'pa55w04d@example.org'
60 client = musicpd.MPDClient()
61 self.assertEqual(client.pwd, 'pa55w04d')
62 self.assertEqual(client.host, 'example.org')
65 os.environ['MPD_HOST'] = 'example.org'
66 client = musicpd.MPDClient()
67 self.assertFalse(client.pwd)
68 self.assertEqual(client.host, 'example.org')
70 # Test password extraction (no host)
71 os.environ['MPD_HOST'] = 'pa55w04d@'
72 with mock.patch('os.path.exists', return_value=False):
73 client = musicpd.MPDClient()
74 self.assertEqual(client.pwd, 'pa55w04d')
75 self.assertEqual(client.host, 'localhost')
77 # Test badly formatted MPD_HOST
78 os.environ['MPD_HOST'] = '@'
79 with mock.patch('os.path.exists', return_value=False):
80 client = musicpd.MPDClient()
81 self.assertEqual(client.pwd, None)
82 self.assertEqual(client.host, 'localhost')
84 # Test unix socket extraction
85 os.environ['MPD_HOST'] = 'pa55w04d@/unix/sock'
86 client = musicpd.MPDClient()
87 self.assertEqual(client.host, '/unix/sock')
89 # Test plain abstract socket extraction
90 os.environ['MPD_HOST'] = '@abstract'
91 client = musicpd.MPDClient()
92 self.assertEqual(client.host, '@abstract')
94 # Test password and abstract socket extraction
95 os.environ['MPD_HOST'] = 'pass@@abstract'
96 client = musicpd.MPDClient()
97 self.assertEqual(client.pwd, 'pass')
98 self.assertEqual(client.host, '@abstract')
100 # Test unix socket fallback
101 os.environ.pop('MPD_HOST', None)
102 os.environ.pop('MPD_PORT', None)
103 os.environ.pop('XDG_RUNTIME_DIR', None)
104 with mock.patch('os.path.exists', return_value=True):
105 client = musicpd.MPDClient()
106 self.assertEqual(client.host, '/run/mpd/socket')
107 os.environ['XDG_RUNTIME_DIR'] = '/run/user/1000'
108 client = musicpd.MPDClient()
109 self.assertEqual(client.host, '/run/user/1000/mpd/socket')
111 os.environ.pop('MPD_HOST', None)
112 os.environ.pop('MPD_PORT', None)
113 os.environ['XDG_RUNTIME_DIR'] = '/run/user/1000/'
114 with mock.patch('os.path.exists', return_value=True):
115 client = musicpd.MPDClient()
116 self.assertEqual(client.host, '/run/user/1000/mpd/socket')
119 os.environ.pop('MPD_TIMEOUT', None)
120 client = musicpd.MPDClient()
121 self.assertEqual(client.mpd_timeout, musicpd.CONNECTION_TIMEOUT)
122 os.environ['MPD_TIMEOUT'] = 'garbage'
123 client = musicpd.MPDClient()
124 self.assertEqual(client.mpd_timeout,
125 musicpd.CONNECTION_TIMEOUT,
126 'Garbage\'s not silently ignore to use default value')
127 os.environ['MPD_TIMEOUT'] = '42'
128 client = musicpd.MPDClient()
129 self.assertEqual(client.mpd_timeout, 42)
132 class TestMPDClient(unittest.TestCase):
135 # last sync: musicpd 0.6.0 unreleased / Fri Feb 19 15:34:53 CET 2021
138 'clearerror': 'nothing',
139 'currentsong': 'object',
144 # Playback Option Commands
145 'consume': 'nothing',
146 'crossfade': 'nothing',
147 'mixrampdb': 'nothing',
148 'mixrampdelay': 'nothing',
154 'replay_gain_mode': 'nothing',
155 'replay_gain_status': 'item',
157 # Playback Control Commands
162 'previous': 'nothing',
165 'seekcur': 'nothing',
172 'deleteid': 'nothing',
175 'playlist': 'playlist',
176 'playlistfind': 'songs',
177 'playlistid': 'songs',
178 'playlistinfo': 'songs',
179 'playlistsearch': 'songs',
180 'plchanges': 'songs',
181 'plchangesposid': 'changes',
184 'rangeid': 'nothing',
185 'shuffle': 'nothing',
188 'addtagid': 'nothing',
189 'cleartagid': 'nothing',
190 # Stored Playlist Commands
191 'listplaylist': 'list',
192 'listplaylistinfo': 'songs',
193 'listplaylists': 'playlists',
195 'playlistadd': 'nothing',
196 'playlistclear': 'nothing',
197 'playlistdelete': 'nothing',
198 'playlistmove': 'nothing',
203 'albumart': 'composite',
205 'getfingerprint': 'object',
207 'findadd': 'nothing',
209 'listall': 'database',
210 'listallinfo': 'database',
211 'listfiles': 'database',
212 'lsinfo': 'database',
213 'readcomments': 'object',
214 'readpicture': 'composite',
216 'searchadd': 'nothing',
217 'searchaddpl': 'nothing',
220 # Mounts and neighbors
222 'unmount': 'nothing',
223 'listmounts': 'mounts',
224 'listneighbors': 'neighbors',
226 'sticker get': 'item',
227 'sticker set': 'nothing',
228 'sticker delete': 'nothing',
229 'sticker list': 'list',
230 'sticker find': 'songs',
231 # Connection Commands
234 'password': 'nothing',
236 'binarylimit': 'nothing',
238 'tagtypes disable': 'nothing',
239 'tagtypes enable': 'nothing',
240 'tagtypes clear': 'nothing',
241 'tagtypes all': 'nothing',
243 'partition': 'nothing',
244 'listpartitions': 'list',
245 'newpartition': 'nothing',
246 'delpartition': 'nothing',
247 'moveoutput': 'nothing',
248 # Audio Output Commands
249 'disableoutput': 'nothing',
250 'enableoutput': 'nothing',
251 'toggleoutput': 'nothing',
252 'outputs': 'outputs',
253 'outputset': 'nothing',
254 # Reflection Commands
257 'notcommands': 'list',
258 'urlhandlers': 'list',
259 'decoders': 'plugins',
261 'subscribe': 'nothing',
262 'unsubscribe': 'nothing',
264 'readmessages': 'messages',
265 'sendmessage': 'nothing',
269 self.socket_patch = mock.patch('musicpd.socket')
270 self.socket_mock = self.socket_patch.start()
271 self.socket_mock.getaddrinfo.return_value = [range(5)]
273 self.socket_mock.socket.side_effect = (
275 # Create a new socket.socket() mock with default attributes,
276 # each time we are calling it back (otherwise, it keeps set
277 # attributes across calls).
278 # That's probably what we want, since reconnecting is like
279 # reinitializing the entire connection, and so, the mock.
280 mock.MagicMock(name='socket.socket'))
282 self.client = musicpd.MPDClient()
283 self.client.connect(TEST_MPD_HOST, TEST_MPD_PORT)
284 self.client._sock.reset_mock()
285 self.MPDWillReturn('ACK don\'t forget to setup your mock\n')
288 self.socket_patch.stop()
290 def MPDWillReturn(self, *lines):
291 # Return what the caller wants first, then do as if the socket was
293 self.client._rfile.readline.side_effect = itertools.chain(
294 lines, itertools.repeat(''))
296 def MPDWillReturnBinary(self, lines):
297 data = bytearray(b''.join(lines))
305 val.append(data.pop(0))
310 while not val.endswith(b'\x0a'):
311 val.append(data.pop(0))
313 self.client._rbfile.readline.side_effect = readline
314 self.client._rbfile.read.side_effect = read
316 def assertMPDReceived(self, *lines):
317 self.client._wfile.write.assert_called_with(*lines)
319 def test_metaclass_commands(self):
320 """Controls client has at least commands as last synchronized in
321 TestMPDClient.commands"""
322 for cmd, ret in TestMPDClient.commands.items():
323 self.assertTrue(hasattr(self.client, cmd), msg='cmd "{}" not available!'.format(cmd))
325 self.assertTrue(hasattr(self.client, cmd.replace(' ', '_')))
327 def test_fetch_nothing(self):
328 self.MPDWillReturn('OK\n')
329 self.assertIsNone(self.client.ping())
330 self.assertMPDReceived('ping\n')
332 def test_fetch_list(self):
333 self.MPDWillReturn('OK\n')
334 self.assertIsInstance(self.client.list('album'), list)
335 self.assertMPDReceived('list "album"\n')
337 def test_fetch_item(self):
338 self.MPDWillReturn('updating_db: 42\n', 'OK\n')
339 self.assertIsNotNone(self.client.update())
341 def test_fetch_object(self):
342 # XXX: _read_objects() doesn't wait for the final OK
343 self.MPDWillReturn('volume: 63\n', 'OK\n')
344 status = self.client.status()
345 self.assertMPDReceived('status\n')
346 self.assertIsInstance(status, dict)
348 # XXX: _read_objects() doesn't wait for the final OK
349 self.MPDWillReturn('OK\n')
350 stats = self.client.stats()
351 self.assertMPDReceived('stats\n')
352 self.assertIsInstance(stats, dict)
354 output = ['outputid: 0\n',
355 'outputname: default detected output\n',
357 'outputenabled: 1\n']
358 self.MPDWillReturn(*output, 'OK\n')
359 outputs = self.client.outputs()
360 self.assertMPDReceived('outputs\n')
361 self.assertIsInstance(outputs, list)
362 self.assertEqual([{'outputid': '0', 'outputname': 'default detected output', 'plugin': 'sndio', 'outputenabled': '1'}], outputs)
364 def test_fetch_songs(self):
365 self.MPDWillReturn('file: my-song.ogg\n', 'Pos: 0\n', 'Id: 66\n', 'OK\n')
366 playlist = self.client.playlistinfo()
368 self.assertMPDReceived('playlistinfo\n')
369 self.assertIsInstance(playlist, list)
370 self.assertEqual(1, len(playlist))
372 self.assertIsInstance(e, dict)
373 self.assertEqual('my-song.ogg', e['file'])
374 self.assertEqual('0', e['pos'])
375 self.assertEqual('66', e['id'])
377 def test_send_and_fetch(self):
378 self.MPDWillReturn('volume: 50\n', 'OK\n')
379 result = self.client.send_status()
380 self.assertEqual(None, result)
381 self.assertMPDReceived('status\n')
383 status = self.client.fetch_status()
384 self.assertEqual(1, self.client._wfile.write.call_count)
385 self.assertEqual({'volume': '50'}, status)
387 def test_iterating(self):
388 self.MPDWillReturn('file: my-song.ogg\n', 'Pos: 0\n', 'Id: 66\n',
389 'file: my-song.ogg\n', 'Pos: 0\n', 'Id: 66\n', 'OK\n')
390 self.client.iterate = True
391 playlist = self.client.playlistinfo()
392 self.assertMPDReceived('playlistinfo\n')
393 self.assertIsInstance(playlist, types.GeneratorType)
394 self.assertTrue(self.client._iterating)
395 for song in playlist:
396 self.assertRaises(musicpd.IteratingError, self.client.status)
397 self.assertIsInstance(song, dict)
398 self.assertEqual('my-song.ogg', song['file'])
399 self.assertEqual('0', song['pos'])
400 self.assertEqual('66', song['id'])
401 self.assertFalse(self.client._iterating)
403 def test_noidle(self):
404 self.MPDWillReturn('OK\n') # nothing changed after idle-ing
405 self.client.send_idle()
406 self.MPDWillReturn('OK\n') # nothing changed after noidle
407 self.assertEqual(self.client.noidle(), [])
408 self.assertMPDReceived('noidle\n')
409 self.MPDWillReturn('volume: 50\n', 'OK\n')
411 self.assertMPDReceived('status\n')
413 def test_noidle_while_idle_started_sending(self):
414 self.MPDWillReturn('OK\n') # nothing changed after idle
415 self.client.send_idle()
416 self.MPDWillReturn('CHANGED: player\n', 'OK\n') # noidle response
417 self.assertEqual(self.client.noidle(), ['player'])
418 self.MPDWillReturn('volume: 50\n', 'OK\n')
419 status = self.client.status()
420 self.assertMPDReceived('status\n')
421 self.assertEqual({'volume': '50'}, status)
423 def test_throw_when_calling_noidle_withoutidling(self):
424 self.assertRaises(musicpd.CommandError, self.client.noidle)
425 self.client.send_status()
426 self.assertRaises(musicpd.CommandError, self.client.noidle)
428 def test_client_to_client(self):
429 self.MPDWillReturn('OK\n')
430 self.assertIsNone(self.client.subscribe("monty"))
431 self.assertMPDReceived('subscribe "monty"\n')
433 self.MPDWillReturn('channel: monty\n', 'OK\n')
434 channels = self.client.channels()
435 self.assertMPDReceived('channels\n')
436 self.assertEqual(['monty'], channels)
438 self.MPDWillReturn('OK\n')
439 self.assertIsNone(self.client.sendmessage('monty', 'SPAM'))
440 self.assertMPDReceived('sendmessage "monty" "SPAM"\n')
442 self.MPDWillReturn('channel: monty\n', 'message: SPAM\n', 'OK\n')
443 msg = self.client.readmessages()
444 self.assertMPDReceived('readmessages\n')
445 self.assertEqual(msg, [{'channel': 'monty', 'message': 'SPAM'}])
447 self.MPDWillReturn('OK\n')
448 self.assertIsNone(self.client.unsubscribe('monty'))
449 self.assertMPDReceived('unsubscribe "monty"\n')
451 self.MPDWillReturn('OK\n')
452 channels = self.client.channels()
453 self.assertMPDReceived('channels\n')
454 self.assertEqual([], channels)
456 def test_ranges_in_command_args(self):
457 self.MPDWillReturn('OK\n')
458 self.client.playlistinfo((10,))
459 self.assertMPDReceived('playlistinfo 10:\n')
461 self.MPDWillReturn('OK\n')
462 self.client.playlistinfo(('10',))
463 self.assertMPDReceived('playlistinfo 10:\n')
465 self.MPDWillReturn('OK\n')
466 self.client.playlistinfo((10, 12))
467 self.assertMPDReceived('playlistinfo 10:12\n')
469 self.MPDWillReturn('OK\n')
470 self.client.rangeid(())
471 self.assertMPDReceived('rangeid :\n')
473 for arg in [(10, 't'), (10, 1, 1), (None,1)]:
474 self.MPDWillReturn('OK\n')
475 with self.assertRaises(musicpd.CommandError):
476 self.client.playlistinfo(arg)
478 def test_numbers_as_command_args(self):
479 self.MPDWillReturn('OK\n')
480 self.client.find('file', 1)
481 self.assertMPDReceived('find "file" "1"\n')
483 def test_commands_without_callbacks(self):
484 self.MPDWillReturn('\n')
486 self.assertMPDReceived('close\n')
488 # XXX: what are we testing here?
490 self.client.connect(TEST_MPD_HOST, TEST_MPD_PORT)
492 def test_connection_lost(self):
493 # Simulate a connection lost: the socket returns empty strings
494 self.MPDWillReturn('')
496 with self.assertRaises(musicpd.ConnectionError):
499 # consistent behaviour
500 # solves <https://github.com/Mic92/python-mpd2/issues/11> (also present
502 with self.assertRaises(musicpd.ConnectionError):
505 self.assertIs(self.client._sock, None)
507 def test_parse_sticker_get_no_sticker(self):
508 self.MPDWillReturn('ACK [50@0] {sticker} no such sticker\n')
509 self.assertRaises(musicpd.CommandError,
510 self.client.sticker_get, 'song', 'baz', 'foo')
512 def test_parse_sticker_list(self):
513 self.MPDWillReturn('sticker: foo=bar\n', 'sticker: lom=bok\n', 'OK\n')
514 res = self.client.sticker_list('song', 'baz')
515 self.assertEqual(['foo=bar', 'lom=bok'], res)
517 # Even with only one sticker, we get a dict
518 self.MPDWillReturn('sticker: foo=bar\n', 'OK\n')
519 res = self.client.sticker_list('song', 'baz')
520 self.assertEqual(['foo=bar'], res)
522 def test_albumart(self):
523 # here is a 34 bytes long data
524 data = bytes('\xff\xd8\xff\xe0\x00\x10JFIF\x00\x01\x01\x00\x00\x01'
525 '\x00\x01\x00\x00\xff\xdb\x00C\x00\x05\x03\x04',
527 read_lines = [b'size: 42\nbinary: 34\n', data, b'\nOK\n']
528 self.MPDWillReturnBinary(read_lines)
529 # Reading albumart / offset 0 should return the data
530 res = self.client.albumart('muse/Raised Fist/2002-Dedication/', 0)
531 self.assertEqual(res.get('data'), data)
533 def test_reading_binary(self):
534 # readpicture when there are no picture returns empty object
535 self.MPDWillReturnBinary([b'OK\n'])
536 res = self.client.readpicture('muse/Raised Fist/2002-Dedication/', 0)
537 self.assertEqual(res, {})
539 def test_command_list(self):
540 self.MPDWillReturn('updating_db: 42\n',
547 self.client.command_list_ok_begin()
550 self.client.repeat(1)
551 self.client.command_list_end()
552 self.assertMPDReceived('command_list_end\n')
554 def test_two_word_commands(self):
555 self.MPDWillReturn('OK\n')
556 self.client.tagtypes_clear()
557 self.assertMPDReceived('tagtypes clear\n')
558 self.MPDWillReturn('OK\n')
559 with self.assertRaises(AttributeError):
560 self.client.foo_bar()
562 class testConnection(unittest.TestCase):
564 def test_exposing_fileno(self):
565 with mock.patch('musicpd.socket') as socket_mock:
566 sock = mock.MagicMock(name='socket')
567 socket_mock.socket.return_value = sock
568 cli = musicpd.MPDClient()
571 cli._sock.fileno.assert_called_with()
573 def test_connect_abstract(self):
574 os.environ['MPD_HOST'] = '@abstract'
575 with mock.patch('musicpd.socket') as socket_mock:
576 sock = mock.MagicMock(name='socket')
577 socket_mock.socket.return_value = sock
578 cli = musicpd.MPDClient()
580 sock.connect.assert_called_with('\0abstract')
582 def test_connect_unix(self):
583 os.environ['MPD_HOST'] = '/run/mpd/socket'
584 with mock.patch('musicpd.socket') as socket_mock:
585 sock = mock.MagicMock(name='socket')
586 socket_mock.socket.return_value = sock
587 cli = musicpd.MPDClient()
589 sock.connect.assert_called_with('/run/mpd/socket')
592 class testException(unittest.TestCase):
594 def test_CommandError_on_newline(self):
595 os.environ['MPD_HOST'] = '/run/mpd/socket'
596 with mock.patch('musicpd.socket') as socket_mock:
597 sock = mock.MagicMock(name='socket')
598 socket_mock.socket.return_value = sock
599 cli = musicpd.MPDClient()
601 with self.assertRaises(musicpd.CommandError):
602 cli.find('(album == "foo\nbar")')
605 if __name__ == '__main__':