Add integration test for failing tasks

tests/test_integration.py

@@ -6,6 +6,7 @@
 import pytest
 from django.urls import reverse
 
+from joeflow.models import Task
 from tests.testapp import models
 
 
@@ -97,3 +98,12 @@ def test_change(self, db, admin_client):
         url = reverse('admin:joeflow_task_change', args=[task.pk])
         response = admin_client.get(url)
         assert response.status_code == 200
+
+
+class TestFailingProcess:
+    def test_fail(self, db):
+        process = models.FailingProcess.start()
+        failed_task = process.task_set.latest()
+        assert failed_task.status == Task.FAILED
+        assert failed_task.exception == 'ValueError: Boom!'
+        assert 'Traceback (most recent call last):' in failed_task.stacktrace

tests/testapp/migrations/0001_initial.py

@@ -64,4 +64,13 @@ class Migration(migrations.Migration):
             },
             bases=('joeflow.process', models.Model),
         ),
+        migrations.CreateModel(
+            name='FailingProcess',
+            fields=[
+                ('process_ptr',
+                 models.OneToOneField(auto_created=True, on_delete=django.db.models.deletion.CASCADE, parent_link=True,
+                                      primary_key=True, serialize=False, to='joeflow.Process')),
+            ],
+            bases=('joeflow.process',),
+        ),
     ]

tests/testapp/models.py

@@ -144,3 +144,14 @@ def end(self):
         (start, wait),
         (wait, end),
     )
+
+
+class FailingProcess(Process):
+    start = tasks.Start()
+
+    def fail(self):
+        raise ValueError("Boom!")
+
+    edges = (
+        (start, fail),
+    )