Testing

Viewflow workflows integrate with Django’s test framework.

Basic Flow Testing

Test the full flow to verify node connections work:

from django.test import TestCase
from django.contrib.auth.models import User, Permission
from myapp.flows import MyFlow
from viewflow.workflow import PROCESS

class MyFlowTests(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(
            username='testuser', password='password'
        )

        # Grant permissions
        for permission in MyFlow.get_required_permissions():
            self.user.user_permissions.add(
                Permission.objects.get(codename=permission.codename)
            )

    def test_flow_execution(self):
        self.client.login(username='testuser', password='password')

        # Start the flow
        response = self.client.post(
            MyFlow.start.reverse('execute'),
            {'text': 'Test message'}
        )
        self.assertEqual(response.status_code, 302)

        # Get the process
        process = MyFlow.process_class.objects.get()
        self.assertEqual(process.text, 'Test message')

        # Find the approval task
        approval_task = process.task_set.get(flow_task=MyFlow.approve)

        # Assign and complete it
        self.client.post(approval_task.reverse('assign'), {})
        self.client.post(
            approval_task.reverse('execute'),
            {'approved': True}
        )

        # Verify completion
        process.refresh_from_db()
        self.assertEqual(process.status, PROCESS.DONE)

Testing Individual Nodes

Test complex nodes in isolation:

from viewflow.workflow import activation

class NodeTests(TestCase):
    def test_approve_node(self):
        process = MyFlow.process_class.objects.create(text='Test')
        task = process.task_set.create(flow_task=MyFlow.approve)

        act = activation.Context(task)
        act.prepare()

        act.process.approved = True
        act.done()

        self.assertTrue(process.approved)
        self.assertTrue(
            process.task_set.filter(flow_task=MyFlow.check_approve).exists()
        )

Testing Branches

Verify each branch path:

def test_approval_branching(self):
    # Test approved branch
    process = MyFlow.process_class.objects.create(text='Test', approved=True)
    task = process.task_set.create(flow_task=MyFlow.check_approve)

    act = activation.Context(task)
    act.prepare()
    act.done()

    self.assertTrue(
        process.task_set.filter(flow_task=MyFlow.send).exists()
    )

    # Test rejected branch
    process = MyFlow.process_class.objects.create(text='Test', approved=False)
    task = process.task_set.create(flow_task=MyFlow.check_approve)

    act = activation.Context(task)
    act.prepare()
    act.done()

    self.assertTrue(
        process.task_set.filter(flow_task=MyFlow.end).exists()
    )

Mocking External Services

from unittest.mock import patch

class ExternalServiceTests(TestCase):
    @patch('myapp.services.external_api.send_notification')
    def test_notification_sending(self, mock_send):
        mock_send.return_value = {'status': 'sent', 'id': '123'}

        process = MyFlow.process_class.objects.create(text='Test')
        task = process.task_set.create(flow_task=MyFlow.send_notification)

        act = activation.Context(task)
        act.prepare()
        act.done()

        mock_send.assert_called_once_with(text='Test')

Testing Parallel Execution

def test_parallel_execution(self):
    process = ParallelFlow.process_class.objects.create()
    task = process.task_set.create(flow_task=ParallelFlow.split)

    act = activation.Context(task)
    act.prepare()
    act.done()

    # Both branches created
    self.assertTrue(
        process.task_set.filter(flow_task=ParallelFlow.task1).exists()
    )
    self.assertTrue(
        process.task_set.filter(flow_task=ParallelFlow.task2).exists()
    )

    # Complete first branch
    task1 = process.task_set.get(flow_task=ParallelFlow.task1)
    act = activation.Context(task1)
    act.prepare()
    act.done()

    # Join not ready yet
    join_task = process.task_set.get(flow_task=ParallelFlow.join)
    self.assertFalse(join_task.ready)

    # Complete second branch
    task2 = process.task_set.get(flow_task=ParallelFlow.task2)
    act = activation.Context(task2)
    act.prepare()
    act.done()

    # Join now ready
    join_task.refresh_from_db()
    self.assertTrue(join_task.ready)

Testing Permissions

def test_permission_enforcement(self):
    regular_user = User.objects.create_user(
        username='regular', password='password'
    )
    self.client.login(username='regular', password='password')

    # Without permission - denied
    response = self.client.post(
        MyFlow.start.reverse('execute'),
        {'text': 'Test message'}
    )
    self.assertEqual(response.status_code, 403)

    # Add permission
    permission = Permission.objects.get(codename='can_start_myflow')
    regular_user.user_permissions.add(permission)

    # Now allowed
    response = self.client.post(
        MyFlow.start.reverse('execute'),
        {'text': 'Test message'}
    )
    self.assertEqual(response.status_code, 302)

Test Fixtures

Create reusable fixtures:

class WorkflowFixturesMixin:
    def create_process_at_approval(self, text='Default text', **kwargs):
        """Create a process at the approval stage."""
        process = MyFlow.process_class.objects.create(text=text, **kwargs)

        process.task_set.create(
            flow_task=MyFlow.start,
            status=TASK.DONE
        )

        approval_task = process.task_set.create(
            flow_task=MyFlow.approve,
            status=TASK.NEW
        )

        return process, approval_task

class ApprovalTests(TestCase, WorkflowFixturesMixin):
    def test_approval_form_validation(self):
        process, task = self.create_process_at_approval()
        # Test the approval form...