X-Git-Url: http://git.kaliko.me/?p=python-musicpd.git;a=blobdiff_plain;f=test.py;h=7f1890fe9f325ec1b4c5c450ea8048d8b5727cca;hp=5922041619214989740a74aa7416da2a4d0a43b5;hb=HEAD;hpb=4b44d5f37ad29e54c356f7dcc97194eaabcecc03 diff --git a/test.py b/test.py index 5922041..299b54a 100755 --- a/test.py +++ b/test.py @@ -1,5 +1,8 @@ #!/usr/bin/env python3 -# -*- coding: utf-8 -*- +# coding: utf-8 +# SPDX-FileCopyrightText: 2012-2024 kaliko +# SPDX-License-Identifier: LGPL-3.0-or-later +# pylint: disable=missing-docstring """ Test suite highly borrowed^Wsteal from python-mpd2 [0] project. @@ -9,21 +12,14 @@ 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') @@ -32,27 +28,69 @@ warnings.simplefilter('default') TEST_MPD_HOST, TEST_MPD_PORT = ('example.com', 10000) -class testEnvVar(unittest.TestCase): +class TestEnvVar(unittest.TestCase): def test_envvar(self): - os.environ.pop('MPD_HOST', None) - os.environ.pop('MPD_PORT', None) - client = musicpd.MPDClient() - self.assertEqual(client.host, 'localhost') - self.assertEqual(client.port, '6600') + # mock "os.path.exists" here to ensure there are no socket in + # XDG_RUNTIME_DIR/mpd or /run/mpd since with test defaults fallbacks + # when : + # * neither MPD_HOST nor XDG_RUNTIME_DIR are not set + # * /run/mpd does not expose a socket + with mock.patch('os.path.exists', return_value=False): + os.environ.pop('MPD_HOST', None) + os.environ.pop('MPD_PORT', None) + client = musicpd.MPDClient() + self.assertEqual(client.host, 'localhost') + self.assertEqual(client.port, '6600') + + os.environ.pop('MPD_HOST', None) + os.environ['MPD_PORT'] = '6666' + client = musicpd.MPDClient() + self.assertEqual(client.pwd, None) + self.assertEqual(client.host, 'localhost') + self.assertEqual(client.port, '6666') + # Test password extraction os.environ['MPD_HOST'] = 'pa55w04d@example.org' client = musicpd.MPDClient() self.assertEqual(client.pwd, 'pa55w04d') self.assertEqual(client.host, 'example.org') - self.assertEqual(client.port, '6600') - os.environ.pop('MPD_HOST', None) - os.environ['MPD_PORT'] = '6666' + # 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): + client = musicpd.MPDClient() + 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') - self.assertEqual(client.port, '6666') + + # Test unix socket extraction + os.environ['MPD_HOST'] = 'pa55w04d@/unix/sock' + client = musicpd.MPDClient() + self.assertEqual(client.host, '/unix/sock') + + # Test plain abstract socket extraction + os.environ['MPD_HOST'] = '@abstract' + client = musicpd.MPDClient() + self.assertEqual(client.host, '@abstract') + + # Test password and abstract socket extraction + os.environ['MPD_HOST'] = 'pass@@abstract' + client = musicpd.MPDClient() + self.assertEqual(client.pwd, 'pass') + self.assertEqual(client.host, '@abstract') # Test unix socket fallback os.environ.pop('MPD_HOST', None) @@ -61,6 +99,9 @@ class testEnvVar(unittest.TestCase): with mock.patch('os.path.exists', return_value=True): client = musicpd.MPDClient() self.assertEqual(client.host, '/run/mpd/socket') + os.environ['XDG_RUNTIME_DIR'] = '/run/user/1000' + client = musicpd.MPDClient() + self.assertEqual(client.host, '/run/user/1000/mpd/socket') os.environ.pop('MPD_HOST', None) os.environ.pop('MPD_PORT', None) @@ -69,10 +110,24 @@ class testEnvVar(unittest.TestCase): client = musicpd.MPDClient() self.assertEqual(client.host, '/run/user/1000/mpd/socket') + # Test MPD_TIMEOUT + os.environ.pop('MPD_TIMEOUT', None) + client = musicpd.MPDClient() + self.assertEqual(client.mpd_timeout, musicpd.CONNECTION_TIMEOUT) + os.environ['MPD_TIMEOUT'] = 'garbage' + client = musicpd.MPDClient() + self.assertEqual(client.mpd_timeout, + musicpd.CONNECTION_TIMEOUT, + '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', @@ -89,6 +144,7 @@ class TestMPDClient(unittest.TestCase): 'random': 'nothing', 'repeat': 'nothing', 'setvol': 'nothing', + 'getvol': 'object', 'single': 'nothing', 'replay_gain_mode': 'nothing', 'replay_gain_status': 'item', @@ -103,7 +159,7 @@ class TestMPDClient(unittest.TestCase): 'seekid': 'nothing', 'seekcur': 'nothing', 'stop': 'nothing', - # Playlist Commands + # Queue Commands 'add': 'nothing', 'addid': 'item', 'clear': 'nothing', @@ -139,19 +195,23 @@ class TestMPDClient(unittest.TestCase): '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', @@ -168,16 +228,28 @@ class TestMPDClient(unittest.TestCase): '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 @@ -216,6 +288,26 @@ class TestMPDClient(unittest.TestCase): self.client._rfile.readline.side_effect = itertools.chain( lines, itertools.repeat('')) + def MPDWillReturnBinary(self, lines): + data = bytearray(b''.join(lines)) + + def read(amount): + val = bytearray() + while amount > 0: + amount -= 1 + # _ = data.pop(0) + # print(hex(_)) + val.append(data.pop(0)) + return val + + def readline(): + val = bytearray() + while not val.endswith(b'\x0a'): + val.append(data.pop(0)) + return val + self.client._rbfile.readline.side_effect = readline + self.client._rbfile.read.side_effect = read + def assertMPDReceived(self, *lines): self.client._wfile.write.assert_called_with(*lines) @@ -254,6 +346,16 @@ class TestMPDClient(unittest.TestCase): 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() @@ -318,9 +420,13 @@ class TestMPDClient(unittest.TestCase): 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') @@ -414,6 +520,217 @@ class TestMPDClient(unittest.TestCase): res = self.client.sticker_list('song', 'baz') self.assertEqual(['foo=bar'], res) + def test_albumart(self): + # here is a 34 bytes long data + data = bytes('\xff\xd8\xff\xe0\x00\x10JFIF\x00\x01\x01\x00\x00\x01' + '\x00\x01\x00\x00\xff\xdb\x00C\x00\x05\x03\x04', + encoding='utf8') + read_lines = [b'size: 42\nbinary: 34\n', data, b'\nOK\n'] + self.MPDWillReturnBinary(read_lines) + # Reading albumart / offset 0 should return the data + 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()