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

[Refactor] Notification plugins (#9735)

Browse Source 
* Refactor notification concept

- Notifications handled by plugins

* Cleanup

* Only send email if template provided in context

* Logic cleanup

* Fix log_error call

* Refactor error logging

- Ensure plugin slug is correctly attached
- Consistent format
- Logic fixes

* More robust plugin lookup

* Refactor calls to tringger_notification

* Tweak for build stock notification

* Low stock notification refactor

- Actually *use* the notification system
- Fix for email template

* Check stock only when build is issued

* Updated documentation

* Add PluginUserSetting class

- Allows plugins to define per-user settings

* Add API endpoints for PluginUserSetting model

* Placeholder for user-plugin-settings page

* Refactoring frontend code

* Placeholder panel

* Adds user interface for changing user-specific plugin settings

* Tweaks

* Remove old model

* Update documentation

* Playwright tests

* Update API version

* Fix unit test

* Fix removed arg

* Fixes for email notifications

- Track status of sending notifications
- Add helper "activate" method for plugin class
- Update unit tests

* Fix barcode tests

* More unit test fixes

* Test fixes

* Fix for settings models with extra fields

* Enhance unit test

* Remove old test file

* Check for null target_fnc

* Improve DB query efficiency

- Provide a flat list of active keys to plugin.is_active
- Prevents DB fetching (in certain circumstances)
- Add registry.active_plugins() method

* Bump query limit up for test

- In practice, this API endpoint is ~10 queries

* Handle potential errors

* Increase query limit for API test

* Increase query limit for some tests

* Bump API version

* Tweak unit test

* Tweak unit test

* Increased allowed queries

* fix user plugin settings

* Fix for unit test

* Update debug msg

* Tweak API

* Fix endpoint

* Remove "active plugin keys" code

* Restore previous behaviour

* Fix unit tests

* Tweak unit test

* Update src/backend/InvenTree/build/tasks.py

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Update src/backend/InvenTree/plugin/base/integration/NotificationMixin.py

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Func updates

* Format

* Add notification settings

* Refactor plugin settings groups

* Fix func type

* Adjust message

* Additional unit tests

* Additional playwright tests

* Additional playwright test

---------

Co-authored-by: Matthias Mair <code@mjmair.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
Oliver
2025-07-26 13:05:59 +10:00
committed by GitHub
parent f862867e2b
commit 1085625af4
60 changed files with 1329 additions and 972 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -6,6 +6,9 @@ INVENTREE_API_VERSION = 372
"""Increment this API version number whenever there is a significant change to the API that any clients need to know about."""
INVENTREE_API_TEXT = """
v358 -> 2025-06-21 : https://github.com/inventree/InvenTree/pull/9735
- Adds PluginUserSetting model (and associated endpoints)
- Remove NotificationSetting model (and associated endpoints)
v372 -> 2025-07-19 : https://github.com/inventree/InvenTree/pull/10056
- Adds BOM validation information to the Part API
@@ -83,7 +86,7 @@ v352 -> 2025-06-18 : https://github.com/inventree/InvenTree/pull/9803
- Add valid fields to ordering field descriptions
v351 -> 2025-06-18 : https://github.com/inventree/InvenTree/pull/9602
- Adds passwort reset API endpoint for admin users
- Adds password reset API endpoint for admin users
v350 -> 2025-06-17 : https://github.com/inventree/InvenTree/pull/9798
- Adds "can_build" field to the part requirements API endpoint

8
src/backend/InvenTree/InvenTree/apps.py
View File

@@ -36,7 +36,6 @@ class InvenTreeConfig(AppConfig):
- Cleaning up tasks
- Starting regular tasks
- Updating exchange rates
- Collecting notification methods
- Collecting state transition methods
- Adding users set in the current environment
"""
@@ -71,7 +70,6 @@ class InvenTreeConfig(AppConfig):
InvenTree.tasks.offload_task(InvenTree.tasks.check_for_migrations)
self.update_site_url()
self.collect_notification_methods()
self.collect_state_transition_methods()
# Ensure the unit registry is loaded
@@ -314,12 +312,6 @@ class InvenTreeConfig(AppConfig):
# do not try again
settings.USER_ADDED_FILE = True
def collect_notification_methods(self):
"""Collect all notification methods."""
from common.notifications import storage
storage.collect()
def collect_state_transition_methods(self):
"""Collect all state transition methods."""
from generic.states import storage

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

@@ -92,7 +92,7 @@ def send_email(
# If we still don't have a valid from_email, then we can't send emails
if not from_email:
if settings.TESTING:
from_email = 'from@test.com'
from_email = 'test@test.inventree.org'
else:
logger.error(
'INVE-W7: send_email failed: DEFAULT_FROM_EMAIL not specified'

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

@@ -1199,10 +1199,11 @@ def notify_staff_users_of_error(instance, label: str, context: dict):
"""Helper function to notify staff users of an error."""
import common.models
import common.notifications
from plugin.builtin.integration.core_notifications import InvenTreeUINotifications
try:
# Get all staff users
staff_users = get_user_model().objects.filter(is_staff=True)
staff_users = get_user_model().objects.filter(is_active=True, is_staff=True)
target_users = []
@@ -1219,7 +1220,7 @@ def notify_staff_users_of_error(instance, label: str, context: dict):
label,
context=context,
targets=target_users,
delivery_methods={common.notifications.UIMessageNotification},
delivery_methods={InvenTreeUINotifications},
)
except Exception as exc:

8
src/backend/InvenTree/InvenTree/permissions.py
View File

@@ -365,6 +365,14 @@ class UserSettingsPermissionsOrScope(OASTokenMixin, permissions.BasePermission):
return user == obj.user
def has_permission(self, request, view):
"""Check that the requesting user is authenticated."""
try:
user = request.user
return user.is_authenticated
except AttributeError:
return False
def get_required_alternate_scopes(self, request, view):
"""Return the required scopes for the current request."""
return map_scope(only_read=True)

30
src/backend/InvenTree/InvenTree/tasks.py
View File

@@ -16,6 +16,7 @@ from django.db import DEFAULT_DB_ALIAS, connections
from django.db.migrations.executor import MigrationExecutor
from django.db.utils import NotSupportedError, OperationalError, ProgrammingError
from django.utils import timezone
from django.utils.translation import gettext_lazy as _
import requests
import structlog
@@ -470,7 +471,10 @@ def delete_old_notifications():
def check_for_updates():
"""Check if there is an update for InvenTree."""
try:
from common.notifications import trigger_superuser_notification
from common.notifications import trigger_notification
from plugin.builtin.integration.core_notifications import (
InvenTreeUINotifications,
)
except AppRegistryNotReady: # pragma: no cover
# Apps not yet loaded!
logger.info("Could not perform 'check_for_updates' - App registry not ready")
@@ -533,19 +537,16 @@ def check_for_updates():
# Send notification if there is a new version
if not isInvenTreeUpToDate():
logger.warning('InvenTree is not up-to-date, sending notification')
plg = registry.get_plugin('InvenTreeCoreNotificationsPlugin')
if not plg:
logger.warning('Cannot send notification - plugin not found')
return
plg = plg.plugin_config()
if not plg:
logger.warning('Cannot send notification - plugin config not found')
return
# Send notification
trigger_superuser_notification(
plg, f'An update for InvenTree to version {tag} is available'
# Send notification to superusers
trigger_notification(
None,
'update_available',
targets=get_user_model().objects.filter(is_superuser=True),
delivery_methods={InvenTreeUINotifications},
context={
'name': _('Update Available'),
'message': _('An update for InvenTree is available'),
},
)
@@ -641,7 +642,6 @@ def check_for_migrations(force: bool = False, reload_registry: bool = True) -> b
Returns bool indicating if migrations are up to date
"""
from plugin import registry
def set_pending_migrations(n: int):
"""Helper function to inform the user about pending migrations."""

5
src/backend/InvenTree/InvenTree/templatetags/inventree_extras.py
View File

@@ -152,11 +152,6 @@ def setting_object(key, *args, **kwargs):
key, plugin=plg, cache=cache
)
elif 'method' in kwargs:
return plugin.models.NotificationUserSetting.get_setting_object(
key, user=kwargs['user'], method=kwargs['method'], cache=cache
)
elif 'user' in kwargs:
return common.models.InvenTreeUserSetting.get_setting_object(
key, user=kwargs['user'], cache=cache

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

@@ -592,13 +592,15 @@ class GeneralApiTests(InvenTreeAPITestCase):
self.assertEqual('InvenTree', data['server'])
# Test with token
token = self.get(url=reverse('api-token')).data['token']
token = self.get(url=reverse('api-token'), max_query_count=275).data['token']
self.client.logout()
# Anon
response = self.get(url)
response = self.get(url, max_query_count=275)
self.assertEqual(response.json()['database'], None)
# Staff
response = self.get(url, headers={'Authorization': f'Token {token}'})
response = self.get(
url, headers={'Authorization': f'Token {token}'}, max_query_count=275
)
self.assertGreater(len(response.json()['database']), 4)

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

@@ -14,7 +14,7 @@ from django_q.models import Schedule, Task
from error_report.models import Error
import InvenTree.tasks
from common.models import InvenTreeSetting
from common.models import InvenTreeSetting, InvenTreeUserSetting
threshold = timezone.now() - timedelta(days=30)
threshold_low = threshold - timedelta(days=1)
@@ -191,7 +191,7 @@ class InvenTreeTaskTests(TestCase):
# Create a staff user (to ensure notifications are sent)
user = User.objects.create_user(
username='staff', password='staffpass', is_staff=False
username='i_am_staff', password='staffpass', is_staff=False, is_active=True
)
n_tasks = Task.objects.count()
@@ -220,6 +220,9 @@ class InvenTreeTaskTests(TestCase):
user.is_staff = True
user.save()
# Ensure error notifications are enabled for this user
InvenTreeUserSetting.set_setting('NOTIFICATION_ERROR_REPORT', True, user=user)
# Create a 'failed' task in the database
# Note: The 'attempt count' is set to 10 to ensure that the task is properly marked as 'failed'
Task.objects.create(id=n_tasks + 2, **test_data)

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

@@ -677,12 +677,15 @@ class AdminTestCase(InvenTreeAPITestCase):
app_app, app_mdl = model._meta.app_label, model._meta.model_name
# 'Test listing
response = self.get(reverse(f'admin:{app_app}_{app_mdl}_changelist'))
response = self.get(
reverse(f'admin:{app_app}_{app_mdl}_changelist'), max_query_count=300
)
self.assertEqual(response.status_code, 200)
# Test change view
response = self.get(
reverse(f'admin:{app_app}_{app_mdl}_change', kwargs={'object_id': obj.pk})
reverse(f'admin:{app_app}_{app_mdl}_change', kwargs={'object_id': obj.pk}),
max_query_count=300,
)
self.assertEqual(response.status_code, 200)
self.assertContains(response, 'Django site admin')

2
src/backend/InvenTree/build/events.py
View File

@@ -13,6 +13,8 @@ class BuildEvents(BaseEventEnum):
COMPLETED = 'build.completed'
OVERDUE = 'build.overdue_build_order'
STOCK_REQUIRED = 'build.stock_required'
# Build output events
OUTPUT_CREATED = 'buildoutput.created'
OUTPUT_COMPLETED = 'buildoutput.completed'

12
src/backend/InvenTree/build/models.py
View File

@@ -760,6 +760,11 @@ class Build(
trigger_event(BuildEvents.ISSUED, id=self.pk)
from build.tasks import check_build_stock
# Run checks on required parts
InvenTree.tasks.offload_task(check_build_stock, self, group='build')
@transaction.atomic
def hold_build(self):
"""Mark the Build as ON HOLD."""
@@ -1504,8 +1509,6 @@ def after_save_build(sender, instance: Build, created: bool, **kwargs):
):
return
from . import tasks as build_tasks
if instance:
if created:
# A new Build has just been created
@@ -1513,11 +1516,6 @@ def after_save_build(sender, instance: Build, created: bool, **kwargs):
# Generate initial BuildLine objects for the Build
instance.create_build_line_items()
# Run checks on required parts
InvenTree.tasks.offload_task(
build_tasks.check_build_stock, instance, group='build'
)
# Notify the responsible users that the build order has been created
InvenTree.helpers_model.notify_responsible(
instance,

47
src/backend/InvenTree/build/tasks.py
View File

@@ -4,16 +4,13 @@ from datetime import timedelta
from decimal import Decimal
from django.contrib.auth.models import User
from django.template.loader import render_to_string
from django.utils.translation import gettext_lazy as _
import structlog
from allauth.account.models import EmailAddress
from opentelemetry import trace
import common.notifications
import InvenTree.helpers
import InvenTree.helpers_email
import InvenTree.helpers_model
import InvenTree.tasks
from build.events import BuildEvents
@@ -175,34 +172,30 @@ def check_build_stock(build):
return
# Are there any users subscribed to these parts?
subscribers = build.part.get_subscribers()
targets = build.part.get_subscribers()
emails = EmailAddress.objects.filter(user__in=subscribers)
if build.responsible:
targets.append(build.responsible)
if len(emails) > 0:
logger.info('Notifying users of stock required for build %s', build.pk)
name = _('Stock required for build order')
context = {
'link': InvenTree.helpers_model.construct_absolute_url(
build.get_absolute_url()
),
'build': build,
'part': build.part,
'lines': lines,
}
context = {
'build': build,
'name': name,
'part': build.part,
'lines': lines,
'link': InvenTree.helpers_model.construct_absolute_url(
build.get_absolute_url()
),
'message': _('Build order {build} requires additional stock').format(
build=build
),
'template': {'html': 'email/build_order_required_stock.html', 'subject': name},
}
# Render the HTML message
html_message = render_to_string(
'email/build_order_required_stock.html', context
)
subject = _('Stock required for build order')
recipients = emails.values_list('email', flat=True)
InvenTree.helpers_email.send_email(
subject, '', recipients, html_message=html_message
)
common.notifications.trigger_notification(
build, BuildEvents.STOCK_REQUIRED, targets=targets, context=context
)
@tracer.start_as_current_span('notify_overdue_build_order')

50
src/backend/InvenTree/common/api.py
View File

@@ -55,8 +55,6 @@ from InvenTree.permissions import (
IsSuperuserOrSuperScope,
UserSettingsPermissionsOrScope,
)
from plugin.models import NotificationUserSetting
from plugin.serializers import NotificationUserSettingSerializer
class CsrfExemptMixin:
@@ -308,36 +306,6 @@ class UserSettingsDetail(RetrieveUpdateAPI):
)
class NotificationUserSettingsList(SettingsList):
"""API endpoint for accessing a list of notification user settings objects."""
queryset = NotificationUserSetting.objects.all()
serializer_class = NotificationUserSettingSerializer
permission_classes = [UserSettingsPermissionsOrScope]
def filter_queryset(self, queryset):
"""Only list settings which apply to the current user."""
try:
user = self.request.user
except AttributeError:
return NotificationUserSetting.objects.none()
queryset = super().filter_queryset(queryset)
queryset = queryset.filter(user=user)
return queryset
class NotificationUserSettingsDetail(RetrieveUpdateAPI):
"""Detail view for an individual "notification user setting" object.
- User can only view / edit settings their own settings objects
"""
queryset = NotificationUserSetting.objects.all()
serializer_class = NotificationUserSettingSerializer
permission_classes = [UserSettingsPermissionsOrScope]
class NotificationMessageMixin:
"""Generic mixin for NotificationMessage."""
@@ -962,24 +930,6 @@ settings_api_urls = [
path('', UserSettingsList.as_view(), name='api-user-setting-list'),
]),
),
# Notification settings
path(
'notification/',
include([
# Notification Settings Detail
path(
'<int:pk>/',
NotificationUserSettingsDetail.as_view(),
name='api-notification-setting-detail',
),
# Notification Settings List
path(
'',
NotificationUserSettingsList.as_view(),
name='api-notification-setting-list',
),
]),
),
# Global settings
path(
'global/',

10
src/backend/InvenTree/common/models.py
View File

@@ -603,7 +603,15 @@ class BaseInvenTreeSetting(models.Model):
if not setting and create:
# Attempt to create a new settings object
default_value = cls.get_setting_default(key, **kwargs)
setting = cls(key=key, value=default_value, **kwargs)
extra_fields = {}
# Provide extra default fields
for field in cls.extra_unique_fields:
if field in kwargs:
extra_fields[field] = kwargs[field]
setting = cls(key=key, value=default_value, **extra_fields)
try:
# Wrap this statement in "atomic", so it can be rolled back if it fails

356
src/backend/InvenTree/common/notifications.py
View File

@@ -12,10 +12,9 @@ from django.utils.translation import gettext_lazy as _
import structlog
import common.models
import InvenTree.helpers
from InvenTree.exceptions import log_error
from InvenTree.ready import isImportingData, isRebuildingData
from plugin import registry
from plugin.models import NotificationUserSetting, PluginConfig
from plugin import PluginMixinEnum, registry
from users.models import Owner
from users.permissions import check_user_permission
@@ -108,194 +107,6 @@ class NotificationMethod:
return context
def get_targets(self):
"""Returns targets for notifications.
Processes `self.targets` to extract all users that should be notified.
"""
raise NotImplementedError('The `get_targets` method must be implemented!')
def setup(self):
"""Set up context before notifications are send.
This is intended to be overridden in method implementations.
"""
return True
def cleanup(self):
"""Clean up context after all notifications were send.
This is intended to be overridden in method implementations.
"""
return True
# region plugins
def get_plugin(self):
"""Returns plugin class."""
return False
def global_setting_disable(self):
"""Check if the method is defined in a plugin and has a global setting."""
# Check if plugin has a setting
if not self.GLOBAL_SETTING:
return False
# Check if plugin is set
plg_cls = self.get_plugin()
if not plg_cls:
return False
# Check if method globally enabled
plg_instance = registry.get_plugin(plg_cls.NAME.lower())
return plg_instance and not plg_instance.get_setting(self.GLOBAL_SETTING)
def usersetting(self, target):
"""Returns setting for this method for a given user."""
return NotificationUserSetting.get_setting(
f'NOTIFICATION_METHOD_{self.METHOD_NAME.upper()}',
user=target,
method=self.METHOD_NAME,
)
# endregion
class SingleNotificationMethod(NotificationMethod):
"""NotificationMethod that sends notifications one by one."""
def send(self, target):
"""This function must be overridden."""
raise NotImplementedError('The `send` method must be overridden!')
class BulkNotificationMethod(NotificationMethod):
"""NotificationMethod that sends all notifications in bulk."""
def send_bulk(self):
"""This function must be overridden."""
raise NotImplementedError('The `send` method must be overridden!')
# endregion
class MethodStorageClass:
"""Class that works as registry for all available notification methods in InvenTree.
Is initialized on startup as one instance named `storage` in this file.
"""
methods_list = None
user_settings = {}
@property
def methods(self):
"""Return all available methods.
This is cached, and stored internally.
"""
if self.methods_list is None:
self.collect()
return self.methods_list
def collect(self, selected_classes=None):
"""Collect all classes in the environment that are notification methods.
Can be filtered to only include provided classes for testing.
Args:
selected_classes (class, optional): References to the classes that should be registered. Defaults to None.
"""
logger.debug('Collecting notification methods...')
current_method = (
InvenTree.helpers.inheritors(NotificationMethod) - IGNORED_NOTIFICATION_CLS
)
# for testing selective loading is made available
if selected_classes:
current_method = [
item for item in current_method if item is selected_classes
]
# make sure only one of each method is added
filtered_list = {}
for item in current_method:
plugin = item.get_plugin(item)
ref = (
f'{plugin.package_path}_{item.METHOD_NAME}'
if plugin
else item.METHOD_NAME
)
item.plugin = plugin() if plugin else None
filtered_list[ref] = item
storage.methods_list = list(filtered_list.values())
logger.info('Found %s notification methods', len(storage.methods_list))
for item in storage.methods_list:
logger.debug(' - %s', str(item))
def get_usersettings(self, user) -> list:
"""Returns all user settings for a specific user.
This is needed to show them in the settings UI.
Args:
user (User): User that should be used as a filter.
Returns:
list: All applicablae notification settings.
"""
methods = []
for item in storage.methods:
if item.USER_SETTING:
new_key = f'NOTIFICATION_METHOD_{item.METHOD_NAME.upper()}'
# make sure the setting exists
self.user_settings[new_key] = item.USER_SETTING
NotificationUserSetting.get_setting(
key=new_key, user=user, method=item.METHOD_NAME
)
# save definition
methods.append({
'key': new_key,
'icon': getattr(item, 'METHOD_ICON', ''),
'method': item.METHOD_NAME,
})
return methods
IGNORED_NOTIFICATION_CLS = {SingleNotificationMethod, BulkNotificationMethod}
storage = MethodStorageClass()
class UIMessageNotification(SingleNotificationMethod):
"""Delivery method for sending specific users notifications in the notification pain in the web UI."""
METHOD_NAME = 'ui_message'
def get_targets(self):
"""Only send notifications for active users."""
return [target for target in self.targets if target.is_active]
def send(self, target):
"""Send a UI notification to a user."""
common.models.NotificationMessage.objects.create(
target_object=self.obj,
source_object=target,
user=target,
category=self.category,
name=self.context['name'],
message=self.context['message'],
)
return True
@dataclass()
class NotificationBody:
@@ -389,16 +200,16 @@ def trigger_notification(
obj_ref_value = None
# Find the first reference that is available
for ref in refs:
if hasattr(obj, ref):
obj_ref_value = getattr(obj, ref)
break
if obj:
for ref in refs:
if hasattr(obj, ref):
obj_ref_value = getattr(obj, ref)
break
# Try with some defaults
if not obj_ref_value:
raise KeyError(
f"Could not resolve an object reference for '{obj!s}' with {','.join(set(refs))}"
)
if not obj_ref_value:
raise KeyError(
f"Could not resolve an object reference for '{obj!s}' with {','.join(set(refs))}"
)
# Check if we have notified recently...
delta = timedelta(days=1)
@@ -419,7 +230,7 @@ def trigger_notification(
target_exclude = set()
# Collect possible targets
if not targets:
if not targets and target_fnc:
targets = target_fnc(*target_args, **target_kwargs)
# Convert list of targets to a list of users
@@ -451,122 +262,45 @@ def trigger_notification(
'Unknown target passed to trigger_notification method: %s', target
)
if target_users:
# Filter out any users who are inactive, or do not have the required model permissions
valid_users = list(
filter(
lambda u: u and u.is_active and check_user_permission(u, obj, 'view'),
list(target_users),
)
# Filter out any users who are inactive, or do not have the required model permissions
valid_users = list(
filter(
lambda u: u
and u.is_active
and (not obj or check_user_permission(u, obj, 'view')),
list(target_users),
)
if len(valid_users) > 0:
logger.info(
"Sending notification '%s' for '%s' to %s users",
category,
str(obj),
len(valid_users),
)
# Collect possible methods
if delivery_methods is None:
delivery_methods = storage.methods or []
else:
delivery_methods = delivery_methods - IGNORED_NOTIFICATION_CLS
for method in delivery_methods:
logger.info("Triggering notification method '%s'", method.METHOD_NAME)
try:
deliver_notification(method, obj, category, valid_users, context)
except NotImplementedError as error:
# Allow any single notification method to fail, without failing the others
logger.error(error)
except Exception as error:
logger.error(error)
# Set delivery flag
common.models.NotificationEntry.notify(category, obj_ref_value)
else:
logger.info("No possible users for notification '%s'", category)
def trigger_superuser_notification(plugin: PluginConfig, msg: str):
"""Trigger a notification to all superusers.
Args:
plugin (PluginConfig): Plugin that is raising the notification
msg (str): Detailed message that should be attached
"""
users = get_user_model().objects.filter(is_superuser=True)
trigger_notification(
plugin,
'inventree.plugin',
context={'error': plugin, 'name': _('Error raised by plugin'), 'message': msg},
targets=users,
delivery_methods={UIMessageNotification},
)
# Track whether any notifications were sent
result = False
def deliver_notification(
cls: NotificationMethod, obj: Model, category: str, targets: list, context: dict
):
"""Send notification with the provided class.
# Send out via all registered notification methods
for plugin in registry.with_mixin(PluginMixinEnum.NOTIFICATION):
# Skip if the plugin is *not* in the "delivery_methods" list?
match = not delivery_methods
Arguments:
cls: The class that should be used to send the notification
obj: The object (model instance) that triggered the notification
category: The category (label) for the notification
targets: List of users that should receive the notification
context: Context dictionary with additional information for the notification
for notification_class in delivery_methods or []:
if type(notification_class) is str:
if plugin.slug == notification_class:
match = True
break
- Initializes the method
- Checks that there are valid targets
- Runs the delivery setup
- Sends notifications either via `send_bulk` or send`
- Runs the delivery cleanup
"""
# Init delivery method
method = cls(obj, category, targets, context)
elif getattr(notification_class, 'SLUG', None) == plugin.slug:
match = True
break
if method.targets and len(method.targets) > 0:
# Log start
logger.info(
"Notify users via '%s' for notification '%s' for '%s'",
method.METHOD_NAME,
category,
str(obj),
)
if not match:
continue
# Run setup for delivery method
method.setup()
try:
# Plugin may optionally filter target users
filtered_users = plugin.filter_targets(list(valid_users))
if plugin.send_notification(obj, category, filtered_users, context):
result = True
except Exception:
log_error('send_notification', plugin=plugin.slug)
# Counters for success logs
success = True
success_count = 0
# Select delivery method and execute it
if hasattr(method, 'send_bulk'):
success = method.send_bulk()
success_count = len(method.targets)
elif hasattr(method, 'send'):
for target in method.targets:
if method.send(target):
success_count += 1
else:
success = False
# Run cleanup for delivery method
method.cleanup()
# Log results
logger.info(
"Notified %s users via '%s' for notification '%s' for '%s' successfully",
success_count,
method.METHOD_NAME,
category,
str(obj),
)
if not success:
logger.info('There were some problems')
# Log the notification entry
if result:
common.models.NotificationEntry.notify(category, obj_ref_value)

149
src/backend/InvenTree/common/test_notifications.py
View File

@@ -1,149 +0,0 @@
"""Tests for basic notification methods and functions in InvenTree."""
from common.notifications import (
BulkNotificationMethod,
NotificationMethod,
SingleNotificationMethod,
)
from part.test_part import BaseNotificationIntegrationTest
class BaseNotificationTests(BaseNotificationIntegrationTest):
"""Tests for basic NotificationMethod."""
def test_NotificationMethod(self):
"""Ensure the implementation requirements are tested."""
class FalseNotificationMethod(NotificationMethod):
METHOD_NAME = 'FalseNotification'
class AnotherFalseNotificationMethod(NotificationMethod):
METHOD_NAME = 'AnotherFalseNotification'
def send(self):
"""A comment so we do not need a pass."""
class NoNameNotificationMethod(NotificationMethod):
def send(self):
"""A comment so we do not need a pass."""
class WrongContextNotificationMethod(NotificationMethod):
METHOD_NAME = 'WrongContextNotification'
CONTEXT_EXTRA = ['aa', ('aa', 'bb'), ('templates', 'ccc'), (123,)]
def send(self):
"""A comment so we do not need a pass."""
# no send / send bulk
with self.assertRaises(NotImplementedError):
FalseNotificationMethod('', '', '', '')
# no METHOD_NAME
with self.assertRaises(NotImplementedError):
NoNameNotificationMethod('', '', '', '')
# a not existent context check
with self.assertRaises(NotImplementedError):
WrongContextNotificationMethod('', '', '', '')
# no get_targets
with self.assertRaises(NotImplementedError):
AnotherFalseNotificationMethod('', '', '', {'name': 1, 'message': 2})
def test_failing_passing(self):
"""Ensure that an error in one deliverymethod is not blocking all mehthods."""
# cover failing delivery
self._notification_run()
def test_errors_passing(self):
"""Ensure that errors do not kill the whole delivery."""
class ErrorImplementation(SingleNotificationMethod):
METHOD_NAME = 'ErrorImplementation'
def get_targets(self):
return [1]
def send(self, target):
raise KeyError('This could be any error')
self._notification_run(ErrorImplementation)
class BulkNotificationMethodTests(BaseNotificationIntegrationTest):
"""Tests for BulkNotificationMethod classes specifically.
General tests for NotificationMethods are in BaseNotificationTests.
"""
def test_BulkNotificationMethod(self):
"""Ensure the implementation requirements are tested.
MixinNotImplementedError needs to raise if the send_bulk() method is not set.
"""
class WrongImplementation(BulkNotificationMethod):
METHOD_NAME = 'WrongImplementationBulk'
def get_targets(self):
return [1]
with self.assertLogs(logger='inventree', level='ERROR'):
self._notification_run(WrongImplementation)
class SingleNotificationMethodTests(BaseNotificationIntegrationTest):
"""Tests for SingleNotificationMethod classes specifically.
General tests for NotificationMethods are in BaseNotificationTests.
"""
def test_SingleNotificationMethod(self):
"""Ensure the implementation requirements are tested.
MixinNotImplementedError needs to raise if the send() method is not set.
"""
class WrongImplementation(SingleNotificationMethod):
METHOD_NAME = 'WrongImplementationSingle'
def get_targets(self):
return [1]
with self.assertLogs(logger='inventree', level='ERROR'):
self._notification_run(WrongImplementation)
# A integration test for notifications is provided in test_part.PartNotificationTest
class NotificationUserSettingTests(BaseNotificationIntegrationTest):
"""Tests for NotificationUserSetting."""
def setUp(self):
"""Setup for all tests."""
super().setUp()
self.client.login(username=self.user.username, password='password')
def test_setting_attributes(self):
"""Check notification method plugin methods: usersettings and tags."""
class SampleImplementation(BulkNotificationMethod):
METHOD_NAME = 'test'
GLOBAL_SETTING = 'ENABLE_NOTIFICATION_TEST'
USER_SETTING = {
'name': 'Enable test notifications',
'description': 'Allow sending of test for event notifications',
'default': True,
'validator': bool,
'units': 'alpha',
}
def get_targets(self):
return [1]
def send_bulk(self):
return True
# run through notification
self._notification_run(SampleImplementation)

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

@@ -32,7 +32,6 @@ from InvenTree.unit_test import (
)
from part.models import Part, PartParameterTemplate
from plugin import registry
from plugin.models import NotificationUserSetting
from .api import WebhookView
from .models import (
@@ -797,28 +796,6 @@ class UserSettingsApiTest(InvenTreeAPITestCase):
response = self.patch(url, {'value': v}, expected_code=400)
class NotificationUserSettingsApiTest(InvenTreeAPITestCase):
"""Tests for the notification user settings API."""
def test_api_list(self):
"""Test list URL."""
url = reverse('api-notification-setting-list')
self.get(url, expected_code=200)
def test_setting(self):
"""Test the string name for NotificationUserSetting."""
NotificationUserSetting.set_setting(
'NOTIFICATION_METHOD_MAIL', True, change_user=self.user, user=self.user
)
test_setting = NotificationUserSetting.get_setting_object(
'NOTIFICATION_METHOD_MAIL', user=self.user
)
self.assertEqual(
str(test_setting), 'NOTIFICATION_METHOD_MAIL (for testuser): True'
)
class PluginSettingsApiTest(PluginMixin, InvenTreeAPITestCase):
"""Tests for the plugin settings API."""

2
src/backend/InvenTree/generic/states/test_transition.py
View File

@@ -26,7 +26,7 @@ def _clean_storage(refs):
class TransitionTests(InvenTreeTestCase):
"""Tests for basic NotificationMethod."""
"""Tests for basic TransitionMethod."""
def test_class(self):
"""Ensure that the class itself works."""

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

@@ -197,7 +197,9 @@ class MachineAPITest(TestMachineRegistryMixin, InvenTreeAPITestCase):
}
# Create a machine
response = self.post(reverse('api-machine-list'), machine_data)
response = self.post(
reverse('api-machine-list'), machine_data, max_query_count=400
)
self.assertEqual(response.data, {**response.data, **machine_data})
pk = response.data['pk']
@@ -231,13 +233,16 @@ 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)
self.get(machine_setting_url, expected_code=404, max_query_count=QUERY_LIMIT)
# Create a machine
machine = MachineConfig.objects.create(
@@ -257,18 +262,22 @@ class MachineAPITest(TestMachineRegistryMixin, InvenTreeAPITestCase):
)
# Get settings
response = self.get(machine_setting_url)
response = self.get(machine_setting_url, max_query_count=QUERY_LIMIT)
self.assertEqual(response.data['value'], '')
response = self.get(driver_setting_url)
response = self.get(driver_setting_url, max_query_count=QUERY_LIMIT)
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)})
response = self.patch(
machine_setting_url,
{'value': str(location.pk)},
max_query_count=QUERY_LIMIT,
)
self.assertEqual(response.data['value'], str(location.pk))
response = self.get(machine_setting_url)
response = self.get(machine_setting_url, max_query_count=QUERY_LIMIT)
self.assertEqual(response.data['value'], str(location.pk))
# Update driver setting
@@ -280,7 +289,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)
response = self.get(settings_url, max_query_count=QUERY_LIMIT)
self.assertEqual(len(response.data), 2)
self.assertEqual(
[('M', 'LOCATION'), ('D', 'TEST_SETTING')],

9
src/backend/InvenTree/machine/tests.py
View File

@@ -157,14 +157,13 @@ class TestDriverMachineInterface(TestMachineRegistryMixin, TestCase):
self.assertEqual(registry.get_drivers('testing-type')[0].SLUG, 'test-driver')
# test that init hooks where called correctly
self.driver_mocks['init_driver'].assert_called_once()
self.assertEqual(self.driver_mocks['init_machine'].call_count, 2)
CALL_COUNT = range(1, 5) # Due to interplay between plugin and machine registry
self.assertIn(self.driver_mocks['init_driver'].call_count, CALL_COUNT)
self.assertIn(self.driver_mocks['init_machine'].call_count, CALL_COUNT)
# Test machine restart hook
registry.restart_machine(self.machine1.machine)
self.driver_mocks['restart_machine'].assert_called_once_with(
self.machine1.machine
)
self.assertEqual(self.machine_mocks[self.machine1]['restart'].call_count, 1)
# Test machine update hook

77
src/backend/InvenTree/part/test_part.py
View File

@@ -7,11 +7,8 @@ from django.core.cache import cache
from django.core.exceptions import ValidationError
from django.test import TestCase
from allauth.account.models import EmailAddress
import part.settings
from common.models import NotificationEntry, NotificationMessage
from common.notifications import UIMessageNotification, storage
from common.settings import get_global_setting, set_global_setting
from InvenTree import version
from InvenTree.templatetags import inventree_extras
@@ -924,67 +921,39 @@ class PartSubscriptionTests(InvenTreeTestCase):
self.assertTrue(self.part.is_starred_by(self.user))
class BaseNotificationIntegrationTest(InvenTreeTestCase):
"""Integration test for notifications."""
class PartNotificationTest(InvenTreeTestCase):
"""Integration test for part notifications."""
fixtures = ['location', 'category', 'part', 'stock']
@classmethod
def setUpTestData(cls):
"""Add an email address as part of initialization."""
super().setUpTestData()
# Add email address
EmailAddress.objects.create(user=cls.user, email='test@testing.com')
# Define part that will be tested
cls.part = Part.objects.get(name='R_2K2_0805')
def _notification_run(self, run_class=None):
"""Run a notification test suit through.
If you only want to test one class pass it to run_class
"""
# reload notification methods
storage.collect(run_class)
def test_low_stock_notification(self):
"""Test that a low stocknotification is generated."""
NotificationEntry.objects.all().delete()
NotificationMessage.objects.all().delete()
# There should be no notification runs
part = Part.objects.get(name='R_2K2_0805')
part.minimum_stock = part.get_stock_count() + 1
part.save()
# There should be no notifications created yet,
# as there are no "subscribed" users for this part
self.assertEqual(NotificationEntry.objects.all().count(), 0)
self.assertEqual(NotificationMessage.objects.all().count(), 0)
# Test that notifications run through without errors
self.part.minimum_stock = (
self.part.get_stock_count() + 1
) # make sure minimum is one higher than current count
self.part.save()
# There should be no notification as no-one is subscribed
self.assertEqual(NotificationEntry.objects.all().count(), 0)
# Subscribe and run again
# Subscribe the user to the part
addUserPermission(self.user, 'part', 'part', 'view')
self.user.is_active = True
self.user.save()
self.part.set_starred(self.user, True)
self.part.save()
part.set_starred(self.user, True)
part.save()
# There should be 1 (or 2) notifications - in some cases an error is generated, which creates a subsequent notification
self.assertIn(NotificationEntry.objects.all().count(), [1, 2])
# Check that a UI notification entry has been created
self.assertGreaterEqual(NotificationEntry.objects.all().count(), 1)
self.assertGreaterEqual(NotificationMessage.objects.all().count(), 1)
# No errors were generated during notification process
from error_report.models import Error
class PartNotificationTest(BaseNotificationIntegrationTest):
"""Integration test for part notifications."""
def test_notification(self):
"""Test that a notification is generated."""
self._notification_run(UIMessageNotification)
# There should be 1 notification message right now
self.assertEqual(NotificationMessage.objects.all().count(), 1)
# Try again -> cover the already send line
self.part.save()
# There should not be more messages
self.assertEqual(NotificationMessage.objects.all().count(), 1)
self.assertEqual(Error.objects.count(), 0)

28
src/backend/InvenTree/plugin/admin.py
View File

@@ -46,6 +46,18 @@ class PluginSettingInline(admin.TabularInline):
return False
class PluginUserSettingInline(admin.TabularInline):
"""Inline admin class for PluginUserSetting."""
model = models.PluginUserSetting
read_only_fields = ['key']
def has_add_permission(self, request, obj):
"""The plugin user settings should not be meddled with manually."""
return False
class PluginConfigAdmin(admin.ModelAdmin):
"""Custom admin with restricted id fields."""
@@ -61,21 +73,9 @@ class PluginConfigAdmin(admin.ModelAdmin):
]
list_filter = ['active']
actions = [plugin_activate, plugin_deactivate]
inlines = [PluginSettingInline]
inlines = [PluginSettingInline, PluginUserSettingInline]
exclude = ['metadata']
class NotificationUserSettingAdmin(admin.ModelAdmin):
"""Admin class for NotificationUserSetting."""
model = models.NotificationUserSetting
read_only_fields = ['key']
def has_add_permission(self, request):
"""Notifications should not be changed."""
return False
search_fields = ['name', 'key']
admin.site.register(models.PluginConfig, PluginConfigAdmin)
admin.site.register(models.NotificationUserSetting, NotificationUserSettingAdmin)

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

@@ -31,7 +31,7 @@ from plugin.base.action.api import ActionPluginView
from plugin.base.barcodes.api import barcode_api_urls
from plugin.base.locate.api import LocatePluginView
from plugin.base.ui.api import ui_plugins_api_urls
from plugin.models import PluginConfig, PluginSetting
from plugin.models import PluginConfig, PluginSetting, PluginUserSetting
from plugin.plugin import InvenTreePlugin
from plugin.registry import registry
@@ -332,19 +332,19 @@ def check_plugin(
# Check that the 'plugin' specified is valid
try:
plugin_cgf = PluginConfig.objects.filter(**filters).first()
plugin_cfg = PluginConfig.objects.filter(**filters).first()
except PluginConfig.DoesNotExist:
raise NotFound(detail=f"Plugin '{ref}' not installed")
if plugin_cgf is None:
if plugin_cfg is None:
# This only occurs if the plugin mechanism broke
raise NotFound(detail=f"Plugin '{ref}' not installed") # pragma: no cover
# Check that the plugin is activated
if not plugin_cgf.active:
if not plugin_cfg.active:
raise NotFound(detail=f"Plugin '{ref}' is not active")
plugin = plugin_cgf.plugin
plugin = plugin_cfg.plugin
if not plugin:
raise NotFound(detail=f"Plugin '{ref}' not installed")
@@ -381,10 +381,7 @@ class PluginAllSettingList(APIView):
class PluginSettingDetail(RetrieveUpdateAPI):
"""Detail endpoint for a plugin-specific setting.
Note that these cannot be created or deleted via the API
"""
"""Detail endpoint for a plugin-specific setting."""
queryset = PluginSetting.objects.all()
serializer_class = PluginSerializers.PluginSettingSerializer
@@ -415,6 +412,65 @@ class PluginSettingDetail(RetrieveUpdateAPI):
permission_classes = [InvenTree.permissions.GlobalSettingsPermissions]
class PluginUserSettingList(APIView):
"""List endpoint for all user settings for a specific plugin.
- GET: return all user settings for a plugin config
"""
queryset = PluginUserSetting.objects.all()
serializer_class = PluginSerializers.PluginUserSettingSerializer
permission_classes = [InvenTree.permissions.UserSettingsPermissionsOrScope]
@extend_schema(
responses={200: PluginSerializers.PluginUserSettingSerializer(many=True)}
)
def get(self, request, plugin):
"""Get all user settings for a plugin config."""
# look up the plugin
plugin = check_plugin(plugin, None)
user_settings = getattr(plugin, 'user_settings', {})
settings_dict = PluginUserSetting.all_settings(
settings_definition=user_settings,
plugin=plugin.plugin_config(),
user=request.user,
)
results = PluginSerializers.PluginUserSettingSerializer(
list(settings_dict.values()), many=True
).data
return Response(results)
class PluginUserSettingDetail(RetrieveUpdateAPI):
"""Detail endpoint for a plugin-specific user setting."""
lookup_field = 'key'
queryset = PluginUserSetting.objects.all()
serializer_class = PluginSerializers.PluginUserSettingSerializer
permission_classes = [InvenTree.permissions.UserSettingsPermissionsOrScope]
def get_object(self):
"""Lookup the plugin user setting object, based on the URL."""
setting_key = self.kwargs['key']
# Look up plugin
plugin = check_plugin(self.kwargs.get('plugin', None), None)
settings = getattr(plugin, 'user_settings', {})
if setting_key not in settings:
raise NotFound(
detail=f"Plugin '{plugin.slug}' has no user setting matching '{setting_key}'"
)
return PluginUserSetting.get_setting_object(
setting_key, plugin=plugin.plugin_config(), user=self.request.user
)
class RegistryStatusView(APIView):
"""Status API endpoint for the plugin registry.
@@ -484,6 +540,21 @@ plugin_api_urls = [
path(
'<str:plugin>/',
include([
path(
'user-settings/',
include([
re_path(
r'^(?P<key>\w+)/',
PluginUserSettingDetail.as_view(),
name='api-plugin-user-setting-detail',
),
path(
'',
PluginUserSettingList.as_view(),
name='api-plugin-user-setting-list',
),
]),
),
path(
'settings/',
include([

4
src/backend/InvenTree/plugin/apps.py
View File

@@ -22,6 +22,10 @@ class PluginAppConfig(AppConfig):
def ready(self):
"""The ready method is extended to initialize plugins."""
self.reload_plugin_registry()
def reload_plugin_registry(self):
"""Reload the plugin registry."""
if not isInMainThread() and not isInWorkerThread():
return

55
src/backend/InvenTree/plugin/base/integration/NotificationMixin.py Normal file
View File

@@ -0,0 +1,55 @@
"""Plugin mixin class for supporting third-party notification methods."""
from typing import TYPE_CHECKING
from django.contrib.auth.models import User
from django.db.models import Model
import structlog
from plugin import PluginMixinEnum
logger = structlog.get_logger('inventree')
if TYPE_CHECKING:
from common.models import SettingsKeyType
else:
class SettingsKeyType:
"""Dummy class, so that python throws no error."""
class NotificationMixin:
"""Plugin mixin class for supporting third-party notification methods."""
class MixinMeta:
"""Meta for mixin."""
MIXIN_NAME = PluginMixinEnum.NOTIFICATION
def __init__(self):
"""Register mixin."""
super().__init__()
self.add_mixin(PluginMixinEnum.NOTIFICATION, True, __class__)
def filter_targets(self, targets: list[User]) -> list[User]:
"""Filter notification targets based on the plugin's logic."""
# Default implementation returns all targets
return targets
def send_notification(
self, target: Model, category: str, users: list, context: dict
) -> bool:
"""Send notification to the specified target users.
Arguments:
target (Model): The target model instance to which the notification relates.
category (str): The category of the notification.
users (list): List of users to send the notification to.
context (dict): Context data for the notification.
Returns:
bool: True if the notification was sent successfully, False otherwise.
"""
# The default implementation does nothing
return False

82
src/backend/InvenTree/plugin/base/integration/SettingsMixin.py
View File

@@ -1,6 +1,6 @@
"""Plugin mixin class for SettingsMixin."""
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Any, Optional
from django.db.utils import OperationalError, ProgrammingError
@@ -12,9 +12,14 @@ logger = structlog.get_logger('inventree')
# import only for typechecking, otherwise this throws a model is unready error
if TYPE_CHECKING:
from django.contrib.auth.models import User
from common.models import SettingsKeyType
else:
class User:
"""Dummy class, so that python throws no error."""
class SettingsKeyType:
"""Dummy class, so that python throws no error."""
@@ -23,6 +28,7 @@ class SettingsMixin:
"""Mixin that enables global settings for the plugin."""
SETTINGS: dict[str, SettingsKeyType] = {}
USER_SETTINGS: dict[str, SettingsKeyType] = {}
class MixinMeta:
"""Meta for mixin."""
@@ -34,6 +40,7 @@ class SettingsMixin:
super().__init__()
self.add_mixin(PluginMixinEnum.SETTINGS, 'has_settings', __class__)
self.settings = getattr(self, 'SETTINGS', {})
self.user_settings = getattr(self, 'USER_SETTINGS', {})
@classmethod
def _activate_mixin(cls, registry, plugins, *args, **kwargs):
@@ -45,12 +52,16 @@ class SettingsMixin:
logger.debug('Activating plugin settings')
registry.mixins_settings = {}
registry.mixins_user_settings = {}
for slug, plugin in plugins:
if plugin.mixin_enabled(PluginMixinEnum.SETTINGS):
plugin_setting = plugin.settings
plugin_setting = plugin.settings or {}
registry.mixins_settings[slug] = plugin_setting
plugin_user_setting = plugin.user_settings or {}
registry.mixins_user_settings[slug] = plugin_user_setting
@classmethod
def _deactivate_mixin(cls, registry, **kwargs):
"""Deactivate all plugin settings."""
@@ -61,14 +72,16 @@ class SettingsMixin:
@property
def has_settings(self):
"""Does this plugin use custom global settings."""
return bool(self.settings)
return bool(self.settings) or bool(self.user_settings)
def get_setting(self, key, cache=False, backup_value=None):
def get_setting(
self, key: str, cache: bool = False, backup_value: Any = None
) -> Any:
"""Return the 'value' of the setting associated with this plugin.
Arguments:
key: The 'name' of the setting value to be retrieved
cache: Whether to use RAM cached value (default = False)
cache: Whether to use cached value (default = False)
backup_value: A backup value to return if the setting is not found
"""
from plugin.models import PluginSetting
@@ -77,8 +90,16 @@ class SettingsMixin:
key, plugin=self.plugin_config(), cache=cache, backup_value=backup_value
)
def set_setting(self, key, value, user=None):
"""Set plugin setting value by key."""
def set_setting(
self, key: str, value: Any, user: Optional[User] = None, **kwargs
) -> None:
"""Set plugin setting value by key.
Arguments:
key: The 'name' of the setting value to be set
value: The value to be set for the setting
user: The user who is making the change (optional)
"""
from plugin.models import PluginSetting
from plugin.registry import registry
@@ -92,7 +113,52 @@ class SettingsMixin:
logger.error("Plugin configuration not found for plugin '%s'", self.slug)
return
PluginSetting.set_setting(key, value, user, plugin=plugin)
PluginSetting.set_setting(key, value, plugin=plugin)
def get_user_setting(
self, key: str, user: User, cache: bool = False, backup_value: Any = None
) -> Any:
"""Return the 'value' of the user setting associated with this plugin.
Arguments:
key: The 'name' of the user setting value to be retrieved
user: The user for which the setting is to be retrieved
cache: Whether to use cached value (default = False)
backup_value: A backup value to return if the setting is not found
"""
from plugin.models import PluginUserSetting
return PluginUserSetting.get_setting(
key,
plugin=self.plugin_config(),
user=user,
cache=cache,
backup_value=backup_value,
settings=self.user_settings,
)
def set_user_setting(self, key: str, value: Any, user: User) -> None:
"""Set user setting value by key.
Arguments:
key: The 'name' of the user setting value to be set
value: The value to be set for the user setting
user: The user for which the setting is to be set
"""
from plugin.models import PluginUserSetting
from plugin.registry import registry
try:
plugin = registry.get_plugin_config(self.plugin_slug(), self.plugin_name())
except (OperationalError, ProgrammingError):
plugin = None
if not plugin: # pragma: no cover
# Cannot find associated plugin model, return
logger.error("Plugin configuration not found for plugin '%s'", self.slug)
return
PluginUserSetting.set_setting(key, value, user, user=user, plugin=plugin)
def check_settings(self):
"""Check if all required settings for this machine are defined.

4
src/backend/InvenTree/plugin/base/label/label.py
View File

@@ -24,6 +24,8 @@ def print_label(plugin_slug: str, **kwargs):
kwargs:
passed through to the plugin.print_label() method
"""
from plugin.builtin.integration.core_notifications import InvenTreeUINotifications
logger.info("Plugin '%s' is printing a label", plugin_slug)
plugin = registry.get_plugin(plugin_slug, active=True)
@@ -53,7 +55,7 @@ def print_label(plugin_slug: str, **kwargs):
'label.printing_failed',
targets=[user],
context=ctx,
delivery_methods={common.notifications.UIMessageNotification},
delivery_methods={InvenTreeUINotifications},
)
if settings.TESTING:

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

@@ -115,11 +115,10 @@ class LabelMixinTests(PrintTestMixins, InvenTreeAPITestCase):
# Plugin is not a label plugin
registry.set_plugin_state('digikeyplugin', True)
no_valid_plg = registry.get_plugin('digikeyplugin').plugin_config()
response = self.post(
url,
{'template': template.pk, 'plugin': no_valid_plg.key, 'items': [1, 2, 3]},
{'template': template.pk, 'plugin': 'digikeyplugin', 'items': [1, 2, 3]},
expected_code=400,
)

290
src/backend/InvenTree/plugin/builtin/integration/core_notifications.py
View File

@@ -1,175 +1,169 @@
"""Core set of Notifications as a Plugin."""
from django.contrib.auth.models import User
from django.db.models import Model
from django.template.loader import render_to_string
from django.utils.translation import gettext_lazy as _
import requests
from allauth.account.models import EmailAddress
import structlog
import common.models
import InvenTree.helpers
import InvenTree.helpers_email
import InvenTree.tasks
from plugin import InvenTreePlugin, registry
from plugin.mixins import BulkNotificationMethod, SettingsMixin
from common.settings import get_global_setting
from plugin import InvenTreePlugin
from plugin.mixins import NotificationMixin, SettingsMixin
logger = structlog.get_logger('inventree')
class PlgMixin:
"""Mixin to access plugin easier.
class InvenTreeUINotifications(NotificationMixin, InvenTreePlugin):
"""Plugin mixin class for supporting UI notification methods."""
This needs to be spit out to reference the class. Perks of python.
"""
def get_plugin(self):
"""Return plugin reference."""
return InvenTreeCoreNotificationsPlugin
class InvenTreeCoreNotificationsPlugin(SettingsMixin, InvenTreePlugin):
"""Core notification methods for InvenTree."""
NAME = 'InvenTreeCoreNotificationsPlugin'
TITLE = _('InvenTree Notifications')
NAME = 'InvenTreeUINotifications'
TITLE = _('InvenTree UI Notifications')
SLUG = 'inventree-ui-notification'
AUTHOR = _('InvenTree contributors')
DESCRIPTION = _('Integrated outgoing notification methods')
DESCRIPTION = _('Integrated UI notification methods')
VERSION = '1.0.0'
def send_notification(
self, target: Model, category: str, users: list[User], context: dict
) -> bool:
"""Create a UI notification entry for specified users."""
from common.models import NotificationMessage
entries = []
if not users:
return False
# Bulk create notification messages for all provided users
for user in users:
entries.append(
NotificationMessage(
target_object=target,
source_object=user,
user=user,
category=category,
name=context['name'],
message=context['message'],
)
)
NotificationMessage.objects.bulk_create(entries)
return True
class InvenTreeEmailNotifications(NotificationMixin, SettingsMixin, InvenTreePlugin):
"""Plugin mixin class for supporting email notification methods."""
NAME = 'InvenTreeEmailNotifications'
TITLE = _('InvenTree Email Notifications')
SLUG = 'inventree-email-notification'
AUTHOR = _('InvenTree contributors')
DESCRIPTION = _('Integrated email notification methods')
VERSION = '1.0.0'
USER_SETTINGS = {
'NOTIFY_BY_EMAIL': {
'name': _('Allow email notifications'),
'description': _('Allow email notifications to be sent to this user'),
'default': True,
'validator': bool,
}
}
def send_notification(
self, target: Model, category: str, users: list[User], context: dict
) -> bool:
"""Send notification to the specified targets."""
# Ignore if there is no template provided to render
if not context.get('template'):
return False
html_message = render_to_string(context['template']['html'], context)
# Prefix the 'instance title' to the email subject
instance_title = get_global_setting('INVENTREE_INSTANCE')
subject = context['template'].get('subject', '')
if instance_title:
subject = f'[{instance_title}] {subject}'
recipients = []
for user in users:
# Skip if the user does not want to receive email notifications
if not self.get_user_setting('NOTIFY_BY_EMAIL', user, backup_value=False):
continue
if email := InvenTree.helpers_email.get_email_for_user(user):
recipients.append(email)
if recipients:
InvenTree.helpers_email.send_email(
subject, '', recipients, html_message=html_message
)
return True
# No recipients found, so we cannot send the email
return False
class InvenTreeSlackNotifications(NotificationMixin, SettingsMixin, InvenTreePlugin):
"""Plugin mixin class for supporting Slack notification methods."""
NAME = 'InvenTreeSlackNotifications'
TITLE = _('InvenTree Slack Notifications')
SLUG = 'inventree-slack-notification'
AUTHOR = _('InvenTree contributors')
DESCRIPTION = _('Integrated Slack notification methods')
VERSION = '1.0.0'
SETTINGS = {
'ENABLE_NOTIFICATION_EMAILS': {
'name': _('Enable email notifications'),
'description': _('Allow sending of emails for event notifications'),
'default': False,
'validator': bool,
},
'ENABLE_NOTIFICATION_SLACK': {
'name': _('Enable slack notifications'),
'description': _(
'Allow sending of slack channel messages for event notifications'
),
'default': False,
'validator': bool,
},
'NOTIFICATION_SLACK_URL': {
'name': _('Slack incoming webhook url'),
'description': _('URL that is used to send messages to a slack channel'),
'protected': True,
},
}
}
def get_settings_content(self, request):
"""Custom settings content for the plugin."""
return """
<p>Setup for Slack:</p>
<ol>
<li>Create a new Slack app on <a href="https://api.slack.com/apps/new" target="_blank">this page</a></li>
<li>Enable <i>Incoming Webhooks</i> for the channel you want the notifications posted to</li>
<li>Set the webhook URL in the settings above</li>
<li>Enable the plugin</li>
"""
def send_notification(
self, target: Model, category: str, users: list[User], context: dict
) -> bool:
"""Send the notifications out via slack."""
url = self.get_setting('NOTIFICATION_SLACK_URL')
class EmailNotification(PlgMixin, BulkNotificationMethod):
"""Notification method for delivery via Email."""
if not url:
return False
METHOD_NAME = 'mail'
METHOD_ICON = 'fa-envelope'
CONTEXT_EXTRA = [('template',), ('template', 'html'), ('template', 'subject')]
GLOBAL_SETTING = 'ENABLE_NOTIFICATION_EMAILS'
USER_SETTING = {
'name': _('Enable email notifications'),
'description': _('Allow sending of emails for event notifications'),
'default': True,
'validator': bool,
}
def get_targets(self):
"""Return a list of target email addresses, only for users which allow email notifications."""
allowed_users = []
for user in self.targets:
if not user.is_active:
# Ignore any users who have been deactivated
continue
allows_emails = InvenTree.helpers.str2bool(self.usersetting(user))
if allows_emails:
allowed_users.append(user)
return EmailAddress.objects.filter(user__in=allowed_users)
def send_bulk(self):
"""Send the notifications out via email."""
html_message = render_to_string(
self.context['template']['html'], self.context
)
targets = self.targets.values_list('email', flat=True)
# Prefix the 'instance title' to the email subject
instance_title = common.models.InvenTreeSetting.get_setting(
'INVENTREE_INSTANCE'
)
subject = self.context['template'].get('subject', '')
if instance_title:
subject = f'[{instance_title}] {subject}'
InvenTree.helpers_email.send_email(
subject, '', targets, html_message=html_message
)
return True
class SlackNotification(PlgMixin, BulkNotificationMethod):
"""Notification method for delivery via Slack channel messages."""
METHOD_NAME = 'slack'
METHOD_ICON = 'fa-envelope'
GLOBAL_SETTING = 'ENABLE_NOTIFICATION_SLACK'
def get_targets(self):
"""Not used by this method."""
return self.targets
def send_bulk(self):
"""Send the notifications out via slack."""
instance = registry.get_plugin(self.get_plugin().NAME.lower())
url = instance.get_setting('NOTIFICATION_SLACK_URL')
if not url:
return False
ret = requests.post(
url,
json={
'text': str(self.context['message']),
'blocks': [
{
'type': 'section',
ret = requests.post(
url,
json={
'text': str(context['message']),
'blocks': [
{
'type': 'section',
'text': {'type': 'plain_text', 'text': str(context['name'])},
},
{
'type': 'section',
'text': {'type': 'mrkdwn', 'text': str(context['message'])},
'accessory': {
'type': 'button',
'text': {
'type': 'plain_text',
'text': str(self.context['name']),
'text': str(_('Open link')),
'emoji': True,
},
'value': f'{category}_{target.pk}' if target else '',
'url': context['link'],
'action_id': 'button-action',
},
{
'type': 'section',
'text': {
'type': 'mrkdwn',
'text': str(self.context['message']),
},
'accessory': {
'type': 'button',
'text': {
'type': 'plain_text',
'text': str(_('Open link')),
'emoji': True,
},
'value': f'{self.category}_{self.obj.pk}',
'url': self.context['link'],
'action_id': 'button-action',
},
},
],
},
)
return ret.ok
},
],
},
)
return ret.ok

67
src/backend/InvenTree/plugin/builtin/integration/test_core_notifications.py
View File

@@ -2,35 +2,64 @@
from django.core import mail
from part.test_part import BaseNotificationIntegrationTest
from common.models import NotificationEntry
from common.notifications import trigger_notification
from InvenTree.unit_test import InvenTreeTestCase
from plugin import registry
from plugin.builtin.integration.core_notifications import (
InvenTreeCoreNotificationsPlugin,
)
from plugin.models import NotificationUserSetting
class CoreNotificationTestTests(BaseNotificationIntegrationTest):
class CoreNotificationTestTests(InvenTreeTestCase):
"""Tests for CoreNotificationsPlugin."""
def setUp(self):
"""Set up the test case."""
super().setUp()
# Ensure that the user has an email address set
self.user.email = 'test.user@demo.inventree.org'
self.user.is_superuser = True
self.user.save()
def notify(self):
"""Send an email notification."""
# Clear all entries to ensure the notification is sent
NotificationEntry.objects.all().delete()
trigger_notification(
self.user,
'test_notification',
targets=[self.user],
context={
'name': 'Test Email Notification',
'message': 'This is a test email notification.',
'template': {
'html': 'email/test_email.html',
'subject': 'Test Email Notification',
},
},
)
def test_email(self):
"""Ensure that the email notifications run."""
# No email should be send
self.assertEqual(len(mail.outbox), 0)
# enable plugin and set mail setting to true
plugin = registry.get_plugin('inventreecorenotificationsplugin')
plugin.set_setting('ENABLE_NOTIFICATION_EMAILS', True)
NotificationUserSetting.set_setting(
key='NOTIFICATION_METHOD_MAIL',
value=True,
change_user=self.user,
user=self.user,
method=InvenTreeCoreNotificationsPlugin.EmailNotification.METHOD_NAME,
)
print('- get email plugin:')
# run through
self._notification_run(InvenTreeCoreNotificationsPlugin.EmailNotification)
plugin = registry.get_plugin('inventree-email-notification')
self.assertIsNotNone(plugin, 'Email notification plugin should be available')
# First, try with setting disabled
plugin.set_user_setting('NOTIFY_BY_EMAIL', False, self.user)
self.notify()
# No email should be sent
self.assertEqual(len(mail.outbox), 0)
# Now, enable the setting
plugin.set_user_setting('NOTIFY_BY_EMAIL', True, self.user)
self.notify()
# Now one mail should be send
self.assertEqual(len(mail.outbox), 1)

14
src/backend/InvenTree/plugin/builtin/suppliers/test_supplier_barcodes.py
View File

@@ -113,6 +113,15 @@ class SupplierBarcodeTests(InvenTreeAPITestCase):
def test_old_mouser_barcode(self):
"""Test old mouser barcode with messed up header."""
registry.set_plugin_state('mouserplugin', False)
# Initial scan should fail - plugin not enabled
self.post(
self.SCAN_URL, data={'barcode': MOUSER_BARCODE_OLD}, expected_code=400
)
registry.set_plugin_state('mouserplugin', True)
result = self.post(
self.SCAN_URL, data={'barcode': MOUSER_BARCODE_OLD}, expected_code=200
)
@@ -171,6 +180,11 @@ class SupplierBarcodePOReceiveTests(InvenTreeAPITestCase):
"""Create supplier part and purchase_order."""
super().setUp()
registry.set_plugin_state('digikeyplugin', True)
registry.set_plugin_state('mouserplugin', True)
registry.set_plugin_state('lcscplugin', True)
registry.set_plugin_state('tmeplugin', True)
self.loc_1 = StockLocation.objects.create(name='Location 1')
self.loc_2 = StockLocation.objects.create(name='Location 2')

59
src/backend/InvenTree/plugin/migrations/0010_pluginusersetting.py Normal file
View File

@@ -0,0 +1,59 @@
# Generated by Django 4.2.22 on 2025-06-09 02:03
from django.conf import settings
from django.db import migrations, models
import django.db.models.deletion
class Migration(migrations.Migration):
dependencies = [
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
("plugin", "0009_alter_pluginconfig_key"),
]
operations = [
migrations.CreateModel(
name="PluginUserSetting",
fields=[
(
"id",
models.AutoField(
auto_created=True,
primary_key=True,
serialize=False,
verbose_name="ID",
),
),
("key", models.CharField(help_text="Settings key", max_length=50)),
(
"value",
models.CharField(
blank=True, help_text="Settings value", max_length=2000
),
),
(
"plugin",
models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE,
related_name="user_settings",
to="plugin.pluginconfig",
verbose_name="Plugin",
),
),
(
"user",
models.ForeignKey(
help_text="User",
on_delete=django.db.models.deletion.CASCADE,
related_name="plugin_settings",
to=settings.AUTH_USER_MODEL,
verbose_name="User",
),
),
],
options={
"unique_together": {("plugin", "user", "key")},
},
),
]

16
src/backend/InvenTree/plugin/migrations/0011_delete_notificationusersetting.py Normal file
View File

@@ -0,0 +1,16 @@
# Generated by Django 4.2.22 on 2025-06-09 06:59
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
("plugin", "0010_pluginusersetting"),
]
operations = [
migrations.DeleteModel(
name="NotificationUserSetting",
),
]

5
src/backend/InvenTree/plugin/mixins/__init__.py
View File

@@ -1,6 +1,5 @@
"""Utility class to enable simpler imports."""
from common.notifications import BulkNotificationMethod, SingleNotificationMethod
from plugin.base.action.mixins import ActionMixin
from plugin.base.barcodes.mixins import BarcodeMixin, SupplierBarcodeMixin
from plugin.base.event.mixins import EventMixin
@@ -10,6 +9,7 @@ from plugin.base.integration.AppMixin import AppMixin
from plugin.base.integration.CurrencyExchangeMixin import CurrencyExchangeMixin
from plugin.base.integration.DataExport import DataExportMixin
from plugin.base.integration.NavigationMixin import NavigationMixin
from plugin.base.integration.NotificationMixin import NotificationMixin
from plugin.base.integration.ReportMixin import ReportMixin
from plugin.base.integration.ScheduleMixin import ScheduleMixin
from plugin.base.integration.SettingsMixin import SettingsMixin
@@ -25,7 +25,6 @@ __all__ = [
'ActionMixin',
'AppMixin',
'BarcodeMixin',
'BulkNotificationMethod',
'CurrencyExchangeMixin',
'DataExportMixin',
'EventMixin',
@@ -34,10 +33,10 @@ __all__ = [
'LocateMixin',
'MailMixin',
'NavigationMixin',
'NotificationMixin',
'ReportMixin',
'ScheduleMixin',
'SettingsMixin',
'SingleNotificationMethod',
'SupplierBarcodeMixin',
'UrlsMixin',
'UserInterfaceMixin',

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

@@ -292,37 +292,60 @@ class PluginSetting(common.models.BaseInvenTreeSetting):
return super().get_setting_definition(key, **kwargs)
class NotificationUserSetting(common.models.BaseInvenTreeSetting):
"""This model represents notification settings for a user."""
class PluginUserSetting(common.models.BaseInvenTreeSetting):
"""This model represents user-specific settings for individual plugins.
typ = 'notification'
extra_unique_fields = ['method', 'user']
In contrast with the PluginSetting model, which holds global settings for plugins,
this model allows for user-specific settings that can be defined by each user.
"""
typ = 'plugin_user'
extra_unique_fields = ['plugin', 'user']
class Meta:
"""Meta for NotificationUserSetting."""
"""Meta for PluginUserSetting."""
unique_together = [('method', 'user', 'key')]
unique_together = [('plugin', 'user', 'key')]
@classmethod
def get_setting_definition(cls, key, **kwargs):
"""Override setting_definition to use notification settings."""
from common.notifications import storage
kwargs['settings'] = storage.user_settings
return super().get_setting_definition(key, **kwargs)
method = models.CharField(max_length=255, verbose_name=_('Method'))
plugin = models.ForeignKey(
PluginConfig,
related_name='user_settings',
null=False,
verbose_name=_('Plugin'),
on_delete=models.CASCADE,
)
user = models.ForeignKey(
User,
on_delete=models.CASCADE,
blank=True,
null=True,
null=False,
verbose_name=_('User'),
help_text=_('User'),
related_name='plugin_settings',
)
def __str__(self) -> str:
"""Nice name of printing."""
return f'{self.key} (for {self.user}): {self.value}'
@classmethod
def get_setting_definition(cls, key, **kwargs):
"""In the BaseInvenTreeSetting class, we have a class attribute named 'SETTINGS', which is a dict object that fully defines all the setting parameters.
Here, unlike the BaseInvenTreeSetting, we do not know the definitions of all settings
'ahead of time' (as they are defined externally in the plugins).
Settings can be provided by the caller, as kwargs['settings'].
If not provided, we'll look at the plugin registry to see what settings are available,
(if the plugin is specified!)
"""
if 'settings' not in kwargs:
plugin = kwargs.pop('plugin', None)
if plugin:
mixin_user_settings = getattr(registry, 'mixins_user_settings', None)
if mixin_user_settings:
kwargs['settings'] = mixin_user_settings.get(plugin.key, {})
return super().get_setting_definition(key, **kwargs)

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

@@ -38,6 +38,7 @@ class PluginMixinEnum(StringEnum):
LOCATE = 'locate'
MAIL = 'mail'
NAVIGATION = 'navigation'
NOTIFICATION = 'notification'
REPORT = 'report'
SCHEDULE = 'schedule'
SETTINGS = 'settings'

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

@@ -92,9 +92,9 @@ class PluginsRegistry:
'inventreebarcode',
'bom-exporter',
'inventree-exporter',
'inventreecorenotificationsplugin',
'inventree-ui-notification',
'inventree-email-notification',
'inventreecurrencyexchange',
'inventreecorenotificationsplugin',
'inventreelabel',
'inventreelabelmachine',
'parameter-exporter',

20
src/backend/InvenTree/plugin/samples/integration/sample.py
View File

@@ -47,6 +47,26 @@ class SampleIntegrationPlugin(
path('ho/', include(he_urls), name='ho'),
]
USER_SETTINGS = {
'USER_SETTING_1': {
'name': _('User Setting 1'),
'description': _('A user setting that can be changed by the user'),
'default': 'Default Value',
},
'USER_SETTING_2': {
'name': _('User Setting 2'),
'description': _('Another user setting'),
'default': True,
'validator': bool,
},
'USER_SETTING_3': {
'name': _('User Setting 3'),
'description': _('A user setting with choices'),
'choices': [('X', 'Choice X'), ('Y', 'Choice Y'), ('Z', 'Choice Z')],
'default': 'X',
},
}
SETTINGS = {
'PO_FUNCTION_ENABLE': {
'name': _('Enable PO'),

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

@@ -8,7 +8,7 @@ from drf_spectacular.utils import extend_schema_field
from rest_framework import serializers
from common.serializers import GenericReferencedSettingSerializer
from plugin.models import NotificationUserSetting, PluginConfig, PluginSetting
from plugin.models import PluginConfig, PluginSetting, PluginUserSetting
class MetadataSerializer(serializers.ModelSerializer):
@@ -275,14 +275,16 @@ class PluginSettingSerializer(GenericReferencedSettingSerializer):
plugin = serializers.CharField(source='plugin.key', read_only=True)
class NotificationUserSettingSerializer(GenericReferencedSettingSerializer):
"""Serializer for the PluginSetting model."""
class PluginUserSettingSerializer(GenericReferencedSettingSerializer):
"""Serializer for the PluginUserSetting model."""
MODEL = NotificationUserSetting
EXTRA_FIELDS = ['method']
MODEL = PluginUserSetting
EXTRA_FIELDS = ['plugin', 'user']
method = serializers.CharField(read_only=True)
typ = serializers.CharField(read_only=True)
plugin = serializers.CharField(source='plugin.key', read_only=True)
user = serializers.PrimaryKeyRelatedField(
read_only=True, help_text=_('The user for which this setting applies')
)
class PluginRegistryErrorSerializer(serializers.Serializer):

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

@@ -315,6 +315,64 @@ class PluginDetailAPITest(PluginMixin, InvenTreeAPITestCase):
self.assertEqual(response.data['value'], '456')
def test_plugin_user_settings(self):
"""Test the PluginUserSetting API endpoints."""
# Fetch user settings for invalid plugin
response = self.get(
reverse(
'api-plugin-user-setting-list', kwargs={'plugin': 'invalid-plugin'}
),
expected_code=404,
)
# Fetch all user settings for the 'email' plugin
url = reverse(
'api-plugin-user-setting-list',
kwargs={'plugin': 'inventree-email-notification'},
)
response = self.get(url, expected_code=200)
settings_keys = [item['key'] for item in response.data]
self.assertIn('NOTIFY_BY_EMAIL', settings_keys)
# Fetch user settings for an invalid key
self.get(
reverse(
'api-plugin-user-setting-detail',
kwargs={'plugin': 'inventree-email-notification', 'key': 'INVALID_KEY'},
),
expected_code=404,
)
# Fetch user setting detail for a valid key
response = self.get(
reverse(
'api-plugin-user-setting-detail',
kwargs={
'plugin': 'inventree-email-notification',
'key': 'NOTIFY_BY_EMAIL',
},
),
expected_code=200,
)
# User ID must match the current user
self.assertEqual(response.data['user'], self.user.pk)
# Check for expected values
for k in [
'pk',
'key',
'value',
'name',
'description',
'type',
'model_name',
'user',
]:
self.assertIn(k, response.data)
def test_plugin_metadata(self):
"""Test metadata endpoint for plugin."""
self.user.is_superuser = True

6
src/backend/InvenTree/templates/email/build_order_required_stock.html
View File

@@ -11,7 +11,7 @@
{% endblock title %}
{% block body %}
<tr colspan='100%' style='height: 2rem; text-align: center;'>{% trans "The following parts are low on required stock" %}</tr>
<tr colspan='3' style='height: 2rem; text-align: center;'>{% trans "The following parts are low on required stock" %}</tr>
<tr style="height: 3rem; border-bottom: 1px solid">
<th>{% trans "Part" %}</th>
@@ -25,9 +25,9 @@
<a href='{{ line.link }}'>{{ line.part.full_name }}</a>{% if line.part.description %} - <em>{{ line.part.description }}</em>{% endif %}
</td>
<td style="text-align: center;">
{% decimal line.required %}
{% decimal line.required %} {% if part.units %} [{{ part.units }}]{% endif %}
</td>
<td style="text-align: center;">{% decimal line.available %}</td>
<td style="text-align: center;">{% decimal line.available %} {% if part.units %} [{{ part.units }}]{% endif %}</td>
</tr>
{% endfor %}

5
src/backend/InvenTree/templates/email/test_email.html Normal file
View File

@@ -0,0 +1,5 @@
{% load i18n %}
{% load static %}
{% load inventree_extras %}
This is a simple test email template used to verify that email functionality is working correctly in InvenTree.

2
src/backend/InvenTree/users/ruleset.py
View File

@@ -79,7 +79,7 @@ def get_ruleset_models() -> dict:
# Plugins
'plugin_pluginconfig',
'plugin_pluginsetting',
'plugin_notificationusersetting',
'plugin_pluginusersetting',
# Misc
'common_barcodescanresult',
'common_newsfeedentry',