beets/test/test_plugins.py

# This file is part of beets.
# Copyright 2016, Thomas Scholtes.
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.


import importlib
import importlib.util
import itertools
import logging
import os
import pkgutil
import re
from unittest.mock import ANY, Mock, patch

import pytest
from mediafile import MediaFile

from beets import config, plugins, ui
from beets.dbcore import types
from beets.importer import (
    Action,
    ArchiveImportTask,
    SentinelImportTask,
    SingletonImportTask,
)
from beets.library import Item
from beets.test import helper
from beets.test.helper import AutotagStub, ImportHelper, TerminalImportMixin
from beets.test.helper import PluginTestCase as BasePluginTestCase
from beets.util import displayable_path, syspath


class PluginLoaderTestCase(BasePluginTestCase):
    def setup_plugin_loader(self):
        # FIXME the mocking code is horrific, but this is the lowest and
        # earliest level of the plugin mechanism we can hook into.
        self._plugin_loader_patch = patch("beets.plugins.load_plugins")
        self._plugin_classes = set()
        load_plugins = self._plugin_loader_patch.start()

        def myload(names=()):
            plugins._classes.update(self._plugin_classes)

        load_plugins.side_effect = myload

    def teardown_plugin_loader(self):
        self._plugin_loader_patch.stop()

    def register_plugin(self, plugin_class):
        self._plugin_classes.add(plugin_class)

    def setUp(self):
        self.setup_plugin_loader()
        super().setUp()

    def tearDown(self):
        self.teardown_plugin_loader()
        super().tearDown()


class PluginImportTestCase(ImportHelper, PluginLoaderTestCase):
    def setUp(self):
        super().setUp()
        self.prepare_album_for_import(2)


class ItemTypesTest(PluginLoaderTestCase):
    def test_flex_field_type(self):
        class RatingPlugin(plugins.BeetsPlugin):
            item_types = {"rating": types.Float()}

        self.register_plugin(RatingPlugin)
        self.config["plugins"] = "rating"

        item = Item(path="apath", artist="aaa")
        item.add(self.lib)

        # Do not match unset values
        out = self.run_with_output("ls", "rating:1..3")
        assert "aaa" not in out

        self.run_command("modify", "rating=2", "--yes")

        # Match in range
        out = self.run_with_output("ls", "rating:1..3")
        assert "aaa" in out

        # Don't match out of range
        out = self.run_with_output("ls", "rating:3..5")
        assert "aaa" not in out


class ItemWriteTest(PluginLoaderTestCase):
    def setUp(self):
        super().setUp()

        class EventListenerPlugin(plugins.BeetsPlugin):
            pass

        self.event_listener_plugin = EventListenerPlugin()
        self.register_plugin(EventListenerPlugin)

    def test_change_tags(self):
        def on_write(item=None, path=None, tags=None):
            if tags["artist"] == "XXX":
                tags["artist"] = "YYY"

        self.register_listener("write", on_write)

        item = self.add_item_fixture(artist="XXX")
        item.write()

        mediafile = MediaFile(syspath(item.path))
        assert mediafile.artist == "YYY"

    def register_listener(self, event, func):
        self.event_listener_plugin.register_listener(event, func)


class ItemTypeConflictTest(PluginLoaderTestCase):
    def test_mismatch(self):
        class EventListenerPlugin(plugins.BeetsPlugin):
            item_types = {"duplicate": types.INTEGER}

        class AdventListenerPlugin(plugins.BeetsPlugin):
            item_types = {"duplicate": types.FLOAT}

        self.event_listener_plugin = EventListenerPlugin
        self.advent_listener_plugin = AdventListenerPlugin
        self.register_plugin(EventListenerPlugin)
        self.register_plugin(AdventListenerPlugin)
        with pytest.raises(plugins.PluginConflictError):
            plugins.types(Item)

    def test_match(self):
        class EventListenerPlugin(plugins.BeetsPlugin):
            item_types = {"duplicate": types.INTEGER}

        class AdventListenerPlugin(plugins.BeetsPlugin):
            item_types = {"duplicate": types.INTEGER}

        self.event_listener_plugin = EventListenerPlugin
        self.advent_listener_plugin = AdventListenerPlugin
        self.register_plugin(EventListenerPlugin)
        self.register_plugin(AdventListenerPlugin)
        assert plugins.types(Item) is not None


class EventsTest(PluginImportTestCase):
    def setUp(self):
        super().setUp()

    def test_import_task_created(self):
        self.importer = self.setup_importer(pretend=True)

        with helper.capture_log() as logs:
            self.importer.run()

        # Exactly one event should have been imported (for the album).
        # Sentinels do not get emitted.
        assert logs.count("Sending event: import_task_created") == 1

        logs = [line for line in logs if not line.startswith("Sending event:")]
        assert logs == [
            f"Album: {displayable_path(os.path.join(self.import_dir, b'album'))}",
            f"  {displayable_path(self.import_media[0].path)}",
            f"  {displayable_path(self.import_media[1].path)}",
        ]

    def test_import_task_created_with_plugin(self):
        class ToSingletonPlugin(plugins.BeetsPlugin):
            def __init__(self):
                super().__init__()

                self.register_listener(
                    "import_task_created", self.import_task_created_event
                )

            def import_task_created_event(self, session, task):
                if (
                    isinstance(task, SingletonImportTask)
                    or isinstance(task, SentinelImportTask)
                    or isinstance(task, ArchiveImportTask)
                ):
                    return task

                new_tasks = []
                for item in task.items:
                    new_tasks.append(SingletonImportTask(task.toppath, item))

                return new_tasks

        to_singleton_plugin = ToSingletonPlugin
        self.register_plugin(to_singleton_plugin)

        self.importer = self.setup_importer(pretend=True)

        with helper.capture_log() as logs:
            self.importer.run()

        # Exactly one event should have been imported (for the album).
        # Sentinels do not get emitted.
        assert logs.count("Sending event: import_task_created") == 1

        logs = [line for line in logs if not line.startswith("Sending event:")]
        assert logs == [
            f"Singleton: {displayable_path(self.import_media[0].path)}",
            f"Singleton: {displayable_path(self.import_media[1].path)}",
        ]


class ListenersTest(PluginLoaderTestCase):
    def test_register(self):
        class DummyPlugin(plugins.BeetsPlugin):
            def __init__(self):
                super().__init__()
                self.register_listener("cli_exit", self.dummy)
                self.register_listener("cli_exit", self.dummy)

            def dummy(self):
                pass

        d = DummyPlugin()
        assert DummyPlugin._raw_listeners["cli_exit"] == [d.dummy]

        d2 = DummyPlugin()
        assert DummyPlugin._raw_listeners["cli_exit"] == [d.dummy, d2.dummy]

        d.register_listener("cli_exit", d2.dummy)
        assert DummyPlugin._raw_listeners["cli_exit"] == [d.dummy, d2.dummy]

    @patch("beets.plugins.find_plugins")
    @patch("inspect.getfullargspec")
    def test_events_called(self, mock_gfa, mock_find_plugins):
        mock_gfa.return_value = Mock(
            args=(),
            varargs="args",
            varkw="kwargs",
        )

        class DummyPlugin(plugins.BeetsPlugin):
            def __init__(self):
                super().__init__()
                self.foo = Mock(__name__="foo")
                self.register_listener("event_foo", self.foo)
                self.bar = Mock(__name__="bar")
                self.register_listener("event_bar", self.bar)

        d = DummyPlugin()
        mock_find_plugins.return_value = (d,)

        plugins.send("event")
        d.foo.assert_has_calls([])
        d.bar.assert_has_calls([])

        plugins.send("event_foo", var="tagada")
        d.foo.assert_called_once_with(var="tagada")
        d.bar.assert_has_calls([])

    @patch("beets.plugins.find_plugins")
    def test_listener_params(self, mock_find_plugins):
        class DummyPlugin(plugins.BeetsPlugin):
            def __init__(self):
                super().__init__()
                for i in itertools.count(1):
                    try:
                        meth = getattr(self, f"dummy{i}")
                    except AttributeError:
                        break
                    self.register_listener(f"event{i}", meth)

            def dummy1(self, foo):
                assert foo == 5

            def dummy2(self, foo=None):
                assert foo == 5

            def dummy3(self):
                # argument cut off
                pass

            def dummy4(self, bar=None):
                # argument cut off
                pass

            def dummy5(self, bar):
                assert not True

            # more complex examples

            def dummy6(self, foo, bar=None):
                assert foo == 5
                assert bar is None

            def dummy7(self, foo, **kwargs):
                assert foo == 5
                assert kwargs == {}

            def dummy8(self, foo, bar, **kwargs):
                assert not True

            def dummy9(self, **kwargs):
                assert kwargs == {"foo": 5}

        d = DummyPlugin()
        mock_find_plugins.return_value = (d,)

        plugins.send("event1", foo=5)
        plugins.send("event2", foo=5)
        plugins.send("event3", foo=5)
        plugins.send("event4", foo=5)

        with pytest.raises(TypeError):
            plugins.send("event5", foo=5)

        plugins.send("event6", foo=5)
        plugins.send("event7", foo=5)

        with pytest.raises(TypeError):
            plugins.send("event8", foo=5)

        plugins.send("event9", foo=5)


class PromptChoicesTest(TerminalImportMixin, PluginImportTestCase):
    def setUp(self):
        super().setUp()
        self.setup_importer()
        self.matcher = AutotagStub(AutotagStub.IDENT).install()
        self.addCleanup(self.matcher.restore)
        # keep track of ui.input_option() calls
        self.input_options_patcher = patch(
            "beets.ui.input_options", side_effect=ui.input_options
        )
        self.mock_input_options = self.input_options_patcher.start()

    def tearDown(self):
        super().tearDown()
        self.input_options_patcher.stop()

    def test_plugin_choices_in_ui_input_options_album(self):
        """Test the presence of plugin choices on the prompt (album)."""

        class DummyPlugin(plugins.BeetsPlugin):
            def __init__(self):
                super().__init__()
                self.register_listener(
                    "before_choose_candidate", self.return_choices
                )

            def return_choices(self, session, task):
                return [
                    ui.commands.PromptChoice("f", "Foo", None),
                    ui.commands.PromptChoice("r", "baR", None),
                ]

        self.register_plugin(DummyPlugin)
        # Default options + extra choices by the plugin ('Foo', 'Bar')
        opts = (
            "Apply",
            "More candidates",
            "Skip",
            "Use as-is",
            "as Tracks",
            "Group albums",
            "Enter search",
            "enter Id",
            "aBort",
        ) + ("Foo", "baR")

        self.importer.add_choice(Action.SKIP)
        self.importer.run()
        self.mock_input_options.assert_called_once_with(
            opts, default="a", require=ANY
        )

    def test_plugin_choices_in_ui_input_options_singleton(self):
        """Test the presence of plugin choices on the prompt (singleton)."""

        class DummyPlugin(plugins.BeetsPlugin):
            def __init__(self):
                super().__init__()
                self.register_listener(
                    "before_choose_candidate", self.return_choices
                )

            def return_choices(self, session, task):
                return [
                    ui.commands.PromptChoice("f", "Foo", None),
                    ui.commands.PromptChoice("r", "baR", None),
                ]

        self.register_plugin(DummyPlugin)
        # Default options + extra choices by the plugin ('Foo', 'Bar')
        opts = (
            "Apply",
            "More candidates",
            "Skip",
            "Use as-is",
            "Enter search",
            "enter Id",
            "aBort",
        ) + ("Foo", "baR")

        config["import"]["singletons"] = True
        self.importer.add_choice(Action.SKIP)
        self.importer.run()
        self.mock_input_options.assert_called_with(
            opts, default="a", require=ANY
        )

    def test_choices_conflicts(self):
        """Test the short letter conflict solving."""

        class DummyPlugin(plugins.BeetsPlugin):
            def __init__(self):
                super().__init__()
                self.register_listener(
                    "before_choose_candidate", self.return_choices
                )

            def return_choices(self, session, task):
                return [
                    ui.commands.PromptChoice("a", "A foo", None),  # dupe
                    ui.commands.PromptChoice("z", "baZ", None),  # ok
                    ui.commands.PromptChoice("z", "Zupe", None),  # dupe
                    ui.commands.PromptChoice("z", "Zoo", None),
                ]  # dupe

        self.register_plugin(DummyPlugin)
        # Default options + not dupe extra choices by the plugin ('baZ')
        opts = (
            "Apply",
            "More candidates",
            "Skip",
            "Use as-is",
            "as Tracks",
            "Group albums",
            "Enter search",
            "enter Id",
            "aBort",
        ) + ("baZ",)
        self.importer.add_choice(Action.SKIP)
        self.importer.run()
        self.mock_input_options.assert_called_once_with(
            opts, default="a", require=ANY
        )

    def test_plugin_callback(self):
        """Test that plugin callbacks are being called upon user choice."""

        class DummyPlugin(plugins.BeetsPlugin):
            def __init__(self):
                super().__init__()
                self.register_listener(
                    "before_choose_candidate", self.return_choices
                )

            def return_choices(self, session, task):
                return [ui.commands.PromptChoice("f", "Foo", self.foo)]

            def foo(self, session, task):
                pass

        self.register_plugin(DummyPlugin)
        # Default options + extra choices by the plugin ('Foo', 'Bar')
        opts = (
            "Apply",
            "More candidates",
            "Skip",
            "Use as-is",
            "as Tracks",
            "Group albums",
            "Enter search",
            "enter Id",
            "aBort",
        ) + ("Foo",)

        # DummyPlugin.foo() should be called once
        with patch.object(DummyPlugin, "foo", autospec=True) as mock_foo:
            with helper.control_stdin("\n".join(["f", "s"])):
                self.importer.run()
            assert mock_foo.call_count == 1

        # input_options should be called twice, as foo() returns None
        assert self.mock_input_options.call_count == 2
        self.mock_input_options.assert_called_with(
            opts, default="a", require=ANY
        )

    def test_plugin_callback_return(self):
        """Test that plugin callbacks that return a value exit the loop."""

        class DummyPlugin(plugins.BeetsPlugin):
            def __init__(self):
                super().__init__()
                self.register_listener(
                    "before_choose_candidate", self.return_choices
                )

            def return_choices(self, session, task):
                return [ui.commands.PromptChoice("f", "Foo", self.foo)]

            def foo(self, session, task):
                return Action.SKIP

        self.register_plugin(DummyPlugin)
        # Default options + extra choices by the plugin ('Foo', 'Bar')
        opts = (
            "Apply",
            "More candidates",
            "Skip",
            "Use as-is",
            "as Tracks",
            "Group albums",
            "Enter search",
            "enter Id",
            "aBort",
        ) + ("Foo",)

        # DummyPlugin.foo() should be called once
        with helper.control_stdin("f\n"):
            self.importer.run()

        # input_options should be called once, as foo() returns SKIP
        self.mock_input_options.assert_called_once_with(
            opts, default="a", require=ANY
        )


def get_available_plugins():
    """Get all available plugins in the beetsplug namespace."""
    namespace_pkg = importlib.import_module("beetsplug")
    plugin_names: list[str] = []
    for _, module_name, _ in pkgutil.iter_modules(namespace_pkg.__path__):
        plugin_names.append(module_name)

    # skip _typing
    return [name for name in plugin_names if name != "_typing"]


class TestImportAllPlugins:
    @pytest.mark.parametrize("plugin_name", get_available_plugins())
    def test_import_plugin(self, caplog, plugin_name):  #
        """Test that a plugin is importable without an error using the
        load_plugins function."""

        # Try to load all plugins
        from beets.plugins import load_plugins

        caplog.set_level(logging.WARNING)
        load_plugins([plugin_name])

        # Check for warnings, is a bit hacky but we can make full use of the beets
        # load_plugins code that way
        records = []
        pattern = r"ModuleNotFoundError: No module named '(.*?)'"
        for record in caplog.records:
            match = re.search(pattern, str(record))
            if match:
                module_name = match.group(1)
                if not self._is_spec_available(module_name):
                    # If the module is not found, we skip it
                    continue
            records.append(record)

        assert len(records) == 0, (
            f"Plugin '{plugin_name}' has issues during import. ",
            records,
        )

    def _is_spec_available(self, spec_name):
        """Check if a module is available by its name."""
        return importlib.util.find_spec(spec_name) is not None