From a5bbe574908438b722f0bd0a9853f3cd7e84bf72 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C5=A0ar=C5=ABnas=20Nejus?= Date: Tue, 15 Jul 2025 09:12:17 +0100 Subject: [PATCH] Fix types in test_player --- beetsplug/bpd/__init__.py | 39 +++++++++++---------------- test/plugins/test_player.py | 53 +++++++------------------------------ 2 files changed, 24 insertions(+), 68 deletions(-) diff --git a/beetsplug/bpd/__init__.py b/beetsplug/bpd/__init__.py index 435368e35..a2ad2835c 100644 --- a/beetsplug/bpd/__init__.py +++ b/beetsplug/bpd/__init__.py @@ -30,7 +30,7 @@ from typing import TYPE_CHECKING import beets import beets.ui -from beets import dbcore, vfs +from beets import dbcore, logging, vfs from beets.library import Item from beets.plugins import BeetsPlugin from beets.util import as_string, bluelet @@ -38,6 +38,17 @@ from beets.util import as_string, bluelet if TYPE_CHECKING: from beets.dbcore.query import Query +log = logging.getLogger(__name__) + + +try: + from . import gstplayer +except ImportError as e: + raise ImportError( + "Gstreamer Python bindings not found." + ' Install "gstreamer1.0" and "python-gi" or similar package to use BPD.' + ) from e + PROTOCOL_VERSION = "0.16.0" BUFSIZE = 1024 @@ -94,11 +105,6 @@ SUBSYSTEMS = [ ] -# Gstreamer import error. -class NoGstreamerError(Exception): - pass - - # Error-handling, exceptions, parameter parsing. @@ -1099,14 +1105,6 @@ class Server(BaseServer): """ def __init__(self, library, host, port, password, ctrl_port, log): - try: - from beetsplug.bpd import gstplayer - except ImportError as e: - # This is a little hacky, but it's the best I know for now. - if e.args[0].endswith(" gst"): - raise NoGstreamerError() - else: - raise log.info("Starting server...") super().__init__(host, port, password, ctrl_port, log) self.lib = library @@ -1616,16 +1614,9 @@ class BPDPlugin(BeetsPlugin): def start_bpd(self, lib, host, port, password, volume, ctrl_port): """Starts a BPD server.""" - try: - server = Server(lib, host, port, password, ctrl_port, self._log) - server.cmd_setvol(None, volume) - server.run() - except NoGstreamerError: - self._log.error("Gstreamer Python bindings not found.") - self._log.error( - 'Install "gstreamer1.0" and "python-gi"' - "or similar package to use BPD." - ) + server = Server(lib, host, port, password, ctrl_port, self._log) + server.cmd_setvol(None, volume) + server.run() def commands(self): cmd = beets.ui.Subcommand( diff --git a/test/plugins/test_player.py b/test/plugins/test_player.py index a7c613d8f..16e424d7e 100644 --- a/test/plugins/test_player.py +++ b/test/plugins/test_player.py @@ -14,19 +14,15 @@ """Tests for BPD's implementation of the MPD protocol.""" -import importlib.util import multiprocessing as mp import os import socket -import sys import tempfile import threading import time import unittest from contextlib import contextmanager - -# Mock GstPlayer so that the forked process doesn't attempt to import gi: -from unittest import mock +from unittest.mock import MagicMock, patch import confuse import pytest @@ -34,43 +30,8 @@ import yaml from beets.test.helper import PluginTestCase from beets.util import bluelet -from beetsplug import bpd -gstplayer = importlib.util.module_from_spec( - importlib.util.find_spec("beetsplug.bpd.gstplayer") -) - - -def _gstplayer_play(*_): - bpd.gstplayer._GstPlayer.playing = True - return mock.DEFAULT - - -gstplayer._GstPlayer = mock.MagicMock( - spec_set=[ - "time", - "volume", - "playing", - "run", - "play_file", - "pause", - "stop", - "seek", - "play", - "get_decoders", - ], - **{ - "playing": False, - "volume": 0, - "time.return_value": (0, 0), - "play_file.side_effect": _gstplayer_play, - "play.side_effect": _gstplayer_play, - "get_decoders.return_value": {"default": ({"audio/mpeg"}, {"mp3"})}, - }, -) -gstplayer.GstPlayer = lambda _: gstplayer._GstPlayer -sys.modules["beetsplug.bpd.gstplayer"] = gstplayer -bpd.gstplayer = gstplayer +bpd = pytest.importorskip("beetsplug.bpd") class CommandParseTest(unittest.TestCase): @@ -256,7 +217,7 @@ def implements(commands, fail=False): bluelet_listener = bluelet.Listener -@mock.patch("beets.util.bluelet.Listener") +@patch("beets.util.bluelet.Listener") def start_server(args, assigned_port, listener_patch): """Start the bpd server, writing the port to `assigned_port`.""" @@ -938,7 +899,7 @@ class BPDPlaylistsTest(BPDTestHelper): response = client.send_command("load", "anything") self._assert_failed(response, bpd.ERROR_NO_EXIST) - @unittest.skip + @unittest.expectedFailure def test_cmd_playlistadd(self): with self.run_bpd() as client: self._bpd_add(client, self.item1, playlist="anything") @@ -1128,7 +1089,7 @@ class BPDConnectionTest(BPDTestHelper): self._assert_ok(response) assert self.TAGTYPES == set(response.data["tagtype"]) - @unittest.skip + @unittest.expectedFailure def test_tagtypes_mask(self): with self.run_bpd() as client: response = client.send_command("tagtypes", "clear") @@ -1169,6 +1130,10 @@ class BPDReflectionTest(BPDTestHelper): fail=True, ) + @patch( + "beetsplug.bpd.gstplayer.GstPlayer.get_decoders", + MagicMock(return_value={"default": ({"audio/mpeg"}, {"mp3"})}), + ) def test_cmd_decoders(self): with self.run_bpd() as client: response = client.send_command("decoders")