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

Add test for GitLab issue #715

Browse Source 
Signed-off-by: Roberto Rosario <roberto.rosario@mayan-edms.com>
This commit is contained in:
Roberto Rosario
2019-12-17 01:14:48 -04:00
parent 52804e6095
commit 8b1c5eb868
3 changed files with 56 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
mayan/apps/sources/models/base.py
View File

@@ -197,7 +197,7 @@ class IntervalBaseModel(OutOfProcessSource):
def _delete_periodic_task(self, pk=None):
try:
periodic_task = PeriodicTask.objects.get(
name=self._get_periodic_task_name(pk)
name=self._get_periodic_task_name(pk=pk)
)
interval_instance = periodic_task.interval
@@ -219,7 +219,7 @@ class IntervalBaseModel(OutOfProcessSource):
def delete(self, *args, **kwargs):
pk = self.pk
super(IntervalBaseModel, self).delete(*args, **kwargs)
self._delete_periodic_task(pk)
self._delete_periodic_task(pk=pk)
def save(self, *args, **kwargs):
new_source = not self.pk

15
mayan/apps/sources/tests/mixins.py
View File

@@ -1,7 +1,8 @@
from __future__ import unicode_literals
from ..literals import SOURCE_CHOICE_WEB_FORM
from ..models import WebFormSource
from ..literals import SOURCE_CHOICE_WEB_FORM, SOURCE_UNCOMPRESS_CHOICE_Y
from ..models.watch_folder_sources import WatchFolderSource
from ..models.webform_sources import WebFormSource
from .literals import TEST_SOURCE_LABEL, TEST_SOURCE_UNCOMPRESS_N
@@ -39,3 +40,13 @@ class SourceViewTestMixin(object):
viewname='sources:setup_source_delete',
kwargs={'pk': self.test_source.pk}
)
class WatchFolderTestMixin(object):
def _create_test_watchfolder(self):
self.test_watch_folder = WatchFolderSource.objects.create(
document_type=self.test_document_type,
folder_path=self.temporary_directory,
include_subdirectories=False,
uncompress=SOURCE_UNCOMPRESS_CHOICE_Y
)

64
mayan/apps/sources/tests/test_models.py
View File

@@ -10,6 +10,8 @@ from pathlib2 import Path
from django.core import mail
from django.utils.encoding import force_text
from django_celery_beat.models import PeriodicTask
from mayan.apps.common.serialization import yaml_dump
from mayan.apps.documents.models import Document
from mayan.apps.documents.tests.base import GenericDocumentTestCase
@@ -23,7 +25,6 @@ from mayan.apps.storage.utils import mkdtemp
from ..literals import SOURCE_UNCOMPRESS_CHOICE_Y
from ..models.email_sources import EmailBaseModel, IMAPEmail, POP3Email
from ..models.watch_folder_sources import WatchFolderSource
from .literals import (
TEST_EMAIL_ATTACHMENT_AND_INLINE, TEST_EMAIL_BASE64_FILENAME,
@@ -32,7 +33,7 @@ from .literals import (
TEST_EMAIL_NO_CONTENT_TYPE_STRING, TEST_EMAIL_ZERO_LENGTH_ATTACHMENT,
TEST_WATCHFOLDER_SUBFOLDER
)
from .mixins import SourceTestMixin
from .mixins import SourceTestMixin, WatchFolderTestMixin
from .mocks import MockIMAPServer, MockPOP3Mailbox
@@ -278,6 +279,31 @@ class IMAPSourceTestCase(GenericDocumentTestCase):
)
class IntervalSourceTestCase(WatchFolderTestMixin, GenericDocumentTestCase):
auto_upload_document = False
def setUp(self):
super(IntervalSourceTestCase, self).setUp()
self.temporary_directory = mkdtemp()
def tearDown(self):
shutil.rmtree(self.temporary_directory)
super(IntervalSourceTestCase, self).tearDown()
def test_periodic_task_create(self):
periodic_task_count = PeriodicTask.objects.count()
self._create_test_watchfolder()
self.assertTrue(PeriodicTask.objects.count() > periodic_task_count)
def test_periodic_task_delete(self):
self._create_test_watchfolder()
periodic_task_count = PeriodicTask.objects.count()
self.test_document_type.delete()
self.assertTrue(PeriodicTask.objects.count() < periodic_task_count)
class POP3SourceTestCase(GenericDocumentTestCase):
auto_upload_document = False
@@ -295,17 +321,9 @@ class POP3SourceTestCase(GenericDocumentTestCase):
)
class WatchFolderTestCase(GenericDocumentTestCase):
class WatchFolderTestCase(WatchFolderTestMixin, GenericDocumentTestCase):
auto_upload_document = False
def _create_test_watchfolder(self):
return WatchFolderSource.objects.create(
document_type=self.test_document_type,
folder_path=self.temporary_directory,
include_subdirectories=False,
uncompress=SOURCE_UNCOMPRESS_CHOICE_Y
)
def setUp(self):
super(WatchFolderTestCase, self).setUp()
self.temporary_directory = mkdtemp()
@@ -315,27 +333,27 @@ class WatchFolderTestCase(GenericDocumentTestCase):
super(WatchFolderTestCase, self).tearDown()
def test_subfolder_support_disabled(self):
watch_folder = self._create_test_watchfolder()
self._create_test_watchfolder()
test_path = Path(self.temporary_directory)
test_subfolder = test_path.joinpath(TEST_WATCHFOLDER_SUBFOLDER)
test_subfolder.mkdir()
shutil.copy(TEST_SMALL_DOCUMENT_PATH, force_text(test_subfolder))
watch_folder.check_source()
self.test_watch_folder.check_source()
self.assertEqual(Document.objects.count(), 0)
def test_subfolder_support_enabled(self):
watch_folder = self._create_test_watchfolder()
watch_folder.include_subdirectories = True
watch_folder.save()
self._create_test_watchfolder()
self.test_watch_folder.include_subdirectories = True
self.test_watch_folder.save()
test_path = Path(self.temporary_directory)
test_subfolder = test_path.joinpath(TEST_WATCHFOLDER_SUBFOLDER)
test_subfolder.mkdir()
shutil.copy(TEST_SMALL_DOCUMENT_PATH, force_text(test_subfolder))
watch_folder.check_source()
self.test_watch_folder.check_source()
self.assertEqual(Document.objects.count(), 1)
document = Document.objects.first()
@@ -353,10 +371,10 @@ class WatchFolderTestCase(GenericDocumentTestCase):
Non-ASCII chars in document name failing in upload via watch folder
gh-issue #163 https://github.com/mayan-edms/mayan-edms/issues/163
"""
watch_folder = self._create_test_watchfolder()
self._create_test_watchfolder()
shutil.copy(TEST_NON_ASCII_DOCUMENT_PATH, self.temporary_directory)
watch_folder.check_source()
self.test_watch_folder.check_source()
self.assertEqual(Document.objects.count(), 1)
document = Document.objects.first()
@@ -373,12 +391,12 @@ class WatchFolderTestCase(GenericDocumentTestCase):
"""
Test Non-ASCII named documents inside Non-ASCII named compressed file
"""
watch_folder = self._create_test_watchfolder()
self._create_test_watchfolder()
shutil.copy(
TEST_NON_ASCII_COMPRESSED_DOCUMENT_PATH, self.temporary_directory
)
watch_folder.check_source()
self.test_watch_folder.check_source()
self.assertEqual(Document.objects.count(), 1)
document = Document.objects.first()
@@ -392,7 +410,7 @@ class WatchFolderTestCase(GenericDocumentTestCase):
self.assertEqual(document.page_count, 1)
def test_locking_support(self):
watch_folder = self._create_test_watchfolder()
self._create_test_watchfolder()
shutil.copy(
TEST_SMALL_DOCUMENT_PATH, self.temporary_directory
@@ -404,7 +422,7 @@ class WatchFolderTestCase(GenericDocumentTestCase):
with path_test_file.open(mode='rb+') as file_object:
fcntl.lockf(file_object, fcntl.LOCK_EX | fcntl.LOCK_NB)
process = Process(target=watch_folder.check_source)
process = Process(target=self.test_watch_folder.check_source)
process.start()
process.join()