From 206776441c5c4854c3768a8215f0eb6c20720a6a Mon Sep 17 00:00:00 2001 From: Roberto Rosario Date: Thu, 23 Feb 2017 23:40:10 -0400 Subject: [PATCH] Add transition ACLs support to the API view and serializer. Signed-off-by: Roberto Rosario --- mayan/apps/document_states/api_views.py | 32 ++-- mayan/apps/document_states/serializers.py | 35 ++-- mayan/apps/document_states/tests/test_api.py | 164 ++++++++++++++++++- 3 files changed, 197 insertions(+), 34 deletions(-) diff --git a/mayan/apps/document_states/api_views.py b/mayan/apps/document_states/api_views.py index 08a1e55d43..4440effedf 100644 --- a/mayan/apps/document_states/api_views.py +++ b/mayan/apps/document_states/api_views.py @@ -590,21 +590,27 @@ class APIWorkflowInstanceLogEntryListView(generics.ListCreateAPIView): ) 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 - ) + if self.request.method == 'GET': + """ + Only test for permission if reading. If writing, the permission + will be checked in the serializer + + IMPROVEMENT: + When writing, add check for permission or ACL for the workflow. + Failing that, check for ACLs for any of the workflow's transitions. + Failing that, then raise PermissionDenied + """ + + 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 diff --git a/mayan/apps/document_states/serializers.py b/mayan/apps/document_states/serializers.py index 4193a46104..3a0daf4ab2 100644 --- a/mayan/apps/document_states/serializers.py +++ b/mayan/apps/document_states/serializers.py @@ -1,5 +1,6 @@ from __future__ import unicode_literals +from django.core.exceptions import ValidationError as DjangoValidationError from django.utils.translation import ugettext_lazy as _ from rest_framework import serializers @@ -328,24 +329,6 @@ class WritableWorkflowInstanceLogEntrySerializer(serializers.ModelSerializer): ) 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=( @@ -353,3 +336,19 @@ class WritableWorkflowInstanceLogEntrySerializer(serializers.ModelSerializer): instance.workflow_instance.pk, ), request=self.context['request'], format=self.context['format'] ) + + def validate(self, attrs): + attrs['user'] = self.context['request'].user + attrs['workflow_instance'] = self.context['workflow_instance'] + attrs['transition'] = WorkflowTransition.objects.get( + pk=attrs.pop('transition_pk') + ) + + instance = WorkflowInstanceLogEntry(**attrs) + + try: + instance.full_clean() + except DjangoValidationError as exception: + raise ValidationError(exception) + + return attrs diff --git a/mayan/apps/document_states/tests/test_api.py b/mayan/apps/document_states/tests/test_api.py index 9ee947abe2..ffc9366894 100644 --- a/mayan/apps/document_states/tests/test_api.py +++ b/mayan/apps/document_states/tests/test_api.py @@ -1,20 +1,27 @@ -from __future__ import unicode_literals +from __future__ import absolute_import, unicode_literals from django.contrib.auth import get_user_model +from django.contrib.auth.models import Group from django.core.urlresolvers import reverse from django.test import override_settings from rest_framework.test import APITestCase +from acls.models import AccessControlList from documents.models import DocumentType from documents.tests.literals import ( TEST_DOCUMENT_TYPE, TEST_SMALL_DOCUMENT_PATH ) -from user_management.tests.literals import ( - TEST_ADMIN_EMAIL, TEST_ADMIN_PASSWORD, TEST_ADMIN_USERNAME +from permissions import Permission +from permissions.models import Role +from permissions.tests.literals import TEST_ROLE_LABEL +from user_management.tests import ( + TEST_ADMIN_EMAIL, TEST_ADMIN_PASSWORD, TEST_ADMIN_USERNAME, TEST_GROUP, + TEST_USER_EMAIL, TEST_USER_USERNAME, TEST_USER_PASSWORD ) from ..models import Workflow +from ..permissions import permission_workflow_transition from .literals import ( TEST_WORKFLOW_LABEL, TEST_WORKFLOW_LABEL_EDITED, @@ -624,3 +631,154 @@ class DocumentWorkflowsAPITestCase(APITestCase): response.data['results'][0]['transition']['label'], TEST_WORKFLOW_TRANSITION_LABEL ) + + +@override_settings(OCR_AUTO_OCR=False) +class DocumentWorkflowsTransitionACLsAPITestCase(APITestCase): + def setUp(self): + self.user = get_user_model().objects.create_user( + username=TEST_USER_USERNAME, email=TEST_USER_EMAIL, + password=TEST_USER_PASSWORD + ) + + self.client.login( + username=TEST_USER_USERNAME, password=TEST_USER_PASSWORD + ) + + self.document_type = DocumentType.objects.create( + label=TEST_DOCUMENT_TYPE + ) + + self.group = Group.objects.create(name=TEST_GROUP) + self.role = Role.objects.create(label=TEST_ROLE_LABEL) + self.group.user_set.add(self.user) + self.role.groups.add(self.group) + Permission.invalidate_cache() + + 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 test_workflow_transition_view_no_permission(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.count(), 0) + + def test_workflow_transition_view_with_permission(self): + self._create_workflow_transition() + self._create_document() + + workflow_instance = self.document.workflows.first() + + self.role.permissions.add( + permission_workflow_transition.stored_permission + ) + + 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_transition_view_with_workflow_acl(self): + self._create_workflow_transition() + self._create_document() + + workflow_instance = self.document.workflows.first() + + acl = AccessControlList.objects.create( + content_object=self.workflow, role=self.role + ) + acl.permissions.add(permission_workflow_transition.stored_permission) + + 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_transition_view_transition_acl(self): + self._create_workflow_transition() + self._create_document() + + workflow_instance = self.document.workflows.first() + + acl = AccessControlList.objects.create( + content_object=self.workflow_transition, role=self.role + ) + acl.permissions.add(permission_workflow_transition.stored_permission) + + 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 + )