Fix types in test_player

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(

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")