Change the way new document version are blocked, moving the logic to the documents app from the checkouts app signal handler. Disable the upload new document version button and have the new version upload view redirect to the document version list with a message. GitLab issue #231.

This commit is contained in:
Roberto Rosario
2015-10-23 06:05:43 -04:00
parent 414c55f824
commit da4c41c5b2
16 changed files with 395 additions and 141 deletions

View File

@@ -13,7 +13,6 @@ from documents.models import Document, DocumentVersion
from mayan.celery import app
from rest_api.classes import APIEndPoint
from .handlers import check_if_new_versions_allowed
from .links import (
link_checkin_document, link_checkout_document, link_checkout_info,
link_checkout_list
@@ -104,8 +103,3 @@ class CheckoutsApp(MayanAppConfig):
'checkouts:checkin_document'
)
)
pre_save.connect(
check_if_new_versions_allowed,
dispatch_uid='document_index_delete', sender=DocumentVersion
)

View File

@@ -10,13 +10,6 @@ class DocumentCheckoutError(Exception):
pass
class DocumentCheckoutWarning(Warning):
"""
Base checkout warning
"""
pass
class DocumentNotCheckedOut(DocumentCheckoutError):
"""
Raised when trying to checkin a document that is not checkedout
@@ -30,11 +23,3 @@ class DocumentAlreadyCheckedOut(DocumentCheckoutError):
"""
def __unicode__(self):
return ugettext('Document already checked out.')
class NewDocumentVersionNotAllowed(DocumentCheckoutWarning):
"""
Uploading new versions for this document is not allowed
Current reasons: Document is in checked out state
"""
pass

View File

@@ -1,16 +0,0 @@
from __future__ import unicode_literals
from django.utils.translation import ugettext_lazy as _
from .exceptions import NewDocumentVersionNotAllowed
from .models import DocumentCheckout
def check_if_new_versions_allowed(sender, **kwargs):
if not DocumentCheckout.objects.are_document_new_versions_allowed(kwargs['instance'].document):
raise NewDocumentVersionNotAllowed(
_(
'New versions not allowed for the checkedout document: %s'
% kwargs['instance'].document
)
)

View File

@@ -10,7 +10,7 @@ from django.utils.encoding import python_2_unicode_compatible
from django.utils.timezone import now
from django.utils.translation import ugettext_lazy as _
from documents.models import Document
from documents.models import Document, NewVersionBlock
from .events import event_document_check_out
from .exceptions import DocumentAlreadyCheckedOut
@@ -50,16 +50,22 @@ class DocumentCheckout(models.Model):
def __str__(self):
return unicode(self.document)
def get_absolute_url(self):
return reverse('checkout:checkout_info', args=(self.document.pk,))
def clean(self):
if self.expiration_datetime < now():
raise ValidationError(
_('Check out expiration date and time must be in the future.')
)
def delete(self, *args, **kwargs):
# TODO: enclose in transaction
NewVersionBlock.objects.unblock(self.document)
super(DocumentCheckout, self).delete(*args, **kwargs)
def get_absolute_url(self):
return reverse('checkout:checkout_info', args=(self.document.pk,))
def save(self, *args, **kwargs):
# TODO: enclose in transaction
new_checkout = not self.pk
if not new_checkout or self.document.is_checked_out():
raise DocumentAlreadyCheckedOut
@@ -69,6 +75,9 @@ class DocumentCheckout(models.Model):
event_document_check_out.commit(
actor=self.user, target=self.document
)
if self.block_new_version:
NewVersionBlock.objects.block(self.document)
logger.info(
'Document "%s" checked out by user "%s"',
self.document, self.user

View File

@@ -8,6 +8,7 @@ from django.core.files import File
from django.test import TestCase, override_settings
from django.utils.timezone import now
from documents.exceptions import NewDocumentVersionNotAllowed
from documents.models import DocumentType
from documents.tests.literals import (
TEST_DOCUMENT_TYPE, TEST_SMALL_DOCUMENT_PATH
@@ -16,10 +17,7 @@ from user_management.tests.literals import (
TEST_ADMIN_USERNAME, TEST_ADMIN_EMAIL, TEST_ADMIN_PASSWORD
)
from ..exceptions import (
DocumentAlreadyCheckedOut, DocumentNotCheckedOut,
NewDocumentVersionNotAllowed
)
from ..exceptions import DocumentAlreadyCheckedOut, DocumentNotCheckedOut
from ..models import DocumentCheckout

View File

@@ -14,75 +14,62 @@ 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_PASSWORD, TEST_ADMIN_USERNAME, TEST_ADMIN_EMAIL
from documents.tests.test_views import GenericDocumentViewTestCase
from sources.links import link_upload_version
from user_management.tests import (
TEST_USER_PASSWORD, TEST_USER_USERNAME, TEST_ADMIN_EMAIL,
TEST_ADMIN_PASSWORD, TEST_ADMIN_USERNAME,
)
from ..models import DocumentCheckout
from ..permissions import (
permission_document_checkin, permission_document_checkout
)
class DocumentCheckoutViewTestCase(TestCase):
def setUp(self):
self.admin_user = User.objects.create_superuser(
username=TEST_ADMIN_USERNAME, email=TEST_ADMIN_EMAIL,
password=TEST_ADMIN_PASSWORD
)
self.client = Client()
# Login the admin user
logged_in = self.client.login(
username=TEST_ADMIN_USERNAME, password=TEST_ADMIN_PASSWORD
)
self.assertTrue(logged_in)
self.assertTrue(self.admin_user.is_authenticated())
self.document_type = DocumentType.objects.create(
label=TEST_DOCUMENT_TYPE
class DocumentCheckoutViewTestCase(GenericDocumentViewTestCase):
def test_checkin_document_view_no_permission(self):
self.login(
username=TEST_USER_USERNAME, password=TEST_USER_PASSWORD
)
with open(TEST_SMALL_DOCUMENT_PATH) as file_object:
self.document = self.document_type.new_document(
file_object=File(file_object)
)
def tearDown(self):
self.document_type.delete()
self.admin_user.delete()
def test_checkout_view(self):
response = self.client.post(
reverse(
'checkouts:checkout_document', args=(self.document.pk,),
), data={
'expiration_datetime_0': 2,
'expiration_datetime_1': TIME_DELTA_UNIT_DAYS,
'block_new_version': True
}, follow=True
)
self.assertEquals(response.status_code, 200)
self.assertTrue(self.document.is_checked_out())
self.assertTrue(
DocumentCheckout.objects.is_document_checked_out(
document=self.document
)
)
def test_checkin_view(self):
expiration_datetime = now() + datetime.timedelta(days=1)
DocumentCheckout.objects.checkout_document(
document=self.document, expiration_datetime=expiration_datetime,
user=self.admin_user, block_new_version=True
user=self.user, block_new_version=True
)
self.assertTrue(self.document.is_checked_out())
response = self.client.post(
reverse(
'checkouts:checkin_document', args=(self.document.pk,),
), follow=True
response = self.post(
'checkouts:checkin_document', args=(self.document.pk,), follow=True
)
self.assertEquals(response.status_code, 403)
self.assertTrue(self.document.is_checked_out())
def test_checkin_document_view_with_permission(self):
self.login(
username=TEST_USER_USERNAME, password=TEST_USER_PASSWORD
)
expiration_datetime = now() + datetime.timedelta(days=1)
DocumentCheckout.objects.checkout_document(
document=self.document, expiration_datetime=expiration_datetime,
user=self.user, block_new_version=True
)
self.assertTrue(self.document.is_checked_out())
self.role.permissions.add(
permission_document_checkin.stored_permission
)
response = self.post(
'checkouts:checkin_document', args=(self.document.pk,), follow=True
)
self.assertEquals(response.status_code, 200)
@@ -94,3 +81,85 @@ class DocumentCheckoutViewTestCase(TestCase):
document=self.document
)
)
def test_checkout_document_view_no_permission(self):
self.login(
username=TEST_USER_USERNAME, password=TEST_USER_PASSWORD
)
response = self.post(
'checkouts:checkout_document', args=(self.document.pk,), data={
'expiration_datetime_0': 2,
'expiration_datetime_1': TIME_DELTA_UNIT_DAYS,
'block_new_version': True
}, follow=True
)
self.assertEquals(response.status_code, 403)
self.assertFalse(self.document.is_checked_out())
def test_checkout_document_view_with_permission(self):
self.login(
username=TEST_USER_USERNAME, password=TEST_USER_PASSWORD
)
self.role.permissions.add(
permission_document_checkout.stored_permission
)
response = self.post(
'checkouts:checkout_document', args=(self.document.pk,), data={
'expiration_datetime_0': 2,
'expiration_datetime_1': TIME_DELTA_UNIT_DAYS,
'block_new_version': True
}, follow=True
)
self.assertEquals(response.status_code, 200)
self.assertTrue(self.document.is_checked_out())
def test_document_new_version_after_checkout(self):
"""
Gitlab issue #231
User shown option to upload new version of a document even though it
is blocked by checkout - v2.0.0b2
Expected results:
- Link to upload version view should not resolve
- Upload version view should reject request
"""
self.login(
username=TEST_ADMIN_USERNAME, password=TEST_ADMIN_PASSWORD
)
expiration_datetime = now() + datetime.timedelta(days=1)
DocumentCheckout.objects.checkout_document(
document=self.document, expiration_datetime=expiration_datetime,
user=self.admin_user, block_new_version=True
)
self.assertTrue(self.document.is_checked_out())
response = self.post(
'sources:upload_version', args=(self.document.pk,),
follow=True
)
self.assertContains(
response, text='blocked from uploading',
status_code=200
)
response = self.get(
'documents:document_version_list', args=(self.document.pk,),
follow=True
)
#Needed by the url view resolver
response.context.current_app = None
resolved_link = link_upload_version.resolve(context=response.context)
self.assertEqual(resolved_link, None)

View File

@@ -0,0 +1,17 @@
from __future__ import unicode_literals
from django.utils.translation import ugettext
class DocumentException(Exception):
"""
Base documents warning
"""
pass
class NewDocumentVersionNotAllowed(DocumentException):
"""
Uploading new versions for this document is not allowed
"""
pass

View File

@@ -13,29 +13,22 @@ from .settings import setting_recent_count
logger = logging.getLogger(__name__)
class RecentDocumentManager(models.Manager):
def add_document_for_user(self, user, document):
if user.is_authenticated():
new_recent, created = self.model.objects.get_or_create(
user=user, document=document
)
if not created:
# document already in the recent list, just save to force
# accessed date and time update
new_recent.save()
class DocumentManager(models.Manager):
def delete_stubs(self):
for stale_stub_document in self.filter(is_stub=True, date_added__lt=now() - timedelta(seconds=STUB_EXPIRATION_INTERVAL)):
stale_stub_document.delete(trash=False)
recent_to_delete = self.filter(user=user).values_list('pk', flat=True)[setting_recent_count.value:]
self.filter(pk__in=list(recent_to_delete)).delete()
def get_by_natural_key(self, uuid):
return self.get(uuid=uuid)
def get_for_user(self, user):
document_model = apps.get_model('documents', 'document')
def get_queryset(self):
return TrashCanQuerySet(
self.model, using=self._db
).filter(in_trash=False)
if user.is_authenticated():
return document_model.objects.filter(
recentdocument__user=user
).order_by('-recentdocument__datetime_accessed')
else:
return document_model.objects.none()
def invalidate_cache(self):
for document in self.model.objects.all():
document.invalidate_cache()
class DocumentTypeManager(models.Manager):
@@ -105,28 +98,46 @@ class DocumentTypeManager(models.Manager):
return self.get(label=label)
class DocumentManager(models.Manager):
def delete_stubs(self):
for stale_stub_document in self.filter(is_stub=True, date_added__lt=now() - timedelta(seconds=STUB_EXPIRATION_INTERVAL)):
stale_stub_document.delete(trash=False)
class NewVersionBlockManager(models.Manager):
def block(self, document):
self.get_or_create(document=document)
def get_by_natural_key(self, uuid):
return self.get(uuid=uuid)
def unblock(self, document):
self.filter(document=document).delete()
def get_queryset(self):
return TrashCanQuerySet(
self.model, using=self._db
).filter(in_trash=False)
def invalidate_cache(self):
for document in self.model.objects.all():
document.invalidate_cache()
def is_blocked(self, document):
return self.filter(document=document).exists()
class PassthroughManager(models.Manager):
pass
class RecentDocumentManager(models.Manager):
def add_document_for_user(self, user, document):
if user.is_authenticated():
new_recent, created = self.model.objects.get_or_create(
user=user, document=document
)
if not created:
# document already in the recent list, just save to force
# accessed date and time update
new_recent.save()
recent_to_delete = self.filter(user=user).values_list('pk', flat=True)[setting_recent_count.value:]
self.filter(pk__in=list(recent_to_delete)).delete()
def get_for_user(self, user):
document_model = apps.get_model('documents', 'document')
if user.is_authenticated():
return document_model.objects.filter(
recentdocument__user=user
).order_by('-recentdocument__datetime_accessed')
else:
return document_model.objects.none()
class TrashCanManager(models.Manager):
def get_queryset(self):
return super(

View File

@@ -0,0 +1,26 @@
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
from django.db import models, migrations
class Migration(migrations.Migration):
dependencies = [
('documents', '0027_auto_20150824_0702'),
]
operations = [
migrations.CreateModel(
name='NewVersionBlock',
fields=[
('id', models.AutoField(verbose_name='ID', serialize=False, auto_created=True, primary_key=True)),
('document', models.ForeignKey(verbose_name='Document', to='documents.Document')),
],
options={
'verbose_name': 'New version block',
'verbose_name_plural': 'New version blocks',
},
bases=(models.Model,),
),
]

View File

@@ -31,10 +31,11 @@ from .events import (
event_document_properties_edit, event_document_type_change,
event_document_version_revert
)
from .exceptions import NewDocumentVersionNotAllowed
from .literals import DEFAULT_DELETE_PERIOD, DEFAULT_DELETE_TIME_UNIT
from .managers import (
DocumentManager, DocumentTypeManager, PassthroughManager,
RecentDocumentManager, TrashCanManager
DocumentManager, DocumentTypeManager, NewVersionBlockManager,
PassthroughManager, RecentDocumentManager, TrashCanManager
)
from .permissions import permission_document_view
from .runtime import cache_storage_backend, storage_backend
@@ -397,6 +398,8 @@ class DocumentVersion(models.Model):
if new_document_version:
logger.info('Creating new version for document: %s', self.document)
if NewVersionBlock.objects.is_blocked(self.document):
raise NewDocumentVersionNotAllowed
try:
with transaction.atomic():
@@ -779,6 +782,17 @@ class DocumentPage(models.Model):
return '{}-{}'.format(self.document_version.uuid, self.pk)
class NewVersionBlock(models.Model):
document = models.ForeignKey(Document, verbose_name=_('Document'))
objects = NewVersionBlockManager()
class Meta:
verbose_name = _('New version block')
verbose_name_plural = _('New version blocks')
@python_2_unicode_compatible
class RecentDocument(models.Model):
"""

View File

@@ -6,8 +6,9 @@ import time
from django.core.files import File
from django.test import TestCase, override_settings
from ..exceptions import NewDocumentVersionNotAllowed
from ..literals import STUB_EXPIRATION_INTERVAL
from ..models import DeletedDocument, Document, DocumentType
from ..models import DeletedDocument, Document, DocumentType, NewVersionBlock
from .literals import (
TEST_DOCUMENT_TYPE, TEST_DOCUMENT_PATH, TEST_MULTI_PAGE_TIFF_PATH,
@@ -257,3 +258,55 @@ class DocumentManagerTestCase(TestCase):
Document.objects.delete_stubs()
self.assertEqual(Document.objects.count(), 0)
@override_settings(OCR_AUTO_OCR=False)
class NewVersionBlockTestCase(TestCase):
def setUp(self):
self.document_type = DocumentType.objects.create(
label=TEST_DOCUMENT_TYPE
)
with open(TEST_SMALL_DOCUMENT_PATH) as file_object:
self.document = self.document_type.new_document(
file_object=File(file_object)
)
def tearDown(self):
self.document.delete()
self.document_type.delete()
def test_blocking(self):
NewVersionBlock.objects.block(document=self.document)
self.assertEqual(NewVersionBlock.objects.count(), 1)
self.assertEqual(
NewVersionBlock.objects.first().document, self.document
)
def test_unblocking(self):
NewVersionBlock.objects.create(document=self.document)
NewVersionBlock.objects.unblock(document=self.document)
self.assertEqual(NewVersionBlock.objects.count(), 0)
def test_is_blocked(self):
NewVersionBlock.objects.create(document=self.document)
self.assertTrue(
NewVersionBlock.objects.is_blocked(document=self.document)
)
NewVersionBlock.objects.all().delete()
self.assertFalse(
NewVersionBlock.objects.is_blocked(document=self.document)
)
def test_blocking_new_versions(self):
NewVersionBlock.objects.block(document=self.document)
with self.assertRaises(NewDocumentVersionNotAllowed):
with open(TEST_SMALL_DOCUMENT_PATH) as file_object:
self.document.new_version(file_object=File(file_object))

View File

@@ -16,7 +16,9 @@ from user_management.tests.literals import (
)
from ..literals import DEFAULT_DELETE_PERIOD, DEFAULT_DELETE_TIME_UNIT
from ..models import DeletedDocument, Document, DocumentType, HASH_FUNCTION
from ..models import (
DeletedDocument, Document, DocumentType, NewVersionBlock, HASH_FUNCTION
)
from ..permissions import (
permission_document_download, permission_document_properties_edit,
permission_document_restore, permission_document_tools,
@@ -575,3 +577,32 @@ class DocumentTypeViewsTestCase(GenericDocumentViewTestCase):
DocumentType.objects.get(pk=self.document_type.pk).label,
TEST_DOCUMENT_TYPE_EDITED_LABEL
)
class NewDocumentVersionBlockViewTestCase(GenericDocumentViewTestCase):
def test_document_new_version_after_checkout(self):
"""
Gitlab issue #231
User shown option to upload new version of a document even though it
is blocked by checkout - v2.0.0b2
Expected results:
- Link to upload version view should not resolve
- Upload version view should reject request
"""
self.login(
username=TEST_ADMIN_USERNAME, password=TEST_ADMIN_PASSWORD
)
NewVersionBlock.objects.block(self.document)
response = self.post(
'sources:upload_version', args=(self.document.pk,),
follow=True
)
self.assertContains(
response, text='blocked from uploading',
status_code=200
)

View File

@@ -2,6 +2,7 @@ from __future__ import absolute_import, unicode_literals
from django.utils.translation import ugettext_lazy as _
from documents.models import NewVersionBlock
from documents.permissions import (
permission_document_create, permission_document_new_version
)
@@ -17,6 +18,10 @@ from .permissions import (
)
def document_new_version_not_blocked(context):
return not NewVersionBlock.objects.is_blocked(context['object'])
link_document_create_multiple = Link(
icon='fa fa-upload', permissions=(permission_document_create,),
text=_('New document'), view='sources:document_create_multiple'
@@ -74,6 +79,7 @@ link_staging_file_delete = Link(
args=('source.pk', 'object.encoded_filename',)
)
link_upload_version = Link(
condition=document_new_version_not_blocked,
permissions=(permission_document_new_version,),
text=_('Upload new version'), view='sources:upload_version',
args='object.pk'

View File

@@ -12,7 +12,7 @@ from documents.models import Document, DocumentType
from documents.tests import (
TEST_COMPRESSED_DOCUMENT_PATH, TEST_DOCUMENT_TYPE,
TEST_NON_ASCII_DOCUMENT_FILENAME, TEST_NON_ASCII_DOCUMENT_PATH,
TEST_NON_ASCII_COMPRESSED_DOCUMENT_PATH
TEST_NON_ASCII_COMPRESSED_DOCUMENT_PATH, TEST_SMALL_DOCUMENT_PATH
)
from user_management.tests import (
TEST_ADMIN_EMAIL, TEST_ADMIN_PASSWORD, TEST_ADMIN_USERNAME

View File

@@ -5,15 +5,18 @@ from django.core.urlresolvers import reverse
from django.test.client import Client
from django.test import TestCase, override_settings
from documents.models import Document, DocumentType
from documents.models import Document, DocumentType, NewVersionBlock
from documents.tests import (
TEST_DOCUMENT_PATH, TEST_SMALL_DOCUMENT_PATH, TEST_DOCUMENT_DESCRIPTION,
TEST_DOCUMENT_TYPE
)
from documents.tests.test_views import GenericDocumentViewTestCase
from user_management.tests import (
TEST_ADMIN_PASSWORD, TEST_ADMIN_USERNAME, TEST_ADMIN_EMAIL
TEST_USER_PASSWORD, TEST_USER_USERNAME, TEST_ADMIN_EMAIL,
TEST_ADMIN_PASSWORD, TEST_ADMIN_USERNAME,
)
from ..links import link_upload_version
from ..literals import SOURCE_CHOICE_WEB_FORM
from ..models import WebFormSource
@@ -128,3 +131,43 @@ class UploadDocumentTestCase(TestCase):
# Fetch document again and test description
document = Document.objects.first()
self.assertEqual(document.description, TEST_DOCUMENT_DESCRIPTION)
class NewDocumentVersionViewTestCase(GenericDocumentViewTestCase):
def test_new_version_block(self):
"""
Gitlab issue #231
User shown option to upload new version of a document even though it
is blocked by checkout - v2.0.0b2
Expected results:
- Link to upload version view should not resolve
- Upload version view should reject request
"""
self.login(
username=TEST_ADMIN_USERNAME, password=TEST_ADMIN_PASSWORD
)
NewVersionBlock.objects.block(self.document)
response = self.post(
'sources:upload_version', args=(self.document.pk,),
follow=True
)
self.assertContains(
response, text='blocked from uploading',
status_code=200
)
response = self.get(
'documents:document_version_list', args=(self.document.pk,),
follow=True
)
#Needed by the url view resolver
response.context.current_app = None
resolved_link = link_upload_version.resolve(context=response.context)
self.assertEqual(resolved_link, None)

View File

@@ -19,7 +19,7 @@ from common.views import (
SingleObjectEditView, SingleObjectListView
)
from common.widgets import two_state_template
from documents.models import DocumentType, Document
from documents.models import DocumentType, Document, NewVersionBlock
from documents.permissions import (
permission_document_create, permission_document_new_version
)
@@ -314,6 +314,20 @@ class UploadInteractiveVersionView(UploadBaseView):
self.subtemplates_list = []
self.document = get_object_or_404(Document, pk=kwargs['document_pk'])
if NewVersionBlock.objects.is_blocked(self.document):
messages.error(
self.request,
_(
'Document "%s" is blocked from uploading new versions.'
) % self.document
)
return HttpResponseRedirect(
reverse(
'documents:document_version_list', args=(self.document.pk,)
)
)
try:
Permission.check_permissions(
self.request.user, (permission_document_new_version,)