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
15 import unittest.mock as mock
21 # show deprecation warnings
22 warnings.simplefilter('default')
25 TEST_MPD_HOST, TEST_MPD_PORT = ('example.com', 10000)
28 class testEnvVar(unittest.TestCase):
30 def test_envvar(self):
31 # mock "os.path.exists" here to ensure there are no socket in
32 # XDG_RUNTIME_DIR/mpd or /run/mpd since with test defaults fallbacks
34 # * neither MPD_HOST nor XDG_RUNTIME_DIR are not set
35 # * /run/mpd does not expose a socket
36 with mock.patch('os.path.exists', return_value=False):
37 os.environ.pop('MPD_HOST', None)
38 os.environ.pop('MPD_PORT', None)
39 client = musicpd.MPDClient()
40 self.assertEqual(client.host, 'localhost')
41 self.assertEqual(client.port, '6600')
43 os.environ.pop('MPD_HOST', None)
44 os.environ['MPD_PORT'] = '6666'
45 client = musicpd.MPDClient()
46 self.assertEqual(client.pwd, None)
47 self.assertEqual(client.host, 'localhost')
48 self.assertEqual(client.port, '6666')
50 # Test password extraction
51 os.environ['MPD_HOST'] = 'pa55w04d@example.org'
52 client = musicpd.MPDClient()
53 self.assertEqual(client.pwd, 'pa55w04d')
54 self.assertEqual(client.host, 'example.org')
57 os.environ['MPD_HOST'] = 'example.org'
58 client = musicpd.MPDClient()
59 self.assertFalse(client.pwd)
60 self.assertEqual(client.host, 'example.org')
62 # Test password extraction (no host)
63 os.environ['MPD_HOST'] = 'pa55w04d@'
64 with mock.patch('os.path.exists', return_value=False):
65 client = musicpd.MPDClient()
66 self.assertEqual(client.pwd, 'pa55w04d')
67 self.assertEqual(client.host, 'localhost')
69 # Test badly formatted MPD_HOST
70 os.environ['MPD_HOST'] = '@'
71 with mock.patch('os.path.exists', return_value=False):
72 client = musicpd.MPDClient()
73 self.assertEqual(client.pwd, None)
74 self.assertEqual(client.host, 'localhost')
76 # Test unix socket extraction
77 os.environ['MPD_HOST'] = 'pa55w04d@/unix/sock'
78 client = musicpd.MPDClient()
79 self.assertEqual(client.host, '/unix/sock')
81 # Test plain abstract socket extraction
82 os.environ['MPD_HOST'] = '@abstract'
83 client = musicpd.MPDClient()
84 self.assertEqual(client.host, '@abstract')
86 # Test password and abstract socket extraction
87 os.environ['MPD_HOST'] = 'pass@@abstract'
88 client = musicpd.MPDClient()
89 self.assertEqual(client.pwd, 'pass')
90 self.assertEqual(client.host, '@abstract')
92 # Test unix socket fallback
93 os.environ.pop('MPD_HOST', None)
94 os.environ.pop('MPD_PORT', None)
95 os.environ.pop('XDG_RUNTIME_DIR', None)
96 with mock.patch('os.path.exists', return_value=True):
97 client = musicpd.MPDClient()
98 self.assertEqual(client.host, '/run/mpd/socket')
99 os.environ['XDG_RUNTIME_DIR'] = '/run/user/1000'
100 client = musicpd.MPDClient()
101 self.assertEqual(client.host, '/run/user/1000/mpd/socket')
103 os.environ.pop('MPD_HOST', None)
104 os.environ.pop('MPD_PORT', None)
105 os.environ['XDG_RUNTIME_DIR'] = '/run/user/1000/'
106 with mock.patch('os.path.exists', return_value=True):
107 client = musicpd.MPDClient()
108 self.assertEqual(client.host, '/run/user/1000/mpd/socket')
111 os.environ.pop('MPD_TIMEOUT', None)
112 client = musicpd.MPDClient()
113 self.assertEqual(client.mpd_timeout, musicpd.CONNECTION_TIMEOUT)
114 os.environ['MPD_TIMEOUT'] = 'garbage'
115 client = musicpd.MPDClient()
116 self.assertEqual(client.mpd_timeout,
117 musicpd.CONNECTION_TIMEOUT,
118 'Garbage is silently ignore to use default value')
119 os.environ['MPD_TIMEOUT'] = '42'
120 client = musicpd.MPDClient()
121 self.assertEqual(client.mpd_timeout, 42)
124 class TestMPDClient(unittest.TestCase):
127 # last sync: musicpd 0.6.0 unreleased / Fri Feb 19 15:34:53 CET 2021
130 'clearerror': 'nothing',
131 'currentsong': 'object',
136 # Playback Option Commands
137 'consume': 'nothing',
138 'crossfade': 'nothing',
139 'mixrampdb': 'nothing',
140 'mixrampdelay': 'nothing',
146 'replay_gain_mode': 'nothing',
147 'replay_gain_status': 'item',
149 # Playback Control Commands
154 'previous': 'nothing',
157 'seekcur': 'nothing',
164 'deleteid': 'nothing',
167 'playlist': 'playlist',
168 'playlistfind': 'songs',
169 'playlistid': 'songs',
170 'playlistinfo': 'songs',
171 'playlistsearch': 'songs',
172 'plchanges': 'songs',
173 'plchangesposid': 'changes',
176 'rangeid': 'nothing',
177 'shuffle': 'nothing',
180 'addtagid': 'nothing',
181 'cleartagid': 'nothing',
182 # Stored Playlist Commands
183 'listplaylist': 'list',
184 'listplaylistinfo': 'songs',
185 'listplaylists': 'playlists',
187 'playlistadd': 'nothing',
188 'playlistclear': 'nothing',
189 'playlistdelete': 'nothing',
190 'playlistmove': 'nothing',
195 'albumart': 'composite',
197 'getfingerprint': 'object',
199 'findadd': 'nothing',
201 'listall': 'database',
202 'listallinfo': 'database',
203 'listfiles': 'database',
204 'lsinfo': 'database',
205 'readcomments': 'object',
206 'readpicture': 'composite',
208 'searchadd': 'nothing',
209 'searchaddpl': 'nothing',
212 # Mounts and neighbors
214 'unmount': 'nothing',
215 'listmounts': 'mounts',
216 'listneighbors': 'neighbors',
218 'sticker get': 'item',
219 'sticker set': 'nothing',
220 'sticker delete': 'nothing',
221 'sticker list': 'list',
222 'sticker find': 'songs',
223 # Connection Commands
226 'password': 'nothing',
228 'binarylimit': 'nothing',
230 'tagtypes disable': 'nothing',
231 'tagtypes enable': 'nothing',
232 'tagtypes clear': 'nothing',
233 'tagtypes all': 'nothing',
235 'partition': 'nothing',
236 'listpartitions': 'list',
237 'newpartition': 'nothing',
238 'delpartition': 'nothing',
239 'moveoutput': 'nothing',
240 # Audio Output Commands
241 'disableoutput': 'nothing',
242 'enableoutput': 'nothing',
243 'toggleoutput': 'nothing',
244 'outputs': 'outputs',
245 'outputset': 'nothing',
246 # Reflection Commands
249 'notcommands': 'list',
250 'urlhandlers': 'list',
251 'decoders': 'plugins',
253 'subscribe': 'nothing',
254 'unsubscribe': 'nothing',
256 'readmessages': 'messages',
257 'sendmessage': 'nothing',
261 self.socket_patch = mock.patch('musicpd.socket')
262 self.socket_mock = self.socket_patch.start()
263 self.socket_mock.getaddrinfo.return_value = [range(5)]
265 self.socket_mock.socket.side_effect = (
267 # Create a new socket.socket() mock with default attributes,
268 # each time we are calling it back (otherwise, it keeps set
269 # attributes across calls).
270 # That's probably what we want, since reconnecting is like
271 # reinitializing the entire connection, and so, the mock.
272 mock.MagicMock(name='socket.socket'))
274 self.client = musicpd.MPDClient()
275 self.client.connect(TEST_MPD_HOST, TEST_MPD_PORT)
276 self.client._sock.reset_mock()
277 self.MPDWillReturn('ACK don\'t forget to setup your mock\n')
280 self.socket_patch.stop()
282 def MPDWillReturn(self, *lines):
283 # Return what the caller wants first, then do as if the socket was
285 self.client._rfile.readline.side_effect = itertools.chain(
286 lines, itertools.repeat(''))
288 def MPDWillReturnBinary(self, lines):
289 data = bytearray(b''.join(lines))
297 val.append(data.pop(0))
302 while not val.endswith(b'\x0a'):
303 val.append(data.pop(0))
305 self.client._rbfile.readline.side_effect = readline
306 self.client._rbfile.read.side_effect = read
308 def assertMPDReceived(self, *lines):
309 self.client._wfile.write.assert_called_with(*lines)
311 def test_metaclass_commands(self):
312 """Controls client has at least commands as last synchronized in
313 TestMPDClient.commands"""
314 for cmd, ret in TestMPDClient.commands.items():
315 self.assertTrue(hasattr(self.client, cmd), msg='cmd "{}" not available!'.format(cmd))
317 self.assertTrue(hasattr(self.client, cmd.replace(' ', '_')))
319 def test_fetch_nothing(self):
320 self.MPDWillReturn('OK\n')
321 self.assertIsNone(self.client.ping())
322 self.assertMPDReceived('ping\n')
324 def test_fetch_list(self):
325 self.MPDWillReturn('OK\n')
326 self.assertIsInstance(self.client.list('album'), list)
327 self.assertMPDReceived('list "album"\n')
329 def test_fetch_item(self):
330 self.MPDWillReturn('updating_db: 42\n', 'OK\n')
331 self.assertIsNotNone(self.client.update())
333 def test_fetch_object(self):
334 # XXX: _read_objects() doesn't wait for the final OK
335 self.MPDWillReturn('volume: 63\n', 'OK\n')
336 status = self.client.status()
337 self.assertMPDReceived('status\n')
338 self.assertIsInstance(status, dict)
340 # XXX: _read_objects() doesn't wait for the final OK
341 self.MPDWillReturn('OK\n')
342 stats = self.client.stats()
343 self.assertMPDReceived('stats\n')
344 self.assertIsInstance(stats, dict)
346 output = ['outputid: 0\n',
347 'outputname: default detected output\n',
349 'outputenabled: 1\n']
350 self.MPDWillReturn(*output, 'OK\n')
351 outputs = self.client.outputs()
352 self.assertMPDReceived('outputs\n')
353 self.assertIsInstance(outputs, list)
354 self.assertEqual([{'outputid': '0', 'outputname': 'default detected output', 'plugin': 'sndio', 'outputenabled': '1'}], outputs)
356 def test_fetch_songs(self):
357 self.MPDWillReturn('file: my-song.ogg\n', 'Pos: 0\n', 'Id: 66\n', 'OK\n')
358 playlist = self.client.playlistinfo()
360 self.assertMPDReceived('playlistinfo\n')
361 self.assertIsInstance(playlist, list)
362 self.assertEqual(1, len(playlist))
364 self.assertIsInstance(e, dict)
365 self.assertEqual('my-song.ogg', e['file'])
366 self.assertEqual('0', e['pos'])
367 self.assertEqual('66', e['id'])
369 def test_send_and_fetch(self):
370 self.MPDWillReturn('volume: 50\n', 'OK\n')
371 result = self.client.send_status()
372 self.assertEqual(None, result)
373 self.assertMPDReceived('status\n')
375 status = self.client.fetch_status()
376 self.assertEqual(1, self.client._wfile.write.call_count)
377 self.assertEqual({'volume': '50'}, status)
379 def test_iterating(self):
380 self.MPDWillReturn('file: my-song.ogg\n', 'Pos: 0\n', 'Id: 66\n',
381 'file: my-song.ogg\n', 'Pos: 0\n', 'Id: 66\n', 'OK\n')
382 self.client.iterate = True
383 playlist = self.client.playlistinfo()
384 self.assertMPDReceived('playlistinfo\n')
385 self.assertIsInstance(playlist, types.GeneratorType)
386 self.assertTrue(self.client._iterating)
387 for song in playlist:
388 self.assertRaises(musicpd.IteratingError, self.client.status)
389 self.assertIsInstance(song, dict)
390 self.assertEqual('my-song.ogg', song['file'])
391 self.assertEqual('0', song['pos'])
392 self.assertEqual('66', song['id'])
393 self.assertFalse(self.client._iterating)
395 def test_noidle(self):
396 self.MPDWillReturn('OK\n') # nothing changed after idle-ing
397 self.client.send_idle()
398 self.MPDWillReturn('OK\n') # nothing changed after noidle
399 self.assertEqual(self.client.noidle(), [])
400 self.assertMPDReceived('noidle\n')
401 self.MPDWillReturn('volume: 50\n', 'OK\n')
403 self.assertMPDReceived('status\n')
405 def test_noidle_while_idle_started_sending(self):
406 self.MPDWillReturn('OK\n') # nothing changed after idle
407 self.client.send_idle()
408 self.MPDWillReturn('CHANGED: player\n', 'OK\n') # noidle response
409 self.assertEqual(self.client.noidle(), ['player'])
410 self.MPDWillReturn('volume: 50\n', 'OK\n')
411 status = self.client.status()
412 self.assertMPDReceived('status\n')
413 self.assertEqual({'volume': '50'}, status)
415 def test_throw_when_calling_noidle_withoutidling(self):
416 self.assertRaises(musicpd.CommandError, self.client.noidle)
417 self.client.send_status()
418 self.assertRaises(musicpd.CommandError, self.client.noidle)
420 def test_client_to_client(self):
421 self.MPDWillReturn('OK\n')
422 self.assertIsNone(self.client.subscribe("monty"))
423 self.assertMPDReceived('subscribe "monty"\n')
425 self.MPDWillReturn('channel: monty\n', 'OK\n')
426 channels = self.client.channels()
427 self.assertMPDReceived('channels\n')
428 self.assertEqual(['monty'], channels)
430 self.MPDWillReturn('OK\n')
431 self.assertIsNone(self.client.sendmessage('monty', 'SPAM'))
432 self.assertMPDReceived('sendmessage "monty" "SPAM"\n')
434 self.MPDWillReturn('channel: monty\n', 'message: SPAM\n', 'OK\n')
435 msg = self.client.readmessages()
436 self.assertMPDReceived('readmessages\n')
437 self.assertEqual(msg, [{'channel': 'monty', 'message': 'SPAM'}])
439 self.MPDWillReturn('OK\n')
440 self.assertIsNone(self.client.unsubscribe('monty'))
441 self.assertMPDReceived('unsubscribe "monty"\n')
443 self.MPDWillReturn('OK\n')
444 channels = self.client.channels()
445 self.assertMPDReceived('channels\n')
446 self.assertEqual([], channels)
448 def test_ranges_in_command_args(self):
449 self.MPDWillReturn('OK\n')
450 self.client.playlistinfo((10,))
451 self.assertMPDReceived('playlistinfo 10:\n')
453 self.MPDWillReturn('OK\n')
454 self.client.playlistinfo(('10',))
455 self.assertMPDReceived('playlistinfo 10:\n')
457 self.MPDWillReturn('OK\n')
458 self.client.playlistinfo((10, 12))
459 self.assertMPDReceived('playlistinfo 10:12\n')
461 self.MPDWillReturn('OK\n')
462 self.client.rangeid(())
463 self.assertMPDReceived('rangeid :\n')
465 for arg in [(10, 't'), (10, 1, 1), (None,1)]:
466 self.MPDWillReturn('OK\n')
467 with self.assertRaises(musicpd.CommandError):
468 self.client.playlistinfo(arg)
470 def test_numbers_as_command_args(self):
471 self.MPDWillReturn('OK\n')
472 self.client.find('file', 1)
473 self.assertMPDReceived('find "file" "1"\n')
475 def test_commands_without_callbacks(self):
476 self.MPDWillReturn('\n')
478 self.assertMPDReceived('close\n')
480 # XXX: what are we testing here?
482 self.client.connect(TEST_MPD_HOST, TEST_MPD_PORT)
484 def test_connection_lost(self):
485 # Simulate a connection lost: the socket returns empty strings
486 self.MPDWillReturn('')
488 with self.assertRaises(musicpd.ConnectionError):
491 # consistent behaviour
492 # solves <https://github.com/Mic92/python-mpd2/issues/11> (also present
494 with self.assertRaises(musicpd.ConnectionError):
497 self.assertIs(self.client._sock, None)
499 def test_parse_sticker_get_no_sticker(self):
500 self.MPDWillReturn('ACK [50@0] {sticker} no such sticker\n')
501 self.assertRaises(musicpd.CommandError,
502 self.client.sticker_get, 'song', 'baz', 'foo')
504 def test_parse_sticker_list(self):
505 self.MPDWillReturn('sticker: foo=bar\n', 'sticker: lom=bok\n', 'OK\n')
506 res = self.client.sticker_list('song', 'baz')
507 self.assertEqual(['foo=bar', 'lom=bok'], res)
509 # Even with only one sticker, we get a dict
510 self.MPDWillReturn('sticker: foo=bar\n', 'OK\n')
511 res = self.client.sticker_list('song', 'baz')
512 self.assertEqual(['foo=bar'], res)
514 def test_albumart(self):
515 # here is a 34 bytes long data
516 data = bytes('\xff\xd8\xff\xe0\x00\x10JFIF\x00\x01\x01\x00\x00\x01'
517 '\x00\x01\x00\x00\xff\xdb\x00C\x00\x05\x03\x04',
519 read_lines = [b'size: 42\nbinary: 34\n', data, b'\nOK\n']
520 self.MPDWillReturnBinary(read_lines)
521 # Reading albumart / offset 0 should return the data
522 res = self.client.albumart('muse/Raised Fist/2002-Dedication/', 0)
523 self.assertEqual(res.get('data'), data)
525 def test_reading_binary(self):
526 # readpicture when there are no picture returns empty object
527 self.MPDWillReturnBinary([b'OK\n'])
528 res = self.client.readpicture('muse/Raised Fist/2002-Dedication/', 0)
529 self.assertEqual(res, {})
531 def test_command_list(self):
532 self.MPDWillReturn('updating_db: 42\n',
539 self.client.command_list_ok_begin()
542 self.client.repeat(1)
543 self.client.command_list_end()
544 self.assertMPDReceived('command_list_end\n')
546 def test_two_word_commands(self):
547 self.MPDWillReturn('OK\n')
548 self.client.tagtypes_clear()
549 self.assertMPDReceived('tagtypes clear\n')
550 self.MPDWillReturn('OK\n')
551 with self.assertRaises(AttributeError):
552 self.client.foo_bar()
554 class testConnection(unittest.TestCase):
556 def test_exposing_fileno(self):
557 with mock.patch('musicpd.socket') as socket_mock:
558 sock = mock.MagicMock(name='socket')
559 socket_mock.socket.return_value = sock
560 cli = musicpd.MPDClient()
563 cli._sock.fileno.assert_called_with()
565 def test_connect_abstract(self):
566 os.environ['MPD_HOST'] = '@abstract'
567 with mock.patch('musicpd.socket') as socket_mock:
568 sock = mock.MagicMock(name='socket')
569 socket_mock.socket.return_value = sock
570 cli = musicpd.MPDClient()
572 sock.connect.assert_called_with('\0abstract')
574 def test_connect_unix(self):
575 os.environ['MPD_HOST'] = '/run/mpd/socket'
576 with mock.patch('musicpd.socket') as socket_mock:
577 sock = mock.MagicMock(name='socket')
578 socket_mock.socket.return_value = sock
579 cli = musicpd.MPDClient()
581 sock.connect.assert_called_with('/run/mpd/socket')
584 class testException(unittest.TestCase):
586 def test_CommandError_on_newline(self):
587 os.environ['MPD_HOST'] = '/run/mpd/socket'
588 with mock.patch('musicpd.socket') as socket_mock:
589 sock = mock.MagicMock(name='socket')
590 socket_mock.socket.return_value = sock
591 cli = musicpd.MPDClient()
593 with self.assertRaises(musicpd.CommandError):
594 cli.find('(album == "foo\nbar")')
596 class testContextManager(unittest.TestCase):
598 def test_enter_exit(self):
599 os.environ['MPD_HOST'] = '@abstract'
600 with mock.patch('musicpd.socket') as socket_mock:
601 sock = mock.MagicMock(name='socket')
602 socket_mock.socket.return_value = sock
603 cli = musicpd.MPDClient()
605 sock.connect.assert_called_with('\0abstract')
606 sock.close.assert_not_called()
607 sock.close.assert_called()
609 if __name__ == '__main__':