[#94] Support class level hooking (#99)

river/core/classworkflowobject.py

@@ -4,6 +4,8 @@
 from django.db.models.functions import Cast
 from django_cte import With
 
+from river.hooking.completed import PostCompletedHooking, PreCompletedHooking
+from river.hooking.transition import PostTransitionHooking, PreTransitionHooking
 from river.models import State, TransitionApprovalMeta, TransitionApproval, PENDING, Workflow
 
 
@@ -69,6 +71,18 @@ def final_states(self):
         final_approvals = TransitionApprovalMeta.objects.filter(workflow=self.workflow, children__isnull=True)
         return State.objects.filter(pk__in=final_approvals.values_list("destination_state", flat=True))
 
+    def hook_post_transition(self, callback, *args, **kwargs):
+        PostTransitionHooking.register(callback, None, self.field_name, *args, **kwargs)
+
+    def hook_pre_transition(self, callback, *args, **kwargs):
+        PreTransitionHooking.register(callback, None, self.field_name, *args, **kwargs)
+
+    def hook_post_complete(self, callback):
+        PostCompletedHooking.register(callback, None, self.field_name)
+
+    def hook_pre_complete(self, callback):
+        PreCompletedHooking.register(callback, None, self.field_name)
+
     def _authorized_approvals(self, as_user):
         group_q = Q()
         for g in as_user.groups.all():

river/hooking/hooking.py

@@ -20,7 +20,9 @@ def dispatch(cls, workflow_object, field_name, *args, **kwargs):
         kwargs.pop('signal', None)
         kwargs.pop('sender', None)
 
-        for callback in callback_backend.get_callbacks(cls, workflow_object, field_name, *args, **kwargs):
+        object_callbacks = callback_backend.get_callbacks(cls, workflow_object, field_name, *args, **kwargs)
+        class_callbacks = callback_backend.get_callbacks(cls, None, field_name, *args, **kwargs)
+        for callback in object_callbacks + class_callbacks:
             exclusions = cls.get_result_exclusions()
             callback(workflow_object, field_name, *args, **{k: v for k, v in kwargs.items() if k not in exclusions})
             LOGGER.debug(

river/tests/hooking/test__completed_hooking.py

@@ -14,7 +14,7 @@ class CompletedHookingTest(TestCase):
     def setUp(self):
         super(CompletedHookingTest, self).setUp()
 
-    def test_shouldInvokeTheRegisteredCallBackWhenFlowIsCompleteForTheObject(self):
+    def test_shouldInvokeTheRegisteredViaInstanceApiCallBackWhenFlowIsCompleteForTheObject(self):
         authorized_permission = PermissionObjectFactory()
         authorized_user = UserObjectFactory(user_permissions=[authorized_permission])
 
@@ -63,3 +63,53 @@ def test_callback(*args, **kwargs):
         assert_that(workflow_object.model.my_field, equal_to(state3))
 
         assert_that(self.test_args, equal_to((workflow_object.model, "my_field")))
+
+    def test_shouldInvokeTheRegisteredViaClassApiCallBackWhenFlowIsCompleteForTheObject(self):
+        authorized_permission = PermissionObjectFactory()
+        authorized_user = UserObjectFactory(user_permissions=[authorized_permission])
+
+        state1 = StateObjectFactory(label="state1")
+        state2 = StateObjectFactory(label="state2")
+        state3 = StateObjectFactory(label="state3")
+
+        content_type = ContentType.objects.get_for_model(BasicTestModel)
+        workflow = WorkflowFactory(initial_state=state1, content_type=content_type, field_name="my_field")
+        TransitionApprovalMetaFactory.create(
+            workflow=workflow,
+            source_state=state1,
+            destination_state=state2,
+            priority=0,
+            permissions=[authorized_permission]
+        )
+
+        TransitionApprovalMetaFactory.create(
+            workflow=workflow,
+            source_state=state2,
+            destination_state=state3,
+            priority=0,
+            permissions=[authorized_permission]
+        )
+
+        workflow_object = BasicTestModelObjectFactory()
+
+        self.test_args = None
+        self.test_kwargs = None
+
+        def test_callback(*args, **kwargs):
+            self.test_args = args
+            self.test_kwargs = kwargs
+
+        BasicTestModel.river.my_field.hook_post_complete(test_callback)
+
+        assert_that(self.test_args, none())
+
+        assert_that(workflow_object.model.my_field, equal_to(state1))
+        workflow_object.model.river.my_field.approve(as_user=authorized_user)
+        assert_that(workflow_object.model.my_field, equal_to(state2))
+
+        assert_that(self.test_args, none())
+
+        workflow_object.model.river.my_field.approve(as_user=authorized_user)
+        assert_that(workflow_object.model.my_field, equal_to(state3))
+
+        assert_that(self.test_args, equal_to((workflow_object.model, "my_field")))

river/tests/hooking/test__transition_hooking.py

@@ -13,7 +13,7 @@
 # noinspection DuplicatedCode
 class TransitionHooking(TestCase):
 
-    def test_shouldInvokeTheRegisteredCallBackWhenATransitionHappens(self):
+    def test_shouldInvokeTheRegisteredViaInstanceApiCallBackWhenATransitionHappens(self):
         self.test_args = None
         self.test_kwargs = None
 
@@ -64,7 +64,58 @@ def test_callback(*args, **kwargs):
         last_approval = TransitionApproval.objects.get(object_id=workflow_object.model.pk, source_state=state1, destination_state=state2, priority=1)
         assert_that(self.test_kwargs, has_entry(equal_to("transition_approval"), equal_to(last_approval)))
 
-    def test_shouldInvokeTheRegisteredCallBackWhenASpecificTransitionHappens(self):
+    def test_shouldInvokeTheRegisteredViaClassApiCallBackWhenATransitionHappens(self):
+        self.test_args = None
+        self.test_kwargs = None
+
+        def test_callback(*args, **kwargs):
+            self.test_args = args
+            self.test_kwargs = kwargs
+
+        authorized_permission = PermissionObjectFactory()
+        authorized_user = UserObjectFactory(user_permissions=[authorized_permission])
+
+        state1 = StateObjectFactory(label="state1")
+        state2 = StateObjectFactory(label="state2")
+
+        content_type = ContentType.objects.get_for_model(BasicTestModel)
+        workflow = WorkflowFactory(initial_state=state1, content_type=content_type, field_name="my_field")
+        TransitionApprovalMetaFactory.create(
+            workflow=workflow,
+            source_state=state1,
+            destination_state=state2,
+            priority=0,
+            permissions=[authorized_permission]
+        )
+
+        TransitionApprovalMetaFactory.create(
+            workflow=workflow,
+            source_state=state1,
+            destination_state=state2,
+            priority=1,
+            permissions=[authorized_permission]
+        )
+
+        workflow_object = BasicTestModelObjectFactory()
+
+        BasicTestModel.river.my_field.hook_post_transition(test_callback)
+
+        assert_that(self.test_args, none())
+
+        assert_that(workflow_object.model.my_field, equal_to(state1))
+        workflow_object.model.river.my_field.approve(as_user=authorized_user)
+        assert_that(workflow_object.model.my_field, equal_to(state1))
+
+        assert_that(self.test_args, none())
+
+        workflow_object.model.river.my_field.approve(as_user=authorized_user)
+        assert_that(workflow_object.model.my_field, equal_to(state2))
+        assert_that(self.test_args, equal_to((workflow_object.model, "my_field")))
+
+        last_approval = TransitionApproval.objects.get(object_id=workflow_object.model.pk, source_state=state1, destination_state=state2, priority=1)
+        assert_that(self.test_kwargs, has_entry(equal_to("transition_approval"), equal_to(last_approval)))
+
+    def test_shouldInvokeTheRegisteredViaInstanceApiCallBackWhenASpecificTransitionHappens(self):
         self.test_args = None
         self.test_kwargs = None
 
@@ -115,3 +166,55 @@ def test_callback(*args, **kwargs):
 
         last_approval = TransitionApproval.objects.get(object_id=workflow_object.model.pk, source_state=state2, destination_state=state3)
         assert_that(self.test_kwargs, has_entry(equal_to("transition_approval"), equal_to(last_approval)))
+
+    def test_shouldInvokeTheRegisteredViaClassApiCallBackWhenASpecificTransitionHappens(self):
+        self.test_args = None
+        self.test_kwargs = None
+
+        def test_callback(*args, **kwargs):
+            self.test_args = args
+            self.test_kwargs = kwargs
+
+        authorized_permission = PermissionObjectFactory()
+        authorized_user = UserObjectFactory(user_permissions=[authorized_permission])
+
+        state1 = StateObjectFactory(label="state1")
+        state2 = StateObjectFactory(label="state2")
+        state3 = StateObjectFactory(label="state3")
+
+        content_type = ContentType.objects.get_for_model(BasicTestModel)
+        workflow = WorkflowFactory(initial_state=state1, content_type=content_type, field_name="my_field")
+        TransitionApprovalMetaFactory.create(
+            workflow=workflow,
+            source_state=state1,
+            destination_state=state2,
+            priority=0,
+            permissions=[authorized_permission]
+        )
+
+        TransitionApprovalMetaFactory.create(
+            workflow=workflow,
+            source_state=state2,
+            destination_state=state3,
+            priority=0,
+            permissions=[authorized_permission]
+        )
+
+        workflow_object = BasicTestModelObjectFactory()
+
+        BasicTestModel.river.my_field.hook_post_transition(test_callback, source_state=state2, destination_state=state3)
+
+        assert_that(self.test_args, none())
+
+        assert_that(workflow_object.model.my_field, equal_to(state1))
+        workflow_object.model.river.my_field.approve(as_user=authorized_user)
+        assert_that(workflow_object.model.my_field, equal_to(state2))
+
+        assert_that(self.test_args, none())
+
+        workflow_object.model.river.my_field.approve(as_user=authorized_user)
+        assert_that(workflow_object.model.my_field, equal_to(state3))
+        assert_that(self.test_args, equal_to((workflow_object.model, "my_field")))
+
+        last_approval = TransitionApproval.objects.get(object_id=workflow_object.model.pk, source_state=state2, destination_state=state3)
+        assert_that(self.test_kwargs, has_entry(equal_to("transition_approval"), equal_to(last_approval)))