diff --git a/docs/docs/plugins/mixins/transition.md b/docs/docs/plugins/mixins/transition.md new file mode 100644 index 0000000000..496bb630d7 --- /dev/null +++ b/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") }}). diff --git a/docs/mkdocs.yml b/docs/mkdocs.yml index fa62adcedf..60f7090edc 100644 --- a/docs/mkdocs.yml +++ b/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 diff --git a/src/backend/InvenTree/InvenTree/apps.py b/src/backend/InvenTree/InvenTree/apps.py index 857aff1d13..751f8eb54b 100644 --- a/src/backend/InvenTree/InvenTree/apps.py +++ b/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 diff --git a/src/backend/InvenTree/build/test_api.py b/src/backend/InvenTree/build/test_api.py index 1ffc8cd031..6070d7e387 100644 --- a/src/backend/InvenTree/build/test_api.py +++ b/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 diff --git a/src/backend/InvenTree/generic/states/__init__.py b/src/backend/InvenTree/generic/states/__init__.py index 90922e9695..35e997fbeb 100644 --- a/src/backend/InvenTree/generic/states/__init__.py +++ b/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', ] diff --git a/src/backend/InvenTree/generic/states/test_transition.py b/src/backend/InvenTree/generic/states/test_transition.py index 9e3e53958b..b8cb3c371d 100644 --- a/src/backend/InvenTree/generic/states/test_transition.py +++ b/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]) + ) diff --git a/src/backend/InvenTree/generic/states/transition.py b/src/backend/InvenTree/generic/states/transition.py index 6abc7f1010..2a2b450d39 100644 --- a/src/backend/InvenTree/generic/states/transition.py +++ b/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) diff --git a/src/backend/InvenTree/plugin/base/integration/TransitionMixin.py b/src/backend/InvenTree/plugin/base/integration/TransitionMixin.py new file mode 100644 index 0000000000..19bc9da176 --- /dev/null +++ b/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 diff --git a/src/backend/InvenTree/plugin/mixins/__init__.py b/src/backend/InvenTree/plugin/mixins/__init__.py index 4ac0c05892..1e3f5e124c 100644 --- a/src/backend/InvenTree/plugin/mixins/__init__.py +++ b/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', diff --git a/src/backend/InvenTree/plugin/plugin.py b/src/backend/InvenTree/plugin/plugin.py index 8245c864d0..1de7946af0 100644 --- a/src/backend/InvenTree/plugin/plugin.py +++ b/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' diff --git a/src/backend/InvenTree/plugin/samples/integration/transition.py b/src/backend/InvenTree/plugin/samples/integration/transition.py index 2057933dc7..6ebd3c4930 100644 --- a/src/backend/InvenTree/plugin/samples/integration/transition.py +++ b/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 []