State transition support for generic states

InvenTree/InvenTree/apps.py

@@ -33,6 +33,7 @@ def ready(self):
         - Starting regular tasks
         - Updating exchange rates
         - Collecting notification methods
+        - Collecting state transition methods
         - Adding users set in the current environment
         """
         # skip loading if plugin registry is not loaded or we run in a background thread
@@ -52,6 +53,7 @@ def ready(self):
                 InvenTree.tasks.offload_task(InvenTree.tasks.check_for_migrations)
 
         self.collect_notification_methods()
+        self.collect_state_transition_methods()
 
         # Ensure the unit registry is loaded
         InvenTree.conversion.get_unit_registry()
@@ -251,3 +253,9 @@ def collect_notification_methods(self):
         from common.notifications import storage
 
         storage.collect()
+
+    def collect_state_transition_methods(self):
+        """Collect all state transition methods."""
+        from generic.states import storage
+
+        storage.collect()

InvenTree/generic/states/__init__.py

@@ -7,7 +7,11 @@
 """
 
 from .states import StatusCode
+from .transition import StateTransitionMixin, TransitionMethod, storage
 
 __all__ = [
     StatusCode,
+    storage,
+    TransitionMethod,
+    StateTransitionMixin,
 ]

InvenTree/generic/states/test_transition.py

@@ -0,0 +1,83 @@
+"""Tests for state transition mechanism."""
+
+from InvenTree.unit_test import InvenTreeTestCase
+
+from .transition import StateTransitionMixin, TransitionMethod, storage
+
+
+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()
+
+
+class TransitionTests(InvenTreeTestCase):
+    """Tests for basic NotificationMethod."""
+
+    def test_class(self):
+        """Ensure that the class itself works."""
+
+        class ErrorImplementation(TransitionMethod):
+            ...
+
+        with self.assertRaises(NotImplementedError):
+            ErrorImplementation()
+
+        _clean_storage([ErrorImplementation])
+
+    def test_storage(self):
+        """Ensure that the storage collection mechanism works."""
+
+        class RaisingImplementation(TransitionMethod):
+            def transition(self, *args, **kwargs):
+                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."""
+
+        # 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):
+                return 1234
+
+        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)
+
+        _clean_storage([ValidImplementationNoEffect, ValidImplementation])
+
+    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')

InvenTree/generic/states/transition.py

@@ -0,0 +1,81 @@
+"""Classes and functions for plugin controlled object state transitions."""
+import InvenTree.helpers
+
+
+class TransitionMethod:
+    """Base class for all transition classes.
+
+    Must implement a method called `transition` that takes both args and kwargs.
+    """
+
+    def __init__(self) -> None:
+        """Check that the method is defined correctly.
+
+        This checks that:
+        - The needed functions are implemented
+        """
+        # Check if a sending fnc is defined
+        if (not hasattr(self, 'transition')):
+            raise NotImplementedError('A TransitionMethod must define a `transition` method')
+
+
+class TransitionMethodStorageClass:
+    """Class that works as registry for all available transition methods in InvenTree.
+
+    Is initialized on startup as one instance named `storage` in this file.
+    """
+
+    list = None
+
+    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()
+
+
+class StateTransitionMixin:
+    """Mixin class to enable state transitions.
+
+    This mixin is used to add state transitions handling to a model. With this you can apply custom logic to state transitions via plugins.
+    ```python
+    class MyModel(StateTransitionMixin, models.Model):
+        def some_dummy_function(self, *args, **kwargs):
+            pass
+
+        def action(self, *args, **kwargs):
+            self.handle_transition(0, 1, self, self.some_dummy_function)
+    ```
+    """
+
+    def handle_transition(self, current_state, target_state, instance, default_action, **kwargs):
+        """Handle a state transition for an object.
+
+        Args:
+            current_state: Current state of instance
+            target_state: Target state of instance
+            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
+
+        # Default action
+        return default_action(current_state, target_state, instance, **kwargs)