inventree/InvenTree
2
0
Fork 0
mirror of https://github.com/inventree/InvenTree.git synced 2025-10-30 20:55:42 +00:00
Code Activity

[plugin] Mandatory plugins (#10094)

Browse Source 
* Add setting for "mandatory" plugins

* Add 'is_active' method to PluginConfig model

* Check against plugin config object by priority

* Prevent plugin from reporting its own 'active' status

* Refactor get_plugin_class for LabelPrint endpoint

* Fix typo

* Mark internal plugin methods as "final"

- Prevent plugins from overriding them

* Enhanced checks for bad actor plugins

* Enhanced unit test for plugin install via API

* Playwright tests for plugin errors

* Test that builtin mandatory plugins are always activated

* Force mandatory plugins to be marked as active on load

* API unit tests

* Unit testing for plugin filtering

* Updated playwright tests

- Force one extra plugin to be mandatory in configuration

* Adjust unit tests

* Updated docs

* Tweak unit test

* Another unit test fix

* Fix with_mixin

- Checking active status first is expensive...

* Make with_mixin call much more efficient

- Pre-load the PluginConfig objects
- Additional unit tests
- Ensure fixed query count

* Fix the 'is_package' method for PluginConfig

* Tweak unit test

* Make api_info endpoint more efficient

- with_mixin is now very quick

* Run just single test

* Disable CI test

* Revert changes to CI pipeline

* Fix typo

* Debug for test

* Style fix

* Additional checks

* Ensure reload

* Ensure plugin registry is ready before running unit tests

* Fix typo

* Add debug statements

* Additional debug output

* Debug logging for MySQL

* Ensure config objects are created?

* Ensure plugin registry is reloaded before running tests

* Remove intentional failure

* Reset debug level

* Fix CI pipeline

* Fix

* Fix test mixins

* Fix test class

* Further updates

* Adjust info view

* Test refactoring

* Fix recursion issue in machine registry

* Force cache behavior

* Reduce API query limits in testing

* Handle potential error case in with_mixin

* remove custom query time code

* Prevent override of is_mandatory()

* Prevent unnecessary reloads

* Tweak unit tests

* Tweak mandatory active save

* Tweak unit test

* Enhanced unit testing

* Exclude lines from coverage

* (final)? cleanup

* Prevent recursive reloads

---------

Co-authored-by: Matthias Mair <code@mjmair.com>
This commit is contained in:
Oliver
2025-07-31 08:26:24 +10:00
committed by GitHub
parent b89a7c45d6
commit b8ea75b2b4
34 changed files with 993 additions and 255 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/backend/InvenTree/InvenTree/api.py
View File

@@ -295,7 +295,6 @@ class InfoView(APIView):
'worker_pending_tasks': self.worker_pending_tasks(),
'plugins_enabled': settings.PLUGINS_ENABLED,
'plugins_install_disabled': settings.PLUGINS_INSTALL_DISABLED,
'active_plugins': plugins_info(),
'email_configured': is_email_configured(),
'debug_mode': settings.DEBUG,
'docker_mode': settings.DOCKER,
@@ -307,6 +306,7 @@ class InfoView(APIView):
'navbar_message': helpers.getCustomOption('navbar_message'),
},
# Following fields are only available to staff users
'active_plugins': plugins_info() if is_staff else None,
'system_health': check_system_health() if is_staff else None,
'database': InvenTree.version.inventreeDatabase() if is_staff else None,
'platform': InvenTree.version.inventreePlatform() if is_staff else None,

9
src/backend/InvenTree/InvenTree/helpers.py
View File

@@ -1095,16 +1095,17 @@ def pui_url(subpath: str) -> str:
def plugins_info(*args, **kwargs):
"""Return information about activated plugins."""
from plugin import PluginMixinEnum
from plugin.registry import registry
# Check if plugins are even enabled
if not settings.PLUGINS_ENABLED:
return False
# Fetch plugins
plug_list = [plg for plg in registry.plugins.values() if plg.plugin_config().active]
# Fetch active plugins
plugins = registry.with_mixin(PluginMixinEnum.BASE)
# Format list
return [
{'name': plg.name, 'slug': plg.slug, 'version': plg.version}
for plg in plug_list
{'name': plg.name, 'slug': plg.slug, 'version': plg.version} for plg in plugins
]

5
src/backend/InvenTree/InvenTree/settings.py
View File

@@ -192,6 +192,10 @@ PLUGINS_ENABLED = get_boolean_setting(
'INVENTREE_PLUGINS_ENABLED', 'plugins_enabled', False
)
PLUGINS_MANDATORY = get_setting(
'INVENTREE_PLUGINS_MANDATORY', 'plugins_mandatory', typecast=list, default_value=[]
)
PLUGINS_INSTALL_DISABLED = get_boolean_setting(
'INVENTREE_PLUGIN_NOINSTALL', 'plugin_noinstall', False
)
@@ -211,6 +215,7 @@ PLUGIN_TESTING_EVENTS = False # Flag if events are tested right now
PLUGIN_TESTING_EVENTS_ASYNC = False # Flag if events are tested asynchronously
PLUGIN_TESTING_RELOAD = False # Flag if plugin reloading is in testing (check_reload)
# Plugin development settings
PLUGIN_DEV_SLUG = (
get_setting('INVENTREE_PLUGIN_DEV_SLUG', 'plugin_dev.slug') if DEBUG else None
)

42
src/backend/InvenTree/InvenTree/test_api.py
View File

@@ -580,9 +580,15 @@ class GeneralApiTests(InvenTreeAPITestCase):
def test_info_view(self):
"""Test that we can read the 'info-view' endpoint."""
from plugin import PluginMixinEnum
from plugin.models import PluginConfig
from plugin.registry import registry
self.ensurePluginsLoaded()
url = reverse('api-inventree-info')
response = self.get(url, max_query_count=275, expected_code=200)
response = self.get(url, max_query_count=20, expected_code=200)
data = response.json()
self.assertIn('server', data)
@@ -592,15 +598,41 @@ class GeneralApiTests(InvenTreeAPITestCase):
self.assertEqual('InvenTree', data['server'])
# Test with token
token = self.get(url=reverse('api-token'), max_query_count=275).data['token']
token = self.get(url=reverse('api-token'), max_query_count=20).data['token']
self.client.logout()
# Anon
response = self.get(url, max_query_count=275)
self.assertEqual(response.json()['database'], None)
response = self.get(url, max_query_count=20)
data = response.json()
self.assertEqual(data['database'], None)
# No active plugin info for anon user
self.assertIsNone(data.get('active_plugins'))
# Staff
response = self.get(
url, headers={'Authorization': f'Token {token}'}, max_query_count=275
url, headers={'Authorization': f'Token {token}'}, max_query_count=20
)
self.assertGreater(len(response.json()['database']), 4)
data = response.json()
# Check for active plugin list
self.assertIn('active_plugins', data)
plugins = data['active_plugins']
# Check that all active plugins are listed
N = len(plugins)
self.assertGreater(N, 0, 'No active plugins found')
self.assertLess(N, PluginConfig.objects.count(), 'Too many plugins found')
self.assertEqual(
N,
len(registry.with_mixin(PluginMixinEnum.BASE, active=True)),
'Incorrect number of active plugins found',
)
keys = [plugin['slug'] for plugin in plugins]
self.assertIn('bom-exporter', keys)
self.assertIn('inventree-ui-notification', keys)
self.assertIn('inventreelabel', keys)

3
src/backend/InvenTree/InvenTree/test_tasks.py
View File

@@ -15,6 +15,7 @@ from error_report.models import Error
import InvenTree.tasks
from common.models import InvenTreeSetting, InvenTreeUserSetting
from InvenTree.unit_test import PluginRegistryMixin
threshold = timezone.now() - timedelta(days=30)
threshold_low = threshold - timedelta(days=1)
@@ -55,7 +56,7 @@ def get_result():
return 'abc'
class InvenTreeTaskTests(TestCase):
class InvenTreeTaskTests(PluginRegistryMixin, TestCase):
"""Unit tests for tasks."""
def test_offloading(self):

54
src/backend/InvenTree/InvenTree/unit_test.py
View File

@@ -320,12 +320,8 @@ class ExchangeRateMixin:
Rate.objects.bulk_create(items)
class InvenTreeTestCase(ExchangeRateMixin, UserMixin, TestCase):
"""Testcase with user setup build in."""
class InvenTreeAPITestCase(ExchangeRateMixin, UserMixin, APITestCase):
"""Base class for running InvenTree API tests."""
class TestQueryMixin:
"""Mixin class for testing query counts."""
# Default query count threshold value
# TODO: This value should be reduced
@@ -375,18 +371,48 @@ class InvenTreeAPITestCase(ExchangeRateMixin, UserMixin, APITestCase):
self.assertLess(n, value, msg=msg)
class PluginRegistryMixin:
"""Mixin to ensure that the plugin registry is ready for tests."""
@classmethod
def setUpTestData(cls):
"""Setup for API tests.
"""Ensure that the plugin registry is ready for tests."""
from time import sleep
- Ensure that all global settings are assigned default values.
"""
from common.models import InvenTreeSetting
from plugin.registry import registry
while not registry.is_ready:
print('Waiting for plugin registry to be ready...')
sleep(0.1)
assert registry.is_ready, 'Plugin registry is not ready'
InvenTreeSetting.build_default_values()
super().setUpTestData()
def ensurePluginsLoaded(self, force: bool = False):
"""Helper function to ensure that plugins are loaded."""
from plugin.models import PluginConfig
if force or PluginConfig.objects.count() == 0:
# Reload the plugin registry at this point to ensure all PluginConfig objects are created
# This is because the django test system may have re-initialized the database (to an empty state)
registry.reload_plugins(full_reload=True, force_reload=True, collect=True)
assert PluginConfig.objects.count() > 0, 'No plugins are installed'
class InvenTreeTestCase(ExchangeRateMixin, PluginRegistryMixin, UserMixin, TestCase):
"""Testcase with user setup build in."""
class InvenTreeAPITestCase(
ExchangeRateMixin, PluginRegistryMixin, TestQueryMixin, UserMixin, APITestCase
):
"""Base class for running InvenTree API tests."""
def check_response(self, url, response, expected_code=None):
"""Debug output for an unexpected response."""
# Check that the response returned the expected status code
@@ -472,15 +498,15 @@ class InvenTreeAPITestCase(ExchangeRateMixin, UserMixin, APITestCase):
url, self.client.delete, expected_code=expected_code, **kwargs
)
def patch(self, url, data, expected_code=200, **kwargs):
def patch(self, url, data=None, expected_code=200, **kwargs):
"""Issue a PATCH request."""
kwargs['data'] = data
kwargs['data'] = data or {}
return self.query(url, self.client.patch, expected_code=expected_code, **kwargs)
def put(self, url, data, expected_code=200, **kwargs):
def put(self, url, data=None, expected_code=200, **kwargs):
"""Issue a PUT request."""
kwargs['data'] = data
kwargs['data'] = data or {}
return self.query(url, self.client.put, expected_code=expected_code, **kwargs)

11
src/backend/InvenTree/build/test_build.py
View File

@@ -7,7 +7,6 @@ from django.contrib.auth import get_user_model
from django.contrib.auth.models import Group
from django.core.exceptions import ValidationError
from django.db.models import Sum
from django.test import TestCase
from django.test.utils import override_settings
from django.urls import reverse
@@ -20,7 +19,11 @@ from build.models import Build, BuildItem, BuildLine, generate_next_build_refere
from build.status_codes import BuildStatus
from common.settings import set_global_setting
from InvenTree import status_codes as status
from InvenTree.unit_test import InvenTreeAPITestCase, findOffloadedEvent
from InvenTree.unit_test import (
InvenTreeAPITestCase,
InvenTreeTestCase,
findOffloadedEvent,
)
from order.models import PurchaseOrder, PurchaseOrderLineItem
from part.models import BomItem, BomItemSubstitute, Part, PartTestTemplate
from stock.models import StockItem, StockItemTestResult, StockLocation
@@ -29,7 +32,7 @@ from users.models import Owner
logger = structlog.get_logger('inventree')
class BuildTestBase(TestCase):
class BuildTestBase(InvenTreeTestCase):
"""Run some tests to ensure that the Build model is working properly."""
fixtures = ['users']
@@ -619,6 +622,8 @@ class BuildTest(BuildTestBase):
def test_overdue_notification(self):
"""Test sending of notifications when a build order is overdue."""
self.ensurePluginsLoaded()
self.build.target_date = datetime.now().date() - timedelta(days=1)
self.build.save()

2
src/backend/InvenTree/common/tests.py
View File

@@ -471,7 +471,7 @@ class SettingsTest(InvenTreeTestCase):
self.assertIsNone(cache.get(cache_key))
# First request should set cache
val = InvenTreeSetting.get_setting(key)
val = InvenTreeSetting.get_setting(key, cache=True)
self.assertEqual(cache.get(cache_key).value, val)
for val in ['A', '{{ part.IPN }}', 'C']:

4
src/backend/InvenTree/machine/registry.py
View File

@@ -153,12 +153,12 @@ class MachineRegistry(
if machine.active:
machine.initialize()
self._update_registry_hash()
logger.info('Initialized %s machines', len(self.machines.keys()))
else:
self._hash = None # reset hash to force reload hash
logger.info('Loaded %s machines', len(self.machines.keys()))
self._update_registry_hash()
def reload_machines(self):
"""Reload all machines from the database."""
self.machines = {}

22
src/backend/InvenTree/machine/test_api.py
View File

@@ -198,8 +198,9 @@ class MachineAPITest(TestMachineRegistryMixin, InvenTreeAPITestCase):
# Create a machine
response = self.post(
reverse('api-machine-list'), machine_data, max_query_count=400
reverse('api-machine-list'), machine_data, max_query_count=150
)
self.assertEqual(response.data, {**response.data, **machine_data})
pk = response.data['pk']
@@ -233,16 +234,13 @@ class MachineAPITest(TestMachineRegistryMixin, InvenTreeAPITestCase):
def test_machine_detail_settings(self):
"""Test machine detail settings API endpoint."""
# TODO: Investigate why these tests need a higher query limit
QUERY_LIMIT = 300
machine_setting_url = reverse(
'api-machine-settings-detail',
kwargs={'pk': self.placeholder_uuid, 'config_type': 'M', 'key': 'LOCATION'},
)
# Test machine settings for non-existent machine
self.get(machine_setting_url, expected_code=404, max_query_count=QUERY_LIMIT)
self.get(machine_setting_url, expected_code=404)
# Create a machine
machine = MachineConfig.objects.create(
@@ -262,22 +260,18 @@ class MachineAPITest(TestMachineRegistryMixin, InvenTreeAPITestCase):
)
# Get settings
response = self.get(machine_setting_url, max_query_count=QUERY_LIMIT)
response = self.get(machine_setting_url)
self.assertEqual(response.data['value'], '')
response = self.get(driver_setting_url, max_query_count=QUERY_LIMIT)
response = self.get(driver_setting_url)
self.assertEqual(response.data['value'], '')
# Update machine setting
location = StockLocation.objects.create(name='Test Location')
response = self.patch(
machine_setting_url,
{'value': str(location.pk)},
max_query_count=QUERY_LIMIT,
)
response = self.patch(machine_setting_url, {'value': str(location.pk)})
self.assertEqual(response.data['value'], str(location.pk))
response = self.get(machine_setting_url, max_query_count=QUERY_LIMIT)
response = self.get(machine_setting_url)
self.assertEqual(response.data['value'], str(location.pk))
# Update driver setting
@@ -289,7 +283,7 @@ class MachineAPITest(TestMachineRegistryMixin, InvenTreeAPITestCase):
# Get list of all settings for a machine
settings_url = reverse('api-machine-settings', kwargs={'pk': machine.pk})
response = self.get(settings_url, max_query_count=QUERY_LIMIT)
response = self.get(settings_url)
self.assertEqual(len(response.data), 2)
self.assertEqual(
[('M', 'LOCATION'), ('D', 'TEST_SETTING')],

7
src/backend/InvenTree/order/test_sales_order.py
View File

@@ -5,13 +5,12 @@ from datetime import datetime, timedelta
from django.contrib.auth import get_user_model
from django.contrib.auth.models import Group
from django.core.exceptions import ValidationError
from django.test import TestCase
import order.tasks
from common.models import InvenTreeSetting, NotificationMessage
from company.models import Company
from InvenTree import status_codes as status
from InvenTree.unit_test import addUserPermission
from InvenTree.unit_test import InvenTreeTestCase, addUserPermission
from order.models import (
SalesOrder,
SalesOrderAllocation,
@@ -24,7 +23,7 @@ from stock.models import StockItem
from users.models import Owner
class SalesOrderTest(TestCase):
class SalesOrderTest(InvenTreeTestCase):
"""Run tests to ensure that the SalesOrder model is working correctly."""
fixtures = ['users']
@@ -319,6 +318,8 @@ class SalesOrderTest(TestCase):
def test_overdue_notification(self):
"""Test overdue sales order notification."""
self.ensurePluginsLoaded()
user = get_user_model().objects.get(pk=3)
addUserPermission(user, 'order', 'salesorder', 'view')

10
src/backend/InvenTree/order/tests.py
View File

@@ -14,7 +14,11 @@ import common.models
import order.tasks
from common.settings import get_global_setting, set_global_setting
from company.models import Company, SupplierPart
from InvenTree.unit_test import ExchangeRateMixin, addUserPermission
from InvenTree.unit_test import (
ExchangeRateMixin,
PluginRegistryMixin,
addUserPermission,
)
from order.status_codes import PurchaseOrderStatus
from part.models import Part
from stock.models import StockItem, StockLocation
@@ -23,7 +27,7 @@ from users.models import Owner
from .models import PurchaseOrder, PurchaseOrderExtraLine, PurchaseOrderLineItem
class OrderTest(TestCase, ExchangeRateMixin):
class OrderTest(ExchangeRateMixin, PluginRegistryMixin, TestCase):
"""Tests to ensure that the order models are functioning correctly."""
fixtures = [
@@ -421,6 +425,8 @@ class OrderTest(TestCase, ExchangeRateMixin):
Ensure that a notification is sent when a PurchaseOrder becomes overdue
"""
self.ensurePluginsLoaded()
po = PurchaseOrder.objects.get(pk=1)
# Ensure that the right users have the right permissions

8
src/backend/InvenTree/plugin/api.py
View File

@@ -94,10 +94,14 @@ class PluginFilter(rest_filters.FilterSet):
def filter_mandatory(self, queryset, name, value):
"""Filter by 'mandatory' flag."""
from django.conf import settings
mandatory_keys = [*registry.MANDATORY_PLUGINS, *settings.PLUGINS_MANDATORY]
if str2bool(value):
return queryset.filter(key__in=registry.MANDATORY_PLUGINS)
return queryset.filter(key__in=mandatory_keys)
else:
return queryset.exclude(key__in=registry.MANDATORY_PLUGINS)
return queryset.exclude(key__in=mandatory_keys)
sample = rest_filters.BooleanFilter(
field_name='sample', label=_('Sample'), method='filter_sample'

3
src/backend/InvenTree/plugin/base/integration/ScheduleMixin.py
View File

@@ -68,8 +68,9 @@ class ScheduleMixin:
if settings.PLUGIN_TESTING or get_global_setting('ENABLE_PLUGINS_SCHEDULE'):
for _key, plugin in plugins:
if (
plugin.mixin_enabled(PluginMixinEnum.SCHEDULE)
plugin
and plugin.is_active()
and plugin.mixin_enabled(PluginMixinEnum.SCHEDULE)
):
# Only active tasks for plugins which are enabled
plugin.register_tasks()

26
src/backend/InvenTree/plugin/base/label/test_label_mixin.py
View File

@@ -60,23 +60,37 @@ class LabelMixinTests(PrintTestMixins, InvenTreeAPITestCase):
def test_api(self):
"""Test that we can filter the API endpoint by mixin."""
self.ensurePluginsLoaded(force=True)
url = reverse('api-plugin-list')
# Try POST (disallowed)
response = self.client.post(url, {})
self.assertEqual(response.status_code, 405)
response = self.client.get(url, {'mixin': 'labels', 'active': True})
response = self.client.get(
url, {'mixin': PluginMixinEnum.LABELS, 'active': True}
)
# No results matching this query!
self.assertEqual(len(response.data), 0)
# Two mandatory label printing plugins
self.assertEqual(len(response.data), 2)
# What about inactive?
response = self.client.get(url, {'mixin': 'labels', 'active': False})
self.assertEqual(len(response.data), 0)
# One builtin, non-mandatory label printing plugin "inventreelabelsheet"
# One sample plugin, "samplelabelprinter"
self.assertEqual(len(response.data), 2)
self.assertEqual(response.data[0]['key'], 'inventreelabelsheet')
self.assertEqual(response.data[1]['key'], 'samplelabelprinter')
with self.assertWarnsMessage(
UserWarning,
'A plugin registry reload was triggered for plugin samplelabelprinter',
):
# Activate the sample label printing plugin
self.do_activate_plugin()
self.do_activate_plugin()
# Should be available via the API now
response = self.client.get(url, {'mixin': 'labels', 'active': True})
@@ -141,7 +155,7 @@ class LabelMixinTests(PrintTestMixins, InvenTreeAPITestCase):
{'template': template.pk, 'plugin': config.key, 'items': [1, 2, 3]},
expected_code=400,
)
self.assertIn('Plugin is not active', str(response.data['plugin']))
self.assertIn('Plugin not found', str(response.data['plugin']))
# Active plugin
self.do_activate_plugin()

22
src/backend/InvenTree/plugin/broken/bad_actor.py Normal file
View File

@@ -0,0 +1,22 @@
"""Attempt to create a bad actor plugin that overrides internal methods."""
from plugin import InvenTreePlugin
class BadActorPlugin(InvenTreePlugin):
"""A plugin that attempts to override internal methods."""
SLUG = 'bad_actor'
def __init__(self, *args, **kwargs):
"""Initialize the plugin."""
super().__init__(*args, **kwargs)
self.add_mixin('settings', 'has_settings', __class__)
def plugin_slug(self) -> str:
"""Return the slug of this plugin."""
return 'bad_actor'
def plugin_name(self) -> str:
"""Return the name of this plugin."""
return 'Bad Actor Plugin'

5
src/backend/InvenTree/plugin/builtin/barcodes/test_inventree_barcode.py
View File

@@ -12,6 +12,11 @@ class TestInvenTreeBarcode(InvenTreeAPITestCase):
fixtures = ['category', 'part', 'location', 'stock', 'company', 'supplier_part']
def setUp(self):
"""Set up the test case."""
super().setUp()
self.ensurePluginsLoaded()
def test_assign_errors(self):
"""Test error cases for assignment action."""

13
src/backend/InvenTree/plugin/installer.py
View File

@@ -341,6 +341,19 @@ def uninstall_plugin(cfg: plugin.models.PluginConfig, user=None, delete_config=T
_('Plugin cannot be uninstalled as it is currently active')
)
if cfg.is_mandatory():
raise ValidationError(_('Plugin cannot be uninstalled as it is mandatory'))
if cfg.is_sample():
raise ValidationError(
_('Plugin cannot be uninstalled as it is a sample plugin')
)
if cfg.is_builtin():
raise ValidationError(
_('Plugin cannot be uninstalled as it is a built-in plugin')
)
if not cfg.is_installed():
raise ValidationError(_('Plugin is not installed'))

34
src/backend/InvenTree/plugin/models.py
View File

@@ -145,17 +145,24 @@ class PluginConfig(InvenTree.models.MetadataMixin, models.Model):
"""Extend save method to reload plugins if the 'active' status changes."""
no_reload = kwargs.pop('no_reload', False) # check if no_reload flag is set
super().save(force_insert, force_update, *args, **kwargs)
mandatory = self.is_mandatory()
if self.is_mandatory():
if mandatory:
# Force active if mandatory plugin
self.active = True
if not no_reload and self.active != self.__org_active:
super().save(force_insert, force_update, *args, **kwargs)
if not no_reload and self.active != self.__org_active and not mandatory:
if settings.PLUGIN_TESTING:
warnings.warn('A plugin registry reload was triggered', stacklevel=2)
warnings.warn(
f'A plugin registry reload was triggered for plugin {self.key}',
stacklevel=2,
)
registry.reload_plugins(full_reload=True, force_reload=True, collect=True)
self.__org_active = self.active
@admin.display(boolean=True, description=_('Installed'))
def is_installed(self) -> bool:
"""Simple check to determine if this plugin is installed.
@@ -184,15 +191,32 @@ class PluginConfig(InvenTree.models.MetadataMixin, models.Model):
@admin.display(boolean=True, description=_('Mandatory Plugin'))
def is_mandatory(self) -> bool:
"""Return True if this plugin is mandatory."""
# List of run-time configured mandatory plugins
if settings.PLUGINS_MANDATORY:
if self.key in settings.PLUGINS_MANDATORY:
return True
# Hard-coded list of mandatory "builtin" plugins
return self.key in registry.MANDATORY_PLUGINS
def is_active(self) -> bool:
"""Return True if this plugin is active.
Note that 'mandatory' plugins are always considered 'active',
"""
return self.is_mandatory() or self.active
@admin.display(boolean=True, description=_('Package Plugin'))
def is_package(self) -> bool:
"""Return True if this is a 'package' plugin."""
if self.package_name:
return True
if not self.plugin:
return False
return getattr(self.plugin, 'is_package', False)
pkg_name = getattr(self.plugin, 'package_name', None)
return pkg_name is not None
@property
def admin_source(self) -> str:

108
src/backend/InvenTree/plugin/plugin.py
View File

@@ -21,6 +21,37 @@ from plugin.helpers import get_git_log
logger = structlog.get_logger('inventree')
def is_method_like(method) -> bool:
"""Check if a method is callable and not a property."""
return any([
callable(method),
isinstance(method, classmethod),
isinstance(method, staticmethod),
isinstance(method, property),
])
def mark_final(method):
"""Decorator to mark a method as 'final'.
This prevents subclasses from overriding this method.
"""
if not is_method_like(method):
raise TypeError('mark_final can only be applied to functions')
method.__final__ = True
return method
def get_final_methods(cls):
"""Find all methods of a class marked with the @mark_final decorator."""
return [
name
for name, method in inspect.getmembers(cls)
if getattr(method, '__final__', False) and is_method_like(method)
]
class PluginMixinEnum(StringEnum):
"""Enumeration of the available plugin mixin types."""
@@ -57,6 +88,7 @@ class MetaBase:
SLUG = None
TITLE = None
@mark_final
def get_meta_value(self, key: str, old_key: Optional[str] = None, __default=None):
"""Reference a meta item with a key.
@@ -87,15 +119,18 @@ class MetaBase:
return __default
return value
@mark_final
def plugin_name(self):
"""Name of plugin."""
return self.get_meta_value('NAME', 'PLUGIN_NAME')
@property
@mark_final
def name(self):
"""Name of plugin."""
return self.plugin_name()
@mark_final
def plugin_slug(self):
"""Slug of plugin.
@@ -108,10 +143,12 @@ class MetaBase:
return slugify(slug.lower())
@property
@mark_final
def slug(self):
"""Slug of plugin."""
return self.plugin_slug()
@mark_final
def plugin_title(self):
"""Title of plugin."""
title = self.get_meta_value('TITLE', 'PLUGIN_TITLE', None)
@@ -120,28 +157,31 @@ class MetaBase:
return self.plugin_name()
@property
@mark_final
def human_name(self):
"""Human readable name of plugin."""
return self.plugin_title()
@mark_final
def plugin_config(self):
"""Return the PluginConfig object associated with this plugin."""
from plugin.registry import registry
return registry.get_plugin_config(self.plugin_slug())
def is_active(self):
@mark_final
def is_active(self) -> bool:
"""Return True if this plugin is currently active."""
# Mandatory plugins are always considered "active"
if self.is_builtin and self.is_mandatory:
if self.is_mandatory():
return True
config = self.plugin_config()
if config:
return config.active
return config.is_active()
return False # pragma: no cover
return False
class MixinBase:
@@ -156,11 +196,13 @@ class MixinBase:
self._mixins = {}
super().__init__(*args, **kwargs)
@mark_final
def mixin(self, key: str) -> bool:
"""Check if mixin is registered."""
key = str(key).lower()
return key in self._mixins
@mark_final
def mixin_enabled(self, key: str) -> bool:
"""Check if mixin is registered, enabled and ready."""
key = str(key).lower()
@@ -181,6 +223,7 @@ class MixinBase:
return False
@mark_final
def add_mixin(self, key: str, fnc_enabled=True, cls=None):
"""Add a mixin to the plugins registry."""
key = str(key).lower()
@@ -188,6 +231,7 @@ class MixinBase:
self._mixins[key] = fnc_enabled
self.setup_mixin(key, cls=cls)
@mark_final
def setup_mixin(self, key, cls=None):
"""Define mixin details for the current mixin -> provides meta details for all active mixins."""
# get human name
@@ -200,6 +244,7 @@ class MixinBase:
# register
self._mixinreg[key] = {'key': key, 'human_name': human_name, 'cls': cls}
@mark_final
def get_registered_mixins(self, with_base: bool = False, with_cls: bool = True):
"""Get all registered mixins for the plugin."""
mixins = getattr(self, '_mixinreg', None)
@@ -220,6 +265,7 @@ class MixinBase:
return mixins
@property
@mark_final
def registered_mixins(self, with_base: bool = False):
"""Get all registered mixins for the plugin."""
return self.get_registered_mixins(with_base=with_base)
@@ -231,6 +277,7 @@ class VersionMixin:
MIN_VERSION = None
MAX_VERSION = None
@mark_final
def check_version(self, latest=None) -> bool:
"""Check if plugin functions for the current InvenTree version."""
from InvenTree import version
@@ -269,11 +316,33 @@ class InvenTreePlugin(VersionMixin, MixinBase, MetaBase):
self.define_package()
def __init_subclass__(cls):
"""Custom code to initialize a subclass of InvenTreePlugin.
This is a security measure to prevent plugins from overriding methods
which are decorated with @mark_final.
"""
final_methods = get_final_methods(InvenTreePlugin)
child_methods = [
name for name, method in cls.__dict__.items() if is_method_like(method)
]
for name in child_methods:
if name in final_methods:
raise TypeError(
f"Plugin '{cls.__name__}' cannot override final method '{name}' from InvenTreePlugin."
)
return super().__init_subclass__()
@mark_final
@classmethod
def file(cls) -> Path:
"""File that contains plugin definition."""
return Path(inspect.getfile(cls))
@mark_final
@classmethod
def path(cls) -> Path:
"""Path to plugins base folder."""
@@ -296,6 +365,7 @@ class InvenTreePlugin(VersionMixin, MixinBase, MetaBase):
# region properties
@property
@mark_final
def description(self):
"""Description of plugin."""
description = self._get_value('DESCRIPTION', 'description')
@@ -304,6 +374,7 @@ class InvenTreePlugin(VersionMixin, MixinBase, MetaBase):
return description
@property
@mark_final
def author(self):
"""Author of plugin - either from plugin settings or git."""
author = self._get_value('AUTHOR', 'author')
@@ -312,6 +383,7 @@ class InvenTreePlugin(VersionMixin, MixinBase, MetaBase):
return author
@property
@mark_final
def pub_date(self):
"""Publishing date of plugin - either from plugin settings or git."""
pub_date = getattr(self, 'PUBLISH_DATE', None)
@@ -323,16 +395,19 @@ class InvenTreePlugin(VersionMixin, MixinBase, MetaBase):
return pub_date
@property
@mark_final
def version(self):
"""Version of plugin."""
return self._get_value('VERSION', 'version')
@property
@mark_final
def website(self):
"""Website of plugin - if set else None."""
return self._get_value('WEBSITE', 'website')
@property
@mark_final
def license(self):
"""License of plugin."""
return self._get_value('LICENSE', 'license')
@@ -340,43 +415,52 @@ class InvenTreePlugin(VersionMixin, MixinBase, MetaBase):
# endregion
@classmethod
@mark_final
def check_is_package(cls):
"""Is the plugin delivered as a package."""
return getattr(cls, 'is_package', False)
@property
@mark_final
def _is_package(self):
"""Is the plugin delivered as a package."""
return getattr(self, 'is_package', False)
@classmethod
@mark_final
def check_is_sample(cls) -> bool:
"""Is this plugin part of the samples?"""
return str(cls.check_package_path()).startswith('plugin/samples/')
@property
@mark_final
def is_sample(self) -> bool:
"""Is this plugin part of the samples?"""
return self.check_is_sample()
@classmethod
@mark_final
def check_is_builtin(cls) -> bool:
"""Determine if a particular plugin class is a 'builtin' plugin."""
return str(cls.check_package_path()).startswith('plugin/builtin')
@property
@mark_final
def is_builtin(self) -> bool:
"""Is this plugin is builtin."""
return self.check_is_builtin()
@property
@mark_final
def is_mandatory(self) -> bool:
"""Is this plugin mandatory (always forced to be active)."""
from plugin.registry import registry
config = self.plugin_config()
if config:
# If the plugin is configured, check if it is marked as mandatory
return config.is_mandatory()
return self.slug in registry.MANDATORY_PLUGINS
return False # pragma: no cover
@classmethod
@mark_final
def check_package_path(cls):
"""Path to the plugin."""
if cls.check_is_package():
@@ -388,11 +472,13 @@ class InvenTreePlugin(VersionMixin, MixinBase, MetaBase):
return cls.file()
@property
@mark_final
def package_path(self):
"""Path to the plugin."""
return self.check_package_path()
@classmethod
@mark_final
def check_package_install_name(cls) -> Union[str, None]:
"""Installable package name of the plugin.
@@ -405,6 +491,7 @@ class InvenTreePlugin(VersionMixin, MixinBase, MetaBase):
return getattr(cls, 'package_name', None)
@property
@mark_final
def package_install_name(self) -> Union[str, None]:
"""Installable package name of the plugin.
@@ -417,6 +504,7 @@ class InvenTreePlugin(VersionMixin, MixinBase, MetaBase):
return self.check_package_install_name()
@property
@mark_final
def settings_url(self) -> str:
"""URL to the settings panel for this plugin."""
if config := self.db:
@@ -424,11 +512,13 @@ class InvenTreePlugin(VersionMixin, MixinBase, MetaBase):
return InvenTree.helpers.pui_url('/settings/admin/plugin/')
# region package info
@mark_final
def _get_package_commit(self):
"""Get last git commit for the plugin."""
return get_git_log(str(self.file()))
@classmethod
@mark_final
def is_editable(cls):
"""Returns if the current part is editable."""
pkg_name = cls.__name__.split('.')[0]
@@ -436,6 +526,7 @@ class InvenTreePlugin(VersionMixin, MixinBase, MetaBase):
return bool(len(dist_info) == 1)
@classmethod
@mark_final
def _get_package_metadata(cls):
"""Get package metadata for plugin."""
# Try simple metadata lookup
@@ -482,6 +573,7 @@ class InvenTreePlugin(VersionMixin, MixinBase, MetaBase):
# endregion
@mark_final
def plugin_static_file(self, *args) -> str:
"""Construct a path to a static file within the plugin directory.

92
src/backend/InvenTree/plugin/registry.py
View File

@@ -34,6 +34,7 @@ from InvenTree.ready import canAppAccessDatabase
from .helpers import (
IntegrationPluginError,
MixinNotImplementedError,
get_entrypoints,
get_plugins,
handle_error,
@@ -144,6 +145,8 @@ class PluginsRegistry:
"""
from common.models import InvenTreeSetting
logger.info('Initializing plugin registry')
self.ready = True
# Install plugins from file (if required)
@@ -184,7 +187,13 @@ class PluginsRegistry:
plg = self.plugins[slug]
if active is not None and active != plg.is_active():
config = self.get_plugin_config(slug)
if not config: # pragma: no cover
logger.warning("Plugin '%s' has no configuration", slug)
return None
if active is not None and active != config.is_active():
return None
if with_mixin is not None and not plg.mixin_enabled(with_mixin):
@@ -209,9 +218,12 @@ class PluginsRegistry:
cfg = PluginConfig.objects.filter(key=slug).first()
if not cfg:
logger.debug(
"get_plugin_config: Creating new PluginConfig for '%s'", slug
)
cfg = PluginConfig.objects.create(key=slug)
except PluginConfig.DoesNotExist:
except PluginConfig.DoesNotExist: # pragma: no cover
return None
except (IntegrityError, OperationalError, ProgrammingError): # pragma: no cover
return None
@@ -285,25 +297,44 @@ class PluginsRegistry:
active (bool, optional): Filter by 'active' status of plugin. Defaults to True.
builtin (bool, optional): Filter by 'builtin' status of plugin. Defaults to None.
"""
try:
# Pre-fetch the PluginConfig objects to avoid multiple database queries
from plugin.models import PluginConfig
plugin_configs = PluginConfig.objects.all()
configs = {config.key: config for config in plugin_configs}
except (ProgrammingError, OperationalError):
# The database is not ready yet
logger.warning('plugin.registry.with_mixin: Database not ready')
return []
mixin = str(mixin).lower().strip()
result = []
plugins = []
for plugin in self.plugins.values():
if plugin.mixin_enabled(mixin):
if active is not None:
# Filter by 'active' status of plugin
if active != plugin.is_active():
continue
try:
if not plugin.mixin_enabled(mixin):
continue
except MixinNotImplementedError:
continue
if builtin is not None:
# Filter by 'builtin' status of plugin
if builtin != plugin.is_builtin:
continue
config = configs.get(plugin.slug) or plugin.plugin_config()
result.append(plugin)
# No config - cannot use this plugin
if not config:
continue
return result
if active is not None and active != config.is_active():
continue
if builtin is not None and builtin != config.is_builtin():
continue
plugins.append(plugin)
return plugins
# endregion
@@ -421,6 +452,15 @@ class PluginsRegistry:
self.update_plugin_hash()
logger.info('Plugin Registry: Loaded %s plugins', len(self.plugins))
# Ensure that each loaded plugin has a valid configuration object in the database
for plugin in self.plugins.values():
config = self.get_plugin_config(plugin.slug)
# Ensure mandatory plugins are marked as active
if config.is_mandatory() and not config.active:
config.active = True
config.save(no_reload=True)
except Exception as e:
logger.exception('Unexpected error during plugin reload: %s', e)
log_error('reload_plugins', scope='plugins')
@@ -611,8 +651,8 @@ class PluginsRegistry:
plugin.db = plg_db
# Check if this is a 'builtin' plugin
builtin = plugin.check_is_builtin()
sample = plugin.check_is_sample()
builtin = plg_db.is_builtin() if plg_db else plugin.check_is_builtin()
sample = plg_db.is_sample() if plg_db else plugin.check_is_sample()
package_name = None
@@ -621,7 +661,7 @@ class PluginsRegistry:
package_name = getattr(plugin, 'package_name', None)
# Auto-enable default builtin plugins
if builtin and plg_db and plg_db.is_mandatory():
if plg_db and plg_db.is_mandatory():
if not plg_db.active:
plg_db.active = True
plg_db.save()
@@ -631,11 +671,16 @@ class PluginsRegistry:
plg_db.package_name = package_name
plg_db.save()
# Check if this plugin is considered 'mandatory'
mandatory = (
plg_key in self.MANDATORY_PLUGINS or plg_key in settings.PLUGINS_MANDATORY
)
# Determine if this plugin should be loaded:
# - If PLUGIN_TESTING is enabled
# - If this is a 'builtin' plugin
# - If this is a 'mandatory' plugin
# - If this plugin has been explicitly enabled by the user
if settings.PLUGIN_TESTING or builtin or (plg_db and plg_db.active):
if settings.PLUGIN_TESTING or mandatory or (plg_db and plg_db.active):
# Initialize package - we can be sure that an admin has activated the plugin
logger.debug('Loading plugin `%s`', plg_name)
@@ -668,6 +713,15 @@ class PluginsRegistry:
plg_i: InvenTreePlugin = plugin()
dt = time.time() - t_start
logger.debug('Loaded plugin `%s` in %.3fs', plg_name, dt)
if mandatory and not plg_db.active: # pragma: no cover
# If this is a mandatory plugin, ensure it is marked as active
logger.info(
'Plugin `%s` is a mandatory plugin - activating', plg_name
)
plg_db.active = True
plg_db.save()
except ModuleNotFoundError as e:
raise e
except Exception as error:

9
src/backend/InvenTree/plugin/serializers.py
View File

@@ -234,9 +234,14 @@ class PluginActivateSerializer(serializers.Serializer):
help_text=_('Activate this plugin'),
)
def update(self, instance, validated_data):
def update(self, instance: PluginConfig, validated_data):
"""Apply the new 'active' value to the plugin instance."""
instance.activate(validated_data.get('active', True))
active = validated_data.get('active', True)
if not active and instance.is_mandatory():
raise ValidationError(_('Mandatory plugin cannot be deactivated'))
instance.activate(active)
return instance

226
src/backend/InvenTree/plugin/test_api.py
View File

@@ -23,12 +23,48 @@ class PluginDetailAPITest(PluginMixin, InvenTreeAPITestCase):
self.PKG_URL = 'git+https://github.com/inventree/inventree-brother-plugin'
super().setUp()
def test_plugin_uninstall(self):
"""Test plugin uninstall command."""
# invalid package name
url = reverse('api-plugin-uninstall', kwargs={'plugin': 'samplexx'})
# Requires superuser permissions
self.patch(url, expected_code=403)
self.user.is_superuser = True
self.user.save()
# Invalid slug (404 error)
self.patch(url, expected_code=404)
url = reverse('api-plugin-uninstall', kwargs={'plugin': 'sample'})
data = self.patch(url, expected_code=400).data
plugs = {
'sample': 'Plugin cannot be uninstalled as it is a sample plugin',
'bom-exporter': 'Plugin cannot be uninstalled as it is currently active',
'inventree-slack-notification': 'Plugin cannot be uninstalled as it is a built-in plugin',
}
for slug, msg in plugs.items():
url = reverse('api-plugin-uninstall', kwargs={'plugin': slug})
data = self.patch(url, expected_code=400).data
self.assertIn(msg, str(data))
with self.settings(PLUGINS_INSTALL_DISABLED=True):
url = reverse('api-plugin-uninstall', kwargs={'plugin': 'bom-exporter'})
data = self.patch(url, expected_code=400).data
self.assertIn(
'Plugin uninstalling is disabled', str(data['non_field_errors'])
)
def test_plugin_install(self):
"""Test the plugin install command."""
url = reverse('api-plugin-install')
# invalid package name
self.post(
data = self.post(
url,
{
'confirm': True,
@@ -36,7 +72,12 @@ class PluginDetailAPITest(PluginMixin, InvenTreeAPITestCase):
},
expected_code=400,
max_query_time=60,
).data
self.assertIn(
'ERROR: Could not find a version that satisfies the requirement', str(data)
)
self.assertIn('ERROR: No matching distribution found for', str(data))
# valid - Pypi
data = self.post(
@@ -69,7 +110,8 @@ class PluginDetailAPITest(PluginMixin, InvenTreeAPITestCase):
# invalid tries
# no input
self.post(url, {}, expected_code=400)
data = self.post(url, {}, expected_code=400).data
self.assertIn('This field is required.', str(data['confirm']))
# no package info
data = self.post(url, {'confirm': True}, expected_code=400).data
@@ -80,7 +122,8 @@ class PluginDetailAPITest(PluginMixin, InvenTreeAPITestCase):
)
# not confirmed
self.post(url, {'packagename': self.PKG_NAME}, expected_code=400)
data = self.post(url, {'packagename': self.PKG_NAME}, expected_code=400).data
self.assertIn('This field is required.', str(data['confirm']))
data = self.post(
url, {'packagename': self.PKG_NAME, 'confirm': False}, expected_code=400
@@ -90,19 +133,41 @@ class PluginDetailAPITest(PluginMixin, InvenTreeAPITestCase):
data['confirm'][0].title().upper(), 'Installation not confirmed'.upper()
)
# install disabled
# Plugin installation disabled
with self.settings(PLUGINS_INSTALL_DISABLED=True):
self.post(url, {}, expected_code=400)
response = self.post(
url,
{'packagename': 'inventree-order-history', 'confirm': True},
expected_code=400,
)
self.assertIn(
'Plugin installation is disabled',
str(response.data['non_field_errors']),
)
def test_plugin_deactivate_mandatory(self):
"""Test deactivating a mandatory plugin."""
self.user.is_superuser = True
self.user.save()
# Get a mandatory plugin
plg = PluginConfig.objects.filter(key='bom-exporter').first()
assert plg is not None
url = reverse('api-plugin-detail-activate', kwargs={'plugin': plg.key})
# Try to deactivate the mandatory plugin
response = self.client.patch(url, {'active': False}, follow=True)
self.assertEqual(response.status_code, 400)
self.assertIn('Mandatory plugin cannot be deactivated', str(response.data))
def test_plugin_activate(self):
"""Test the plugin activate."""
test_plg = self.plugin_confs.first()
assert test_plg is not None
def assert_plugin_active(self, active):
plgs = PluginConfig.objects.all().first()
assert plgs is not None
self.assertEqual(plgs.active, active)
"""Test the plugin activation API endpoint."""
test_plg = PluginConfig.objects.get(key='samplelocate')
self.assertIsNotNone(test_plg, 'Test plugin not found')
self.assertFalse(test_plg.is_active())
self.assertFalse(test_plg.is_builtin())
self.assertFalse(test_plg.is_mandatory())
url = reverse('api-plugin-detail-activate', kwargs={'plugin': test_plg.key})
@@ -119,20 +184,27 @@ class PluginDetailAPITest(PluginMixin, InvenTreeAPITestCase):
test_plg.save()
# Activate plugin with detail url
assert_plugin_active(self, False)
test_plg.refresh_from_db()
self.assertFalse(test_plg.is_active())
response = self.client.patch(url, {}, follow=True)
self.assertEqual(response.status_code, 200)
assert_plugin_active(self, True)
test_plg.refresh_from_db()
self.assertTrue(test_plg.is_active())
# Deactivate plugin
test_plg.active = False
test_plg.save()
# Activate plugin
assert_plugin_active(self, False)
test_plg.refresh_from_db()
self.assertFalse(test_plg.active)
response = self.client.patch(url, {}, follow=True)
self.assertEqual(response.status_code, 200)
assert_plugin_active(self, True)
test_plg.refresh_from_db()
self.assertTrue(test_plg.is_active())
def test_pluginCfg_delete(self):
"""Test deleting a config."""
@@ -228,7 +300,21 @@ class PluginDetailAPITest(PluginMixin, InvenTreeAPITestCase):
plg_inactive.active = True
plg_inactive.save()
self.assertEqual(cm.warning.args[0], 'A plugin registry reload was triggered')
self.assertEqual(
cm.warning.args[0],
f'A plugin registry reload was triggered for plugin {plg_inactive.key}',
)
# Set active state back to False
with self.assertWarns(Warning) as cm:
plg_inactive.active = False
plg_inactive.save()
self.assertEqual(
cm.warning.args[0],
f'A plugin registry reload was triggered for plugin {plg_inactive.key}',
)
def test_check_plugin(self):
"""Test check_plugin function."""
@@ -404,3 +490,107 @@ class PluginDetailAPITest(PluginMixin, InvenTreeAPITestCase):
self.user.is_superuser = False
self.user.save()
def test_plugin_filter_by_mixin(self):
"""Test filtering plugins by mixin."""
from plugin import PluginMixinEnum
from plugin.registry import registry
# Ensure we have some plugins loaded
registry.reload_plugins(full_reload=True, collect=True)
url = reverse('api-plugin-list')
# Filter by 'mixin' parameter
mixin_results = {
PluginMixinEnum.BARCODE: 5,
PluginMixinEnum.EXPORTER: 3,
PluginMixinEnum.ICON_PACK: 1,
PluginMixinEnum.MAIL: 1,
PluginMixinEnum.NOTIFICATION: 3,
PluginMixinEnum.USER_INTERFACE: 1,
}
for mixin, expected_count in mixin_results.items():
data = self.get(url, {'mixin': mixin}).data
self.assertEqual(len(data), expected_count)
if expected_count > 0:
for item in data:
self.assertIn(mixin, item['mixins'])
def test_plugin_filters(self):
"""Unit testing for plugin API filters."""
from plugin.models import PluginConfig
from plugin.registry import registry
PluginConfig.objects.all().delete()
registry.reload_plugins(full_reload=True, collect=True)
N = PluginConfig.objects.count()
self.assertGreater(N, 0)
url = reverse('api-plugin-list')
data = self.get(url).data
self.assertGreater(len(data), 0)
self.assertEqual(len(data), N)
# Filter by 'builtin' plugins
data = self.get(url, {'builtin': 'true'}).data
Y_BUILTIN = len(data)
for item in data:
self.assertTrue(item['is_builtin'])
data = self.get(url, {'builtin': 'false'}).data
N_BUILTIN = len(data)
for item in data:
self.assertFalse(item['is_builtin'])
self.assertGreater(Y_BUILTIN, 0)
self.assertGreater(N_BUILTIN, 0)
self.assertEqual(N_BUILTIN + Y_BUILTIN, N)
# Filter by 'active' status
Y_ACTIVE = len(self.get(url, {'active': 'true'}).data)
N_ACTIVE = len(self.get(url, {'active': 'false'}).data)
self.assertGreater(Y_ACTIVE, 0)
self.assertGreater(N_ACTIVE, 0)
self.assertEqual(Y_ACTIVE + N_ACTIVE, N)
# Filter by 'sample' status
Y_SAMPLE = len(self.get(url, {'sample': 'true'}).data)
N_SAMPLE = len(self.get(url, {'sample': 'false'}).data)
self.assertGreater(Y_SAMPLE, 0)
self.assertGreater(N_SAMPLE, 0)
self.assertEqual(Y_SAMPLE + N_SAMPLE, N)
# Filter by 'mandatory' status`
Y_MANDATORY = len(self.get(url, {'mandatory': 'true'}).data)
N_MANDATORY = len(self.get(url, {'mandatory': 'false'}).data)
self.assertGreater(Y_MANDATORY, 0)
self.assertGreater(N_MANDATORY, 0)
self.assertEqual(Y_MANDATORY + N_MANDATORY, N)
# Add in a new mandatory plugin
with self.settings(PLUGINS_MANDATORY=['samplelocate']):
registry.reload_plugins(full_reload=True, collect=True)
Y_MANDATORY_2 = len(self.get(url, {'mandatory': 'true'}).data)
N_MANDATORY_2 = len(self.get(url, {'mandatory': 'false'}).data)
self.assertEqual(Y_MANDATORY_2, Y_MANDATORY + 1)
self.assertEqual(N_MANDATORY_2, N_MANDATORY - 1)

242
src/backend/InvenTree/plugin/test_plugin.py
View File

@@ -13,7 +13,9 @@ from unittest.mock import patch
from django.test import TestCase, override_settings
import plugin.templatetags.plugin_extras as plugin_tags
from plugin import InvenTreePlugin, registry
from InvenTree.unit_test import PluginRegistryMixin, TestQueryMixin
from plugin import InvenTreePlugin, PluginMixinEnum
from plugin.registry import registry
from plugin.samples.integration.another_sample import (
NoIntegrationPlugin,
WrongIntegrationPlugin,
@@ -24,7 +26,7 @@ from plugin.samples.integration.sample import SampleIntegrationPlugin
PLUGIN_TEST_DIR = '_testfolder/test_plugins'
class PluginTagTests(TestCase):
class PluginTagTests(PluginRegistryMixin, TestCase):
"""Tests for the plugin extras."""
def setUp(self):
@@ -60,7 +62,9 @@ class PluginTagTests(TestCase):
def test_mixin_available(self):
"""Check that mixin_available works."""
self.assertEqual(plugin_tags.mixin_available('barcode'), True)
from plugin import PluginMixinEnum
self.assertEqual(plugin_tags.mixin_available(PluginMixinEnum.BARCODE), True)
self.assertEqual(plugin_tags.mixin_available('wrong'), False)
def test_tag_safe_url(self):
@@ -203,7 +207,7 @@ class InvenTreePluginTests(TestCase):
self.assertEqual(plug.is_active(), False)
class RegistryTests(TestCase):
class RegistryTests(TestQueryMixin, PluginRegistryMixin, TestCase):
"""Tests for registry loading methods."""
def mockDir(self) -> str:
@@ -282,18 +286,76 @@ class RegistryTests(TestCase):
self.assertEqual(len(registry.errors), 3)
# There should be at least one discovery error in the module `broken_file`
self.assertGreater(len(registry.errors.get('discovery')), 0)
self.assertEqual(
registry.errors.get('discovery')[0]['broken_file'],
"name 'bb' is not defined",
errors = registry.errors
def find_error(group: str, key: str) -> str:
"""Find a matching error in the registry errors."""
for error in errors.get(group, []):
if key in error:
return error[key]
return None
# Check for expected errors in the registry
self.assertIn(
"Plugin 'BadActorPlugin' cannot override final method 'plugin_slug'",
find_error('discovery', 'bad_actor'),
)
# There should be at least one load error with an intentional KeyError
self.assertGreater(len(registry.errors.get('Test:init_plugin')), 0)
self.assertEqual(
registry.errors.get('Test:init_plugin')[0]['broken_sample'],
"'This is a dummy error'",
self.assertIn(
"name 'bb' is not defined", find_error('discovery', 'broken_file')
)
self.assertIn(
'This is a dummy error', find_error('Test:init_plugin', 'broken_sample')
)
def test_plugin_override_mandatory(self):
"""Test that a plugin cannot override the is_mandatory method."""
with self.assertRaises(TypeError) as e:
# Attempt to create a class which overrides the 'is_mandatory' method
class MyDummyPlugin(InvenTreePlugin):
"""A dummy plugin for testing."""
NAME = 'MyDummyPlugin'
SLUG = 'mydummyplugin'
TITLE = 'My Dummy Plugin'
VERSION = '1.0.0'
def is_mandatory(self):
"""Override is_mandatory to always return True."""
return True
# Check that the error message is as expected
self.assertIn(
"Plugin 'MyDummyPlugin' cannot override final method 'is_mandatory' from InvenTreePlugin",
str(e.exception),
)
def test_plugin_override_active(self):
"""Test that the plugin override works as expected."""
with self.assertRaises(TypeError) as e:
# Attempt to create a class which overrides the 'is_active' method
class MyDummyPlugin(InvenTreePlugin):
"""A dummy plugin for testing."""
NAME = 'MyDummyPlugin'
SLUG = 'mydummyplugin'
TITLE = 'My Dummy Plugin'
VERSION = '1.0.0'
def is_active(self):
"""Override is_active to always return True."""
return True
def __init_subclass__(cls):
"""Override __init_subclass__."""
# Ensure that overriding the __init_subclass__ method
# does not prevent the TypeError from being raised
# Check that the error message is as expected
self.assertIn(
"Plugin 'MyDummyPlugin' cannot override final method 'is_active' from InvenTreePlugin",
str(e.exception),
)
@override_settings(PLUGIN_TESTING=True, PLUGIN_TESTING_SETUP=True)
@@ -418,3 +480,155 @@ class RegistryTests(TestCase):
# Check that changed hashes run through
registry.registry_hash = 'abc'
self.assertTrue(registry.check_reload())
def test_builtin_mandatory_plugins(self):
"""Test that mandatory builtin plugins are always loaded."""
from plugin.models import PluginConfig
from plugin.registry import registry
# Start with a 'clean slate'
PluginConfig.objects.all().delete()
registry.reload_plugins(full_reload=True, collect=True)
mandatory = registry.MANDATORY_PLUGINS
self.assertEqual(len(mandatory), 9)
# Check that the mandatory plugins are loaded
self.assertEqual(
PluginConfig.objects.filter(active=True).count(), len(mandatory)
)
for key in mandatory:
cfg = registry.get_plugin_config(key)
self.assertIsNotNone(cfg, f"Mandatory plugin '{key}' not found in config")
self.assertTrue(cfg.is_mandatory())
self.assertTrue(cfg.active, f"Mandatory plugin '{key}' is not active")
self.assertTrue(cfg.is_active())
self.assertTrue(cfg.is_builtin())
plg = registry.get_plugin(key)
self.assertIsNotNone(plg, f"Mandatory plugin '{key}' not found")
self.assertTrue(
plg.is_mandatory, f"Plugin '{key}' is not marked as mandatory"
)
slug = 'bom-exporter'
self.assertIn(slug, mandatory)
cfg = registry.get_plugin_config(slug)
# Try to disable the mandatory plugin
cfg.active = False
cfg.save()
cfg.refresh_from_db()
# Mandatory plugin cannot be disabled!
self.assertTrue(cfg.active)
self.assertTrue(cfg.is_active())
def test_mandatory_plugins(self):
"""Test that plugins marked as 'mandatory' are always active."""
from plugin.models import PluginConfig
from plugin.registry import registry
# Start with a 'clean slate'
PluginConfig.objects.all().delete()
self.assertEqual(PluginConfig.objects.count(), 0)
registry.reload_plugins(full_reload=True, collect=True)
N_CONFIG = PluginConfig.objects.count()
N_ACTIVE = PluginConfig.objects.filter(active=True).count()
# Run checks across the registered plugin configurations
self.assertGreater(N_CONFIG, 0, 'No plugin configs found after reload')
self.assertGreater(N_ACTIVE, 0, 'No active plugin configs found after reload')
self.assertLess(
N_ACTIVE, N_CONFIG, 'All plugins are installed, but only some are active'
)
self.assertEqual(
N_ACTIVE,
len(registry.MANDATORY_PLUGINS),
'Not all mandatory plugins are active',
)
# Next, mark some additional plugins as mandatory
# These are a mix of "builtin" and "sample" plugins
mandatory_slugs = ['sampleui', 'validator', 'digikeyplugin', 'autocreatebuilds']
with self.settings(PLUGINS_MANDATORY=mandatory_slugs):
# Reload the plugins to apply the mandatory settings
registry.reload_plugins(full_reload=True, collect=True)
self.assertEqual(N_CONFIG, PluginConfig.objects.count())
self.assertEqual(
N_ACTIVE + 4, PluginConfig.objects.filter(active=True).count()
)
# Check that the mandatory plugins are active
for slug in mandatory_slugs:
cfg = registry.get_plugin_config(slug)
self.assertIsNotNone(
cfg, f"Mandatory plugin '{slug}' not found in config"
)
self.assertTrue(cfg.is_mandatory())
self.assertTrue(cfg.active, f"Mandatory plugin '{slug}' is not active")
self.assertTrue(cfg.is_active())
plg = registry.get_plugin(slug)
self.assertTrue(plg.is_active(), f"Plugin '{slug}' is not active")
self.assertIsNotNone(plg, f"Mandatory plugin '{slug}' not found")
self.assertTrue(
plg.is_mandatory, f"Plugin '{slug}' is not marked as mandatory"
)
def test_with_mixin(self):
"""Tests for the 'with_mixin' registry method."""
from plugin.models import PluginConfig
from plugin.registry import registry
self.ensurePluginsLoaded()
N_CONFIG = PluginConfig.objects.count()
self.assertGreater(N_CONFIG, 0, 'No plugin configs found')
# Test that the 'with_mixin' method is query efficient
for mixin in PluginMixinEnum:
with self.assertNumQueriesLessThan(3):
registry.with_mixin(mixin)
# Test for the 'base' mixin - we expect that this returns "all" plugins
base = registry.with_mixin(PluginMixinEnum.BASE, active=None, builtin=None)
self.assertEqual(len(base), N_CONFIG, 'Base mixin does not return all plugins')
# Next, fetch only "active" plugins
n_active = len(registry.with_mixin(PluginMixinEnum.BASE, active=True))
self.assertGreater(n_active, 0, 'No active plugins found with base mixin')
self.assertLess(n_active, N_CONFIG, 'All plugins are active with base mixin')
n_inactive = len(registry.with_mixin(PluginMixinEnum.BASE, active=False))
self.assertGreater(n_inactive, 0, 'No inactive plugins found with base mixin')
self.assertLess(n_inactive, N_CONFIG, 'All plugins are active with base mixin')
self.assertEqual(
n_active + n_inactive, N_CONFIG, 'Active and inactive plugins do not match'
)
# Filter by 'builtin' status
plugins = registry.with_mixin(PluginMixinEnum.LABELS, builtin=True, active=True)
self.assertEqual(len(plugins), 2)
keys = [p.slug for p in plugins]
self.assertIn('inventreelabel', keys)
self.assertIn('inventreelabelmachine', keys)
def test_config_attributes(self):
"""Test attributes for PluginConfig objects."""
self.ensurePluginsLoaded()
cfg = registry.get_plugin_config('bom-exporter')
self.assertIsNotNone(cfg, 'PluginConfig for bom-exporter not found')
self.assertTrue(cfg.is_mandatory())
self.assertTrue(cfg.is_active())
self.assertTrue(cfg.is_builtin())
self.assertFalse(cfg.is_package())
self.assertFalse(cfg.is_sample())

12
src/backend/InvenTree/report/api.py
View File

@@ -100,26 +100,18 @@ class LabelPrint(GenericAPIView):
def get_plugin_class(self, plugin_slug: str, raise_error=False):
"""Return the plugin class for the given plugin key."""
from plugin.models import PluginConfig
from plugin import registry
if not plugin_slug:
# Use the default label printing plugin
plugin_slug = InvenTreeLabelPlugin.NAME.lower()
plugin = None
try:
plugin_config = PluginConfig.objects.get(key=plugin_slug)
plugin = plugin_config.plugin
except (ValueError, PluginConfig.DoesNotExist):
pass
plugin = registry.get_plugin(plugin_slug, active=True)
error = None
if not plugin:
error = _('Plugin not found')
elif not plugin.is_active():
error = _('Plugin is not active')
elif not plugin.mixin_enabled(PluginMixinEnum.LABELS):
error = _('Plugin does not support label printing')

2
src/backend/InvenTree/report/tests.py
View File

@@ -393,7 +393,7 @@ class PrintTestMixins:
},
expected_code=201,
max_query_time=15,
max_query_count=500 * len(qs),
max_query_count=150 * len(qs),
)
# Test with wrong dimensions

3
src/backend/InvenTree/stock/test_api.py
View File

@@ -1593,13 +1593,10 @@ class StockItemTest(StockAPITestCase):
self.assertIn('This field is required', str(response.data['location']))
# TODO: Return to this and work out why it is taking so long
# Ref: https://github.com/inventree/InvenTree/pull/7157
response = self.post(
url,
{'location': '1', 'notes': 'Returned from this customer for testing'},
expected_code=201,
max_query_time=5.0,
)
item.refresh_from_db()