Transition plugin (#10088)

* Add new enum

* Define StateTransitionMixin class

* Import

* Rename plugin mixin class

- Avoid naming clash

* Offload transitions to plugin registry

* fix imports

* Fix default value

* Fix sample transition class

* Refactor unit test

* Add docs

* Fix type hinting

* Check for expected message

* Tests for various failure modes

* Tweak query count limit

* Exclude lines from coverage

* Remove debug code

---------

Co-authored-by: Matthias Mair <code@mjmair.com>

docs/docs/plugins/mixins/transition.md

@@ -0,0 +1,16 @@
+---
+title: Transition Mixin
+---
+
+## TransitionMixin
+
+The `TransitionMixin` allows plugins to provide custom state transition logic for InvenTree models.
+
+Model types which support transition between different "states" (e.g. orders), can be extended to support custom transition logic by implementing the `TransitionMixin` class.
+
+This allows for custom functionality to be executed when a state transition occurs, such as sending notifications, updating related models, or performing other actions.
+
+Additionally, the mixin can be used to prevent certain transitions from occurring, or to modify the transition logic based on custom conditions.
+
+!!! info "More Info"
+    For more information on this plugin mixin, refer to the InvenTree source code. [A working example is available as a starting point]({{ sourcefile("/src/backend/InvenTree/plugin/samples/integration/transition.py") }}).

docs/mkdocs.yml

@@ -233,6 +233,7 @@ nav:
       - Report Mixin: plugins/mixins/report.md
       - Schedule Mixin: plugins/mixins/schedule.md
       - Settings Mixin: plugins/mixins/settings.md
+      - Transition Mixin: plugins/mixins/transition.md
       - URL Mixin: plugins/mixins/urls.md
       - User Interface Mixin: plugins/mixins/ui.md
       - Validation Mixin: plugins/mixins/validation.md

src/backend/InvenTree/InvenTree/apps.py

@@ -70,7 +70,6 @@ class InvenTreeConfig(AppConfig):
                 InvenTree.tasks.offload_task(InvenTree.tasks.check_for_migrations)
 
         self.update_site_url()
-        self.collect_state_transition_methods()
 
         # Ensure the unit registry is loaded
         InvenTree.conversion.get_unit_registry()
@@ -312,12 +311,6 @@ class InvenTreeConfig(AppConfig):
         # do not try again
         settings.USER_ADDED_FILE = True
 
-    def collect_state_transition_methods(self):
-        """Collect all state transition methods."""
-        from generic.states import storage
-
-        storage.collect()
-
     def ensure_migrations_done(self=None):
         """Ensures there are no open migrations, stop if inconsistent state."""
         global MIGRATIONS_CHECK_DONE

src/backend/InvenTree/build/test_api.py

@@ -996,7 +996,7 @@ class BuildOverallocationTest(BuildAPITest):
             self.url,
             {'accept_overallocated': 'trim'},
             expected_code=201,
-            max_query_count=375,
+            max_query_count=400,
         )
 
         # Note: Large number of queries is due to pricing recalculation for each stock item

src/backend/InvenTree/generic/states/__init__.py

@@ -7,7 +7,7 @@ States can be extended with custom options for each InvenTree instance - those o
 """
 
 from .states import ColorEnum, StatusCode, StatusCodeMixin
-from .transition import StateTransitionMixin, TransitionMethod, storage
+from .transition import StateTransitionMixin, TransitionMethod
 
 __all__ = [
     'ColorEnum',
@@ -15,5 +15,4 @@ __all__ = [
     'StatusCode',
     'StatusCodeMixin',
     'TransitionMethod',
-    'storage',
 ]

src/backend/InvenTree/generic/states/test_transition.py

@@ -1,107 +1,123 @@
 """Tests for state transition mechanism."""
 
+from django.core.exceptions import ValidationError
+
 from InvenTree.unit_test import InvenTreeTestCase
-
-from .transition import StateTransitionMixin, TransitionMethod, storage
-
-# Global variables to determine which transition classes raises an exception
-raise_storage = False
-raise_function = False
-
-
-class MyPrivateError(NotImplementedError):
-    """Error for testing purposes."""
-
-
-def dflt(*args, **kwargs):
-    """Default function for testing."""
-    raise MyPrivateError('dflt')
-
-
-def _clean_storage(refs):
-    """Clean the storage."""
-    for ref in refs:
-        del ref
-    storage.collect()
+from order.models import ReturnOrder
+from order.status_codes import ReturnOrderStatus
+from plugin import registry
 
 
 class TransitionTests(InvenTreeTestCase):
-    """Tests for basic TransitionMethod."""
+    """Tests for custom state transition logic."""
 
-    def test_class(self):
-        """Ensure that the class itself works."""
+    fixtures = ['company', 'return_order', 'part', 'stock', 'location', 'category']
 
-        class ErrorImplementation(TransitionMethod): ...
+    def setUp(self):
+        """Set up the test environment."""
+        super().setUp()
+        self.ensurePluginsLoaded()
 
-        with self.assertRaises(NotImplementedError):
-            ErrorImplementation()
+    def test_return_order(self):
+        """Test transition of a return order."""
+        # Ensure plugin is enabled
+        registry.set_plugin_state('sample-transition', True)
 
-        _clean_storage([ErrorImplementation])
+        ro = ReturnOrder.objects.get(pk=2)
+        self.assertEqual(ro.status, ReturnOrderStatus.IN_PROGRESS.value)
 
-    def test_storage(self):
-        """Ensure that the storage collection mechanism works."""
-        global raise_storage
-        global raise_function
+        # Attempt to transition to COMPLETE state
+        # This should fail - due to the StateTransitionMixin logic
+        with self.assertRaises(ValidationError) as e:
+            ro.complete_order()
 
-        raise_storage = True
-        raise_function = False
-
-        class RaisingImplementation(TransitionMethod):
-            def transition(self, *args, **kwargs):
-                """Custom transition method."""
-                global raise_storage
-
-                if raise_storage:
-                    raise MyPrivateError('RaisingImplementation')
-
-        # Ensure registering works
-        storage.collect()
-
-        # Ensure the class is registered
-        self.assertIn(RaisingImplementation, storage.list)
-
-        # Ensure stuff is passed to the class
-        with self.assertRaises(MyPrivateError) as exp:
-            StateTransitionMixin.handle_transition(0, 1, self, self, dflt)
-        self.assertEqual(str(exp.exception), 'RaisingImplementation')
-
-        _clean_storage([RaisingImplementation])
-
-    def test_function(self):
-        """Ensure that a TransitionMethod's function is called."""
-        global raise_storage
-        global raise_function
-
-        raise_storage = False
-        raise_function = True
-
-        # Setup
-        class ValidImplementationNoEffect(TransitionMethod):
-            def transition(self, *args, **kwargs):
-                return False  # Return false to test that that work too
-
-        class ValidImplementation(TransitionMethod):
-            def transition(self, *args, **kwargs):
-                global raise_function
-
-                if raise_function:
-                    return 1234
-                else:
-                    return False  # pragma: no cover # Return false to keep other transitions working
-
-        storage.collect()
-        self.assertIn(ValidImplementationNoEffect, storage.list)
-        self.assertIn(ValidImplementation, storage.list)
-
-        # Ensure that the function is called
-        self.assertEqual(
-            StateTransitionMixin.handle_transition(0, 1, self, self, dflt), 1234
+        self.assertIn(
+            'Return order without responsible owner can not be completed',
+            str(e.exception),
         )
 
-        _clean_storage([ValidImplementationNoEffect, ValidImplementation])
+        # Now disable the plugin
+        registry.set_plugin_state('sample-transition', False)
 
-    def test_default_function(self):
-        """Ensure that the default function is called."""
-        with self.assertRaises(MyPrivateError) as exp:
-            StateTransitionMixin.handle_transition(0, 1, self, self, dflt)
-        self.assertEqual(str(exp.exception), 'dflt')
+        # Attempt to transition again
+        ro.complete_order()
+        ro.refresh_from_db()
+
+        self.assertEqual(ro.status, ReturnOrderStatus.COMPLETE.value)
+
+    def test_broken_transition_plugin(self):
+        """Test handling of an intentionally broken transition plugin.
+
+        This test uses a custom plugin which is designed to fail in various ways.
+        """
+        from error_report.models import Error
+
+        Error.objects.all().delete()
+
+        # Ensure the correct plugin is enabled
+        registry.set_plugin_state('sample-transition', False)
+        registry.set_plugin_state('sample-broken-transition', True)
+
+        ro = ReturnOrder.objects.get(pk=2)
+        self.assertEqual(ro.status, ReturnOrderStatus.IN_PROGRESS.value)
+
+        # Transition to "ON HOLD" state
+        ro.hold_order()
+
+        # Ensure plugin starts in a known state
+        plugin = registry.get_plugin('sample-broken-transition')
+        plugin.set_setting('BROKEN_GET_METHOD', False)
+        plugin.set_setting('WRONG_RETURN_TYPE', False)
+        plugin.set_setting('WRONG_RETURN_VALUE', False)
+
+        # Expect a "warning" message on each run
+        # This assures us that the transition handler is being called
+        msg = 'get_transition_handlers is intentionally broken in this plugin'
+
+        with self.assertWarnsMessage(UserWarning, msg):
+            # No error should occur here
+            ro.complete_order()
+            self.assertEqual(ro.status, ReturnOrderStatus.ON_HOLD.value)
+
+        # No error should be logged
+        self.assertEqual(0, Error.objects.count())
+
+        # Now, enable the "WRONG_RETURN_VALUE" setting
+        plugin.set_setting('WRONG_RETURN_VALUE', True)
+
+        with self.assertLogs('inventree', level='ERROR') as cm:
+            with self.assertWarnsMessage(UserWarning, msg):
+                # No error should occur here
+                ro.issue_order()
+                self.assertEqual(ro.status, ReturnOrderStatus.IN_PROGRESS.value)
+
+            # Ensure correct eroror was logged
+            self.assertIn('Invalid transition handler type: 1', str(cm.output[0]))
+
+        # Now, enable the "WRONG_RETURN_TYPE" setting
+        plugin.set_setting('WRONG_RETURN_TYPE', True)
+
+        with self.assertLogs('inventree', level='ERROR') as cm:
+            with self.assertWarnsMessage(UserWarning, msg):
+                # No error should occur here
+                ro.hold_order()
+                self.assertEqual(ro.status, ReturnOrderStatus.ON_HOLD.value)
+
+            # Ensure correct error was logged
+            self.assertIn(
+                'Plugin sample-broken-transition returned invalid type for transition handlers',
+                str(cm.output[0]),
+            )
+
+        # Now, enable the "BROKEN_GET_METHOD" setting
+        plugin.set_setting('BROKEN_GET_METHOD', True)
+
+        with self.assertLogs('inventree', level='ERROR') as cm:
+            with self.assertWarnsMessage(UserWarning, msg):
+                ro.issue_order()
+                self.assertEqual(ro.status, ReturnOrderStatus.IN_PROGRESS.value)
+
+            # Ensure correct error was logged
+            self.assertIn(
+                "ValueError('This is a broken transition plugin!')", str(cm.output[0])
+            )

src/backend/InvenTree/generic/states/transition.py

@@ -1,6 +1,10 @@
 """Classes and functions for plugin controlled object state transitions."""
 
-import InvenTree.helpers
+from django.db.models import Model
+
+import structlog
+
+logger = structlog.get_logger('inventree')
 
 
 class TransitionMethod:
@@ -16,39 +20,52 @@ class TransitionMethod:
         - The needed functions are implemented
         """
         # Check if a sending fnc is defined
-        if not hasattr(self, 'transition'):
+        if not hasattr(self, 'transition'):  # pragma: no cover
             raise NotImplementedError(
                 'A TransitionMethod must define a `transition` method'
             )
 
+    def transition(
+        self,
+        current_state: int,
+        target_state: int,
+        instance: Model,
+        default_action: callable,
+        **kwargs,
+    ) -> bool:
+        """Perform a state transition.
 
-class TransitionMethodStorageClass:
-    """Class that works as registry for all available transition methods in InvenTree.
+        Success:
+            - The custom transition logic succeeded
+            - Return True result
+            - No further transitions are attempted
+        Ignore:
+            - The custom transition logic did not apply
+            - Return False result
+            - Further transitions are attempted (if available)
+            - Default action is called if no transition was successful
+        Failure:
+            - The custom transition logic failed
+            - Raise a ValidationError
+            - No further transitions are attempted
+            - Default action is not called
 
-    Is initialized on startup as one instance named `storage` in this file.
-    """
+        Arguments:
+            current_state: int - Current state of the instance.
+            target_state: int - Target state to transition to.
+            instance: Model - The object instance to transition.
+            default_action: callable - Default action to be taken if no transition is successful.
+            **kwargs: Additional keyword arguments for custom logic.
 
-    list = None
+        Returns:
+            result: bool - True if the transition method was successful, False otherwise.
 
-    def collect(self):
-        """Collect all classes in the environment that are transition methods."""
-        filtered_list = {}
-        for item in InvenTree.helpers.inheritors(TransitionMethod):
-            # Try if valid
-            try:
-                item()
-            except NotImplementedError:
-                continue
-            filtered_list[f'{item.__module__}.{item.__qualname__}'] = item
-
-        self.list = list(filtered_list.values())
-
-        # Ensure the list has items
-        if not self.list:
-            self.list = []
-
-
-storage = TransitionMethodStorageClass()
+        Raises:
+            ValidationError: Alert the user that the transition failued
+        """
+        raise NotImplementedError(
+            'TransitionMethod.transition must be implemented'
+        )  # pragma: no cover
 
 
 class StateTransitionMixin:
@@ -76,13 +93,37 @@ class StateTransitionMixin:
             instance: Object instance
             default_action: Default action to be taken if none of the transitions returns a boolean true value
         """
-        # Check if there is a custom override function for this transition
-        for override in storage.list:
-            rslt = override.transition(
-                current_state, target_state, instance, default_action, **kwargs
-            )
-            if rslt:
-                return rslt
+        from InvenTree.exceptions import log_error
+        from plugin import PluginMixinEnum, registry
+
+        transition_plugins = registry.with_mixin(PluginMixinEnum.STATE_TRANSITION)
+
+        for plugin in transition_plugins:
+            try:
+                handlers = plugin.get_transition_handlers()
+            except Exception:
+                log_error('get_transition_handlers', plugin=plugin)
+                continue
+
+            if type(handlers) is not list:
+                logger.error(
+                    'Plugin %s returned invalid type for transition handlers',
+                    plugin.slug,
+                )
+                continue
+
+            for handler in handlers:
+                if not isinstance(handler, TransitionMethod):
+                    logger.error('Invalid transition handler type: %s', handler)
+                    continue
+
+                # Call the transition method
+                result = handler.transition(
+                    current_state, target_state, instance, default_action, **kwargs
+                )
+
+                if result:
+                    return result
 
         # Default action
         return default_action(current_state, target_state, instance, **kwargs)

src/backend/InvenTree/plugin/base/integration/TransitionMixin.py

@@ -0,0 +1,63 @@
+"""Plugin mixin for handling state transitions."""
+
+import structlog
+
+from generic.states.transition import TransitionMethod
+from plugin import PluginMixinEnum
+
+logger = structlog.get_logger('inventree')
+
+
+class TransitionMixin:
+    """Mixin class for handling state transitions.
+
+    - This mixin allows plugins to define custom state transition methods.
+    - These methods can be used to apply custom logic when an object transitions from one state to another.
+    """
+
+    # List of transition methods that can be used by plugins
+    TRANSITION_HANDLERS: list[TransitionMethod]
+
+    class MixinMeta:
+        """Meta class for the StateTransitionMixin."""
+
+        MIXIN_NAME = 'StateTransition'
+
+    def __init__(self):
+        """Initialize the mixin and register it."""
+        super().__init__()
+        self.add_mixin(
+            PluginMixinEnum.STATE_TRANSITION, 'has_transition_handlers', __class__
+        )
+
+    @property
+    def has_transition_handlers(self) -> bool:
+        """Check if there are any transition handlers defined."""
+        try:
+            result = bool(self.get_transition_handlers())
+            return result
+        except Exception:
+            return False
+
+    def get_transition_handlers(self) -> list[TransitionMethod]:
+        """Get the list of transition methods available in this mixin.
+
+        The default implementation returns the TRANSITION_HANDLERS list.
+        """
+        handlers = getattr(self, 'TRANSITION_HANDLERS', None) or []
+
+        if not isinstance(handlers, list):
+            raise TypeError(
+                'TRANSITION_HANDLERS must be a list of TransitionMethod instances'
+            )
+
+        handler_methods = []
+
+        for handler in handlers:
+            if not isinstance(handler, TransitionMethod):
+                logger.error('Invalid transition handler type: %s', handler)
+                continue
+
+            handler_methods.append(handler)
+
+        return handler_methods

src/backend/InvenTree/plugin/mixins/__init__.py

@@ -13,6 +13,7 @@ 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
+from plugin.base.integration.TransitionMixin import TransitionMixin
 from plugin.base.integration.UrlsMixin import UrlsMixin
 from plugin.base.integration.ValidationMixin import ValidationMixin
 from plugin.base.label.mixins import LabelPrintingMixin
@@ -38,6 +39,7 @@ __all__ = [
     'ScheduleMixin',
     'SettingsMixin',
     'SupplierBarcodeMixin',
+    'TransitionMixin',
     'UrlsMixin',
     'UserInterfaceMixin',
     'ValidationMixin',

src/backend/InvenTree/plugin/plugin.py

@@ -74,6 +74,7 @@ class PluginMixinEnum(StringEnum):
     SCHEDULE = 'schedule'
     SETTINGS = 'settings'
     SETTINGS_CONTENT = 'settingscontent'
+    STATE_TRANSITION = 'statetransition'
     SUPPLIER_BARCODE = 'supplier-barcode'
     URLS = 'urls'
     USER_INTERFACE = 'ui'

src/backend/InvenTree/plugin/samples/integration/transition.py

@@ -1,21 +1,29 @@
 """Sample implementation of state transition implementation."""
 
+import warnings
+
+from django.core.exceptions import ValidationError
+
 from common.notifications import trigger_notification
 from generic.states import TransitionMethod
 from order.models import ReturnOrder
 from order.status_codes import ReturnOrderStatus
 from plugin import InvenTreePlugin
+from plugin.mixins import SettingsMixin, TransitionMixin
 
 
-class SampleTransitionPlugin(InvenTreePlugin):
+class SampleTransitionPlugin(TransitionMixin, InvenTreePlugin):
     """A sample plugin which shows how state transitions might be implemented."""
 
     NAME = 'SampleTransitionPlugin'
+    SLUG = 'sample-transition'
 
     class ReturnChangeHandler(TransitionMethod):
-        """Transition method for PurchaseOrder objects."""
+        """Transition method for ReturnOrder objects."""
 
-        def transition(current_state, target_state, instance, default_action, **kwargs):  # noqa: N805
+        def transition(
+            self, current_state, target_state, instance, default_action, **kwargs
+        ) -> bool:
             """Example override function for state transition."""
             # Only act on ReturnOrders that should be completed
             if (
@@ -26,15 +34,71 @@ class SampleTransitionPlugin(InvenTreePlugin):
 
             # Only allow proceeding if the return order has a responsible user assigned
             if not instance.responsible:
+                msg = 'Return order without responsible owner can not be completed!'
+
                 # Trigger whoever created the return order
                 instance.created_by
                 trigger_notification(
                     instance,
                     'sampel_123_456',
                     targets=[instance.created_by],
-                    context={
-                        'message': 'Return order without responsible owner can not be completed!'
-                    },
+                    context={'message': msg},
                 )
-                return True  # True means nothing will happen
+
+                raise ValidationError(msg)
+
             return False  # Do not act
+
+    TRANSITION_HANDLERS = [ReturnChangeHandler()]
+
+
+class BrokenTransitionPlugin(SettingsMixin, TransitionMixin, InvenTreePlugin):
+    """An intentionally broken plugin to test error handling."""
+
+    NAME = 'BrokenTransitionPlugin'
+    SLUG = 'sample-broken-transition'
+
+    SETTINGS = {
+        'BROKEN_GET_METHOD': {
+            'name': 'Broken Get Method',
+            'description': 'If set, the get_transition_handlers method will raise an error.',
+            'validator': bool,
+            'default': False,
+        },
+        'WRONG_RETURN_TYPE': {
+            'name': 'Wrong Return Type',
+            'description': 'If set, the get_transition_handlers method will return an incorrect type.',
+            'validator': bool,
+            'default': False,
+        },
+        'WRONG_RETURN_VALUE': {
+            'name': 'Wrong Return Value',
+            'description': 'If set, the get_transition_handlers method will return an incorrect value.',
+            'validator': bool,
+            'default': False,
+        },
+    }
+
+    @property
+    def has_transition_handlers(self) -> bool:
+        """Ensure that this plugin always has handlers."""
+        return True
+
+    def get_transition_handlers(self) -> list[TransitionMethod]:
+        """Return transition handlers for the given instance."""
+        warnings.warn(
+            'get_transition_handlers is intentionally broken in this plugin',
+            stacklevel=2,
+        )
+
+        if self.get_setting('BROKEN_GET_METHOD', backup_value=False, cache=False):
+            raise ValueError('This is a broken transition plugin!')
+
+        if self.get_setting('WRONG_RETURN_TYPE', backup_value=False, cache=False):
+            return 'This is not a list of handlers!'
+
+        if self.get_setting('WRONG_RETURN_VALUE', backup_value=False, cache=False):
+            return [1, 2, 3]
+
+        # Return a valid handler list (empty)
+        return []