Messages endpoint (#13247)

* Adds /api/status/ endpoint for admin-facing system messages

* update alerts menu

* right-align badges

* Rename SERVER_RESTART_REQUIRED to _SERVER_RESTART_REQUIRED

* Adjust API endpoint

- Move to /api/system/status/
- Make messages a sub-list
- Remove "system_state" from /api/

* move staff-only messages to new endpoint (and adjust UI accordingly)

* system dialogs staff only

* Reduce scope and revert most changes

* ignore reports in frontend

* clean up blob

---------

Co-authored-by: Matthias Mair <code@mjmair.com>
This commit is contained in:
OliverandMatthias Mair authored and GitHub committed 2026-11-12 11:14:01 +11:00
1 parent a0ec8e4c71
commit af5055b33e
11 files changed
+742 -157

No files matched your search

@@ -1,11 +1,14 @@
"""InvenTree API version information."""
# InvenTree API version
INVENTREE_API_VERSION = 575
INVENTREE_API_VERSION = 576
"""Increment this API version number whenever there is a significant change to the API that any clients need to know about."""
INVENTREE_API_TEXT = """
v576 -> 2026-11-11 : https://github.com/inventree/InvenTree/pull/13247
- Adds /api/system/messages/ endpoint: staff-only, read-only; classified system messages are provided in the "messages" array
v575 -> 2026-11-10 : https://github.com/inventree/InvenTree/pull/13208
- Adds optional read-only "consumed_by_detail" to the StockItem API
@@ -0,0 +1,180 @@
"""Staff-only system messages, displayed in the admin center.
Each message describes the outcome of a single system check. Messages are
generated on demand (they are not stored in the database), and are exposed
via the read-only /api/system/messages/ endpoint.
"""
from dataclasses import dataclass
from enum import StrEnum
from typing import Optional
from django.conf import settings
from django.utils.translation import gettext_lazy as _
import structlog
from common.settings import get_global_setting
logger = structlog.get_logger('inventree')
class AdminMessageClassifier(StrEnum):
"""Severity classifier for an admin message."""
ERROR = 'error'
WARNING = 'warning'
INFO = 'info'
SUCCESS = 'success'
# Ordering used when sorting messages (most severe first)
CLASSIFIER_RANK = {
AdminMessageClassifier.ERROR: 0,
AdminMessageClassifier.WARNING: 1,
AdminMessageClassifier.INFO: 2,
AdminMessageClassifier.SUCCESS: 3,
}
@dataclass
class AdminMessage:
"""A single staff-facing system message.
Attributes:
key: Unique identifier for the message (e.g. 'worker')
title: Short, human readable title
message: Detailed message text
classifier: Severity of the message
code: Optional InvenTree error code (e.g. 'INVE-W5'), used to link to documentation
plugin: Slug of the plugin which provided this message (None for core messages)
"""
key: str
title: str
message: str
classifier: AdminMessageClassifier
code: Optional[str] = None
plugin: Optional[str] = None
def _check(
key: str,
title: str,
message: str,
condition: bool,
code: Optional[str] = None,
classifier: AdminMessageClassifier = AdminMessageClassifier.WARNING,
) -> AdminMessage:
"""Construct a message for a simple pass/fail system check.
Arguments:
key: Unique message key
title: Message title
message: Message text to display if the check fails
condition: True if the check has *failed* (i.e. the issue is present)
code: Optional error code
classifier: Classifier to use if the check fails
"""
if condition:
return AdminMessage(
key=key, title=title, message=message, classifier=classifier, code=code
)
return AdminMessage(
key=key,
title=title,
message=_('No issues detected'),
classifier=AdminMessageClassifier.SUCCESS,
code=code,
)
def _pending_migrations() -> int:
"""Return the number of pending database migrations (as recorded by the background worker)."""
try:
return int(get_global_setting('_PENDING_MIGRATIONS', 0) or 0)
except (TypeError, ValueError):
return 0
def get_core_admin_messages() -> list[AdminMessage]:
"""Return the set of core InvenTree system messages."""
# Imported here to avoid circular imports at app load
from InvenTree.helpers_email import is_email_configured
from InvenTree.status import is_worker_running
return [
_check(
'debug',
_('Debug Mode'),
_('The server is running in debug mode.'),
bool(settings.DEBUG),
code='INVE-W4',
),
_check(
'worker',
_('Background Worker'),
_('The background worker process is not running.'),
not is_worker_running(),
code='INVE-W5',
),
_check(
'restart',
_('Server Restart'),
_('The server requires a restart to apply changes.'),
bool(get_global_setting('SERVER_RESTART_REQUIRED', False, cache=False)),
code='INVE-W6',
),
_check(
'email',
_('Email settings'),
_('Email settings not configured.'),
not is_email_configured(),
code='INVE-W7',
),
_check(
'migrations',
_('Database Migrations'),
_('There are pending database migrations.'),
_pending_migrations() > 0,
code='INVE-W8',
),
_check(
'cors',
_('CORS Settings'),
_('CORS is set to allow all origins.'),
bool(settings.CORS_ALLOW_ALL_ORIGINS),
code='INVE-W16',
),
_check(
'basic_auth',
_('Basic Authentication'),
_('Basic (username:password) authentication is enabled for the API.'),
bool(settings.BASIC_AUTH_ENABLED),
code='INVE-W18',
),
]
def get_plugin_admin_messages() -> list[AdminMessage]:
"""Return system messages provided by plugins.
PLACEHOLDER: There is not yet a plugin mixin for providing admin messages.
The intended design is a dedicated mixin (e.g. 'AdminMessageMixin') with a
method returning a list of AdminMessage objects. Each returned message
should be tagged with the providing plugin's slug (AdminMessage.plugin),
and have its key namespaced (e.g. '<slug>.<key>') to avoid collisions
with core messages. Any exception raised by a plugin must be caught and
logged here, so that a faulty plugin cannot break the endpoint.
"""
return []
def get_admin_messages() -> list[AdminMessage]:
"""Return all admin messages (core and plugin), sorted most severe first."""
messages = get_core_admin_messages() + get_plugin_admin_messages()
# Stable sort - preserves definition order within each classifier
return sorted(messages, key=lambda msg: CLASSIFIER_RANK[msg.classifier])
+29 -1
View File
@@ -1741,6 +1741,32 @@ class HealthCheckStatusSerializer(serializers.Serializer):
)
class AdminMessageList(APIView):
"""Read-only, staff-only list of classified system messages (error / warning / info / success).
Messages are generated on demand from core system checks (and potentially plugins).
"""
permission_classes = [IsAdminOrAdminScope]
@extend_schema(
responses={
200: OpenApiResponse(
response=common.serializers.AdminMessageListSerializer,
description='System messages for staff users',
)
}
)
def get(self, request, *args, **kwargs):
"""Return system messages (most severe first)."""
from common.admin_messages import get_admin_messages
serializer = common.serializers.AdminMessageListSerializer({
'messages': get_admin_messages()
})
return Response(serializer.data)
class HealthCheckView(APIView):
"""Simple JSON endpoint for InvenTree health check.
@@ -2170,7 +2196,9 @@ common_api_urls = [
'system/',
include([
# Health check
path('health/', HealthCheckView.as_view(), name='api-system-health')
path('health/', HealthCheckView.as_view(), name='api-system-health'),
# Staff-only system status messages
path('messages/', AdminMessageList.as_view(), name='api-system-messages'),
]),
),
# Internal System APIs - DO NOT USE
@@ -16,6 +16,7 @@ from rest_framework import serializers
from rest_framework.exceptions import PermissionDenied
from taggit.models import Tag
import common.admin_messages
import common.filters
import common.models as common_models
import common.validators
@@ -506,6 +507,31 @@ class CustomStateSerializer(DataImportExportSerializerMixin, InvenTreeModelSeria
)
class AdminMessageSerializer(serializers.Serializer):
"""Serializer for a (read-only) staff-facing system message.
Instances are common.admin_messages.AdminMessage dataclass objects.
"""
key = serializers.CharField(read_only=True)
title = serializers.CharField(read_only=True)
message = serializers.CharField(read_only=True)
classifier = serializers.ChoiceField(
read_only=True,
choices=[
(c.value, c.value) for c in common.admin_messages.AdminMessageClassifier
],
)
code = serializers.CharField(read_only=True, allow_null=True)
plugin = serializers.CharField(read_only=True, allow_null=True)
class AdminMessageListSerializer(serializers.Serializer):
"""Serializer for the staff-only system messages endpoint."""
messages = AdminMessageSerializer(many=True, read_only=True)
class FlagSerializer(serializers.Serializer):
"""Serializer for feature flags."""
+117
View File
@@ -3592,3 +3592,120 @@ class GenericMetadataAuthorizationTests(InvenTreeAPITestCase):
request = factory.patch('/')
request.user = self.user
self.assertTrue(perm.has_object_permission(request, None, unit))
class AdminMessageAPITests(InvenTreeAPITestCase):
"""Tests for the staff-only /api/system/messages/ endpoint."""
roles = []
CORE_KEYS = {
'debug',
'worker',
'restart',
'email',
'migrations',
'cors',
'basic_auth',
}
@property
def url(self):
"""Return the endpoint URL."""
return reverse('api-system-messages')
def _messages(self) -> dict:
"""Fetch messages, keyed by message key."""
response = self.get(self.url, expected_code=200)
return {msg['key']: msg for msg in response.data['messages']}
def test_permissions(self):
"""Only staff users may access the endpoint."""
self.get(self.url, expected_code=200)
self.user.is_staff = False
self.user.save()
self.get(self.url, expected_code=403)
self.logout()
self.get(self.url, expected_code=401)
def test_read_only(self):
"""The endpoint does not accept write methods."""
self.post(self.url, {}, expected_code=405)
self.put(self.url, {}, expected_code=405)
self.patch(self.url, {}, expected_code=405)
self.delete(self.url, expected_code=405)
def test_core_messages(self):
"""All core system checks are reported, with valid structure."""
response = self.get(self.url, expected_code=200)
self.assertEqual(set(response.data.keys()), {'messages'})
self.assertIsInstance(response.data['messages'], list)
messages = self._messages()
self.assertTrue(self.CORE_KEYS.issubset(messages.keys()))
for msg in messages.values():
self.assertIn(msg['classifier'], ['error', 'warning', 'info', 'success'])
self.assertTrue(msg['title'])
self.assertTrue(msg['message'])
self.assertIsNone(msg['plugin'])
self.assertEqual(messages['basic_auth']['code'], 'INVE-W18')
def test_ordering(self):
"""Messages are returned most-severe first."""
rank = {'error': 0, 'warning': 1, 'info': 2, 'success': 3}
response = self.get(self.url, expected_code=200)
ranks = [rank[msg['classifier']] for msg in response.data['messages']]
self.assertEqual(ranks, sorted(ranks))
def test_conditions(self):
"""Message classifiers reflect the underlying system state."""
with override_settings(CORS_ALLOW_ALL_ORIGINS=True, BASIC_AUTH_ENABLED=True):
messages = self._messages()
self.assertEqual(messages['cors']['classifier'], 'warning')
self.assertEqual(messages['basic_auth']['classifier'], 'warning')
with override_settings(CORS_ALLOW_ALL_ORIGINS=False, BASIC_AUTH_ENABLED=False):
messages = self._messages()
self.assertEqual(messages['cors']['classifier'], 'success')
self.assertEqual(messages['basic_auth']['classifier'], 'success')
set_global_setting('_PENDING_MIGRATIONS', 3, None)
set_global_setting('SERVER_RESTART_REQUIRED', True, None)
messages = self._messages()
self.assertEqual(messages['migrations']['classifier'], 'warning')
self.assertEqual(messages['restart']['classifier'], 'warning')
set_global_setting('_PENDING_MIGRATIONS', 0, None)
set_global_setting('SERVER_RESTART_REQUIRED', False, None)
messages = self._messages()
self.assertEqual(messages['migrations']['classifier'], 'success')
self.assertEqual(messages['restart']['classifier'], 'success')
def test_plugin_messages(self):
"""Plugin-provided messages are merged in and sorted by severity."""
from unittest import mock
from common.admin_messages import AdminMessage, AdminMessageClassifier
plugin_msg = AdminMessage(
key='sample.error',
title='Plugin Error',
message='Something is wrong',
classifier=AdminMessageClassifier.ERROR,
plugin='sample',
)
with mock.patch(
'common.admin_messages.get_plugin_admin_messages', return_value=[plugin_msg]
):
response = self.get(self.url, expected_code=200)
first = response.data['messages'][0]
self.assertEqual(first['key'], 'sample.error')
self.assertEqual(first['classifier'], 'error')
self.assertEqual(first['plugin'], 'sample')