cthulhu/tests/test_plugin_system_manager_regressions.py

import sys
import tempfile
import textwrap
import types
import unittest
from pathlib import Path
from unittest import mock

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))

import cthulhu as cthulhuPackage

missingModule = object()
previousInputEventManagerModule = sys.modules.get(
    "cthulhu.input_event_manager",
    missingModule,
)
previousInputEventManagerAttribute = getattr(
    cthulhuPackage,
    "input_event_manager",
    missingModule,
)
previousPluginSystemManagerModule = sys.modules.get(
    "cthulhu.plugin_system_manager",
    missingModule,
)
previousPluginSystemManagerAttribute = getattr(
    cthulhuPackage,
    "plugin_system_manager",
    missingModule,
)
inputEventManagerStub = types.ModuleType("cthulhu.input_event_manager")
inputEventManagerStub.get_manager = mock.Mock(return_value=mock.Mock())
sys.modules["cthulhu.input_event_manager"] = inputEventManagerStub
cthulhuPackage.input_event_manager = inputEventManagerStub

try:
    from cthulhu import plugin_system_manager
    from cthulhu.plugin_system_manager import PluginInfo, PluginSystemManager
finally:
    if previousInputEventManagerModule is missingModule:
        sys.modules.pop("cthulhu.input_event_manager", None)
    else:
        sys.modules["cthulhu.input_event_manager"] = previousInputEventManagerModule

    if previousInputEventManagerAttribute is missingModule:
        delattr(cthulhuPackage, "input_event_manager")
    else:
        cthulhuPackage.input_event_manager = previousInputEventManagerAttribute

    if previousPluginSystemManagerModule is missingModule:
        sys.modules.pop("cthulhu.plugin_system_manager", None)
    else:
        sys.modules["cthulhu.plugin_system_manager"] = previousPluginSystemManagerModule

    if previousPluginSystemManagerAttribute is missingModule:
        delattr(cthulhuPackage, "plugin_system_manager")
    else:
        cthulhuPackage.plugin_system_manager = previousPluginSystemManagerAttribute


PLUGIN_TEMPLATE = """
from cthulhu.plugin import Plugin, cthulhu_hookimpl


class {class_name}(Plugin):
    def __init__(self):
        super().__init__()
        self.activation_count = 0
        self.deactivation_count = 0

    @cthulhu_hookimpl
    def activate(self, plugin=None):
        self.activation_count += 1

    @cthulhu_hookimpl
    def deactivate(self, plugin=None):
        self.deactivation_count += 1
"""

PLUGIN_TEMPLATE_PLAIN_SIGNATURE = """
from cthulhu.plugin import Plugin, cthulhu_hookimpl


class {class_name}(Plugin):
    def __init__(self):
        super().__init__()
        self.activation_count = 0
        self.deactivation_count = 0

    @cthulhu_hookimpl
    def activate(self):
        self.activation_count += 1

    @cthulhu_hookimpl
    def deactivate(self):
        self.deactivation_count += 1
"""


class PluginSystemManagerRegressionTests(unittest.TestCase):
    def _create_plugin_info(
        self,
        root_dir: str,
        module_name: str,
        template: str = PLUGIN_TEMPLATE,
    ) -> PluginInfo:
        plugin_dir = Path(root_dir) / module_name
        plugin_dir.mkdir()
        (plugin_dir / "plugin.py").write_text(
            textwrap.dedent(template.format(class_name=module_name)),
            encoding="utf-8",
        )
        return PluginInfo(
            module_name,
            module_name,
            str(plugin_dir),
            canonical_name=module_name,
            source_id="test",
            origin="test",
        )

    def _create_manager(self) -> PluginSystemManager:
        app = mock.Mock()
        app.getSignalManager.return_value = mock.Mock()
        return PluginSystemManager(app)

    @mock.patch("cthulhu.plugin_system_manager.dbus_service.get_remote_controller")
    def test_loading_second_plugin_does_not_reactivate_first_plugin(self, remote_controller):
        remote_controller.return_value = mock.Mock()

        with tempfile.TemporaryDirectory() as temp_dir:
            manager = self._create_manager()
            first_plugin = self._create_plugin_info(temp_dir, "FirstPlugin")
            second_plugin = self._create_plugin_info(temp_dir, "SecondPlugin")

            self.assertTrue(manager.loadPlugin(first_plugin))
            self.assertEqual(first_plugin.instance.activation_count, 1)

            self.assertTrue(manager.loadPlugin(second_plugin))

            self.assertEqual(first_plugin.instance.activation_count, 1)
            self.assertEqual(second_plugin.instance.activation_count, 1)

    @mock.patch("cthulhu.plugin_system_manager.dbus_service.get_remote_controller")
    def test_unloading_plugin_does_not_deactivate_other_loaded_plugins(self, remote_controller):
        remote_controller.return_value = mock.Mock()

        with tempfile.TemporaryDirectory() as temp_dir:
            manager = self._create_manager()
            first_plugin = self._create_plugin_info(temp_dir, "FirstPlugin")
            second_plugin = self._create_plugin_info(temp_dir, "SecondPlugin")

            self.assertTrue(manager.loadPlugin(first_plugin))
            self.assertTrue(manager.loadPlugin(second_plugin))

            second_instance = second_plugin.instance

            self.assertTrue(manager.unloadPlugin(second_plugin))

            self.assertEqual(first_plugin.instance.deactivation_count, 0)
            self.assertEqual(second_instance.deactivation_count, 1)

    @mock.patch("cthulhu.plugin_system_manager.dbus_service.get_remote_controller")
    def test_loading_plugin_with_plain_activate_signature_succeeds(self, remote_controller):
        remote_controller.return_value = mock.Mock()

        with tempfile.TemporaryDirectory() as temp_dir:
            manager = self._create_manager()
            plugin_info = self._create_plugin_info(
                temp_dir,
                "PlainSignaturePlugin",
                template=PLUGIN_TEMPLATE_PLAIN_SIGNATURE,
            )

            self.assertTrue(manager.loadPlugin(plugin_info))
            instance = plugin_info.instance
            self.assertEqual(instance.activation_count, 1)

            self.assertTrue(manager.unloadPlugin(plugin_info))
            self.assertEqual(instance.deactivation_count, 1)
            self.assertEqual(plugin_info.instance, None)

    @mock.patch("cthulhu.plugin_system_manager.dbus_service.get_remote_controller")
    def test_obsolete_plugin_manager_active_entry_is_dropped(self, remote_controller):
        remote_controller.return_value = mock.Mock()

        with tempfile.TemporaryDirectory() as temp_dir:
            manager = self._create_manager()
            plugin_info = self._create_plugin_info(temp_dir, "OCR")
            manager._plugins["OCR"] = plugin_info
            manager._plugin_name_index["OCR"] = ["OCR"]

            with mock.patch.object(manager, "syncAllPluginsActive"):
                manager.setActivePlugins(["PluginManager", "OCR"])

            self.assertEqual(manager.getActivePlugins(), ["OCR"])


if __name__ == "__main__":
    unittest.main()