Add override feature to proccess admin

Author: codingjoe

joeflow/admin.py

@@ -1,9 +1,10 @@
 from django.contrib import admin, messages
 from django.contrib.auth import get_permission_codename
+from django.db import transaction
 from django.utils.html import format_html
 from django.utils.translation import ugettext_lazy as t
 
-from . import models
+from . import forms, models
 from .contrib.reversion import VersionAdmin
 
 __all__ = ("ProcessAdmin",)
@@ -118,13 +119,43 @@ def child_tasks(self, obj):
     )
 
 
-class ProcessAdmin(VersionAdmin):
-    readonly_fields = (
-        "modified",
+class TaskInlineAdmin(admin.TabularInline):
+    model = models.Task
+    readonly_fields = [
+        "name",
+        "type",
+        "assignees",
         "created",
-    )
+        "completed",
+        "completed_by_user",
+        "status",
+    ]
+    fields = readonly_fields
+    extra = 0
+    can_delete = False
+    show_change_link = True
+    classes = ["collapse"]
 
+
+class ProcessAdmin(VersionAdmin):
     list_filter = (
         "modified",
         "created",
     )
+    form = forms.OverrideForm
+
+    def get_inlines(self, *args, **kwargs):
+        return [*super().get_inlines(*args, **kwargs), TaskInlineAdmin]
+
+    def get_readonly_fields(self, *args, **kwargs):
+        return [
+            "get_instance_graph_svg",
+            *super().get_readonly_fields(*args, **kwargs),
+            "modified",
+            "created",
+        ]
+
+    @transaction.atomic()
+    def save_model(self, request, obj, form, change):
+        super().save_model(request, obj, form, change)
+        form.start_next_tasks(request.user)

joeflow/forms.py

@@ -0,0 +1,39 @@
+from django import forms
+from django.utils.translation import gettext_lazy as t
+
+from . import models
+
+
+class OverrideForm(forms.ModelForm):
+    next_tasks = forms.MultipleChoiceField(
+        label=t("Next tasks"), choices=[], required=False,
+    )
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.fields["next_tasks"].choices = [
+            (name, name) for name in dict(self._meta.model.get_nodes()).keys()
+        ]
+
+    def get_next_task_nodes(self):
+        names = self.cleaned_data["next_tasks"]
+        for name in names:
+            yield self._meta.model.get_node(name)
+
+    def start_next_tasks(self, user=None):
+        active_tasks = list(self.instance.task_set.filter(completed=None))
+        for task in active_tasks:
+            task.cancel(user)
+        if active_tasks:
+            parent_tasks = active_tasks
+        else:
+            try:
+                parent_tasks = [self.instance.task_set.latest()]
+            except models.Task.DoesNotExist:
+                parent_tasks = []
+        override_task = self.instance.task_set.create(
+            name="override", type=models.Task.HUMAN,
+        )
+        override_task.parent_task_set.set(parent_tasks)
+        override_task.finish(user=user)
+        override_task.start_next_tasks(next_nodes=self.get_next_task_nodes())

joeflow/utils.py

@@ -5,6 +5,7 @@
 def get_processes() -> types.GeneratorType:
     """Return all registered processes."""
     from django.apps import apps
+
     from .models import Process
 
     apps.check_models_ready()

joeflow/views.py

@@ -1,11 +1,10 @@
-from django import forms
 from django.contrib.auth.mixins import PermissionRequiredMixin
 from django.db import transaction
+from django.forms import modelform_factory
 from django.shortcuts import get_object_or_404
-from django.utils.translation import gettext_lazy as t
 from django.views import generic
 
-from . import models
+from . import forms, models
 from .contrib.reversion import RevisionMixin
 
 
@@ -38,7 +37,7 @@ def get_task(self):
                 models.Task, pk=self.kwargs["pk"], name=self.name, completed=None,
             )
         except KeyError:
-            return models.Task(name=self.name,)
+            return models.Task(name=self.name, type=models.Task.HUMAN)
 
     def get_object(self, queryset=None):
         task = self.get_task()
@@ -47,11 +46,15 @@ def get_object(self, queryset=None):
     @transaction.atomic
     def post(self, request, *args, **kwargs):
         response = super().post(request, *args, **kwargs)
+        self.create_task(request)
+        return response
+
+    def create_task(self, request):
         task = self.get_task()
-        task.process = self.object
+        task.process = self.model._base_manager.get(pk=self.object.pk)
         task.finish(request.user)
         task.start_next_tasks()
-        return response
+        return task
 
 
 class ProcessDetailView(ProcessTemplateNameViewMixin, generic.DetailView):
@@ -66,37 +69,14 @@ class OverrideView(
 ):
     permission_required = "override"
     name = "override"
+    form_class = forms.OverrideForm
     fields = "__all__"
 
-    @staticmethod
-    def get_task_choices(process):
-        for name in dict(process.get_nodes()).keys():
-            yield name, name
-
     def get_form_class(self):
-        form_class = super().get_form_class()
-
-        class OverrideForm(form_class):
-            next_tasks = forms.MultipleChoiceField(
-                label=t("Next tasks"), choices=self.get_task_choices(self.object),
-            )
-
-        return OverrideForm
-
-    def get_next_task_nodes(self, form):
-        names = form.cleaned_data["next_tasks"]
-        for name in names:
-            yield self.object.get_node(name)
+        return modelform_factory(self.model, form=self.form_class, fields=self.fields)
 
     @transaction.atomic()
     def form_valid(self, form):
-        next_nodes = self.get_next_task_nodes(form)
         response = super().form_valid(form)
-        active_tasks = list(self.object.task_set.filter(completed=None))
-        for task in active_tasks:
-            task.finish()
-        override_task = self.object.task_set.create(name="override",)
-        override_task.parent_task_set.set(active_tasks)
-        override_task.finish()
-        override_task.start_next_tasks(next_nodes=next_nodes)
+        form.start_next_tasks(self.request.user)
         return response

tests/test_forms.py

@@ -0,0 +1,63 @@
+from joeflow import forms
+from tests.testapp.models import SimpleProcess
+
+
+class TestOverrideForm:
+    def test_get_next_task_nodes(self):
+        class SimpleProcessForm(forms.OverrideForm):
+            class Meta:
+                model = SimpleProcess
+                fields = "__all__"
+
+        form = SimpleProcessForm({"next_tasks": ["end"]})
+        assert form.is_valid()
+        assert list(form.get_next_task_nodes()) == [SimpleProcess.end]
+
+    def test_start_next_tasks(self, db, admin_user):
+        process = SimpleProcess.start_method()
+        assert process.task_set.scheduled().exists()
+
+        class SimpleProcessForm(forms.OverrideForm):
+            class Meta:
+                model = SimpleProcess
+                fields = "__all__"
+
+        form = SimpleProcessForm({"next_tasks": ["end"]}, instance=process)
+        assert form.is_valid()
+        form.start_next_tasks()
+
+        assert process.task_set.scheduled()[0].name == "end"
+
+    def test_start_next_tasks__user(self, db, admin_user):
+        process = SimpleProcess.start_method()
+        assert process.task_set.scheduled().exists()
+
+        class SimpleProcessForm(forms.OverrideForm):
+            class Meta:
+                model = SimpleProcess
+                fields = "__all__"
+
+        form = SimpleProcessForm({"next_tasks": ["end"]}, instance=process)
+        assert form.is_valid()
+        form.start_next_tasks(admin_user)
+
+        assert process.task_set.canceled()[0].completed_by_user == admin_user
+        assert (
+            process.task_set.filter(name="override")[0].completed_by_user == admin_user
+        )
+        assert process.task_set.filter(name="override")[0].status == "succeeded"
+
+    def test_start_next_tasks__no_next_task(self, db, admin_user):
+        process = SimpleProcess.start_method()
+        assert process.task_set.scheduled().exists()
+
+        class SimpleProcessForm(forms.OverrideForm):
+            class Meta:
+                model = SimpleProcess
+                fields = "__all__"
+
+        form = SimpleProcessForm({"next_tasks": []}, instance=process)
+        assert form.is_valid()
+        form.start_next_tasks()
+
+        assert not process.task_set.scheduled().exists()