3 # SPDX-FileCopyrightText: 2012-2021 kaliko <kaliko@azylum.org>
4 # SPDX-License-Identifier: GPL-3.0-or-later
5 # pylint: disable=missing-docstring
7 Test suite highly borrowed^Wsteal from python-mpd2 [0] project.
9 [0] https://github.com/Mic92/python-mpd2
17 import unittest.mock as mock
23 # show deprecation warnings
24 warnings.simplefilter('default')
27 TEST_MPD_HOST, TEST_MPD_PORT = ('example.com', 10000)
30 class testEnvVar(unittest.TestCase):
32 def test_envvar(self):
33 # mock "os.path.exists" here to ensure there are no socket in
34 # XDG_RUNTIME_DIR/mpd or /run/mpd since with test defaults fallbacks
36 # * neither MPD_HOST nor XDG_RUNTIME_DIR are not set
37 # * /run/mpd does not expose a socket
38 with mock.patch('os.path.exists', return_value=False):
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.pop('MPD_HOST', None)
46 os.environ['MPD_PORT'] = '6666'
47 client = musicpd.MPDClient()
48 self.assertEqual(client.pwd, None)
49 self.assertEqual(client.host, 'localhost')
50 self.assertEqual(client.port, '6666')
52 # Test password extraction
53 os.environ['MPD_HOST'] = 'pa55w04d@example.org'
54 client = musicpd.MPDClient()
55 self.assertEqual(client.pwd, 'pa55w04d')
56 self.assertEqual(client.host, 'example.org')
59 os.environ['MPD_HOST'] = 'example.org'
60 client = musicpd.MPDClient()
61 self.assertFalse(client.pwd)
62 self.assertEqual(client.host, 'example.org')
64 # Test password extraction (no host)
65 os.environ['MPD_HOST'] = 'pa55w04d@'
66 with mock.patch('os.path.exists', return_value=False):
67 client = musicpd.MPDClient()
68 self.assertEqual(client.pwd, 'pa55w04d')
69 self.assertEqual(client.host, 'localhost')
71 # Test badly formatted MPD_HOST
72 os.environ['MPD_HOST'] = '@'
73 with mock.patch('os.path.exists', return_value=False):
74 client = musicpd.MPDClient()
75 self.assertEqual(client.pwd, None)
76 self.assertEqual(client.host, 'localhost')
78 # Test unix socket extraction
79 os.environ['MPD_HOST'] = 'pa55w04d@/unix/sock'
80 client = musicpd.MPDClient()
81 self.assertEqual(client.host, '/unix/sock')
83 # Test plain abstract socket extraction
84 os.environ['MPD_HOST'] = '@abstract'
85 client = musicpd.MPDClient()
86 self.assertEqual(client.host, '@abstract')
88 # Test password and abstract socket extraction
89 os.environ['MPD_HOST'] = 'pass@@abstract'
90 client = musicpd.MPDClient()
91 self.assertEqual(client.pwd, 'pass')
92 self.assertEqual(client.host, '@abstract')
94 # Test unix socket fallback
95 os.environ.pop('MPD_HOST', None)
96 os.environ.pop('MPD_PORT', None)
97 os.environ.pop('XDG_RUNTIME_DIR', None)
98 with mock.patch('os.path.exists', return_value=True):
99 client = musicpd.MPDClient()
100 self.assertEqual(client.host, '/run/mpd/socket')
101 os.environ['XDG_RUNTIME_DIR'] = '/run/user/1000'
102 client = musicpd.MPDClient()
103 self.assertEqual(client.host, '/run/user/1000/mpd/socket')
105 os.environ.pop('MPD_HOST', None)
106 os.environ.pop('MPD_PORT', None)
107 os.environ['XDG_RUNTIME_DIR'] = '/run/user/1000/'
108 with mock.patch('os.path.exists', return_value=True):
109 client = musicpd.MPDClient()
110 self.assertEqual(client.host, '/run/user/1000/mpd/socket')
113 os.environ.pop('MPD_TIMEOUT', None)
114 client = musicpd.MPDClient()
115 self.assertEqual(client.mpd_timeout, musicpd.CONNECTION_TIMEOUT)
116 os.environ['MPD_TIMEOUT'] = 'garbage'
117 client = musicpd.MPDClient()
118 self.assertEqual(client.mpd_timeout,
119 musicpd.CONNECTION_TIMEOUT,
120 'Garbage is silently ignore to use default value')
121 os.environ['MPD_TIMEOUT'] = '42'
122 client = musicpd.MPDClient()
123 self.assertEqual(client.mpd_timeout, 42)
126 class TestMPDClient(unittest.TestCase):
129 # last sync: musicpd 0.6.0 unreleased / Fri Feb 19 15:34:53 CET 2021
132 'clearerror': 'nothing',
133 'currentsong': 'object',
138 # Playback Option Commands
139 'consume': 'nothing',
140 'crossfade': 'nothing',
141 'mixrampdb': 'nothing',
142 'mixrampdelay': 'nothing',
148 'replay_gain_mode': 'nothing',
149 'replay_gain_status': 'item',
151 # Playback Control Commands
156 'previous': 'nothing',
159 'seekcur': 'nothing',
166 'deleteid': 'nothing',
169 'playlist': 'playlist',
170 'playlistfind': 'songs',
171 'playlistid': 'songs',
172 'playlistinfo': 'songs',
173 'playlistsearch': 'songs',
174 'plchanges': 'songs',
175 'plchangesposid': 'changes',
178 'rangeid': 'nothing',
179 'shuffle': 'nothing',
182 'addtagid': 'nothing',
183 'cleartagid': 'nothing',
184 # Stored Playlist Commands
185 'listplaylist': 'list',
186 'listplaylistinfo': 'songs',
187 'listplaylists': 'playlists',
189 'playlistadd': 'nothing',
190 'playlistclear': 'nothing',
191 'playlistdelete': 'nothing',
192 'playlistmove': 'nothing',
197 'albumart': 'composite',
199 'getfingerprint': 'object',
201 'findadd': 'nothing',
203 'listall': 'database',
204 'listallinfo': 'database',
205 'listfiles': 'database',
206 'lsinfo': 'database',
207 'readcomments': 'object',
208 'readpicture': 'composite',
210 'searchadd': 'nothing',
211 'searchaddpl': 'nothing',
214 # Mounts and neighbors
216 'unmount': 'nothing',
217 'listmounts': 'mounts',
218 'listneighbors': 'neighbors',
220 'sticker get': 'item',
221 'sticker set': 'nothing',
222 'sticker delete': 'nothing',
223 'sticker list': 'list',
224 'sticker find': 'songs',
225 # Connection Commands
228 'password': 'nothing',
230 'binarylimit': 'nothing',
232 'tagtypes disable': 'nothing',
233 'tagtypes enable': 'nothing',
234 'tagtypes clear': 'nothing',
235 'tagtypes all': 'nothing',
237 'partition': 'nothing',
238 'listpartitions': 'list',
239 'newpartition': 'nothing',
240 'delpartition': 'nothing',
241 'moveoutput': 'nothing',
242 # Audio Output Commands
243 'disableoutput': 'nothing',
244 'enableoutput': 'nothing',
245 'toggleoutput': 'nothing',
246 'outputs': 'outputs',
247 'outputset': 'nothing',
248 # Reflection Commands
251 'notcommands': 'list',
252 'urlhandlers': 'list',
253 'decoders': 'plugins',
255 'subscribe': 'nothing',
256 'unsubscribe': 'nothing',
258 'readmessages': 'messages',
259 'sendmessage': 'nothing',
263 self.socket_patch = mock.patch('musicpd.socket')
264 self.socket_mock = self.socket_patch.start()
265 self.socket_mock.getaddrinfo.return_value = [range(5)]
267 self.socket_mock.socket.side_effect = (
269 # Create a new socket.socket() mock with default attributes,
270 # each time we are calling it back (otherwise, it keeps set
271 # attributes across calls).
272 # That's probably what we want, since reconnecting is like
273 # reinitializing the entire connection, and so, the mock.
274 mock.MagicMock(name='socket.socket'))
276 self.client = musicpd.MPDClient()
277 self.client.connect(TEST_MPD_HOST, TEST_MPD_PORT)
278 self.client._sock.reset_mock()
279 self.MPDWillReturn('ACK don\'t forget to setup your mock\n')
282 self.socket_patch.stop()
284 def MPDWillReturn(self, *lines):
285 # Return what the caller wants first, then do as if the socket was
287 self.client._rfile.readline.side_effect = itertools.chain(
288 lines, itertools.repeat(''))
290 def MPDWillReturnBinary(self, lines):
291 data = bytearray(b''.join(lines))
299 val.append(data.pop(0))
304 while not val.endswith(b'\x0a'):
305 val.append(data.pop(0))
307 self.client._rbfile.readline.side_effect = readline
308 self.client._rbfile.read.side_effect = read
310 def assertMPDReceived(self, *lines):
311 self.client._wfile.write.assert_called_with(*lines)
313 def test_metaclass_commands(self):
314 """Controls client has at least commands as last synchronized in
315 TestMPDClient.commands"""
316 for cmd, ret in TestMPDClient.commands.items():
317 self.assertTrue(hasattr(self.client, cmd), msg='cmd "{}" not available!'.format(cmd))
319 self.assertTrue(hasattr(self.client, cmd.replace(' ', '_')))
321 def test_fetch_nothing(self):
322 self.MPDWillReturn('OK\n')
323 self.assertIsNone(self.client.ping())
324 self.assertMPDReceived('ping\n')
326 def test_fetch_list(self):
327 self.MPDWillReturn('OK\n')
328 self.assertIsInstance(self.client.list('album'), list)
329 self.assertMPDReceived('list "album"\n')
331 def test_fetch_item(self):
332 self.MPDWillReturn('updating_db: 42\n', 'OK\n')
333 self.assertIsNotNone(self.client.update())
335 def test_fetch_object(self):
336 # XXX: _read_objects() doesn't wait for the final OK
337 self.MPDWillReturn('volume: 63\n', 'OK\n')
338 status = self.client.status()
339 self.assertMPDReceived('status\n')
340 self.assertIsInstance(status, dict)
342 # XXX: _read_objects() doesn't wait for the final OK
343 self.MPDWillReturn('OK\n')
344 stats = self.client.stats()
345 self.assertMPDReceived('stats\n')
346 self.assertIsInstance(stats, dict)
348 output = ['outputid: 0\n',
349 'outputname: default detected output\n',
351 'outputenabled: 1\n']
352 self.MPDWillReturn(*output, 'OK\n')
353 outputs = self.client.outputs()
354 self.assertMPDReceived('outputs\n')
355 self.assertIsInstance(outputs, list)
356 self.assertEqual([{'outputid': '0', 'outputname': 'default detected output', 'plugin': 'sndio', 'outputenabled': '1'}], outputs)
358 def test_fetch_songs(self):
359 self.MPDWillReturn('file: my-song.ogg\n', 'Pos: 0\n', 'Id: 66\n', 'OK\n')
360 playlist = self.client.playlistinfo()
362 self.assertMPDReceived('playlistinfo\n')
363 self.assertIsInstance(playlist, list)
364 self.assertEqual(1, len(playlist))
366 self.assertIsInstance(e, dict)
367 self.assertEqual('my-song.ogg', e['file'])
368 self.assertEqual('0', e['pos'])
369 self.assertEqual('66', e['id'])
371 def test_send_and_fetch(self):
372 self.MPDWillReturn('volume: 50\n', 'OK\n')
373 result = self.client.send_status()
374 self.assertEqual(None, result)
375 self.assertMPDReceived('status\n')
377 status = self.client.fetch_status()
378 self.assertEqual(1, self.client._wfile.write.call_count)
379 self.assertEqual({'volume': '50'}, status)
381 def test_iterating(self):
382 self.MPDWillReturn('file: my-song.ogg\n', 'Pos: 0\n', 'Id: 66\n',
383 'file: my-song.ogg\n', 'Pos: 0\n', 'Id: 66\n', 'OK\n')
384 self.client.iterate = True
385 playlist = self.client.playlistinfo()
386 self.assertMPDReceived('playlistinfo\n')
387 self.assertIsInstance(playlist, types.GeneratorType)
388 self.assertTrue(self.client._iterating)
389 for song in playlist:
390 self.assertRaises(musicpd.IteratingError, self.client.status)
391 self.assertIsInstance(song, dict)
392 self.assertEqual('my-song.ogg', song['file'])
393 self.assertEqual('0', song['pos'])
394 self.assertEqual('66', song['id'])
395 self.assertFalse(self.client._iterating)
397 def test_noidle(self):
398 self.MPDWillReturn('OK\n') # nothing changed after idle-ing
399 self.client.send_idle()
400 self.MPDWillReturn('OK\n') # nothing changed after noidle
401 self.assertEqual(self.client.noidle(), [])
402 self.assertMPDReceived('noidle\n')
403 self.MPDWillReturn('volume: 50\n', 'OK\n')
405 self.assertMPDReceived('status\n')
407 def test_noidle_while_idle_started_sending(self):
408 self.MPDWillReturn('OK\n') # nothing changed after idle
409 self.client.send_idle()
410 self.MPDWillReturn('CHANGED: player\n', 'OK\n') # noidle response
411 self.assertEqual(self.client.noidle(), ['player'])
412 self.MPDWillReturn('volume: 50\n', 'OK\n')
413 status = self.client.status()
414 self.assertMPDReceived('status\n')
415 self.assertEqual({'volume': '50'}, status)
417 def test_throw_when_calling_noidle_withoutidling(self):
418 self.assertRaises(musicpd.CommandError, self.client.noidle)
419 self.client.send_status()
420 self.assertRaises(musicpd.CommandError, self.client.noidle)
422 def test_client_to_client(self):
423 self.MPDWillReturn('OK\n')
424 self.assertIsNone(self.client.subscribe("monty"))
425 self.assertMPDReceived('subscribe "monty"\n')
427 self.MPDWillReturn('channel: monty\n', 'OK\n')
428 channels = self.client.channels()
429 self.assertMPDReceived('channels\n')
430 self.assertEqual(['monty'], channels)
432 self.MPDWillReturn('OK\n')
433 self.assertIsNone(self.client.sendmessage('monty', 'SPAM'))
434 self.assertMPDReceived('sendmessage "monty" "SPAM"\n')
436 self.MPDWillReturn('channel: monty\n', 'message: SPAM\n', 'OK\n')
437 msg = self.client.readmessages()
438 self.assertMPDReceived('readmessages\n')
439 self.assertEqual(msg, [{'channel': 'monty', 'message': 'SPAM'}])
441 self.MPDWillReturn('OK\n')
442 self.assertIsNone(self.client.unsubscribe('monty'))
443 self.assertMPDReceived('unsubscribe "monty"\n')
445 self.MPDWillReturn('OK\n')
446 channels = self.client.channels()
447 self.assertMPDReceived('channels\n')
448 self.assertEqual([], channels)
450 def test_ranges_in_command_args(self):
451 self.MPDWillReturn('OK\n')
452 self.client.playlistinfo((10,))
453 self.assertMPDReceived('playlistinfo 10:\n')
455 self.MPDWillReturn('OK\n')
456 self.client.playlistinfo(('10',))
457 self.assertMPDReceived('playlistinfo 10:\n')
459 self.MPDWillReturn('OK\n')
460 self.client.playlistinfo((10, 12))
461 self.assertMPDReceived('playlistinfo 10:12\n')
463 self.MPDWillReturn('OK\n')
464 self.client.rangeid(())
465 self.assertMPDReceived('rangeid :\n')
467 for arg in [(10, 't'), (10, 1, 1), (None,1)]:
468 self.MPDWillReturn('OK\n')
469 with self.assertRaises(musicpd.CommandError):
470 self.client.playlistinfo(arg)
472 def test_numbers_as_command_args(self):
473 self.MPDWillReturn('OK\n')
474 self.client.find('file', 1)
475 self.assertMPDReceived('find "file" "1"\n')
477 def test_commands_without_callbacks(self):
478 self.MPDWillReturn('\n')
480 self.assertMPDReceived('close\n')
482 # XXX: what are we testing here?
484 self.client.connect(TEST_MPD_HOST, TEST_MPD_PORT)
486 def test_connection_lost(self):
487 # Simulate a connection lost: the socket returns empty strings
488 self.MPDWillReturn('')
490 with self.assertRaises(musicpd.ConnectionError):
493 # consistent behaviour
494 # solves <https://github.com/Mic92/python-mpd2/issues/11> (also present
496 with self.assertRaises(musicpd.ConnectionError):
499 self.assertIs(self.client._sock, None)
501 def test_parse_sticker_get_no_sticker(self):
502 self.MPDWillReturn('ACK [50@0] {sticker} no such sticker\n')
503 self.assertRaises(musicpd.CommandError,
504 self.client.sticker_get, 'song', 'baz', 'foo')
506 def test_parse_sticker_list(self):
507 self.MPDWillReturn('sticker: foo=bar\n', 'sticker: lom=bok\n', 'OK\n')
508 res = self.client.sticker_list('song', 'baz')
509 self.assertEqual(['foo=bar', 'lom=bok'], res)
511 # Even with only one sticker, we get a dict
512 self.MPDWillReturn('sticker: foo=bar\n', 'OK\n')
513 res = self.client.sticker_list('song', 'baz')
514 self.assertEqual(['foo=bar'], res)
516 def test_albumart(self):
517 # here is a 34 bytes long data
518 data = bytes('\xff\xd8\xff\xe0\x00\x10JFIF\x00\x01\x01\x00\x00\x01'
519 '\x00\x01\x00\x00\xff\xdb\x00C\x00\x05\x03\x04',
521 read_lines = [b'size: 42\nbinary: 34\n', data, b'\nOK\n']
522 self.MPDWillReturnBinary(read_lines)
523 # Reading albumart / offset 0 should return the data
524 res = self.client.albumart('muse/Raised Fist/2002-Dedication/', 0)
525 self.assertEqual(res.get('data'), data)
527 def test_reading_binary(self):
528 # readpicture when there are no picture returns empty object
529 self.MPDWillReturnBinary([b'OK\n'])
530 res = self.client.readpicture('muse/Raised Fist/2002-Dedication/', 0)
531 self.assertEqual(res, {})
533 def test_command_list(self):
534 self.MPDWillReturn('updating_db: 42\n',
541 self.client.command_list_ok_begin()
544 self.client.repeat(1)
545 self.client.command_list_end()
546 self.assertMPDReceived('command_list_end\n')
548 def test_two_word_commands(self):
549 self.MPDWillReturn('OK\n')
550 self.client.tagtypes_clear()
551 self.assertMPDReceived('tagtypes clear\n')
552 self.MPDWillReturn('OK\n')
553 with self.assertRaises(AttributeError):
554 self.client.foo_bar()
556 class testConnection(unittest.TestCase):
558 def test_exposing_fileno(self):
559 with mock.patch('musicpd.socket') as socket_mock:
560 sock = mock.MagicMock(name='socket')
561 socket_mock.socket.return_value = sock
562 cli = musicpd.MPDClient()
565 cli._sock.fileno.assert_called_with()
567 def test_connect_abstract(self):
568 os.environ['MPD_HOST'] = '@abstract'
569 with mock.patch('musicpd.socket') as socket_mock:
570 sock = mock.MagicMock(name='socket')
571 socket_mock.socket.return_value = sock
572 cli = musicpd.MPDClient()
574 sock.connect.assert_called_with('\0abstract')
576 def test_connect_unix(self):
577 os.environ['MPD_HOST'] = '/run/mpd/socket'
578 with mock.patch('musicpd.socket') as socket_mock:
579 sock = mock.MagicMock(name='socket')
580 socket_mock.socket.return_value = sock
581 cli = musicpd.MPDClient()
583 sock.connect.assert_called_with('/run/mpd/socket')
586 class testException(unittest.TestCase):
588 def test_CommandError_on_newline(self):
589 os.environ['MPD_HOST'] = '/run/mpd/socket'
590 with mock.patch('musicpd.socket') as socket_mock:
591 sock = mock.MagicMock(name='socket')
592 socket_mock.socket.return_value = sock
593 cli = musicpd.MPDClient()
595 with self.assertRaises(musicpd.CommandError):
596 cli.find('(album == "foo\nbar")')
598 class testContextManager(unittest.TestCase):
600 def test_enter_exit(self):
601 os.environ['MPD_HOST'] = '@abstract'
602 with mock.patch('musicpd.socket') as socket_mock:
603 sock = mock.MagicMock(name='socket')
604 socket_mock.socket.return_value = sock
605 cli = musicpd.MPDClient()
607 sock.connect.assert_called_with('\0abstract')
608 sock.close.assert_not_called()
609 sock.close.assert_called()
611 if __name__ == '__main__':