matthias/mayan-edms
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Add workflow instance API endpoints and tests.

Browse Source
This commit is contained in:
Roberto Rosario
2017-02-10 03:13:26 -04:00
parent 0ff0841826
commit 75b77d6059
7 changed files with 405 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

127
mayan/apps/document_states/api_views.py
View File

@@ -6,6 +6,7 @@ from django.shortcuts import get_object_or_404
from rest_framework import generics
from acls.models import AccessControlList
from documents.models import Document
from documents.permissions import permission_document_type_view
from permissions import Permission
from rest_api.filters import MayanObjectPermissionsFilter
@@ -14,12 +15,15 @@ from rest_api.permissions import MayanPermission
from .models import Workflow
from .permissions import (
permission_workflow_create, permission_workflow_delete,
permission_workflow_edit, permission_workflow_view
permission_workflow_edit, permission_workflow_transition,
permission_workflow_view
)
from .serializers import (
NewWorkflowDocumentTypeSerializer, WorkflowDocumentTypeSerializer,
WorkflowInstanceSerializer, WorkflowInstanceLogEntrySerializer,
WorkflowSerializer, WorkflowStateSerializer, WorkflowTransitionSerializer,
WritableWorkflowSerializer, WritableWorkflowTransitionSerializer
WritableWorkflowInstanceLogEntrySerializer, WritableWorkflowSerializer,
WritableWorkflowTransitionSerializer
)
@@ -488,3 +492,122 @@ class APIWorkflowTransitionView(generics.RetrieveUpdateDestroyAPIView):
"""
return super(APIWorkflowTransitionView, self).put(*args, **kwargs)
# Document workflow views
class APIWorkflowInstanceListView(generics.ListAPIView):
serializer_class = WorkflowInstanceSerializer
def get(self, *args, **kwargs):
"""
Returns a list of all the document workflows.
"""
return super(APIWorkflowInstanceListView, self).get(*args, **kwargs)
def get_document(self):
document = get_object_or_404(Document, pk=self.kwargs['pk'])
try:
Permission.check_permissions(
self.request.user, (permission_workflow_view,)
)
except PermissionDenied:
AccessControlList.objects.check_access(
permission_workflow_view, self.request.user, document
)
return document
def get_queryset(self):
return self.get_document().workflows.all()
class APIWorkflowInstanceView(generics.RetrieveAPIView):
lookup_url_kwarg = 'workflow_pk'
serializer_class = WorkflowInstanceSerializer
def get(self, *args, **kwargs):
"""
Return the details of the selected document workflow.
"""
return super(APIWorkflowInstanceView, self).get(*args, **kwargs)
def get_document(self):
document = get_object_or_404(Document, pk=self.kwargs['pk'])
try:
Permission.check_permissions(
self.request.user, (permission_workflow_view,)
)
except PermissionDenied:
AccessControlList.objects.check_access(
permission_workflow_view, self.request.user, document
)
return document
def get_queryset(self):
return self.get_document().workflows.all()
class APIWorkflowInstanceLogEntryListView(generics.ListCreateAPIView):
def get(self, *args, **kwargs):
"""
Returns a list of all the document workflows log entries.
"""
return super(APIWorkflowInstanceLogEntryListView, self).get(
*args, **kwargs
)
def get_document(self):
if self.request.method == 'GET':
permission_required = permission_workflow_view
else:
permission_required = permission_workflow_transition
document = get_object_or_404(Document, pk=self.kwargs['pk'])
try:
Permission.check_permissions(
self.request.user, (permission_required,)
)
except PermissionDenied:
AccessControlList.objects.check_access(
permission_required, self.request.user, document
)
return document
def get_serializer_class(self):
if self.request.method == 'GET':
return WorkflowInstanceLogEntrySerializer
else:
return WritableWorkflowInstanceLogEntrySerializer
def get_serializer_context(self):
return {
'format': self.format_kwarg,
'request': self.request,
'workflow_instance': self.get_workflow_instance(),
'view': self
}
def get_queryset(self):
return self.get_workflow_instance().log_entries.all()
def get_workflow_instance(self):
workflow = get_object_or_404(
self.get_document().workflows, pk=self.kwargs['workflow_pk']
)
return workflow
def post(self, *args, **kwargs):
"""
Transition a document workflow by creating a new document workflow
log entry.
"""
return super(APIWorkflowInstanceLogEntryListView, self).post(*args, **kwargs)

18
mayan/apps/document_states/models.py
View File

@@ -3,6 +3,7 @@ from __future__ import unicode_literals
import logging
from django.conf import settings
from django.core.exceptions import ValidationError
from django.core.urlresolvers import reverse
from django.db import IntegrityError, models
from django.utils.encoding import python_2_unicode_compatible
@@ -166,7 +167,18 @@ class WorkflowInstance(models.Model):
return None
def get_transition_choices(self):
return self.get_current_state().origin_transitions.all()
current_state = self.get_current_state()
if current_state:
return current_state.origin_transitions.all()
else:
"""
This happens when a workflow has no initial state and a document
whose document type has this workflow is created. We return an
empty transition queryset.
"""
return WorkflowTransition.objects.none()
class Meta:
unique_together = ('document', 'workflow')
@@ -195,3 +207,7 @@ class WorkflowInstanceLogEntry(models.Model):
class Meta:
verbose_name = _('Workflow instance log entry')
verbose_name_plural = _('Workflow instance log entries')
def clean(self):
if self.transition not in self.workflow_instance.get_transition_choices():
raise ValidationError(_('Not a valid transition choice.'))

3
mayan/apps/document_states/permissions.py
View File

@@ -22,6 +22,5 @@ permission_workflow_view = namespace.add_permission(
# 'transition workflows' from one state to another, to move the workflow
# forwards
permission_workflow_transition = namespace.add_permission(
name='workflow_transition',
label=_('Transition workflows')
name='workflow_transition', label=_('Transition workflows')
)

115
mayan/apps/document_states/serializers.py
View File

@@ -3,12 +3,17 @@ from __future__ import unicode_literals
from django.utils.translation import ugettext_lazy as _
from rest_framework import serializers
from rest_framework.exceptions import ValidationError
from rest_framework.reverse import reverse
from documents.models import DocumentType
from documents.serializers import DocumentTypeSerializer
from user_management.serializers import UserSerializer
from .models import Workflow, WorkflowState, WorkflowTransition
from .models import (
Workflow, WorkflowInstance, WorkflowInstanceLogEntry, WorkflowState,
WorkflowTransition
)
class NewWorkflowDocumentTypeSerializer(serializers.Serializer):
@@ -182,6 +187,71 @@ class WorkflowSerializer(serializers.HyperlinkedModelSerializer):
model = Workflow
class WorkflowInstanceLogEntrySerializer(serializers.ModelSerializer):
document_workflow_url = serializers.SerializerMethodField()
transition = WorkflowTransitionSerializer(read_only=True)
user = UserSerializer(read_only=True)
class Meta:
fields = (
'comment', 'datetime', 'document_workflow_url', 'transition',
'user'
)
model = WorkflowInstanceLogEntry
def get_document_workflow_url(self, instance):
return reverse(
'rest_api:workflowinstance-detail', args=(
instance.workflow_instance.document.pk,
instance.workflow_instance.pk,
), request=self.context['request'], format=self.context['format']
)
class WorkflowInstanceSerializer(serializers.ModelSerializer):
current_state = WorkflowStateSerializer(
read_only=True, source='get_current_state'
)
document_workflow_url = serializers.SerializerMethodField(
help_text=_(
'API URL pointing to a workflow in relation to the '
'document to which it is attached. This URL is different than '
'the canonical workflow URL.'
)
)
last_log_entry = WorkflowInstanceLogEntrySerializer(
read_only=True, source='get_last_log_entry'
)
log_entries_url = serializers.SerializerMethodField(
help_text=_('A link to the entire history of this workflow.')
)
transition_choices = WorkflowTransitionSerializer(
many=True, read_only=True, source='get_transition_choices'
)
workflow = WorkflowSerializer(read_only=True)
class Meta:
fields = (
'current_state', 'document_workflow_url', 'last_log_entry',
'log_entries_url', 'transition_choices', 'workflow',
)
model = WorkflowInstance
def get_document_workflow_url(self, instance):
return reverse(
'rest_api:workflowinstance-detail', args=(
instance.document.pk, instance.pk,
), request=self.context['request'], format=self.context['format']
)
def get_log_entries_url(self, instance):
return reverse(
'rest_api:workflowinstancelogentry-list', args=(
instance.document.pk, instance.pk,
), request=self.context['request'], format=self.context['format']
)
class WritableWorkflowSerializer(serializers.ModelSerializer):
document_types_pk_list = serializers.CharField(
help_text=_(
@@ -240,3 +310,46 @@ class WritableWorkflowSerializer(serializers.ModelSerializer):
)
return instance
class WritableWorkflowInstanceLogEntrySerializer(serializers.ModelSerializer):
document_workflow_url = serializers.SerializerMethodField()
transition_pk = serializers.IntegerField(
help_text=_('Primary key of the transition to be added.'),
write_only=True
)
transition = WorkflowTransitionSerializer(read_only=True)
user = UserSerializer(read_only=True)
class Meta:
fields = (
'comment', 'datetime', 'document_workflow_url', 'transition',
'transition_pk', 'user'
)
model = WorkflowInstanceLogEntry
def create(self, validated_data):
validated_data['transition'] = WorkflowTransition.objects.get(
pk=validated_data.pop('transition_pk')
)
validated_data['user'] = self.context['request'].user
validated_data['workflow_instance'] = self.context['workflow_instance']
if validated_data['transition'] not in validated_data['workflow_instance'].get_transition_choices():
raise ValidationError(
{
'transition_pk': _('Not a valid transition choice.')
}
)
return super(WritableWorkflowInstanceLogEntrySerializer, self).create(
validated_data
)
def get_document_workflow_url(self, instance):
return reverse(
'rest_api:workflowinstance-detail', args=(
instance.workflow_instance.document.pk,
instance.workflow_instance.pk,
), request=self.context['request'], format=self.context['format']
)

1
mayan/apps/document_states/tests/literals.py
View File

@@ -4,6 +4,7 @@ TEST_WORKFLOW_LABEL = 'test workflow label'
TEST_WORKFLOW_LABEL_EDITED = 'test workflow label edited'
TEST_WORKFLOW_INITIAL_STATE_LABEL = 'test initial state'
TEST_WORKFLOW_INITIAL_STATE_COMPLETION = 33
TEST_WORKFLOW_INSTANCE_LOG_ENTRY_COMMENT = 'test workflow instance log entry comment'
TEST_WORKFLOW_STATE_LABEL = 'test state label'
TEST_WORKFLOW_STATE_LABEL_EDITED = 'test state label edited'
TEST_WORKFLOW_STATE_COMPLETION = 66

134
mayan/apps/document_states/tests/test_api.py
View File

@@ -3,7 +3,6 @@ from __future__ import unicode_literals
from django.contrib.auth import get_user_model
from django.core.urlresolvers import reverse
from django.test import override_settings
from django.utils.encoding import force_text
from rest_framework.test import APITestCase
@@ -20,9 +19,9 @@ from ..models import Workflow
from .literals import (
TEST_WORKFLOW_LABEL, TEST_WORKFLOW_LABEL_EDITED,
TEST_WORKFLOW_INITIAL_STATE_COMPLETION, TEST_WORKFLOW_INITIAL_STATE_LABEL,
TEST_WORKFLOW_STATE_COMPLETION, TEST_WORKFLOW_STATE_LABEL,
TEST_WORKFLOW_STATE_LABEL_EDITED, TEST_WORKFLOW_TRANSITION_LABEL,
TEST_WORKFLOW_TRANSITION_LABEL_EDITED
TEST_WORKFLOW_INSTANCE_LOG_ENTRY_COMMENT, TEST_WORKFLOW_STATE_COMPLETION,
TEST_WORKFLOW_STATE_LABEL, TEST_WORKFLOW_STATE_LABEL_EDITED,
TEST_WORKFLOW_TRANSITION_LABEL, TEST_WORKFLOW_TRANSITION_LABEL_EDITED
)
@@ -485,3 +484,130 @@ class WorkflowTransitionsAPITestCase(APITestCase):
self.workflow_transition.destination_state,
self.workflow_state_1
)
@override_settings(OCR_AUTO_OCR=False)
class DocumentWorkflowsAPITestCase(APITestCase):
def setUp(self):
self.admin_user = get_user_model().objects.create_superuser(
username=TEST_ADMIN_USERNAME, email=TEST_ADMIN_EMAIL,
password=TEST_ADMIN_PASSWORD
)
self.client.login(
username=TEST_ADMIN_USERNAME, password=TEST_ADMIN_PASSWORD
)
self.document_type = DocumentType.objects.create(
label=TEST_DOCUMENT_TYPE
)
def tearDown(self):
if hasattr(self, 'document_type'):
self.document_type.delete()
def _create_document(self):
with open(TEST_SMALL_DOCUMENT_PATH) as file_object:
self.document = self.document_type.new_document(
file_object=file_object
)
def _create_workflow(self):
self.workflow = Workflow.objects.create(label=TEST_WORKFLOW_LABEL)
self.workflow.document_types.add(self.document_type)
def _create_workflow_states(self):
self._create_workflow()
self.workflow_state_1 = self.workflow.states.create(
completion=TEST_WORKFLOW_INITIAL_STATE_COMPLETION,
initial=True, label=TEST_WORKFLOW_INITIAL_STATE_LABEL
)
self.workflow_state_2 = self.workflow.states.create(
completion=TEST_WORKFLOW_STATE_COMPLETION,
label=TEST_WORKFLOW_STATE_LABEL
)
def _create_workflow_transition(self):
self._create_workflow_states()
self.workflow_transition = self.workflow.transitions.create(
label=TEST_WORKFLOW_TRANSITION_LABEL,
origin_state=self.workflow_state_1,
destination_state=self.workflow_state_2,
)
def _create_workflow_instance_log_entry(self):
self.document.workflows.first().log_entries.create(
comment=TEST_WORKFLOW_INSTANCE_LOG_ENTRY_COMMENT, transition=self.workflow_transition,
user=self.admin_user
)
def test_workflow_instance_detail_view(self):
self._create_workflow_transition()
self._create_document()
response = self.client.get(
reverse(
'rest_api:workflowinstance-detail', args=(
self.document.pk, self.document.workflows.first().pk
)
),
)
self.assertEqual(
response.data['workflow']['label'],
TEST_WORKFLOW_LABEL
)
def test_workflow_instance_list_view(self):
self._create_workflow_transition()
self._create_document()
response = self.client.get(
reverse(
'rest_api:workflowinstance-list', args=(self.document.pk,)
),
)
self.assertEqual(
response.data['results'][0]['workflow']['label'],
TEST_WORKFLOW_LABEL
)
def test_workflow_instance_log_entries_create_view(self):
self._create_workflow_transition()
self._create_document()
workflow_instance = self.document.workflows.first()
self.client.post(
reverse(
'rest_api:workflowinstancelogentry-list', args=(
self.document.pk, workflow_instance.pk
),
), data={'transition_pk': self.workflow_transition.pk}
)
workflow_instance.refresh_from_db()
self.assertEqual(
workflow_instance.log_entries.first().transition.label,
TEST_WORKFLOW_TRANSITION_LABEL
)
def test_workflow_instance_log_entries_list_view(self):
self._create_workflow_transition()
self._create_document()
self._create_workflow_instance_log_entry()
response = self.client.get(
reverse(
'rest_api:workflowinstancelogentry-list', args=(
self.document.pk, self.document.workflows.first().pk
)
),
)
self.assertEqual(
response.data['results'][0]['transition']['label'],
TEST_WORKFLOW_TRANSITION_LABEL
)

20
mayan/apps/document_states/urls.py
View File

@@ -4,9 +4,10 @@ from django.conf.urls import patterns, url
from .api_views import (
APIWorkflowDocumentTypeList, APIWorkflowDocumentTypeView,
APIWorkflowListView, APIWorkflowStateListView, APIWorkflowStateView,
APIWorkflowTransitionListView, APIWorkflowTransitionView,
APIWorkflowView
APIWorkflowInstanceListView, APIWorkflowInstanceView,
APIWorkflowInstanceLogEntryListView, APIWorkflowListView,
APIWorkflowStateListView, APIWorkflowStateView,
APIWorkflowTransitionListView, APIWorkflowTransitionView, APIWorkflowView
)
from .views import (
DocumentWorkflowInstanceListView, SetupWorkflowCreateView,
@@ -136,4 +137,17 @@ api_urls = [
r'^workflows/(?P<pk>[0-9]+)/transitions/(?P<transition_pk>[0-9]+)/$',
APIWorkflowTransitionView.as_view(), name='workflowtransition-detail'
),
url(
r'^document/(?P<pk>[0-9]+)/workflows/$',
APIWorkflowInstanceListView.as_view(), name='workflowinstance-list'
),
url(
r'^document/(?P<pk>[0-9]+)/workflows/(?P<workflow_pk>[0-9]+)/$',
APIWorkflowInstanceView.as_view(), name='workflowinstance-detail'
),
url(
r'^document/(?P<pk>[0-9]+)/workflows/(?P<workflow_pk>[0-9]+)/log_entries/$',
APIWorkflowInstanceLogEntryListView.as_view(),
name='workflowinstancelogentry-list'
),
]