2
0
mirror of https://github.com/inventree/InvenTree.git synced 2025-07-10 06:54:15 +00:00

[bug] Logic fix for plugins (#9934)

* Logic fix for plugins

- Prevent tasks being run for disabled plugins

* Adjust default value for "get_plugin" method

* Fix return type

* Update typing

* Tweak unit test

* Update unit tests

* More test updates
This commit is contained in:
Oliver
2025-07-06 10:15:33 +10:00
committed by GitHub
parent 82dfe561ee
commit 945cb46f32
17 changed files with 45 additions and 40 deletions

View File

@ -30,7 +30,7 @@ class InvenTreeExchange(SimpleExchangeBackend):
# Find the selected exchange rate plugin # Find the selected exchange rate plugin
slug = get_global_setting('CURRENCY_UPDATE_PLUGIN', create=False) slug = get_global_setting('CURRENCY_UPDATE_PLUGIN', create=False)
plugin = registry.get_plugin(slug) if slug else None plugin = registry.get_plugin(slug, active=True) if slug else None
if not plugin: if not plugin:
# Find the first active currency exchange plugin # Find the first active currency exchange plugin

View File

@ -35,7 +35,7 @@ def export_data(
""" """
from plugin import registry from plugin import registry
if (plugin := registry.get_plugin(plugin_key)) is None: if (plugin := registry.get_plugin(plugin_key, active=True)) is None:
logger.warning("export_data: Plugin '%s' not found", plugin_key) logger.warning("export_data: Plugin '%s' not found", plugin_key)
return return

View File

@ -105,10 +105,10 @@ def process_event(plugin_slug, event, *args, **kwargs):
This function is run by the background worker process. This function is run by the background worker process.
This function may queue multiple functions to be handled by the background worker. This function may queue multiple functions to be handled by the background worker.
""" """
plugin = registry.get_plugin(plugin_slug) plugin = registry.get_plugin(plugin_slug, active=True)
if plugin is None: # pragma: no cover if plugin is None: # pragma: no cover
logger.error("Could not find matching plugin for '%s'", plugin_slug) logger.error("Could not find matching active plugin for '%s'", plugin_slug)
return return
logger.debug("Plugin '%s' is processing triggered event '%s'", plugin_slug, event) logger.debug("Plugin '%s' is processing triggered event '%s'", plugin_slug, event)

View File

@ -26,10 +26,10 @@ def print_label(plugin_slug: str, **kwargs):
""" """
logger.info("Plugin '%s' is printing a label", plugin_slug) logger.info("Plugin '%s' is printing a label", plugin_slug)
plugin = registry.get_plugin(plugin_slug) plugin = registry.get_plugin(plugin_slug, active=True)
if plugin is None: # pragma: no cover if plugin is None: # pragma: no cover
logger.error("Could not find matching plugin for '%s'", plugin_slug) logger.error("Could not find matching active plugin for '%s'", plugin_slug)
return return
try: try:

View File

@ -124,7 +124,7 @@ class LabelMixinTests(PrintTestMixins, InvenTreeAPITestCase):
plugins = registry.with_mixin(PluginMixinEnum.LABELS) plugins = registry.with_mixin(PluginMixinEnum.LABELS)
self.assertGreater(len(plugins), 0) self.assertGreater(len(plugins), 0)
plugin = registry.get_plugin('samplelabelprinter') plugin = registry.get_plugin('samplelabelprinter', active=None)
self.assertIsNotNone(plugin) self.assertIsNotNone(plugin)
config = plugin.plugin_config() config = plugin.plugin_config()

View File

@ -16,9 +16,7 @@ class LocatePluginTests(InvenTreeAPITestCase):
super().setUp() super().setUp()
# Activate plugin # Activate plugin
config = registry.get_plugin('samplelocate').plugin_config() registry.set_plugin_state('samplelocate', True)
config.active = True
config.save()
fixtures = ['category', 'part', 'location', 'stock'] fixtures = ['category', 'part', 'location', 'stock']

View File

@ -101,16 +101,18 @@ class PluginsRegistry:
self.installed_apps = [] # Holds all added plugin_paths self.installed_apps = [] # Holds all added plugin_paths
@property @property
def is_loading(self): def is_loading(self) -> bool:
"""Return True if the plugin registry is currently loading.""" """Return True if the plugin registry is currently loading."""
return self.loading_lock.locked() return self.loading_lock.locked()
def get_plugin(self, slug, active=None, with_mixin=None): def get_plugin(
self, slug: str, active: bool = True, with_mixin: Optional[str] = None
) -> InvenTreePlugin:
"""Lookup plugin by slug (unique key). """Lookup plugin by slug (unique key).
Args: Args:
slug (str): The slug of the plugin to look up. slug (str): The slug of the plugin to look up.
active (bool, optional): Filter by 'active' status of the plugin. If None, no filtering is applied. Defaults to None. active (bool, optional): Filter by 'active' status of the plugin. If None, no filtering is applied. Defaults to True.
with_mixin (str, optional): Filter by mixin name. If None, no filtering is applied. Defaults to None. with_mixin (str, optional): Filter by mixin name. If None, no filtering is applied. Defaults to None.
Returns: Returns:
@ -136,7 +138,7 @@ class PluginsRegistry:
def get_plugin_config(self, slug: str, name: Union[str, None] = None): def get_plugin_config(self, slug: str, name: Union[str, None] = None):
"""Return the matching PluginConfig instance for a given plugin. """Return the matching PluginConfig instance for a given plugin.
Args: Arguments:
slug: The plugin slug slug: The plugin slug
name: The plugin name (optional) name: The plugin name (optional)
""" """
@ -167,10 +169,10 @@ class PluginsRegistry:
return cfg return cfg
def set_plugin_state(self, slug, state): def set_plugin_state(self, slug: str, state: bool):
"""Set the state(active/inactive) of a plugin. """Set the state(active/inactive) of a plugin.
Args: Arguments:
slug (str): Plugin slug slug (str): Plugin slug
state (bool): Plugin state - true = active, false = inactive state (bool): Plugin state - true = active, false = inactive
""" """
@ -374,7 +376,7 @@ class PluginsRegistry:
# Ensure the lock is released always # Ensure the lock is released always
self.loading_lock.release() self.loading_lock.release()
def plugin_dirs(self): def plugin_dirs(self) -> list[str]:
"""Construct a list of directories from where plugins can be loaded.""" """Construct a list of directories from where plugins can be loaded."""
# Builtin plugins are *always* loaded # Builtin plugins are *always* loaded
dirs = ['plugin.builtin'] dirs = ['plugin.builtin']

View File

@ -15,9 +15,7 @@ class EventPluginSampleTests(TestCase):
def test_run_event(self): def test_run_event(self):
"""Check if the event is issued.""" """Check if the event is issued."""
# Activate plugin # Activate plugin
config = registry.get_plugin('sampleevent').plugin_config() registry.set_plugin_state('sampleevent', True)
config.active = True
config.save()
InvenTreeSetting.set_setting('ENABLE_PLUGINS_EVENTS', True, change_user=None) InvenTreeSetting.set_setting('ENABLE_PLUGINS_EVENTS', True, change_user=None)

View File

@ -13,9 +13,7 @@ class FilteredEventPluginSampleTests(TestCase):
def test_run_event(self): def test_run_event(self):
"""Check if the event is issued.""" """Check if the event is issued."""
# Activate plugin # Activate plugin
config = registry.get_plugin('filteredsampleevent').plugin_config() registry.set_plugin_state('filteredsampleevent', True)
config.active = True
config.save()
InvenTreeSetting.set_setting('ENABLE_PLUGINS_EVENTS', True, change_user=None) InvenTreeSetting.set_setting('ENABLE_PLUGINS_EVENTS', True, change_user=None)
@ -29,9 +27,7 @@ class FilteredEventPluginSampleTests(TestCase):
def test_ignore_event(self): def test_ignore_event(self):
"""Check if the event is issued.""" """Check if the event is issued."""
# Activate plugin # Activate plugin
config = registry.get_plugin('filteredsampleevent').plugin_config() registry.set_plugin_state('filteredsampleevent', True)
config.active = True
config.save()
InvenTreeSetting.set_setting('ENABLE_PLUGINS_EVENTS', True, change_user=None) InvenTreeSetting.set_setting('ENABLE_PLUGINS_EVENTS', True, change_user=None)

View File

@ -14,9 +14,7 @@ class SampleIconPackPluginTests(InvenTreeAPITestCase):
def test_get_icons_api(self): def test_get_icons_api(self):
"""Check get icons api.""" """Check get icons api."""
# Activate plugin # Activate plugin
config = registry.get_plugin('sampleicons').plugin_config() registry.set_plugin_state('sampleicons', True)
config.active = True
config.save()
response = self.get(reverse('api-icon-list'), expected_code=200) response = self.get(reverse('api-icon-list'), expected_code=200)
self.assertEqual(len(response.data), 2) self.assertEqual(len(response.data), 2)

View File

@ -45,6 +45,7 @@ class SampleIntegrationPluginTests(InvenTreeTestCase):
def test_settings(self): def test_settings(self):
"""Check the SettingsMixin.check_settings function.""" """Check the SettingsMixin.check_settings function."""
registry.set_plugin_state('sample', True)
plugin = registry.get_plugin('sample') plugin = registry.get_plugin('sample')
self.assertIsNotNone(plugin) self.assertIsNotNone(plugin)
@ -57,13 +58,19 @@ class SampleIntegrationPluginTests(InvenTreeTestCase):
def test_settings_validator(self): def test_settings_validator(self):
"""Test settings validator for plugins.""" """Test settings validator for plugins."""
registry.set_plugin_state('sample', False)
self.assertIsNone(registry.get_plugin('sample'))
registry.set_plugin_state('sample', True)
plugin = registry.get_plugin('sample') plugin = registry.get_plugin('sample')
self.assertIsNotNone(plugin)
valid_json = '{"ts": 13}' valid_json = '{"ts": 13}'
not_valid_json = '{"ts""13"}' not_valid_json = '{"ts""13"}'
# no error, should pass validator # no error, should pass validator
plugin.set_setting('VALIDATOR_SETTING', valid_json) plugin.set_setting('VALIDATOR_SETTING', valid_json)
# should throw an error # This should throw an error
with self.assertRaises(ValidationError): with self.assertRaises(ValidationError):
plugin.set_setting('VALIDATOR_SETTING', not_valid_json) plugin.set_setting('VALIDATOR_SETTING', not_valid_json)

View File

@ -76,7 +76,15 @@ class ExampleScheduledTaskPluginTests(TestCase):
def test_calling(self): def test_calling(self):
"""Test calling of plugin functions by name.""" """Test calling of plugin functions by name."""
# Check with right parameters # First, plugin is *not* enabled
registry.set_plugin_state('schedule', False)
with self.assertRaises(AttributeError):
self.assertEqual(call_plugin_function('schedule', 'member_func'), False)
registry.set_plugin_state('schedule', True)
# Should work now
self.assertEqual(call_plugin_function('schedule', 'member_func'), False) self.assertEqual(call_plugin_function('schedule', 'member_func'), False)
# Check with wrong key # Check with wrong key

View File

@ -8,7 +8,7 @@ from plugin.helpers import MixinNotImplementedError
from plugin.mixins import LocateMixin from plugin.mixins import LocateMixin
class SampleLocatePlugintests(InvenTreeAPITestCase): class SampleLocatePluginTests(InvenTreeAPITestCase):
"""Tests for SampleLocatePlugin.""" """Tests for SampleLocatePlugin."""
fixtures = ['location', 'category', 'part', 'stock'] fixtures = ['location', 'category', 'part', 'stock']
@ -16,7 +16,7 @@ class SampleLocatePlugintests(InvenTreeAPITestCase):
def test_run_locator(self): def test_run_locator(self):
"""Check if the event is issued.""" """Check if the event is issued."""
# Activate plugin # Activate plugin
config = registry.get_plugin('samplelocate').plugin_config() config = registry.get_plugin('samplelocate', active=None).plugin_config()
config.active = True config.active = True
config.save() config.save()

View File

@ -15,7 +15,7 @@ class MailPluginSampleTests(TestCase):
def activate_plugin(self): def activate_plugin(self):
"""Activate the sample mail plugin.""" """Activate the sample mail plugin."""
config = registry.get_plugin('samplemail').plugin_config() config = registry.get_plugin('samplemail', active=None).plugin_config()
config.active = True config.active = True
config.save() config.save()

View File

@ -217,6 +217,7 @@ class RegistryTests(TestCase):
with mock.patch.dict(os.environ, envs): with mock.patch.dict(os.environ, envs):
# Reload to rediscover plugins # Reload to rediscover plugins
registry.reload_plugins(full_reload=True, collect=True) registry.reload_plugins(full_reload=True, collect=True)
registry.set_plugin_state('simple', True)
# Depends on the meta set in InvenTree/plugin/mock/simple:SimplePlugin # Depends on the meta set in InvenTree/plugin/mock/simple:SimplePlugin
plg = registry.get_plugin('simple') plg = registry.get_plugin('simple')
@ -264,7 +265,7 @@ class RegistryTests(TestCase):
registry.reload_plugins(full_reload=True, collect=True) registry.reload_plugins(full_reload=True, collect=True)
# Test that plugin was installed # Test that plugin was installed
plg = registry.get_plugin('zapier') plg = registry.get_plugin('zapier', active=None)
self.assertEqual(plg.slug, 'zapier') self.assertEqual(plg.slug, 'zapier')
self.assertEqual(plg.name, 'inventree_zapier') self.assertEqual(plg.name, 'inventree_zapier')

View File

@ -68,7 +68,7 @@ def print_labels(
model = template.get_model() model = template.get_model()
items = model.objects.filter(pk__in=item_ids) items = model.objects.filter(pk__in=item_ids)
plugin = registry.get_plugin(plugin_slug) plugin = registry.get_plugin(plugin_slug, active=True)
if not plugin: if not plugin:
logger.warning("Label printing plugin '%s' not found", plugin_slug) logger.warning("Label printing plugin '%s' not found", plugin_slug)

View File

@ -357,12 +357,9 @@ class PrintTestMixins:
def do_activate_plugin(self): def do_activate_plugin(self):
"""Activate the 'samplelabel' plugin.""" """Activate the 'samplelabel' plugin."""
registry.set_plugin_state(self.plugin_ref, True)
plugin = registry.get_plugin(self.plugin_ref) plugin = registry.get_plugin(self.plugin_ref)
self.assertIsNotNone(plugin) self.assertIsNotNone(plugin)
config = plugin.plugin_config()
self.assertIsNotNone(config)
config.active = True
config.save()
def run_print_test(self, qs, model_type, label: bool = True): def run_print_test(self, qs, model_type, label: bool = True):
"""Run tests on single and multiple page printing. """Run tests on single and multiple page printing.