#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
+# coding: utf-8
+# SPDX-FileCopyrightText: 2012-2024 kaliko <kaliko@azylum.org>
+# SPDX-License-Identifier: LGPL-3.0-or-later
# pylint: disable=missing-docstring
"""
Test suite highly borrowed^Wsteal from python-mpd2 [0] project.
import itertools
import os
-import sys
import types
import unittest
+import unittest.mock
import warnings
import musicpd
-try:
- import unittest.mock as mock
-except ImportError:
- try:
- import mock
- except ImportError:
- print("Please install mock from PyPI to run tests!")
- sys.exit(1)
+mock = unittest.mock
# show deprecation warnings
warnings.simplefilter('default')
TEST_MPD_HOST, TEST_MPD_PORT = ('example.com', 10000)
-class testEnvVar(unittest.TestCase):
+class TestEnvVar(unittest.TestCase):
def test_envvar(self):
# mock "os.path.exists" here to ensure there are no socket in
self.assertEqual(client.pwd, 'pa55w04d')
self.assertEqual(client.host, 'example.org')
+ # Test host alone
+ os.environ['MPD_HOST'] = 'example.org'
+ client = musicpd.MPDClient()
+ self.assertFalse(client.pwd)
+ self.assertEqual(client.host, 'example.org')
+
# Test password extraction (no host)
os.environ['MPD_HOST'] = 'pa55w04d@'
with mock.patch('os.path.exists', return_value=False):
self.assertEqual(client.pwd, 'pa55w04d')
self.assertEqual(client.host, 'localhost')
+ # Test badly formatted MPD_HOST
+ os.environ['MPD_HOST'] = '@'
+ with mock.patch('os.path.exists', return_value=False):
+ client = musicpd.MPDClient()
+ self.assertEqual(client.pwd, None)
+ self.assertEqual(client.host, 'localhost')
+
# Test unix socket extraction
os.environ['MPD_HOST'] = 'pa55w04d@/unix/sock'
client = musicpd.MPDClient()
client = musicpd.MPDClient()
self.assertEqual(client.mpd_timeout,
musicpd.CONNECTION_TIMEOUT,
- 'Garbage\'s not silently ignore to use default value')
+ 'Garbage is silently ignore to use default value')
os.environ['MPD_TIMEOUT'] = '42'
client = musicpd.MPDClient()
self.assertEqual(client.mpd_timeout, 42)
class TestMPDClient(unittest.TestCase):
longMessage = True
- # last sync: musicpd 0.4.2 unreleased / Mon Nov 17 21:45:22 CET 2014
+ # last sync: musicpd 0.6.0 unreleased / Fri Feb 19 15:34:53 CET 2021
commands = {
# Status Commands
'clearerror': 'nothing',
'random': 'nothing',
'repeat': 'nothing',
'setvol': 'nothing',
+ 'getvol': 'object',
'single': 'nothing',
'replay_gain_mode': 'nothing',
'replay_gain_status': 'item',
'seekid': 'nothing',
'seekcur': 'nothing',
'stop': 'nothing',
- # Playlist Commands
+ # Queue Commands
'add': 'nothing',
'addid': 'item',
'clear': 'nothing',
'rm': 'nothing',
'save': 'nothing',
# Database Commands
+ 'albumart': 'composite',
'count': 'object',
+ 'getfingerprint': 'object',
'find': 'songs',
'findadd': 'nothing',
'list': 'list',
'listall': 'database',
'listallinfo': 'database',
+ 'listfiles': 'database',
'lsinfo': 'database',
+ 'readcomments': 'object',
+ 'readpicture': 'composite',
'search': 'songs',
'searchadd': 'nothing',
'searchaddpl': 'nothing',
'update': 'item',
'rescan': 'item',
- 'readcomments': 'object',
# Mounts and neighbors
'mount': 'nothing',
'unmount': 'nothing',
'kill': None,
'password': 'nothing',
'ping': 'nothing',
+ 'binarylimit': 'nothing',
+ 'tagtypes': 'list',
+ 'tagtypes disable': 'nothing',
+ 'tagtypes enable': 'nothing',
+ 'tagtypes clear': 'nothing',
+ 'tagtypes all': 'nothing',
# Partition Commands
'partition': 'nothing',
'listpartitions': 'list',
'newpartition': 'nothing',
+ 'delpartition': 'nothing',
+ 'moveoutput': 'nothing',
# Audio Output Commands
'disableoutput': 'nothing',
'enableoutput': 'nothing',
'toggleoutput': 'nothing',
'outputs': 'outputs',
+ 'outputset': 'nothing',
# Reflection Commands
'config': 'object',
'commands': 'list',
'notcommands': 'list',
- 'tagtypes': 'list',
'urlhandlers': 'list',
'decoders': 'plugins',
# Client to Client
self.assertMPDReceived('stats\n')
self.assertIsInstance(stats, dict)
+ output = ['outputid: 0\n',
+ 'outputname: default detected output\n',
+ 'plugin: sndio\n',
+ 'outputenabled: 1\n']
+ self.MPDWillReturn(*output, 'OK\n')
+ outputs = self.client.outputs()
+ self.assertMPDReceived('outputs\n')
+ self.assertIsInstance(outputs, list)
+ self.assertEqual([{'outputid': '0', 'outputname': 'default detected output', 'plugin': 'sndio', 'outputenabled': '1'}], outputs)
+
def test_fetch_songs(self):
self.MPDWillReturn('file: my-song.ogg\n', 'Pos: 0\n', 'Id: 66\n', 'OK\n')
playlist = self.client.playlistinfo()
self.client.send_status()
self.assertRaises(musicpd.CommandError, self.client.noidle)
- def test_client_to_client(self):
- # client to client is at this time in beta!
+ def test_send_noidle_calls_noidle(self):
+ self.MPDWillReturn('OK\n') # nothing changed after idle
+ self.client.send_idle()
+ self.client.send_noidle()
+ self.assertMPDReceived('noidle\n')
+ def test_client_to_client(self):
self.MPDWillReturn('OK\n')
self.assertIsNone(self.client.subscribe("monty"))
self.assertMPDReceived('subscribe "monty"\n')
res = self.client.albumart('muse/Raised Fist/2002-Dedication/', 0)
self.assertEqual(res.get('data'), data)
+ def test_reading_binary(self):
+ # readpicture when there are no picture returns empty object
+ self.MPDWillReturnBinary([b'OK\n'])
+ res = self.client.readpicture('muse/Raised Fist/2002-Dedication/', 0)
+ self.assertEqual(res, {})
+
+ def test_command_list(self):
+ self.MPDWillReturn('updating_db: 42\n',
+ f'{musicpd.NEXT}\n',
+ 'repeat: 0\n',
+ 'random: 0\n',
+ f'{musicpd.NEXT}\n',
+ f'{musicpd.NEXT}\n',
+ 'OK\n')
+ self.client.command_list_ok_begin()
+ self.client.update()
+ self.client.status()
+ self.client.repeat(1)
+ self.client.command_list_end()
+ self.assertMPDReceived('command_list_end\n')
+
+ def test_two_word_commands(self):
+ self.MPDWillReturn('OK\n')
+ self.client.tagtypes_clear()
+ self.assertMPDReceived('tagtypes clear\n')
+ self.MPDWillReturn('OK\n')
+ with self.assertRaises(AttributeError):
+ self.client.foo_bar()
+
+class TestConnection(unittest.TestCase):
+
+ def test_exposing_fileno(self):
+ with mock.patch('musicpd.socket') as socket_mock:
+ sock = mock.MagicMock(name='socket')
+ socket_mock.socket.return_value = sock
+ cli = musicpd.MPDClient()
+ cli.connect()
+ cli.fileno()
+ cli._sock.fileno.assert_called_with()
+
+ def test_connect_abstract(self):
+ os.environ['MPD_HOST'] = '@abstract'
+ with mock.patch('musicpd.socket') as socket_mock:
+ sock = mock.MagicMock(name='socket')
+ socket_mock.socket.return_value = sock
+ cli = musicpd.MPDClient()
+ cli.connect()
+ sock.connect.assert_called_with('\0abstract')
+
+ def test_connect_unix(self):
+ os.environ['MPD_HOST'] = '/run/mpd/socket'
+ with mock.patch('musicpd.socket') as socket_mock:
+ sock = mock.MagicMock(name='socket')
+ socket_mock.socket.return_value = sock
+ cli = musicpd.MPDClient()
+ cli.connect()
+ sock.connect.assert_called_with('/run/mpd/socket')
+
+ def test_sockettimeout(self):
+ with mock.patch('musicpd.socket') as socket_mock:
+ sock = mock.MagicMock(name='socket')
+ socket_mock.socket.return_value = sock
+ cli = musicpd.MPDClient()
+ # Default is no socket timeout
+ cli.connect()
+ sock.settimeout.assert_called_with(None)
+ cli.disconnect()
+ # set a socket timeout before connection
+ cli.socket_timeout = 10
+ cli.connect()
+ sock.settimeout.assert_called_with(10)
+ # Set socket timeout while already connected
+ cli.socket_timeout = 42
+ sock.settimeout.assert_called_with(42)
+ # set a socket timeout using str
+ cli.socket_timeout = '10'
+ sock.settimeout.assert_called_with(10)
+ # Set socket timeout to None
+ cli.socket_timeout = None
+ sock.settimeout.assert_called_with(None)
+ # Set socket timeout Raises Exception
+ with self.assertRaises(ValueError):
+ cli.socket_timeout = 'foo'
+ with self.assertRaises(ValueError,
+ msg='socket_timeout expects a non zero positive integer'):
+ cli.socket_timeout = '0'
+ with self.assertRaises(ValueError,
+ msg='socket_timeout expects a non zero positive integer'):
+ cli.socket_timeout = '-1'
+
+
+class TestConnectionError(unittest.TestCase):
+
+ @mock.patch('socket.socket')
+ def test_connect_unix(self, socket_mock):
+ """Unix socket socket.error should raise a musicpd.ConnectionError"""
+ mocked_socket = socket_mock.return_value
+ mocked_socket.connect.side_effect = musicpd.socket.error(42, 'err 42')
+ os.environ['MPD_HOST'] = '/run/mpd/socket'
+ cli = musicpd.MPDClient()
+ with self.assertRaises(musicpd.ConnectionError) as cme:
+ cli.connect()
+ self.assertEqual('[Errno 42] err 42', str(cme.exception))
+
+ def test_non_available_unix_socket(self):
+ delattr(musicpd.socket, 'AF_UNIX')
+ os.environ['MPD_HOST'] = '/run/mpd/socket'
+ cli = musicpd.MPDClient()
+ with self.assertRaises(musicpd.ConnectionError) as cme:
+ cli.connect()
+ self.assertEqual('Unix domain sockets not supported on this platform',
+ str(cme.exception))
+
+ @mock.patch('socket.getaddrinfo')
+ def test_connect_tcp_getaddrinfo(self, gai_mock):
+ """TCP socket.gaierror should raise a musicpd.ConnectionError"""
+ gai_mock.side_effect = musicpd.socket.error(42, 'gaierr 42')
+ cli = musicpd.MPDClient()
+ with self.assertRaises(musicpd.ConnectionError) as cme:
+ cli.connect(host=TEST_MPD_HOST)
+ self.assertEqual('[Errno 42] gaierr 42', str(cme.exception))
+
+ @mock.patch('socket.getaddrinfo')
+ @mock.patch('socket.socket')
+ def test_connect_tcp_connect(self, socket_mock, gai_mock):
+ """A socket.error should raise a musicpd.ConnectionError
+ Mocking getaddrinfo to prevent network access (DNS)
+ """
+ gai_mock.return_value = [range(5)]
+ mocked_socket = socket_mock.return_value
+ mocked_socket.connect.side_effect = musicpd.socket.error(42, 'tcp conn err 42')
+ cli = musicpd.MPDClient()
+ with self.assertRaises(musicpd.ConnectionError) as cme:
+ cli.connect(host=TEST_MPD_HOST)
+ self.assertEqual('[Errno 42] tcp conn err 42', str(cme.exception))
+
+ @mock.patch('socket.getaddrinfo')
+ def test_connect_tcp_connect_empty_gai(self, gai_mock):
+ """An empty getaddrinfo should raise a musicpd.ConnectionError"""
+ gai_mock.return_value = []
+ cli = musicpd.MPDClient()
+ with self.assertRaises(musicpd.ConnectionError) as cme:
+ cli.connect(host=TEST_MPD_HOST)
+ self.assertEqual('getaddrinfo returns an empty list', str(cme.exception))
+
+
+class TestCommandErrorException(unittest.TestCase):
+
+ def test_error_on_newline(self):
+ os.environ['MPD_HOST'] = '/run/mpd/socket'
+ with mock.patch('musicpd.socket') as socket_mock:
+ sock = mock.MagicMock(name='socket')
+ socket_mock.socket.return_value = sock
+ cli = musicpd.MPDClient()
+ cli.connect()
+ with self.assertRaises(musicpd.CommandError):
+ cli.find('(album == "foo\nbar")')
+
+class testContextManager(unittest.TestCase):
+
+ def test_enter_exit(self):
+ os.environ['MPD_HOST'] = '@abstract'
+ with mock.patch('musicpd.socket') as socket_mock:
+ sock = mock.MagicMock(name='socket')
+ socket_mock.socket.return_value = sock
+ cli = musicpd.MPDClient()
+ with cli as c:
+ sock.connect.assert_called_with('\0abstract')
+ sock.close.assert_not_called()
+ sock.close.assert_called()
+
+class testRange(unittest.TestCase):
+
+ def test_range(self):
+ tests = [
+ ((), ':'),
+ ((None,None), ':'),
+ (('',''), ':'),
+ (('',), ':'),
+ ((42,42), '42:42'),
+ ((42,), '42:'),
+ (('42',), '42:'),
+ (('42',None), '42:'),
+ (('42',''), '42:'),
+ ]
+ for tpl, result in tests:
+ self.assertEqual(str(musicpd.Range(tpl)), result)
+ with self.assertRaises(musicpd.CommandError):
+ #CommandError: Integer expected to start the range: (None, 42)
+ musicpd.Range((None,'42'))
+ with self.assertRaises(musicpd.CommandError):
+ # CommandError: Not an integer: "foo"
+ musicpd.Range(('foo',))
+ with self.assertRaises(musicpd.CommandError):
+ # CommandError: Wrong range: 42 > 41
+ musicpd.Range(('42',41))
+ with self.assertRaises(musicpd.CommandError):
+ # CommandError: Wrong range: 42 > 41
+ musicpd.Range(('42','42','42'))
+
if __name__ == '__main__':
unittest.main()
home / dev / repositories / python-musicpd.git / blobdiff