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

Add workflow transition API views

Browse Source 
Signed-off-by: Roberto Rosario <Roberto.Rosario@mayan-edms.com>
This commit is contained in:
Roberto Rosario
2019-03-17 17:57:18 -04:00
parent 62c92ba6fd
commit 2fbe4625c0
5 changed files with 336 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -19,7 +19,8 @@ from .permissions import (
permission_workflow_edit, permission_workflow_view
)
from .serializers import (
WorkflowSerializer, WorkflowStateSerializer
WorkflowSerializer, WorkflowStateSerializer,
WorkflowTransitionSerializer, WorkflowTransitionWritableSerializer
)
from .settings import settings_workflow_image_cache_time
from .storages import storage_workflowimagecache
@@ -72,6 +73,42 @@ class WorkflowStateAPIViewSet(ExternalObjectAPIViewSetMixin, MayanAPIModelViewSe
return context
class WorkflowTransitionAPIViewSet(ExternalObjectAPIViewSetMixin, MayanAPIModelViewSet):
external_object_class = Workflow
external_object_pk_url_kwarg = 'workflow_id'
lookup_url_kwarg = 'workflow_transition_id'
def get_external_object_permission(self):
action = getattr(self, 'action', None)
if action is None:
return None
elif action in ['create', 'destroy', 'partial_update', 'update']:
return permission_workflow_edit
else:
return permission_workflow_view
def get_queryset(self):
return self.get_external_object().transitions.all()
def get_serializer_class(self):
action = getattr(self, 'action', None)
if action is None:
return None
if action in ['create', 'partial_update', 'update']:
return WorkflowTransitionWritableSerializer
else:
return WorkflowTransitionSerializer
def get_serializer_context(self):
context = super(WorkflowTransitionAPIViewSet, self).get_serializer_context()
if self.kwargs:
context.update(
{
'workflow': self.get_external_object(),
}
)
return context
'''

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

@@ -9,25 +9,30 @@ from rest_framework.reverse import reverse
from mayan.apps.documents.models import DocumentType
from mayan.apps.documents.serializers import DocumentTypeSerializer
from mayan.apps.rest_api.relations import MultiKwargHyperlinkedIdentityField
from mayan.apps.rest_api.relations import (
FilteredPrimaryKeyRelatedField, MultiKwargHyperlinkedIdentityField
)
from mayan.apps.user_management.serializers import UserSerializer
from .models import (
Workflow, WorkflowInstance, WorkflowInstanceLogEntry, WorkflowState,
WorkflowTransition
)
from .permissions import permission_workflow_edit
class WorkflowSerializer(serializers.HyperlinkedModelSerializer):
#document_types_url = serializers.HyperlinkedIdentityField(
# view_name='rest_api:workflow-document-type-list'
#)
#image_url = serializers.SerializerMethodField()
#transitions = WorkflowTransitionSerializer(many=True, required=False)
state_list_url = serializers.HyperlinkedIdentityField(
lookup_url_kwarg='workflow_id',
view_name='rest_api:workflow-state-list'
)
transition_list_url = serializers.HyperlinkedIdentityField(
lookup_url_kwarg='workflow_id',
view_name='rest_api:workflow-transition-list'
)
class Meta:
extra_kwargs = {
@@ -38,8 +43,8 @@ class WorkflowSerializer(serializers.HyperlinkedModelSerializer):
}
fields = (
#'document_types_url', 'image_url',
'id', 'internal_name', 'label', 'state_list_url', 'url'
#'transitions',
'id', 'internal_name', 'label', 'state_list_url',
'transition_list_url', 'url'
)
model = Workflow
@@ -71,6 +76,82 @@ class WorkflowStateSerializer(serializers.HyperlinkedModelSerializer):
return super(WorkflowStateSerializer, self).create(validated_data)
class WorkflowTransitionSerializer(serializers.HyperlinkedModelSerializer):
destination_state = WorkflowStateSerializer(read_only=True)
origin_state = WorkflowStateSerializer(read_only=True)
workflow = WorkflowSerializer(read_only=True)
url = MultiKwargHyperlinkedIdentityField(
view_kwargs=(
{
'lookup_field': 'workflow_id',
'lookup_url_kwarg': 'workflow_id',
},
{
'lookup_field': 'pk',
'lookup_url_kwarg': 'workflow_transition_id',
}
),
view_name='rest_api:workflow-transition-detail'
)
class Meta:
fields = (
'destination_state', 'id', 'label', 'origin_state', 'url',
'workflow',
)
model = WorkflowTransition
class WorkflowTransitionWritableSerializer(WorkflowTransitionSerializer):
destination_state = FilteredPrimaryKeyRelatedField(
label=_('Destination state'),
source_permission=permission_workflow_edit,
write_only=True
)
origin_state = FilteredPrimaryKeyRelatedField(
label=_('Source state'),
source_permission=permission_workflow_edit,
write_only=True
)
def create(self, validated_data):
validated_data['workflow'] = self.context['workflow']
return super(WorkflowTransitionWritableSerializer, self).create(
validated_data=validated_data
)
def get_destination_state_queryset(self):
return self.context['workflow'].states.all()
def get_origin_state_queryset(self):
return self.context['workflow'].states.all()
def update(self, instance, validated_data):
return super(WorkflowTransitionWritableSerializer, self).update(
instance=instance, validated_data=validated_data
)
"""
def validate(self, attrs):
attrs['document'] = self.context['document']
instance = DocumentMetadata(**attrs)
try:
instance.full_clean()
except DjangoValidationError as exception:
raise ValidationError(
{
api_settings.NON_FIELD_ERRORS_KEY: exception.messages
}, code='invalid'
)
return attrs
"""
"""
class NewWorkflowDocumentTypeSerializer(serializers.Serializer):
document_type_pk = serializers.IntegerField(
@@ -109,97 +190,6 @@ class WorkflowDocumentTypeSerializer(DocumentTypeSerializer):
)
class WorkflowTransitionSerializer(serializers.HyperlinkedModelSerializer):
destination_state = WorkflowStateSerializer()
origin_state = WorkflowStateSerializer()
url = serializers.SerializerMethodField()
workflow_url = serializers.SerializerMethodField()
class Meta:
fields = (
'destination_state', 'id', 'label', 'origin_state', 'url',
'workflow_url',
)
model = WorkflowTransition
def get_url(self, instance):
return reverse(
viewname='rest_api:workflowtransition-detail', kwargs={
'workflow_pk': instance.workflow.pk, 'transition_pk': instance.pk
}, request=self.context['request'], format=self.context['format']
)
def get_workflow_url(self, instance):
return reverse(
viewname='rest_api:workflow-detail', kwargs={
'workflow_pk': instance.workflow.pk,
}, request=self.context['request'], format=self.context['format']
)
class WritableWorkflowTransitionSerializer(serializers.ModelSerializer):
destination_state_pk = serializers.IntegerField(
help_text=_('Primary key of the destination state to be added.'),
write_only=True
)
origin_state_pk = serializers.IntegerField(
help_text=_('Primary key of the origin state to be added.'),
write_only=True
)
url = serializers.SerializerMethodField()
workflow_url = serializers.SerializerMethodField()
class Meta:
fields = (
'destination_state_pk', 'id', 'label', 'origin_state_pk', 'url',
'workflow_url',
)
model = WorkflowTransition
def create(self, validated_data):
validated_data['destination_state'] = WorkflowState.objects.get(
pk=validated_data.pop('destination_state_pk')
)
validated_data['origin_state'] = WorkflowState.objects.get(
pk=validated_data.pop('origin_state_pk')
)
validated_data['workflow'] = self.context['workflow']
return super(WritableWorkflowTransitionSerializer, self).create(
validated_data
)
def get_url(self, instance):
return reverse(
viewname='rest_api:workflowtransition-detail', kwargs={
'workflow_pk': instance.workflow.pk, 'transition_pk': instance.pk
}, request=self.context['request'], format=self.context['format']
)
def get_workflow_url(self, instance):
return reverse(
viewname='rest_api:workflow-detail', kwargs={
'workflow_pk': instance.workflow.pk,
}, request=self.context['request'], format=self.context['format']
)
def update(self, instance, validated_data):
validated_data['destination_state'] = WorkflowState.objects.get(
pk=validated_data.pop('destination_state_pk')
)
validated_data['origin_state'] = WorkflowState.objects.get(
pk=validated_data.pop('origin_state_pk')
)
return super(WritableWorkflowTransitionSerializer, self).update(
instance, validated_data
)
class WorkflowInstanceLogEntrySerializer(serializers.ModelSerializer):
document_workflow_url = serializers.SerializerMethodField()
transition = WorkflowTransitionSerializer(read_only=True)

17
mayan/apps/document_states/tests/mixins.py
View File

@@ -25,26 +25,27 @@ class WorkflowTestMixin(object):
def _create_test_workflow_states(self):
self.test_workflow_initial_state = WorkflowState.objects.create(
workflow=self.workflow, label=TEST_WORKFLOW_INITIAL_STATE_LABEL,
workflow=self.test_workflow, label=TEST_WORKFLOW_INITIAL_STATE_LABEL,
completion=TEST_WORKFLOW_INITIAL_STATE_COMPLETION, initial=True
)
self.test_workflow_state = WorkflowState.objects.create(
workflow=self.workflow, label=TEST_WORKFLOW_STATE_LABEL,
workflow=self.test_workflow, label=TEST_WORKFLOW_STATE_LABEL,
completion=TEST_WORKFLOW_STATE_COMPLETION
)
def _create_test_workflow_transition(self):
self._create_test_workflow_states()
self.test_workflow_transition = WorkflowTransition.objects.create(
workflow=self.workflow, label=TEST_WORKFLOW_TRANSITION_LABEL,
origin_state=self.workflow_initial_state,
destination_state=self.workflow_state
workflow=self.test_workflow, label=TEST_WORKFLOW_TRANSITION_LABEL,
origin_state=self.test_workflow_initial_state,
destination_state=self.test_workflow_state
)
def _create_test_workflow_transitions(self):
self.workflow_transition = WorkflowTransition.objects.create(
workflow=self.workflow, label=TEST_WORKFLOW_TRANSITION_LABEL,
origin_state=self.workflow_initial_state,
destination_state=self.workflow_state
workflow=self.test_workflow, label=TEST_WORKFLOW_TRANSITION_LABEL,
origin_state=self.test_workflow_initial_state,
destination_state=self.test_workflow_state
)
self.workflow_transition_2 = WorkflowTransition.objects.create(

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

@@ -400,9 +400,203 @@ class WorkflowStateAPIViewTestCase(WorkflowTestMixin, BaseAPITestCase):
)
class WorkflowTransitionAPIViewTestCase(WorkflowTestMixin, BaseAPITestCase):
def setUp(self):
super(WorkflowTransitionAPIViewTestCase, self).setUp()
self._create_test_workflow()
def _request_workflow_transition_create_api_view(self):
self._create_test_workflow_states()
return self.post(
viewname='rest_api:workflow-transition-list',
kwargs={'workflow_id': self.test_workflow.pk}, data={
'label': TEST_WORKFLOW_TRANSITION_LABEL,
'origin_state': self.test_workflow_initial_state.pk,
'destination_state': self.test_workflow_state.pk
}
)
def test_workflow_transition_create_api_view_no_access(self):
response = self._request_workflow_transition_create_api_view()
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
self.assertEqual(self.test_workflow.transitions.count(), 0)
def test_workflow_transition_create_api_view_with_permission(self):
self.grant_access(
obj=self.test_workflow, permission=permission_workflow_edit
)
response = self._request_workflow_transition_create_api_view()
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
self.assertEqual(
response.data['label'], TEST_WORKFLOW_TRANSITION_LABEL
)
self.assertEqual(self.test_workflow.transitions.count(), 1)
def _request_workflow_transition_delete_api_view(self):
return self.delete(
viewname='rest_api:workflow-transition-detail', kwargs={
'workflow_id': self.test_workflow.pk,
'workflow_transition_id': self.test_workflow_transition.pk
}
)
def test_workflow_transition_delete_api_view_no_access(self):
self._create_test_workflow_transition()
workflow_transition_count = self.test_workflow.transitions.count()
response = self._request_workflow_transition_delete_api_view()
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
self.assertEqual(
workflow_transition_count, self.test_workflow.transitions.count()
)
def test_workflow_transition_delete_view_with_access(self):
self.expected_content_type = None
self._create_test_workflow_transition()
workflow_transition_count = self.test_workflow.transitions.count()
self.grant_access(obj=self.test_workflow, permission=permission_workflow_edit)
response = self._request_workflow_transition_delete_api_view()
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
self.assertNotEqual(
workflow_transition_count, self.test_workflow.transitions.count()
)
def _request_workflow_transition_detail_api_view(self):
return self.get(
viewname='rest_api:workflow-transition-detail', kwargs={
'workflow_id': self.test_workflow.pk,
'workflow_transition_id': self.test_workflow_transition.pk
}
)
def test_workflow_transition_detail_api_view_no_access(self):
self._create_test_workflow_transition()
response = self._request_workflow_transition_detail_api_view()
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
self.assertFalse('label' in response.json())
def test_workflow_transition_detail_api_view_with_access(self):
self._create_test_workflow_transition()
self.grant_access(
obj=self.test_workflow, permission=permission_workflow_view
)
response = self._request_workflow_transition_detail_api_view()
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(
response.json()['label'], TEST_WORKFLOW_TRANSITION_LABEL
)
def _request_workflow_transition_edit_patch_api_view(self):
return self.patch(
viewname='rest_api:workflow-transition-detail', kwargs={
'workflow_id': self.test_workflow.pk,
'workflow_transition_id': self.test_workflow_transition.pk
}, data={'label': TEST_WORKFLOW_TRANSITION_LABEL_EDITED}
)
def test_workflow_transition_edit_patch_api_view_no_access(self):
self._create_test_workflow_transition()
test_workflow_transition = copy.copy(self.test_workflow_transition)
response = self._request_workflow_transition_edit_patch_api_view()
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
self.test_workflow_transition.refresh_from_db()
self.assertEqual(
test_workflow_transition.label, self.test_workflow_transition.label
)
def test_workflow_transition_edit_patch_api_view_with_access(self):
self._create_test_workflow_transition()
test_workflow_transition = copy.copy(self.test_workflow_transition)
self.grant_access(
obj=self.test_workflow, permission=permission_workflow_edit
)
response = self._request_workflow_transition_edit_patch_api_view()
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.test_workflow_transition.refresh_from_db()
self.assertNotEqual(
test_workflow_transition.label, self.test_workflow_transition.label
)
def _request_workflow_transition_edit_put_api_view(self):
return self.put(
viewname='rest_api:workflow-transition-detail', kwargs={
'workflow_id': self.test_workflow.pk,
'workflow_transition_id': self.test_workflow_transition.pk
}, data={
'label': TEST_WORKFLOW_TRANSITION_LABEL_EDITED,
'origin_state': self.test_workflow_initial_state.pk,
'destination_state': self.test_workflow_state.pk
}
)
def test_workflow_transition_edit_put_api_view_no_access(self):
self._create_test_workflow_transition()
test_workflow_transition = copy.copy(self.test_workflow_transition)
response = self._request_workflow_transition_edit_put_api_view()
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
self.test_workflow_transition.refresh_from_db()
self.assertEqual(
test_workflow_transition.label, self.test_workflow_transition.label
)
def test_workflow_transition_edit_put_api_view_with_access(self):
self._create_test_workflow_transition()
test_workflow_transition = copy.copy(self.test_workflow_transition)
self.grant_access(
obj=self.test_workflow, permission=permission_workflow_edit
)
response = self._request_workflow_transition_edit_put_api_view()
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.test_workflow_transition.refresh_from_db()
self.assertNotEqual(
test_workflow_transition.label, self.test_workflow_transition.label
)
def _request_workflow_transition_list_api_view(self):
return self.get(
viewname='rest_api:workflow-transition-list', kwargs={
'workflow_id': self.test_workflow.pk
}
)
def test_workflow_transition_list_api_view_no_access(self):
self._create_test_workflow_transition()
response = self._request_workflow_transition_list_api_view()
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
self.assertFalse('label' in response.data)
def test_workflow_transition_list_api_view_with_access(self):
self._create_test_workflow_transition()
self.grant_access(
obj=self.test_workflow, permission=permission_workflow_view
)
response = self._request_workflow_transition_list_api_view()
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(
response.data['results'][0]['label'], self.test_workflow_transition.label
)
"""
def _request_workflow_create_view_with_document_type(self):
return self.post(
viewname='rest_api:workflow-list', data={

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

@@ -3,8 +3,8 @@ from __future__ import unicode_literals
from django.conf.urls import url
from .api_views import (
WorkflowAPIViewSet,
WorkflowStateAPIViewSet
WorkflowAPIViewSet, WorkflowStateAPIViewSet,
WorkflowTransitionAPIViewSet
)
from .views import (
DocumentWorkflowInstanceListView, ToolLaunchAllWorkflows,
@@ -185,6 +185,11 @@ api_router_entries = (
{
'prefix': r'workflows/(?P<workflow_id>[^/.]+)/states',
'viewset': WorkflowStateAPIViewSet, 'basename': 'workflow-state'
},
{
'prefix': r'workflows/(?P<workflow_id>[^/.]+)/transitions',
'viewset': WorkflowTransitionAPIViewSet,
'basename': 'workflow-transition'
}
#{
# 'prefix': r'metadata_types/(?P<metadata_type_id>[^/.]+)/document_type_relations',