paradizelost/paperless-ngx
1
0
Fork 0
mirror of https://github.com/paperless-ngx/paperless-ngx.git synced 2025-07-28 18:24:38 -05:00
Code Issues Packages Projects Releases Wiki Activity

Format Python code with black

Browse Source
This commit is contained in:
kpj
2022-02-27 15:26:41 +01:00
parent f0ffc69010
commit c56cb25b5f
136 changed files with 6142 additions and 3811 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/documents/tests/factories.py
View File

@@ -5,7 +5,6 @@ from ..models import Document, Correspondent
class CorrespondentFactory(DjangoModelFactory):
class Meta:
model = Correspondent
@@ -13,6 +12,5 @@ class CorrespondentFactory(DjangoModelFactory):
class DocumentFactory(DjangoModelFactory):
class Meta:
model = Document

11
src/documents/tests/test_admin.py
View File

@@ -11,7 +11,6 @@ from documents.tests.utils import DirectoriesMixin
class TestDocumentAdmin(DirectoriesMixin, TestCase):
def get_document_from_index(self, doc):
ix = index.open_index()
with ix.searcher() as searcher:
@@ -27,7 +26,7 @@ class TestDocumentAdmin(DirectoriesMixin, TestCase):
doc.title = "new title"
self.doc_admin.save_model(None, doc, None, None)
self.assertEqual(Document.objects.get(id=doc.id).title, "new title")
self.assertEqual(self.get_document_from_index(doc)['id'], doc.id)
self.assertEqual(self.get_document_from_index(doc)["id"], doc.id)
def test_delete_model(self):
doc = Document.objects.create(title="test")
@@ -42,7 +41,9 @@ class TestDocumentAdmin(DirectoriesMixin, TestCase):
def test_delete_queryset(self):
docs = []
for i in range(42):
doc = Document.objects.create(title="Many documents with the same title", checksum=f"{i:02}")
doc = Document.objects.create(
title="Many documents with the same title", checksum=f"{i:02}"
)
docs.append(doc)
index.add_or_update_document(doc)
@@ -59,5 +60,7 @@ class TestDocumentAdmin(DirectoriesMixin, TestCase):
self.assertIsNone(self.get_document_from_index(doc))
def test_created(self):
doc = Document.objects.create(title="test", created=timezone.datetime(2020, 4, 12))
doc = Document.objects.create(
title="test", created=timezone.datetime(2020, 4, 12)
)
self.assertEqual(self.doc_admin.created_(doc), "2020-04-12")

1484
src/documents/tests/test_api.py
View File

File diff suppressed because it is too large Load Diff

14
src/documents/tests/test_checks.py
View File

@@ -11,7 +11,6 @@ from ..models import Document
class ChecksTestCase(TestCase):
def test_changed_password_check_empty_db(self):
self.assertEqual(changed_password_check(None), [])
@@ -23,8 +22,15 @@ class ChecksTestCase(TestCase):
self.assertEqual(parser_check(None), [])
with mock.patch('documents.checks.document_consumer_declaration.send') as m:
with mock.patch("documents.checks.document_consumer_declaration.send") as m:
m.return_value = []
self.assertEqual(parser_check(None), [Error("No parsers found. This is a bug. The consumer won't be "
"able to consume any documents without parsers.")])
self.assertEqual(
parser_check(None),
[
Error(
"No parsers found. This is a bug. The consumer won't be "
"able to consume any documents without parsers."
)
],
)

197
src/documents/tests/test_classifier.py
View File

@@ -7,30 +7,60 @@ import pytest
from django.conf import settings
from django.test import TestCase, override_settings
from documents.classifier import DocumentClassifier, IncompatibleClassifierVersionError, load_classifier
from documents.classifier import (
DocumentClassifier,
IncompatibleClassifierVersionError,
load_classifier,
)
from documents.models import Correspondent, Document, Tag, DocumentType
from documents.tests.utils import DirectoriesMixin
class TestClassifier(DirectoriesMixin, TestCase):
def setUp(self):
super(TestClassifier, self).setUp()
self.classifier = DocumentClassifier()
def generate_test_data(self):
self.c1 = Correspondent.objects.create(name="c1", matching_algorithm=Correspondent.MATCH_AUTO)
self.c1 = Correspondent.objects.create(
name="c1", matching_algorithm=Correspondent.MATCH_AUTO
)
self.c2 = Correspondent.objects.create(name="c2")
self.c3 = Correspondent.objects.create(name="c3", matching_algorithm=Correspondent.MATCH_AUTO)
self.t1 = Tag.objects.create(name="t1", matching_algorithm=Tag.MATCH_AUTO, pk=12)
self.t2 = Tag.objects.create(name="t2", matching_algorithm=Tag.MATCH_ANY, pk=34, is_inbox_tag=True)
self.t3 = Tag.objects.create(name="t3", matching_algorithm=Tag.MATCH_AUTO, pk=45)
self.dt = DocumentType.objects.create(name="dt", matching_algorithm=DocumentType.MATCH_AUTO)
self.dt2 = DocumentType.objects.create(name="dt2", matching_algorithm=DocumentType.MATCH_AUTO)
self.c3 = Correspondent.objects.create(
name="c3", matching_algorithm=Correspondent.MATCH_AUTO
)
self.t1 = Tag.objects.create(
name="t1", matching_algorithm=Tag.MATCH_AUTO, pk=12
)
self.t2 = Tag.objects.create(
name="t2", matching_algorithm=Tag.MATCH_ANY, pk=34, is_inbox_tag=True
)
self.t3 = Tag.objects.create(
name="t3", matching_algorithm=Tag.MATCH_AUTO, pk=45
)
self.dt = DocumentType.objects.create(
name="dt", matching_algorithm=DocumentType.MATCH_AUTO
)
self.dt2 = DocumentType.objects.create(
name="dt2", matching_algorithm=DocumentType.MATCH_AUTO
)
self.doc1 = Document.objects.create(title="doc1", content="this is a document from c1", correspondent=self.c1, checksum="A", document_type=self.dt)
self.doc2 = Document.objects.create(title="doc1", content="this is another document, but from c2", correspondent=self.c2, checksum="B")
self.doc_inbox = Document.objects.create(title="doc235", content="aa", checksum="C")
self.doc1 = Document.objects.create(
title="doc1",
content="this is a document from c1",
correspondent=self.c1,
checksum="A",
document_type=self.dt,
)
self.doc2 = Document.objects.create(
title="doc1",
content="this is another document, but from c2",
correspondent=self.c2,
checksum="B",
)
self.doc_inbox = Document.objects.create(
title="doc235", content="aa", checksum="C"
)
self.doc1.tags.add(self.t1)
self.doc2.tags.add(self.t1)
@@ -59,17 +89,29 @@ class TestClassifier(DirectoriesMixin, TestCase):
def testTrain(self):
self.generate_test_data()
self.classifier.train()
self.assertListEqual(list(self.classifier.correspondent_classifier.classes_), [-1, self.c1.pk])
self.assertListEqual(list(self.classifier.tags_binarizer.classes_), [self.t1.pk, self.t3.pk])
self.assertListEqual(
list(self.classifier.correspondent_classifier.classes_), [-1, self.c1.pk]
)
self.assertListEqual(
list(self.classifier.tags_binarizer.classes_), [self.t1.pk, self.t3.pk]
)
def testPredict(self):
self.generate_test_data()
self.classifier.train()
self.assertEqual(self.classifier.predict_correspondent(self.doc1.content), self.c1.pk)
self.assertEqual(
self.classifier.predict_correspondent(self.doc1.content), self.c1.pk
)
self.assertEqual(self.classifier.predict_correspondent(self.doc2.content), None)
self.assertListEqual(self.classifier.predict_tags(self.doc1.content), [self.t1.pk])
self.assertListEqual(self.classifier.predict_tags(self.doc2.content), [self.t1.pk, self.t3.pk])
self.assertEqual(self.classifier.predict_document_type(self.doc1.content), self.dt.pk)
self.assertListEqual(
self.classifier.predict_tags(self.doc1.content), [self.t1.pk]
)
self.assertListEqual(
self.classifier.predict_tags(self.doc2.content), [self.t1.pk, self.t3.pk]
)
self.assertEqual(
self.classifier.predict_document_type(self.doc1.content), self.dt.pk
)
self.assertEqual(self.classifier.predict_document_type(self.doc2.content), None)
def testDatasetHashing(self):
@@ -90,7 +132,9 @@ class TestClassifier(DirectoriesMixin, TestCase):
classifier2 = DocumentClassifier()
current_ver = DocumentClassifier.FORMAT_VERSION
with mock.patch("documents.classifier.DocumentClassifier.FORMAT_VERSION", current_ver+1):
with mock.patch(
"documents.classifier.DocumentClassifier.FORMAT_VERSION", current_ver + 1
):
# assure that we won't load old classifiers.
self.assertRaises(IncompatibleClassifierVersionError, classifier2.load)
@@ -112,7 +156,9 @@ class TestClassifier(DirectoriesMixin, TestCase):
new_classifier.load()
self.assertFalse(new_classifier.train())
@override_settings(MODEL_FILE=os.path.join(os.path.dirname(__file__), "data", "model.pickle"))
@override_settings(
MODEL_FILE=os.path.join(os.path.dirname(__file__), "data", "model.pickle")
)
def test_load_and_classify(self):
self.generate_test_data()
@@ -122,38 +168,67 @@ class TestClassifier(DirectoriesMixin, TestCase):
self.assertCountEqual(new_classifier.predict_tags(self.doc2.content), [45, 12])
def test_one_correspondent_predict(self):
c1 = Correspondent.objects.create(name="c1", matching_algorithm=Correspondent.MATCH_AUTO)
doc1 = Document.objects.create(title="doc1", content="this is a document from c1", correspondent=c1, checksum="A")
c1 = Correspondent.objects.create(
name="c1", matching_algorithm=Correspondent.MATCH_AUTO
)
doc1 = Document.objects.create(
title="doc1",
content="this is a document from c1",
correspondent=c1,
checksum="A",
)
self.classifier.train()
self.assertEqual(self.classifier.predict_correspondent(doc1.content), c1.pk)
def test_one_correspondent_predict_manydocs(self):
c1 = Correspondent.objects.create(name="c1", matching_algorithm=Correspondent.MATCH_AUTO)
doc1 = Document.objects.create(title="doc1", content="this is a document from c1", correspondent=c1, checksum="A")
doc2 = Document.objects.create(title="doc2", content="this is a document from noone", checksum="B")
c1 = Correspondent.objects.create(
name="c1", matching_algorithm=Correspondent.MATCH_AUTO
)
doc1 = Document.objects.create(
title="doc1",
content="this is a document from c1",
correspondent=c1,
checksum="A",
)
doc2 = Document.objects.create(
title="doc2", content="this is a document from noone", checksum="B"
)
self.classifier.train()
self.assertEqual(self.classifier.predict_correspondent(doc1.content), c1.pk)
self.assertIsNone(self.classifier.predict_correspondent(doc2.content))
def test_one_type_predict(self):
dt = DocumentType.objects.create(name="dt", matching_algorithm=DocumentType.MATCH_AUTO)
dt = DocumentType.objects.create(
name="dt", matching_algorithm=DocumentType.MATCH_AUTO
)
doc1 = Document.objects.create(title="doc1", content="this is a document from c1",
checksum="A", document_type=dt)
doc1 = Document.objects.create(
title="doc1",
content="this is a document from c1",
checksum="A",
document_type=dt,
)
self.classifier.train()
self.assertEqual(self.classifier.predict_document_type(doc1.content), dt.pk)
def test_one_type_predict_manydocs(self):
dt = DocumentType.objects.create(name="dt", matching_algorithm=DocumentType.MATCH_AUTO)
dt = DocumentType.objects.create(
name="dt", matching_algorithm=DocumentType.MATCH_AUTO
)
doc1 = Document.objects.create(title="doc1", content="this is a document from c1",
checksum="A", document_type=dt)
doc1 = Document.objects.create(
title="doc1",
content="this is a document from c1",
checksum="A",
document_type=dt,
)
doc2 = Document.objects.create(title="doc1", content="this is a document from c2",
checksum="B")
doc2 = Document.objects.create(
title="doc1", content="this is a document from c2", checksum="B"
)
self.classifier.train()
self.assertEqual(self.classifier.predict_document_type(doc1.content), dt.pk)
@@ -162,7 +237,9 @@ class TestClassifier(DirectoriesMixin, TestCase):
def test_one_tag_predict(self):
t1 = Tag.objects.create(name="t1", matching_algorithm=Tag.MATCH_AUTO, pk=12)
doc1 = Document.objects.create(title="doc1", content="this is a document from c1", checksum="A")
doc1 = Document.objects.create(
title="doc1", content="this is a document from c1", checksum="A"
)
doc1.tags.add(t1)
self.classifier.train()
@@ -171,7 +248,9 @@ class TestClassifier(DirectoriesMixin, TestCase):
def test_one_tag_predict_unassigned(self):
t1 = Tag.objects.create(name="t1", matching_algorithm=Tag.MATCH_AUTO, pk=12)
doc1 = Document.objects.create(title="doc1", content="this is a document from c1", checksum="A")
doc1 = Document.objects.create(
title="doc1", content="this is a document from c1", checksum="A"
)
self.classifier.train()
self.assertListEqual(self.classifier.predict_tags(doc1.content), [])
@@ -180,7 +259,9 @@ class TestClassifier(DirectoriesMixin, TestCase):
t1 = Tag.objects.create(name="t1", matching_algorithm=Tag.MATCH_AUTO, pk=12)
t2 = Tag.objects.create(name="t2", matching_algorithm=Tag.MATCH_AUTO, pk=121)
doc4 = Document.objects.create(title="doc1", content="this is a document from c4", checksum="D")
doc4 = Document.objects.create(
title="doc1", content="this is a document from c4", checksum="D"
)
doc4.tags.add(t1)
doc4.tags.add(t2)
@@ -191,10 +272,18 @@ class TestClassifier(DirectoriesMixin, TestCase):
t1 = Tag.objects.create(name="t1", matching_algorithm=Tag.MATCH_AUTO, pk=12)
t2 = Tag.objects.create(name="t2", matching_algorithm=Tag.MATCH_AUTO, pk=121)
doc1 = Document.objects.create(title="doc1", content="this is a document from c1", checksum="A")
doc2 = Document.objects.create(title="doc1", content="this is a document from c2", checksum="B")
doc3 = Document.objects.create(title="doc1", content="this is a document from c3", checksum="C")
doc4 = Document.objects.create(title="doc1", content="this is a document from c4", checksum="D")
doc1 = Document.objects.create(
title="doc1", content="this is a document from c1", checksum="A"
)
doc2 = Document.objects.create(
title="doc1", content="this is a document from c2", checksum="B"
)
doc3 = Document.objects.create(
title="doc1", content="this is a document from c3", checksum="C"
)
doc4 = Document.objects.create(
title="doc1", content="this is a document from c4", checksum="D"
)
doc1.tags.add(t1)
doc2.tags.add(t2)
@@ -210,8 +299,12 @@ class TestClassifier(DirectoriesMixin, TestCase):
def test_one_tag_predict_multi(self):
t1 = Tag.objects.create(name="t1", matching_algorithm=Tag.MATCH_AUTO, pk=12)
doc1 = Document.objects.create(title="doc1", content="this is a document from c1", checksum="A")
doc2 = Document.objects.create(title="doc2", content="this is a document from c2", checksum="B")
doc1 = Document.objects.create(
title="doc1", content="this is a document from c1", checksum="A"
)
doc2 = Document.objects.create(
title="doc2", content="this is a document from c2", checksum="B"
)
doc1.tags.add(t1)
doc2.tags.add(t1)
@@ -222,8 +315,12 @@ class TestClassifier(DirectoriesMixin, TestCase):
def test_one_tag_predict_multi_2(self):
t1 = Tag.objects.create(name="t1", matching_algorithm=Tag.MATCH_AUTO, pk=12)
doc1 = Document.objects.create(title="doc1", content="this is a document from c1", checksum="A")
doc2 = Document.objects.create(title="doc2", content="this is a document from c2", checksum="B")
doc1 = Document.objects.create(
title="doc1", content="this is a document from c1", checksum="A"
)
doc2 = Document.objects.create(
title="doc2", content="this is a document from c2", checksum="B"
)
doc1.tags.add(t1)
self.classifier.train()
@@ -240,9 +337,15 @@ class TestClassifier(DirectoriesMixin, TestCase):
self.assertIsNotNone(load_classifier())
load.assert_called_once()
@override_settings(CACHES={'default': {'BACKEND': 'django.core.cache.backends.locmem.LocMemCache'}})
@override_settings(MODEL_FILE=os.path.join(os.path.dirname(__file__), "data", "model.pickle"))
@pytest.mark.skip(reason="Disabled caching due to high memory usage - need to investigate.")
@override_settings(
CACHES={"default": {"BACKEND": "django.core.cache.backends.locmem.LocMemCache"}}
)
@override_settings(
MODEL_FILE=os.path.join(os.path.dirname(__file__), "data", "model.pickle")
)
@pytest.mark.skip(
reason="Disabled caching due to high memory usage - need to investigate."
)
def test_load_classifier_cached(self):
classifier = load_classifier()
self.assertIsNotNone(classifier)

232
src/documents/tests/test_consumer.py
View File

@@ -31,21 +31,14 @@ class TestAttributes(TestCase):
self.assertEqual(tuple([t.name for t in file_info.tags]), tags, filename)
def test_guess_attributes_from_name_when_title_starts_with_dash(self):
self._test_guess_attributes_from_name(
'- weird but should not break.pdf',
None,
'- weird but should not break',
()
"- weird but should not break.pdf", None, "- weird but should not break", ()
)
def test_guess_attributes_from_name_when_title_ends_with_dash(self):
self._test_guess_attributes_from_name(
'weird but should not break -.pdf',
None,
'weird but should not break -',
()
"weird but should not break -.pdf", None, "weird but should not break -", ()
)
@@ -55,19 +48,13 @@ class TestFieldPermutations(TestCase):
"20150102030405Z",
"20150102Z",
)
valid_correspondents = [
"timmy",
"Dr. McWheelie",
"Dash Gor-don",
"ο Θερμαστής",
""
]
valid_correspondents = ["timmy", "Dr. McWheelie", "Dash Gor-don", "ο Θερμαστής", ""]
valid_titles = ["title", "Title w Spaces", "Title a-dash", "Τίτλος", ""]
valid_tags = ["tag", "tig,tag", "tag1,tag2,tag-3"]
def _test_guessed_attributes(self, filename, created=None,
correspondent=None, title=None,
tags=None):
def _test_guessed_attributes(
self, filename, created=None, correspondent=None, title=None, tags=None
):
info = FileInfo.from_filename(filename)
@@ -92,13 +79,10 @@ class TestFieldPermutations(TestCase):
if tags is None:
self.assertEqual(info.tags, (), filename)
else:
self.assertEqual(
[t.name for t in info.tags], tags.split(','),
filename
)
self.assertEqual([t.name for t in info.tags], tags.split(","), filename)
def test_just_title(self):
template = '{title}.pdf'
template = "{title}.pdf"
for title in self.valid_titles:
spec = dict(title=title)
filename = template.format(**spec)
@@ -109,12 +93,8 @@ class TestFieldPermutations(TestCase):
for created in self.valid_dates:
for title in self.valid_titles:
spec = {
"created": created,
"title": title
}
self._test_guessed_attributes(
template.format(**spec), **spec)
spec = {"created": created, "title": title}
self._test_guessed_attributes(template.format(**spec), **spec)
def test_invalid_date_format(self):
info = FileInfo.from_filename("06112017Z - title.pdf")
@@ -127,7 +107,7 @@ class TestFieldPermutations(TestCase):
all_patt = re.compile("^.*$")
none_patt = re.compile("$a")
exact_patt = re.compile("^([a-z0-9,]+)_(\\d{8})_(\\d{6})_([0-9]+)\\.")
repl1 = " - \\4 - \\1." # (empty) corrspondent, title and tags
repl1 = " - \\4 - \\1." # (empty) corrspondent, title and tags
repl2 = "\\2Z - " + repl1 # creation date + repl1
# No transformations configured (= default)
@@ -137,36 +117,37 @@ class TestFieldPermutations(TestCase):
self.assertIsNone(info.created)
# Pattern doesn't match (filename unaltered)
with self.settings(
FILENAME_PARSE_TRANSFORMS=[(none_patt, "none.gif")]):
with self.settings(FILENAME_PARSE_TRANSFORMS=[(none_patt, "none.gif")]):
info = FileInfo.from_filename(filename)
self.assertEqual(info.title, "tag1,tag2_20190908_180610_0001")
# Simple transformation (match all)
with self.settings(
FILENAME_PARSE_TRANSFORMS=[(all_patt, "all.gif")]):
with self.settings(FILENAME_PARSE_TRANSFORMS=[(all_patt, "all.gif")]):
info = FileInfo.from_filename(filename)
self.assertEqual(info.title, "all")
# Multiple transformations configured (first pattern matches)
with self.settings(
FILENAME_PARSE_TRANSFORMS=[
(all_patt, "all.gif"),
(all_patt, "anotherall.gif")]):
FILENAME_PARSE_TRANSFORMS=[
(all_patt, "all.gif"),
(all_patt, "anotherall.gif"),
]
):
info = FileInfo.from_filename(filename)
self.assertEqual(info.title, "all")
# Multiple transformations configured (second pattern matches)
with self.settings(
FILENAME_PARSE_TRANSFORMS=[
(none_patt, "none.gif"),
(all_patt, "anotherall.gif")]):
FILENAME_PARSE_TRANSFORMS=[
(none_patt, "none.gif"),
(all_patt, "anotherall.gif"),
]
):
info = FileInfo.from_filename(filename)
self.assertEqual(info.title, "anotherall")
class DummyParser(DocumentParser):
def get_thumbnail(self, document_path, mime_type, file_name=None):
# not important during tests
raise NotImplementedError()
@@ -184,7 +165,6 @@ class DummyParser(DocumentParser):
class CopyParser(DocumentParser):
def get_thumbnail(self, document_path, mime_type, file_name=None):
return self.fake_thumb
@@ -202,7 +182,6 @@ class CopyParser(DocumentParser):
class FaultyParser(DocumentParser):
def get_thumbnail(self, document_path, mime_type, file_name=None):
# not important during tests
raise NotImplementedError()
@@ -233,8 +212,15 @@ def fake_magic_from_file(file, mime=False):
@mock.patch("documents.consumer.magic.from_file", fake_magic_from_file)
class TestConsumer(DirectoriesMixin, TestCase):
def _assert_first_last_send_progress(self, first_status="STARTING", last_status="SUCCESS", first_progress=0, first_progress_max=100, last_progress=100, last_progress_max=100):
def _assert_first_last_send_progress(
self,
first_status="STARTING",
last_status="SUCCESS",
first_progress=0,
first_progress_max=100,
last_progress=100,
last_progress_max=100,
):
self._send_progress.assert_called()
@@ -243,13 +229,17 @@ class TestConsumer(DirectoriesMixin, TestCase):
self.assertEqual(args[1], first_progress_max)
self.assertEqual(args[2], first_status)
args, kwargs = self._send_progress.call_args_list[len(self._send_progress.call_args_list) - 1]
args, kwargs = self._send_progress.call_args_list[
len(self._send_progress.call_args_list) - 1
]
self.assertEqual(args[0], last_progress)
self.assertEqual(args[1], last_progress_max)
self.assertEqual(args[2], last_status)
def make_dummy_parser(self, logging_group, progress_callback=None):
return DummyParser(logging_group, self.dirs.scratch_dir, self.get_test_archive_file())
return DummyParser(
logging_group, self.dirs.scratch_dir, self.get_test_archive_file()
)
def make_faulty_parser(self, logging_group, progress_callback=None):
return FaultyParser(logging_group, self.dirs.scratch_dir)
@@ -259,11 +249,16 @@ class TestConsumer(DirectoriesMixin, TestCase):
patcher = mock.patch("documents.parsers.document_consumer_declaration.send")
m = patcher.start()
m.return_value = [(None, {
"parser": self.make_dummy_parser,
"mime_types": {"application/pdf": ".pdf"},
"weight": 0
})]
m.return_value = [
(
None,
{
"parser": self.make_dummy_parser,
"mime_types": {"application/pdf": ".pdf"},
"weight": 0,
},
)
]
self.addCleanup(patcher.stop)
# this prevents websocket message reports during testing.
@@ -274,13 +269,21 @@ class TestConsumer(DirectoriesMixin, TestCase):
self.consumer = Consumer()
def get_test_file(self):
src = os.path.join(os.path.dirname(__file__), "samples", "documents", "originals", "0000001.pdf")
src = os.path.join(
os.path.dirname(__file__),
"samples",
"documents",
"originals",
"0000001.pdf",
)
dst = os.path.join(self.dirs.scratch_dir, "sample.pdf")
shutil.copy(src, dst)
return dst
def get_test_archive_file(self):
src = os.path.join(os.path.dirname(__file__), "samples", "documents", "archive", "0000001.pdf")
src = os.path.join(
os.path.dirname(__file__), "samples", "documents", "archive", "0000001.pdf"
)
dst = os.path.join(self.dirs.scratch_dir, "sample_archive.pdf")
shutil.copy(src, dst)
return dst
@@ -292,23 +295,19 @@ class TestConsumer(DirectoriesMixin, TestCase):
document = self.consumer.try_consume_file(filename)
self.assertEqual(document.content, "The Text")
self.assertEqual(document.title, os.path.splitext(os.path.basename(filename))[0])
self.assertEqual(
document.title, os.path.splitext(os.path.basename(filename))[0]
)
self.assertIsNone(document.correspondent)
self.assertIsNone(document.document_type)
self.assertEqual(document.filename, "0000001.pdf")
self.assertEqual(document.archive_filename, "0000001.pdf")
self.assertTrue(os.path.isfile(
document.source_path
))
self.assertTrue(os.path.isfile(document.source_path))
self.assertTrue(os.path.isfile(
document.thumbnail_path
))
self.assertTrue(os.path.isfile(document.thumbnail_path))
self.assertTrue(os.path.isfile(
document.archive_path
))
self.assertTrue(os.path.isfile(document.archive_path))
self.assertEqual(document.checksum, "42995833e01aea9b3edee44bbfdd7ce1")
self.assertEqual(document.archive_checksum, "62acb0bcbfbcaa62ca6ad3668e4e404b")
@@ -330,40 +329,45 @@ class TestConsumer(DirectoriesMixin, TestCase):
document = self.consumer.try_consume_file(filename)
self.assertTrue(os.path.isfile(
document.source_path
))
self.assertTrue(os.path.isfile(document.source_path))
self.assertFalse(os.path.isfile(shadow_file))
self.assertFalse(os.path.isfile(filename))
def testOverrideFilename(self):
filename = self.get_test_file()
override_filename = "Statement for November.pdf"
document = self.consumer.try_consume_file(filename, override_filename=override_filename)
document = self.consumer.try_consume_file(
filename, override_filename=override_filename
)
self.assertEqual(document.title, "Statement for November")
self._assert_first_last_send_progress()
def testOverrideTitle(self):
document = self.consumer.try_consume_file(self.get_test_file(), override_title="Override Title")
document = self.consumer.try_consume_file(
self.get_test_file(), override_title="Override Title"
)
self.assertEqual(document.title, "Override Title")
self._assert_first_last_send_progress()
def testOverrideCorrespondent(self):
c = Correspondent.objects.create(name="test")
document = self.consumer.try_consume_file(self.get_test_file(), override_correspondent_id=c.pk)
document = self.consumer.try_consume_file(
self.get_test_file(), override_correspondent_id=c.pk
)
self.assertEqual(document.correspondent.id, c.id)
self._assert_first_last_send_progress()
def testOverrideDocumentType(self):
dt = DocumentType.objects.create(name="test")
document = self.consumer.try_consume_file(self.get_test_file(), override_document_type_id=dt.pk)
document = self.consumer.try_consume_file(
self.get_test_file(), override_document_type_id=dt.pk
)
self.assertEqual(document.document_type.id, dt.id)
self._assert_first_last_send_progress()
@@ -371,7 +375,9 @@ class TestConsumer(DirectoriesMixin, TestCase):
t1 = Tag.objects.create(name="t1")
t2 = Tag.objects.create(name="t2")
t3 = Tag.objects.create(name="t3")
document = self.consumer.try_consume_file(self.get_test_file(), override_tag_ids=[t1.id, t3.id])
document = self.consumer.try_consume_file(
self.get_test_file(), override_tag_ids=[t1.id, t3.id]
)
self.assertIn(t1, document.tags.all())
self.assertNotIn(t2, document.tags.all())
@@ -384,7 +390,7 @@ class TestConsumer(DirectoriesMixin, TestCase):
ConsumerError,
"File not found",
self.consumer.try_consume_file,
"non-existing-file"
"non-existing-file",
)
self._assert_first_last_send_progress(last_status="FAILED")
@@ -396,7 +402,7 @@ class TestConsumer(DirectoriesMixin, TestCase):
ConsumerError,
"It is a duplicate",
self.consumer.try_consume_file,
self.get_test_file()
self.get_test_file(),
)
self._assert_first_last_send_progress(last_status="FAILED")
@@ -408,7 +414,7 @@ class TestConsumer(DirectoriesMixin, TestCase):
ConsumerError,
"It is a duplicate",
self.consumer.try_consume_file,
self.get_test_archive_file()
self.get_test_archive_file(),
)
self._assert_first_last_send_progress(last_status="FAILED")
@@ -425,25 +431,29 @@ class TestConsumer(DirectoriesMixin, TestCase):
ConsumerError,
"sample.pdf: Unsupported mime type application/pdf",
self.consumer.try_consume_file,
self.get_test_file()
self.get_test_file(),
)
self._assert_first_last_send_progress(last_status="FAILED")
@mock.patch("documents.parsers.document_consumer_declaration.send")
def testFaultyParser(self, m):
m.return_value = [(None, {
"parser": self.make_faulty_parser,
"mime_types": {"application/pdf": ".pdf"},
"weight": 0
})]
m.return_value = [
(
None,
{
"parser": self.make_faulty_parser,
"mime_types": {"application/pdf": ".pdf"},
"weight": 0,
},
)
]
self.assertRaisesMessage(
ConsumerError,
"sample.pdf: Error while consuming document sample.pdf: Does not compute.",
self.consumer.try_consume_file,
self.get_test_file()
self.get_test_file(),
)
self._assert_first_last_send_progress(last_status="FAILED")
@@ -457,7 +467,7 @@ class TestConsumer(DirectoriesMixin, TestCase):
ConsumerError,
"sample.pdf: The following error occured while consuming sample.pdf: NO.",
self.consumer.try_consume_file,
filename
filename,
)
self._assert_first_last_send_progress(last_status="FAILED")
@@ -491,7 +501,7 @@ class TestConsumer(DirectoriesMixin, TestCase):
filenames.insert(0, f)
return f
m.side_effect = lambda f, archive_filename = False: get_filename()
m.side_effect = lambda f, archive_filename=False: get_filename()
filename = self.get_test_file()
@@ -565,17 +575,37 @@ class TestConsumer(DirectoriesMixin, TestCase):
@override_settings(PAPERLESS_FILENAME_FORMAT="{title}")
@mock.patch("documents.parsers.document_consumer_declaration.send")
def test_similar_filenames(self, m):
shutil.copy(os.path.join(os.path.dirname(__file__), "samples", "simple.pdf"), os.path.join(settings.CONSUMPTION_DIR, "simple.pdf"))
shutil.copy(os.path.join(os.path.dirname(__file__), "samples", "simple.png"), os.path.join(settings.CONSUMPTION_DIR, "simple.png"))
shutil.copy(os.path.join(os.path.dirname(__file__), "samples", "simple-noalpha.png"), os.path.join(settings.CONSUMPTION_DIR, "simple.png.pdf"))
m.return_value = [(None, {
"parser": CopyParser,
"mime_types": {"application/pdf": ".pdf", "image/png": ".png"},
"weight": 0
})]
doc1 = self.consumer.try_consume_file(os.path.join(settings.CONSUMPTION_DIR, "simple.png"))
doc2 = self.consumer.try_consume_file(os.path.join(settings.CONSUMPTION_DIR, "simple.pdf"))
doc3 = self.consumer.try_consume_file(os.path.join(settings.CONSUMPTION_DIR, "simple.png.pdf"))
shutil.copy(
os.path.join(os.path.dirname(__file__), "samples", "simple.pdf"),
os.path.join(settings.CONSUMPTION_DIR, "simple.pdf"),
)
shutil.copy(
os.path.join(os.path.dirname(__file__), "samples", "simple.png"),
os.path.join(settings.CONSUMPTION_DIR, "simple.png"),
)
shutil.copy(
os.path.join(os.path.dirname(__file__), "samples", "simple-noalpha.png"),
os.path.join(settings.CONSUMPTION_DIR, "simple.png.pdf"),
)
m.return_value = [
(
None,
{
"parser": CopyParser,
"mime_types": {"application/pdf": ".pdf", "image/png": ".png"},
"weight": 0,
},
)
]
doc1 = self.consumer.try_consume_file(
os.path.join(settings.CONSUMPTION_DIR, "simple.png")
)
doc2 = self.consumer.try_consume_file(
os.path.join(settings.CONSUMPTION_DIR, "simple.pdf")
)
doc3 = self.consumer.try_consume_file(
os.path.join(settings.CONSUMPTION_DIR, "simple.png.pdf")
)
self.assertEqual(doc1.filename, "simple.png")
self.assertEqual(doc1.archive_filename, "simple.pdf")
@@ -588,7 +618,6 @@ class TestConsumer(DirectoriesMixin, TestCase):
class PreConsumeTestCase(TestCase):
@mock.patch("documents.consumer.Popen")
@override_settings(PRE_CONSUME_SCRIPT=None)
def test_no_pre_consume_script(self, m):
@@ -625,7 +654,6 @@ class PreConsumeTestCase(TestCase):
class PostConsumeTestCase(TestCase):
@mock.patch("documents.consumer.Popen")
@override_settings(POST_CONSUME_SCRIPT=None)
def test_no_post_consume_script(self, m):
@@ -662,7 +690,9 @@ class PostConsumeTestCase(TestCase):
with tempfile.NamedTemporaryFile() as script:
with override_settings(POST_CONSUME_SCRIPT=script.name):
c = Correspondent.objects.create(name="my_bank")
doc = Document.objects.create(title="Test", mime_type="application/pdf", correspondent=c)
doc = Document.objects.create(
title="Test", mime_type="application/pdf", correspondent=c
)
tag1 = Tag.objects.create(name="a")
tag2 = Tag.objects.create(name="b")
doc.tags.add(tag1)

67
src/documents/tests/test_date_parsing.py
View File

@@ -12,7 +12,9 @@ from documents.parsers import parse_date
class TestDate(TestCase):
SAMPLE_FILES = os.path.join(os.path.dirname(__file__), "../../paperless_tesseract/tests/samples")
SAMPLE_FILES = os.path.join(
os.path.dirname(__file__), "../../paperless_tesseract/tests/samples"
)
SCRATCH = "/tmp/paperless-tests-{}".format(str(uuid4())[:8])
def setUp(self):
@@ -38,24 +40,15 @@ class TestDate(TestCase):
date = parse_date("", text)
self.assertEqual(
date,
datetime.datetime(
2018, 2, 13, 0, 0,
tzinfo=tz.gettz(settings.TIME_ZONE)
)
datetime.datetime(2018, 2, 13, 0, 0, tzinfo=tz.gettz(settings.TIME_ZONE)),
)
def test_date_format_5(self):
text = (
"lorem ipsum 130218, 2018, 20180213 and lorem 13.02.2018 lorem "
"ipsum"
)
text = "lorem ipsum 130218, 2018, 20180213 and lorem 13.02.2018 lorem " "ipsum"
date = parse_date("", text)
self.assertEqual(
date,
datetime.datetime(
2018, 2, 13, 0, 0,
tzinfo=tz.gettz(settings.TIME_ZONE)
)
datetime.datetime(2018, 2, 13, 0, 0, tzinfo=tz.gettz(settings.TIME_ZONE)),
)
def test_date_format_6(self):
@@ -73,18 +66,11 @@ class TestDate(TestCase):
self.assertEqual(parse_date("", text), None)
def test_date_format_7(self):
text = (
"lorem ipsum\n"
"März 2019\n"
"lorem ipsum"
)
text = "lorem ipsum\n" "März 2019\n" "lorem ipsum"
date = parse_date("", text)
self.assertEqual(
date,
datetime.datetime(
2019, 3, 1, 0, 0,
tzinfo=tz.gettz(settings.TIME_ZONE)
)
datetime.datetime(2019, 3, 1, 0, 0, tzinfo=tz.gettz(settings.TIME_ZONE)),
)
def test_date_format_8(self):
@@ -102,26 +88,15 @@ class TestDate(TestCase):
)
self.assertEqual(
parse_date("", text),
datetime.datetime(
2020, 3, 1, 0, 0,
tzinfo=tz.gettz(settings.TIME_ZONE)
)
datetime.datetime(2020, 3, 1, 0, 0, tzinfo=tz.gettz(settings.TIME_ZONE)),
)
@override_settings(SCRATCH_DIR=SCRATCH)
def test_date_format_9(self):
text = (
"lorem ipsum\n"
"27. Nullmonth 2020\n"
"März 2020\n"
"lorem ipsum"
)
text = "lorem ipsum\n" "27. Nullmonth 2020\n" "März 2020\n" "lorem ipsum"
self.assertEqual(
parse_date("", text),
datetime.datetime(
2020, 3, 1, 0, 0,
tzinfo=tz.gettz(settings.TIME_ZONE)
)
datetime.datetime(2020, 3, 1, 0, 0, tzinfo=tz.gettz(settings.TIME_ZONE)),
)
def test_crazy_date_past(self, *args):
@@ -135,19 +110,17 @@ class TestDate(TestCase):
@override_settings(FILENAME_DATE_ORDER="YMD")
def test_filename_date_parse_invalid(self, *args):
self.assertIsNone(parse_date("/tmp/20 408000l 2475 - test.pdf", "No date in here"))
@override_settings(IGNORE_DATES=(datetime.date(2019, 11, 3), datetime.date(2020, 1, 17)))
def test_ignored_dates(self, *args):
text = (
"lorem ipsum 110319, 20200117 and lorem 13.02.2018 lorem "
"ipsum"
self.assertIsNone(
parse_date("/tmp/20 408000l 2475 - test.pdf", "No date in here")
)
@override_settings(
IGNORE_DATES=(datetime.date(2019, 11, 3), datetime.date(2020, 1, 17))
)
def test_ignored_dates(self, *args):
text = "lorem ipsum 110319, 20200117 and lorem 13.02.2018 lorem " "ipsum"
date = parse_date("", text)
self.assertEqual(
date,
datetime.datetime(
2018, 2, 13, 0, 0,
tzinfo=tz.gettz(settings.TIME_ZONE)
)
datetime.datetime(2018, 2, 13, 0, 0, tzinfo=tz.gettz(settings.TIME_ZONE)),
)

27
src/documents/tests/test_document_model.py
View File

@@ -10,7 +10,6 @@ from ..models import Document, Correspondent
class TestDocument(TestCase):
def setUp(self) -> None:
self.originals_dir = tempfile.mkdtemp()
self.thumb_dir = tempfile.mkdtemp()
@@ -30,7 +29,7 @@ class TestDocument(TestCase):
title="Title",
content="content",
checksum="checksum",
mime_type="application/pdf"
mime_type="application/pdf",
)
file_path = document.source_path
@@ -47,20 +46,36 @@ class TestDocument(TestCase):
def test_file_name(self):
doc = Document(mime_type="application/pdf", title="test", created=timezone.datetime(2020, 12, 25))
doc = Document(
mime_type="application/pdf",
title="test",
created=timezone.datetime(2020, 12, 25),
)
self.assertEqual(doc.get_public_filename(), "2020-12-25 test.pdf")
def test_file_name_jpg(self):
doc = Document(mime_type="image/jpeg", title="test", created=timezone.datetime(2020, 12, 25))
doc = Document(
mime_type="image/jpeg",
title="test",
created=timezone.datetime(2020, 12, 25),
)
self.assertEqual(doc.get_public_filename(), "2020-12-25 test.jpg")
def test_file_name_unknown(self):
doc = Document(mime_type="application/zip", title="test", created=timezone.datetime(2020, 12, 25))
doc = Document(
mime_type="application/zip",
title="test",
created=timezone.datetime(2020, 12, 25),
)
self.assertEqual(doc.get_public_filename(), "2020-12-25 test.zip")
def test_file_name_invalid_type(self):
doc = Document(mime_type="image/jpegasd", title="test", created=timezone.datetime(2020, 12, 25))
doc = Document(
mime_type="image/jpegasd",
title="test",
created=timezone.datetime(2020, 12, 25),
)
self.assertEqual(doc.get_public_filename(), "2020-12-25 test")

327
src/documents/tests/test_file_handling.py
View File

@@ -13,13 +13,16 @@ from django.test import TestCase, override_settings
from django.utils import timezone
from .utils import DirectoriesMixin
from ..file_handling import generate_filename, create_source_path_directory, delete_empty_directories, \
generate_unique_filename
from ..file_handling import (
generate_filename,
create_source_path_directory,
delete_empty_directories,
generate_unique_filename,
)
from ..models import Document, Correspondent, Tag, DocumentType
class TestFileHandling(DirectoriesMixin, TestCase):
@override_settings(PAPERLESS_FILENAME_FORMAT="")
def test_generate_source_filename(self):
document = Document()
@@ -30,8 +33,9 @@ class TestFileHandling(DirectoriesMixin, TestCase):
self.assertEqual(generate_filename(document), "{:07d}.pdf".format(document.pk))
document.storage_type = Document.STORAGE_TYPE_GPG
self.assertEqual(generate_filename(document),
"{:07d}.pdf.gpg".format(document.pk))
self.assertEqual(
generate_filename(document), "{:07d}.pdf.gpg".format(document.pk)
)
@override_settings(PAPERLESS_FILENAME_FORMAT="{correspondent}/{correspondent}")
def test_file_renaming(self):
@@ -41,7 +45,10 @@ class TestFileHandling(DirectoriesMixin, TestCase):
document.save()
# Test default source_path
self.assertEqual(document.source_path, settings.ORIGINALS_DIR + "/{:07d}.pdf".format(document.pk))
self.assertEqual(
document.source_path,
settings.ORIGINALS_DIR + "/{:07d}.pdf".format(document.pk),
)
document.filename = generate_filename(document)
@@ -51,8 +58,7 @@ class TestFileHandling(DirectoriesMixin, TestCase):
# Enable encryption and check again
document.storage_type = Document.STORAGE_TYPE_GPG
document.filename = generate_filename(document)
self.assertEqual(document.filename,
"none/none.pdf.gpg")
self.assertEqual(document.filename, "none/none.pdf.gpg")
document.save()
@@ -68,7 +74,9 @@ class TestFileHandling(DirectoriesMixin, TestCase):
# Check proper handling of files
self.assertEqual(os.path.isdir(settings.ORIGINALS_DIR + "/test"), True)
self.assertEqual(os.path.isdir(settings.ORIGINALS_DIR + "/none"), False)
self.assertEqual(os.path.isfile(settings.ORIGINALS_DIR + "/test/test.pdf.gpg"), True)
self.assertEqual(
os.path.isfile(settings.ORIGINALS_DIR + "/test/test.pdf.gpg"), True
)
@override_settings(PAPERLESS_FILENAME_FORMAT="{correspondent}/{correspondent}")
def test_file_renaming_missing_permissions(self):
@@ -79,13 +87,14 @@ class TestFileHandling(DirectoriesMixin, TestCase):
# Ensure that filename is properly generated
document.filename = generate_filename(document)
self.assertEqual(document.filename,
"none/none.pdf")
self.assertEqual(document.filename, "none/none.pdf")
create_source_path_directory(document.source_path)
Path(document.source_path).touch()
# Test source_path
self.assertEqual(document.source_path, settings.ORIGINALS_DIR + "/none/none.pdf")
self.assertEqual(
document.source_path, settings.ORIGINALS_DIR + "/none/none.pdf"
)
# Make the folder read- and execute-only (no writing and no renaming)
os.chmod(settings.ORIGINALS_DIR + "/none", 0o555)
@@ -95,7 +104,9 @@ class TestFileHandling(DirectoriesMixin, TestCase):
document.save()
# Check proper handling of files
self.assertEqual(os.path.isfile(settings.ORIGINALS_DIR + "/none/none.pdf"), True)
self.assertEqual(
os.path.isfile(settings.ORIGINALS_DIR + "/none/none.pdf"), True
)
self.assertEqual(document.filename, "none/none.pdf")
os.chmod(settings.ORIGINALS_DIR + "/none", 0o777)
@@ -103,7 +114,11 @@ class TestFileHandling(DirectoriesMixin, TestCase):
@override_settings(PAPERLESS_FILENAME_FORMAT="{correspondent}/{correspondent}")
def test_file_renaming_database_error(self):
document1 = Document.objects.create(mime_type="application/pdf", storage_type=Document.STORAGE_TYPE_UNENCRYPTED, checksum="AAAAA")
document1 = Document.objects.create(
mime_type="application/pdf",
storage_type=Document.STORAGE_TYPE_UNENCRYPTED,
checksum="AAAAA",
)
document = Document()
document.mime_type = "application/pdf"
@@ -113,8 +128,7 @@ class TestFileHandling(DirectoriesMixin, TestCase):
# Ensure that filename is properly generated
document.filename = generate_filename(document)
self.assertEqual(document.filename,
"none/none.pdf")
self.assertEqual(document.filename, "none/none.pdf")
create_source_path_directory(document.source_path)
Path(document.source_path).touch()
@@ -122,8 +136,7 @@ class TestFileHandling(DirectoriesMixin, TestCase):
self.assertTrue(os.path.isfile(document.source_path))
# Set a correspondent and save the document
document.correspondent = Correspondent.objects.get_or_create(
name="test")[0]
document.correspondent = Correspondent.objects.get_or_create(name="test")[0]
with mock.patch("documents.signals.handlers.Document.objects.filter") as m:
m.side_effect = DatabaseError()
@@ -131,7 +144,9 @@ class TestFileHandling(DirectoriesMixin, TestCase):
# Check proper handling of files
self.assertTrue(os.path.isfile(document.source_path))
self.assertEqual(os.path.isfile(settings.ORIGINALS_DIR + "/none/none.pdf"), True)
self.assertEqual(
os.path.isfile(settings.ORIGINALS_DIR + "/none/none.pdf"), True
)
self.assertEqual(document.filename, "none/none.pdf")
@override_settings(PAPERLESS_FILENAME_FORMAT="{correspondent}/{correspondent}")
@@ -143,8 +158,7 @@ class TestFileHandling(DirectoriesMixin, TestCase):
# Ensure that filename is properly generated
document.filename = generate_filename(document)
self.assertEqual(document.filename,
"none/none.pdf")
self.assertEqual(document.filename, "none/none.pdf")
create_source_path_directory(document.source_path)
Path(document.source_path).touch()
@@ -152,10 +166,15 @@ class TestFileHandling(DirectoriesMixin, TestCase):
# Ensure file deletion after delete
pk = document.pk
document.delete()
self.assertEqual(os.path.isfile(settings.ORIGINALS_DIR + "/none/none.pdf"), False)
self.assertEqual(
os.path.isfile(settings.ORIGINALS_DIR + "/none/none.pdf"), False
)
self.assertEqual(os.path.isdir(settings.ORIGINALS_DIR + "/none"), False)
@override_settings(PAPERLESS_FILENAME_FORMAT="{correspondent}/{correspondent}", TRASH_DIR=tempfile.mkdtemp())
@override_settings(
PAPERLESS_FILENAME_FORMAT="{correspondent}/{correspondent}",
TRASH_DIR=tempfile.mkdtemp(),
)
def test_document_delete_trash(self):
document = Document()
document.mime_type = "application/pdf"
@@ -164,8 +183,7 @@ class TestFileHandling(DirectoriesMixin, TestCase):
# Ensure that filename is properly generated
document.filename = generate_filename(document)
self.assertEqual(document.filename,
"none/none.pdf")
self.assertEqual(document.filename, "none/none.pdf")
create_source_path_directory(document.source_path)
Path(document.source_path).touch()
@@ -173,7 +191,9 @@ class TestFileHandling(DirectoriesMixin, TestCase):
# Ensure file was moved to trash after delete
self.assertEqual(os.path.isfile(settings.TRASH_DIR + "/none/none.pdf"), False)
document.delete()
self.assertEqual(os.path.isfile(settings.ORIGINALS_DIR + "/none/none.pdf"), False)
self.assertEqual(
os.path.isfile(settings.ORIGINALS_DIR + "/none/none.pdf"), False
)
self.assertEqual(os.path.isdir(settings.ORIGINALS_DIR + "/none"), False)
self.assertEqual(os.path.isfile(settings.TRASH_DIR + "/none.pdf"), True)
self.assertEqual(os.path.isfile(settings.TRASH_DIR + "/none_01.pdf"), False)
@@ -207,8 +227,7 @@ class TestFileHandling(DirectoriesMixin, TestCase):
# Ensure that filename is properly generated
document.filename = generate_filename(document)
self.assertEqual(document.filename,
"none/none.pdf")
self.assertEqual(document.filename, "none/none.pdf")
create_source_path_directory(document.source_path)
@@ -238,8 +257,18 @@ class TestFileHandling(DirectoriesMixin, TestCase):
@override_settings(PAPERLESS_FILENAME_FORMAT="{asn} - {title}")
def test_asn(self):
d1 = Document.objects.create(title="the_doc", mime_type="application/pdf", archive_serial_number=652, checksum="A")
d2 = Document.objects.create(title="the_doc", mime_type="application/pdf", archive_serial_number=None, checksum="B")
d1 = Document.objects.create(
title="the_doc",
mime_type="application/pdf",
archive_serial_number=652,
checksum="A",
)
d2 = Document.objects.create(
title="the_doc",
mime_type="application/pdf",
archive_serial_number=None,
checksum="B",
)
self.assertEqual(generate_filename(d1), "652 - the_doc.pdf")
self.assertEqual(generate_filename(d2), "none - the_doc.pdf")
@@ -256,8 +285,7 @@ class TestFileHandling(DirectoriesMixin, TestCase):
document.save()
# Ensure that filename is properly generated
self.assertEqual(generate_filename(document),
"demo.pdf")
self.assertEqual(generate_filename(document), "demo.pdf")
@override_settings(PAPERLESS_FILENAME_FORMAT="{tags[type]}")
def test_tags_with_dash(self):
@@ -272,8 +300,7 @@ class TestFileHandling(DirectoriesMixin, TestCase):
document.save()
# Ensure that filename is properly generated
self.assertEqual(generate_filename(document),
"demo.pdf")
self.assertEqual(generate_filename(document), "demo.pdf")
@override_settings(PAPERLESS_FILENAME_FORMAT="{tags[type]}")
def test_tags_malformed(self):
@@ -288,8 +315,7 @@ class TestFileHandling(DirectoriesMixin, TestCase):
document.save()
# Ensure that filename is properly generated
self.assertEqual(generate_filename(document),
"none.pdf")
self.assertEqual(generate_filename(document), "none.pdf")
@override_settings(PAPERLESS_FILENAME_FORMAT="{tags[0]}")
def test_tags_all(self):
@@ -303,8 +329,7 @@ class TestFileHandling(DirectoriesMixin, TestCase):
document.save()
# Ensure that filename is properly generated
self.assertEqual(generate_filename(document),
"demo.pdf")
self.assertEqual(generate_filename(document), "demo.pdf")
@override_settings(PAPERLESS_FILENAME_FORMAT="{tags[1]}")
def test_tags_out_of_bounds(self):
@@ -318,8 +343,7 @@ class TestFileHandling(DirectoriesMixin, TestCase):
document.save()
# Ensure that filename is properly generated
self.assertEqual(generate_filename(document),
"none.pdf")
self.assertEqual(generate_filename(document), "none.pdf")
@override_settings(PAPERLESS_FILENAME_FORMAT="{tags}")
def test_tags_without_args(self):
@@ -338,7 +362,9 @@ class TestFileHandling(DirectoriesMixin, TestCase):
self.assertEqual(generate_filename(doc), "doc1 tag1,tag2.pdf")
doc = Document.objects.create(title="doc2", checksum="B", mime_type="application/pdf")
doc = Document.objects.create(
title="doc2", checksum="B", mime_type="application/pdf"
)
self.assertEqual(generate_filename(doc), "doc2.pdf")
@@ -348,12 +374,19 @@ class TestFileHandling(DirectoriesMixin, TestCase):
doc.filename = generate_filename(doc)
doc.save()
self.assertEqual(doc.source_path, os.path.join(settings.ORIGINALS_DIR, "etc", "something", "doc1.pdf"))
self.assertEqual(
doc.source_path,
os.path.join(settings.ORIGINALS_DIR, "etc", "something", "doc1.pdf"),
)
@override_settings(PAPERLESS_FILENAME_FORMAT="{created_year}-{created_month}-{created_day}")
@override_settings(
PAPERLESS_FILENAME_FORMAT="{created_year}-{created_month}-{created_day}"
)
def test_created_year_month_day(self):
d1 = timezone.make_aware(datetime.datetime(2020, 3, 6, 1, 1, 1))
doc1 = Document.objects.create(title="doc1", mime_type="application/pdf", created=d1)
doc1 = Document.objects.create(
title="doc1", mime_type="application/pdf", created=d1
)
self.assertEqual(generate_filename(doc1), "2020-03-06.pdf")
@@ -361,10 +394,14 @@ class TestFileHandling(DirectoriesMixin, TestCase):
self.assertEqual(generate_filename(doc1), "2020-11-16.pdf")
@override_settings(PAPERLESS_FILENAME_FORMAT="{added_year}-{added_month}-{added_day}")
@override_settings(
PAPERLESS_FILENAME_FORMAT="{added_year}-{added_month}-{added_day}"
)
def test_added_year_month_day(self):
d1 = timezone.make_aware(datetime.datetime(232, 1, 9, 1, 1, 1))
doc1 = Document.objects.create(title="doc1", mime_type="application/pdf", added=d1)
doc1 = Document.objects.create(
title="doc1", mime_type="application/pdf", added=d1
)
self.assertEqual(generate_filename(doc1), "232-01-09.pdf")
@@ -372,7 +409,9 @@ class TestFileHandling(DirectoriesMixin, TestCase):
self.assertEqual(generate_filename(doc1), "2020-11-16.pdf")
@override_settings(PAPERLESS_FILENAME_FORMAT="{correspondent}/{correspondent}/{correspondent}")
@override_settings(
PAPERLESS_FILENAME_FORMAT="{correspondent}/{correspondent}/{correspondent}"
)
def test_nested_directory_cleanup(self):
document = Document()
document.mime_type = "application/pdf"
@@ -391,7 +430,9 @@ class TestFileHandling(DirectoriesMixin, TestCase):
pk = document.pk
document.delete()
self.assertEqual(os.path.isfile(settings.ORIGINALS_DIR + "/none/none/none.pdf"), False)
self.assertEqual(
os.path.isfile(settings.ORIGINALS_DIR + "/none/none/none.pdf"), False
)
self.assertEqual(os.path.isdir(settings.ORIGINALS_DIR + "/none/none"), False)
self.assertEqual(os.path.isdir(settings.ORIGINALS_DIR + "/none"), False)
self.assertEqual(os.path.isdir(settings.ORIGINALS_DIR), True)
@@ -414,12 +455,12 @@ class TestFileHandling(DirectoriesMixin, TestCase):
Path(os.path.join(tmp, "notempty", "file")).touch()
os.makedirs(os.path.join(tmp, "notempty", "empty"))
delete_empty_directories(os.path.join(tmp, "notempty", "empty"), root=settings.ORIGINALS_DIR)
delete_empty_directories(
os.path.join(tmp, "notempty", "empty"), root=settings.ORIGINALS_DIR
)
self.assertEqual(os.path.isdir(os.path.join(tmp, "notempty")), True)
self.assertEqual(os.path.isfile(
os.path.join(tmp, "notempty", "file")), True)
self.assertEqual(os.path.isdir(
os.path.join(tmp, "notempty", "empty")), False)
self.assertEqual(os.path.isfile(os.path.join(tmp, "notempty", "file")), True)
self.assertEqual(os.path.isdir(os.path.join(tmp, "notempty", "empty")), False)
@override_settings(PAPERLESS_FILENAME_FORMAT="{created/[title]")
def test_invalid_format(self):
@@ -441,8 +482,12 @@ class TestFileHandling(DirectoriesMixin, TestCase):
@override_settings(PAPERLESS_FILENAME_FORMAT="{title}")
def test_duplicates(self):
document = Document.objects.create(mime_type="application/pdf", title="qwe", checksum="A", pk=1)
document2 = Document.objects.create(mime_type="application/pdf", title="qwe", checksum="B", pk=2)
document = Document.objects.create(
mime_type="application/pdf", title="qwe", checksum="A", pk=1
)
document2 = Document.objects.create(
mime_type="application/pdf", title="qwe", checksum="B", pk=2
)
Path(document.source_path).touch()
Path(document2.source_path).touch()
document.filename = "0000001.pdf"
@@ -480,11 +525,17 @@ class TestFileHandling(DirectoriesMixin, TestCase):
self.assertTrue(os.path.isfile(document.source_path))
self.assertEqual(document2.filename, "qwe.pdf")
@override_settings(PAPERLESS_FILENAME_FORMAT="{title}")
@mock.patch("documents.signals.handlers.Document.objects.filter")
def test_no_update_without_change(self, m):
doc = Document.objects.create(title="document", filename="document.pdf", archive_filename="document.pdf", checksum="A", archive_checksum="B", mime_type="application/pdf")
doc = Document.objects.create(
title="document",
filename="document.pdf",
archive_filename="document.pdf",
checksum="A",
archive_checksum="B",
mime_type="application/pdf",
)
Path(doc.source_path).touch()
Path(doc.archive_path).touch()
@@ -493,16 +544,20 @@ class TestFileHandling(DirectoriesMixin, TestCase):
m.assert_not_called()
class TestFileHandlingWithArchive(DirectoriesMixin, TestCase):
@override_settings(PAPERLESS_FILENAME_FORMAT=None)
def test_create_no_format(self):
original = os.path.join(settings.ORIGINALS_DIR, "0000001.pdf")
archive = os.path.join(settings.ARCHIVE_DIR, "0000001.pdf")
Path(original).touch()
Path(archive).touch()
doc = Document.objects.create(mime_type="application/pdf", filename="0000001.pdf", checksum="A", archive_filename="0000001.pdf", archive_checksum="B")
doc = Document.objects.create(
mime_type="application/pdf",
filename="0000001.pdf",
checksum="A",
archive_filename="0000001.pdf",
archive_checksum="B",
)
self.assertTrue(os.path.isfile(original))
self.assertTrue(os.path.isfile(archive))
@@ -515,21 +570,39 @@ class TestFileHandlingWithArchive(DirectoriesMixin, TestCase):
archive = os.path.join(settings.ARCHIVE_DIR, "0000001.pdf")
Path(original).touch()
Path(archive).touch()
doc = Document.objects.create(mime_type="application/pdf", title="my_doc", filename="0000001.pdf", checksum="A", archive_checksum="B", archive_filename="0000001.pdf")
doc = Document.objects.create(
mime_type="application/pdf",
title="my_doc",
filename="0000001.pdf",
checksum="A",
archive_checksum="B",
archive_filename="0000001.pdf",
)
self.assertFalse(os.path.isfile(original))
self.assertFalse(os.path.isfile(archive))
self.assertTrue(os.path.isfile(doc.source_path))
self.assertTrue(os.path.isfile(doc.archive_path))
self.assertEqual(doc.source_path, os.path.join(settings.ORIGINALS_DIR, "none", "my_doc.pdf"))
self.assertEqual(doc.archive_path, os.path.join(settings.ARCHIVE_DIR, "none", "my_doc.pdf"))
self.assertEqual(
doc.source_path, os.path.join(settings.ORIGINALS_DIR, "none", "my_doc.pdf")
)
self.assertEqual(
doc.archive_path, os.path.join(settings.ARCHIVE_DIR, "none", "my_doc.pdf")
)
@override_settings(PAPERLESS_FILENAME_FORMAT="{correspondent}/{title}")
def test_move_archive_gone(self):
original = os.path.join(settings.ORIGINALS_DIR, "0000001.pdf")
archive = os.path.join(settings.ARCHIVE_DIR, "0000001.pdf")
Path(original).touch()
doc = Document.objects.create(mime_type="application/pdf", title="my_doc", filename="0000001.pdf", checksum="A", archive_checksum="B", archive_filename="0000001.pdf")
doc = Document.objects.create(
mime_type="application/pdf",
title="my_doc",
filename="0000001.pdf",
checksum="A",
archive_checksum="B",
archive_filename="0000001.pdf",
)
self.assertTrue(os.path.isfile(original))
self.assertFalse(os.path.isfile(archive))
@@ -545,7 +618,14 @@ class TestFileHandlingWithArchive(DirectoriesMixin, TestCase):
Path(archive).touch()
os.makedirs(os.path.join(settings.ARCHIVE_DIR, "none"))
Path(existing_archive_file).touch()
doc = Document.objects.create(mime_type="application/pdf", title="my_doc", filename="0000001.pdf", checksum="A", archive_checksum="B", archive_filename="0000001.pdf")
doc = Document.objects.create(
mime_type="application/pdf",
title="my_doc",
filename="0000001.pdf",
checksum="A",
archive_checksum="B",
archive_filename="0000001.pdf",
)
self.assertFalse(os.path.isfile(original))
self.assertFalse(os.path.isfile(archive))
@@ -561,8 +641,14 @@ class TestFileHandlingWithArchive(DirectoriesMixin, TestCase):
Path(original).touch()
Path(archive).touch()
doc = Document.objects.create(mime_type="application/pdf", title="document", filename="document_01.pdf", checksum="A",
archive_checksum="B", archive_filename="document.pdf")
doc = Document.objects.create(
mime_type="application/pdf",
title="document",
filename="document_01.pdf",
checksum="A",
archive_checksum="B",
archive_filename="document.pdf",
)
self.assertEqual(doc.filename, "document.pdf")
self.assertEqual(doc.archive_filename, "document.pdf")
@@ -577,8 +663,14 @@ class TestFileHandlingWithArchive(DirectoriesMixin, TestCase):
Path(original).touch()
Path(archive).touch()
doc = Document.objects.create(mime_type="application/pdf", title="document", filename="document.pdf", checksum="A",
archive_checksum="B", archive_filename="document_01.pdf")
doc = Document.objects.create(
mime_type="application/pdf",
title="document",
filename="document.pdf",
checksum="A",
archive_checksum="B",
archive_filename="document_01.pdf",
)
self.assertEqual(doc.filename, "document.pdf")
self.assertEqual(doc.archive_filename, "document.pdf")
@@ -589,7 +681,6 @@ class TestFileHandlingWithArchive(DirectoriesMixin, TestCase):
@override_settings(PAPERLESS_FILENAME_FORMAT="{correspondent}/{title}")
@mock.patch("documents.signals.handlers.os.rename")
def test_move_archive_error(self, m):
def fake_rename(src, dst):
if "archive" in src:
raise OSError()
@@ -603,7 +694,14 @@ class TestFileHandlingWithArchive(DirectoriesMixin, TestCase):
archive = os.path.join(settings.ARCHIVE_DIR, "0000001.pdf")
Path(original).touch()
Path(archive).touch()
doc = Document.objects.create(mime_type="application/pdf", title="my_doc", filename="0000001.pdf", checksum="A", archive_checksum="B", archive_filename="0000001.pdf")
doc = Document.objects.create(
mime_type="application/pdf",
title="my_doc",
filename="0000001.pdf",
checksum="A",
archive_checksum="B",
archive_filename="0000001.pdf",
)
m.assert_called()
self.assertTrue(os.path.isfile(original))
@@ -615,9 +713,16 @@ class TestFileHandlingWithArchive(DirectoriesMixin, TestCase):
def test_move_file_gone(self):
original = os.path.join(settings.ORIGINALS_DIR, "0000001.pdf")
archive = os.path.join(settings.ARCHIVE_DIR, "0000001.pdf")
#Path(original).touch()
# Path(original).touch()
Path(archive).touch()
doc = Document.objects.create(mime_type="application/pdf", title="my_doc", filename="0000001.pdf", archive_filename="0000001.pdf", checksum="A", archive_checksum="B")
doc = Document.objects.create(
mime_type="application/pdf",
title="my_doc",
filename="0000001.pdf",
archive_filename="0000001.pdf",
checksum="A",
archive_checksum="B",
)
self.assertFalse(os.path.isfile(original))
self.assertTrue(os.path.isfile(archive))
@@ -627,7 +732,6 @@ class TestFileHandlingWithArchive(DirectoriesMixin, TestCase):
@override_settings(PAPERLESS_FILENAME_FORMAT="{correspondent}/{title}")
@mock.patch("documents.signals.handlers.os.rename")
def test_move_file_error(self, m):
def fake_rename(src, dst):
if "original" in src:
raise OSError()
@@ -641,7 +745,14 @@ class TestFileHandlingWithArchive(DirectoriesMixin, TestCase):
archive = os.path.join(settings.ARCHIVE_DIR, "0000001.pdf")
Path(original).touch()
Path(archive).touch()
doc = Document.objects.create(mime_type="application/pdf", title="my_doc", filename="0000001.pdf", archive_filename="0000001.pdf", checksum="A", archive_checksum="B")
doc = Document.objects.create(
mime_type="application/pdf",
title="my_doc",
filename="0000001.pdf",
archive_filename="0000001.pdf",
checksum="A",
archive_checksum="B",
)
m.assert_called()
self.assertTrue(os.path.isfile(original))
@@ -655,7 +766,14 @@ class TestFileHandlingWithArchive(DirectoriesMixin, TestCase):
archive = os.path.join(settings.ARCHIVE_DIR, "0000001.pdf")
Path(original).touch()
Path(archive).touch()
doc = Document.objects.create(mime_type="application/pdf", title="my_doc", filename="0000001.pdf", checksum="A", archive_checksum="B", archive_filename="0000001.pdf")
doc = Document.objects.create(
mime_type="application/pdf",
title="my_doc",
filename="0000001.pdf",
checksum="A",
archive_checksum="B",
archive_filename="0000001.pdf",
)
self.assertTrue(os.path.isfile(original))
self.assertTrue(os.path.isfile(archive))
@@ -678,8 +796,20 @@ class TestFileHandlingWithArchive(DirectoriesMixin, TestCase):
Path(original2).touch()
Path(archive).touch()
doc1 = Document.objects.create(mime_type="image/png", title="document", filename="document.png", checksum="A", archive_checksum="B", archive_filename="0000001.pdf")
doc2 = Document.objects.create(mime_type="application/pdf", title="0000001", filename="0000001.pdf", checksum="C")
doc1 = Document.objects.create(
mime_type="image/png",
title="document",
filename="document.png",
checksum="A",
archive_checksum="B",
archive_filename="0000001.pdf",
)
doc2 = Document.objects.create(
mime_type="application/pdf",
title="0000001",
filename="0000001.pdf",
checksum="C",
)
self.assertTrue(os.path.isfile(doc1.source_path))
self.assertTrue(os.path.isfile(doc1.archive_path))
@@ -698,7 +828,14 @@ class TestFileHandlingWithArchive(DirectoriesMixin, TestCase):
archive = os.path.join(settings.ARCHIVE_DIR, "0000001.pdf")
Path(original).touch()
Path(archive).touch()
doc = Document(mime_type="application/pdf", title="my_doc", filename="0000001.pdf", checksum="A", archive_filename="0000001.pdf", archive_checksum="B")
doc = Document(
mime_type="application/pdf",
title="my_doc",
filename="0000001.pdf",
checksum="A",
archive_filename="0000001.pdf",
archive_checksum="B",
)
with mock.patch("documents.signals.handlers.Document.objects.filter") as m:
m.side_effect = DatabaseError()
doc.save()
@@ -710,28 +847,38 @@ class TestFileHandlingWithArchive(DirectoriesMixin, TestCase):
class TestFilenameGeneration(TestCase):
@override_settings(
PAPERLESS_FILENAME_FORMAT="{title}"
)
@override_settings(PAPERLESS_FILENAME_FORMAT="{title}")
def test_invalid_characters(self):
doc = Document.objects.create(title="This. is the title.", mime_type="application/pdf", pk=1, checksum="1")
doc = Document.objects.create(
title="This. is the title.", mime_type="application/pdf", pk=1, checksum="1"
)
self.assertEqual(generate_filename(doc), "This. is the title.pdf")
doc = Document.objects.create(title="my\\invalid/../title:yay", mime_type="application/pdf", pk=2, checksum="2")
doc = Document.objects.create(
title="my\\invalid/../title:yay",
mime_type="application/pdf",
pk=2,
checksum="2",
)
self.assertEqual(generate_filename(doc), "my-invalid-..-title-yay.pdf")
@override_settings(
PAPERLESS_FILENAME_FORMAT="{created}"
)
@override_settings(PAPERLESS_FILENAME_FORMAT="{created}")
def test_date(self):
doc = Document.objects.create(title="does not matter", created=timezone.make_aware(datetime.datetime(2020,5,21, 7,36,51, 153)), mime_type="application/pdf", pk=2, checksum="2")
doc = Document.objects.create(
title="does not matter",
created=timezone.make_aware(datetime.datetime(2020, 5, 21, 7, 36, 51, 153)),
mime_type="application/pdf",
pk=2,
checksum="2",
)
self.assertEqual(generate_filename(doc), "2020-05-21.pdf")
def run():
doc = Document.objects.create(checksum=str(uuid.uuid4()), title=str(uuid.uuid4()), content="wow")
doc = Document.objects.create(
checksum=str(uuid.uuid4()), title=str(uuid.uuid4()), content="wow"
)
doc.filename = generate_unique_filename(doc)
Path(doc.thumbnail_path).touch()
with open(doc.source_path, "w") as f:

17
src/documents/tests/test_importer.py
View File

@@ -6,14 +6,14 @@ from ..management.commands.document_importer import Command
class TestImporter(TestCase):
def __init__(self, *args, **kwargs):
TestCase.__init__(self, *args, **kwargs)
def test_check_manifest_exists(self):
cmd = Command()
self.assertRaises(
CommandError, cmd._check_manifest_exists, "/tmp/manifest.json")
CommandError, cmd._check_manifest_exists, "/tmp/manifest.json"
)
def test_check_manifest(self):
@@ -23,15 +23,14 @@ class TestImporter(TestCase):
cmd.manifest = [{"model": "documents.document"}]
with self.assertRaises(CommandError) as cm:
cmd._check_manifest()
self.assertTrue(
'The manifest file contains a record' in str(cm.exception))
self.assertTrue("The manifest file contains a record" in str(cm.exception))
cmd.manifest = [{
"model": "documents.document",
EXPORTER_FILE_NAME: "noexist.pdf"
}]
cmd.manifest = [
{"model": "documents.document", EXPORTER_FILE_NAME: "noexist.pdf"}
]
# self.assertRaises(CommandError, cmd._check_manifest)
with self.assertRaises(CommandError) as cm:
cmd._check_manifest()
self.assertTrue(
'The manifest file refers to "noexist.pdf"' in str(cm.exception))
'The manifest file refers to "noexist.pdf"' in str(cm.exception)
)

13
src/documents/tests/test_index.py
View File

@@ -6,10 +6,11 @@ from documents.tests.utils import DirectoriesMixin
class TestAutoComplete(DirectoriesMixin, TestCase):
def test_auto_complete(self):
doc1 = Document.objects.create(title="doc1", checksum="A", content="test test2 test3")
doc1 = Document.objects.create(
title="doc1", checksum="A", content="test test2 test3"
)
doc2 = Document.objects.create(title="doc2", checksum="B", content="test test2")
doc3 = Document.objects.create(title="doc3", checksum="C", content="test2")
@@ -19,7 +20,11 @@ class TestAutoComplete(DirectoriesMixin, TestCase):
ix = index.open_index()
self.assertListEqual(index.autocomplete(ix, "tes"), [b"test3", b"test", b"test2"])
self.assertListEqual(index.autocomplete(ix, "tes", limit=3), [b"test3", b"test", b"test2"])
self.assertListEqual(
index.autocomplete(ix, "tes"), [b"test3", b"test", b"test2"]
)
self.assertListEqual(
index.autocomplete(ix, "tes", limit=3), [b"test3", b"test", b"test2"]
)
self.assertListEqual(index.autocomplete(ix, "tes", limit=1), [b"test3"])
self.assertListEqual(index.autocomplete(ix, "tes", limit=0), [])

89
src/documents/tests/test_management.py
View File

@@ -22,21 +22,29 @@ sample_file = os.path.join(os.path.dirname(__file__), "samples", "simple.pdf")
@override_settings(PAPERLESS_FILENAME_FORMAT="{correspondent}/{title}")
class TestArchiver(DirectoriesMixin, TestCase):
def make_models(self):
return Document.objects.create(checksum="A", title="A", content="first document", mime_type="application/pdf")
return Document.objects.create(
checksum="A",
title="A",
content="first document",
mime_type="application/pdf",
)
def test_archiver(self):
doc = self.make_models()
shutil.copy(sample_file, os.path.join(self.dirs.originals_dir, f"{doc.id:07}.pdf"))
shutil.copy(
sample_file, os.path.join(self.dirs.originals_dir, f"{doc.id:07}.pdf")
)
call_command('document_archiver')
call_command("document_archiver")
def test_handle_document(self):
doc = self.make_models()
shutil.copy(sample_file, os.path.join(self.dirs.originals_dir, f"{doc.id:07}.pdf"))
shutil.copy(
sample_file, os.path.join(self.dirs.originals_dir, f"{doc.id:07}.pdf")
)
handle_document(doc.pk)
@@ -66,10 +74,24 @@ class TestArchiver(DirectoriesMixin, TestCase):
@override_settings(PAPERLESS_FILENAME_FORMAT="{title}")
def test_naming_priorities(self):
doc1 = Document.objects.create(checksum="A", title="document", content="first document", mime_type="application/pdf", filename="document.pdf")
doc2 = Document.objects.create(checksum="B", title="document", content="second document", mime_type="application/pdf", filename="document_01.pdf")
doc1 = Document.objects.create(
checksum="A",
title="document",
content="first document",
mime_type="application/pdf",
filename="document.pdf",
)
doc2 = Document.objects.create(
checksum="B",
title="document",
content="second document",
mime_type="application/pdf",
filename="document_01.pdf",
)
shutil.copy(sample_file, os.path.join(self.dirs.originals_dir, f"document.pdf"))
shutil.copy(sample_file, os.path.join(self.dirs.originals_dir, f"document_01.pdf"))
shutil.copy(
sample_file, os.path.join(self.dirs.originals_dir, f"document_01.pdf")
)
handle_document(doc2.pk)
handle_document(doc1.pk)
@@ -82,12 +104,11 @@ class TestArchiver(DirectoriesMixin, TestCase):
class TestDecryptDocuments(TestCase):
@override_settings(
ORIGINALS_DIR=os.path.join(os.path.dirname(__file__), "samples", "originals"),
THUMBNAIL_DIR=os.path.join(os.path.dirname(__file__), "samples", "thumb"),
PASSPHRASE="test",
PAPERLESS_FILENAME_FORMAT=None
PAPERLESS_FILENAME_FORMAT=None,
)
@mock.patch("documents.management.commands.decrypt_documents.input")
def test_decrypt(self, m):
@@ -99,17 +120,39 @@ class TestDecryptDocuments(TestCase):
os.makedirs(thumb_dir, exist_ok=True)
override_settings(
ORIGINALS_DIR=originals_dir,
THUMBNAIL_DIR=thumb_dir,
PASSPHRASE="test"
ORIGINALS_DIR=originals_dir, THUMBNAIL_DIR=thumb_dir, PASSPHRASE="test"
).enable()
doc = Document.objects.create(checksum="82186aaa94f0b98697d704b90fd1c072", title="wow", filename="0000004.pdf.gpg", mime_type="application/pdf", storage_type=Document.STORAGE_TYPE_GPG)
doc = Document.objects.create(
checksum="82186aaa94f0b98697d704b90fd1c072",
title="wow",
filename="0000004.pdf.gpg",
mime_type="application/pdf",
storage_type=Document.STORAGE_TYPE_GPG,
)
shutil.copy(os.path.join(os.path.dirname(__file__), "samples", "documents", "originals", "0000004.pdf.gpg"), os.path.join(originals_dir, "0000004.pdf.gpg"))
shutil.copy(os.path.join(os.path.dirname(__file__), "samples", "documents", "thumbnails", f"0000004.png.gpg"), os.path.join(thumb_dir, f"{doc.id:07}.png.gpg"))
shutil.copy(
os.path.join(
os.path.dirname(__file__),
"samples",
"documents",
"originals",
"0000004.pdf.gpg",
),
os.path.join(originals_dir, "0000004.pdf.gpg"),
)
shutil.copy(
os.path.join(
os.path.dirname(__file__),
"samples",
"documents",
"thumbnails",
f"0000004.png.gpg",
),
os.path.join(thumb_dir, f"{doc.id:07}.png.gpg"),
)
call_command('decrypt_documents')
call_command("decrypt_documents")
doc.refresh_from_db()
@@ -126,7 +169,6 @@ class TestDecryptDocuments(TestCase):
class TestMakeIndex(TestCase):
@mock.patch("documents.management.commands.document_index.index_reindex")
def test_reindex(self, m):
call_command("document_index", "reindex")
@@ -139,7 +181,6 @@ class TestMakeIndex(TestCase):
class TestRenamer(DirectoriesMixin, TestCase):
@override_settings(PAPERLESS_FILENAME_FORMAT="")
def test_rename(self):
doc = Document.objects.create(title="test", mime_type="image/jpeg")
@@ -164,8 +205,9 @@ class TestRenamer(DirectoriesMixin, TestCase):
class TestCreateClassifier(TestCase):
@mock.patch("documents.management.commands.document_create_classifier.train_classifier")
@mock.patch(
"documents.management.commands.document_create_classifier.train_classifier"
)
def test_create_classifier(self, m):
call_command("document_create_classifier")
@@ -173,7 +215,6 @@ class TestCreateClassifier(TestCase):
class TestSanityChecker(DirectoriesMixin, TestCase):
def test_no_issues(self):
with self.assertLogs() as capture:
call_command("document_sanity_checker")
@@ -182,7 +223,9 @@ class TestSanityChecker(DirectoriesMixin, TestCase):
self.assertIn("Sanity checker detected no issues.", capture.output[0])
def test_errors(self):
doc = Document.objects.create(title="test", content="test", filename="test.pdf", checksum="abc")
doc = Document.objects.create(
title="test", content="test", filename="test.pdf", checksum="abc"
)
Path(doc.source_path).touch()
Path(doc.thumbnail_path).touch()

74
src/documents/tests/test_management_consumer.py
View File

@@ -16,7 +16,6 @@ from documents.tests.utils import DirectoriesMixin
class ConsumerThread(Thread):
def __init__(self):
super().__init__()
self.cmd = document_consumer.Command()
@@ -31,7 +30,7 @@ class ConsumerThread(Thread):
def chunked(size, source):
for i in range(0, len(source), size):
yield source[i:i+size]
yield source[i : i + size]
class ConsumerMixin:
@@ -41,7 +40,9 @@ class ConsumerMixin:
def setUp(self) -> None:
super(ConsumerMixin, self).setUp()
self.t = None
patcher = mock.patch("documents.management.commands.document_consumer.async_task")
patcher = mock.patch(
"documents.management.commands.document_consumer.async_task"
)
self.task_mock = patcher.start()
self.addCleanup(patcher.stop)
@@ -81,13 +82,13 @@ class ConsumerMixin:
print("Consumed a perfectly valid file.")
def slow_write_file(self, target, incomplete=False):
with open(self.sample_file, 'rb') as f:
with open(self.sample_file, "rb") as f:
pdf_bytes = f.read()
if incomplete:
pdf_bytes = pdf_bytes[:len(pdf_bytes) - 100]
pdf_bytes = pdf_bytes[: len(pdf_bytes) - 100]
with open(target, 'wb') as f:
with open(target, "wb") as f:
# this will take 2 seconds, since the file is about 20k.
print("Start writing file.")
for b in chunked(1000, pdf_bytes):
@@ -97,7 +98,6 @@ class ConsumerMixin:
class TestConsumer(DirectoriesMixin, ConsumerMixin, TransactionTestCase):
def test_consume_file(self):
self.t_start()
@@ -195,23 +195,35 @@ class TestConsumer(DirectoriesMixin, ConsumerMixin, TransactionTestCase):
@override_settings(CONSUMPTION_DIR="does_not_exist")
def test_consumption_directory_invalid(self):
self.assertRaises(CommandError, call_command, 'document_consumer', '--oneshot')
self.assertRaises(CommandError, call_command, "document_consumer", "--oneshot")
@override_settings(CONSUMPTION_DIR="")
def test_consumption_directory_unset(self):
self.assertRaises(CommandError, call_command, 'document_consumer', '--oneshot')
self.assertRaises(CommandError, call_command, "document_consumer", "--oneshot")
def test_mac_write(self):
self.task_mock.side_effect = self.bogus_task
self.t_start()
shutil.copy(self.sample_file, os.path.join(self.dirs.consumption_dir, ".DS_STORE"))
shutil.copy(self.sample_file, os.path.join(self.dirs.consumption_dir, "my_file.pdf"))
shutil.copy(self.sample_file, os.path.join(self.dirs.consumption_dir, "._my_file.pdf"))
shutil.copy(self.sample_file, os.path.join(self.dirs.consumption_dir, "my_second_file.pdf"))
shutil.copy(self.sample_file, os.path.join(self.dirs.consumption_dir, "._my_second_file.pdf"))
shutil.copy(
self.sample_file, os.path.join(self.dirs.consumption_dir, ".DS_STORE")
)
shutil.copy(
self.sample_file, os.path.join(self.dirs.consumption_dir, "my_file.pdf")
)
shutil.copy(
self.sample_file, os.path.join(self.dirs.consumption_dir, "._my_file.pdf")
)
shutil.copy(
self.sample_file,
os.path.join(self.dirs.consumption_dir, "my_second_file.pdf"),
)
shutil.copy(
self.sample_file,
os.path.join(self.dirs.consumption_dir, "._my_second_file.pdf"),
)
sleep(5)
@@ -219,15 +231,20 @@ class TestConsumer(DirectoriesMixin, ConsumerMixin, TransactionTestCase):
self.assertEqual(2, self.task_mock.call_count)
fnames = [os.path.basename(args[1]) for args, _ in self.task_mock.call_args_list]
fnames = [
os.path.basename(args[1]) for args, _ in self.task_mock.call_args_list
]
self.assertCountEqual(fnames, ["my_file.pdf", "my_second_file.pdf"])
def test_is_ignored(self):
test_paths = [
(os.path.join(self.dirs.consumption_dir, "foo.pdf"), False),
(os.path.join(self.dirs.consumption_dir, "foo","bar.pdf"), False),
(os.path.join(self.dirs.consumption_dir, "foo", "bar.pdf"), False),
(os.path.join(self.dirs.consumption_dir, ".DS_STORE", "foo.pdf"), True),
(os.path.join(self.dirs.consumption_dir, "foo", ".DS_STORE", "bar.pdf"), True),
(
os.path.join(self.dirs.consumption_dir, "foo", ".DS_STORE", "bar.pdf"),
True,
),
(os.path.join(self.dirs.consumption_dir, ".stfolder", "foo.pdf"), True),
(os.path.join(self.dirs.consumption_dir, "._foo.pdf"), True),
(os.path.join(self.dirs.consumption_dir, "._foo", "bar.pdf"), False),
@@ -236,10 +253,13 @@ class TestConsumer(DirectoriesMixin, ConsumerMixin, TransactionTestCase):
self.assertEqual(
expected_ignored,
document_consumer._is_ignored(file_path),
f'_is_ignored("{file_path}") != {expected_ignored}')
f'_is_ignored("{file_path}") != {expected_ignored}',
)
@override_settings(CONSUMER_POLLING=1, CONSUMER_POLLING_DELAY=3, CONSUMER_POLLING_RETRY_COUNT=20)
@override_settings(
CONSUMER_POLLING=1, CONSUMER_POLLING_DELAY=3, CONSUMER_POLLING_RETRY_COUNT=20
)
class TestConsumerPolling(TestConsumer):
# just do all the tests with polling
pass
@@ -251,21 +271,27 @@ class TestConsumerRecursive(TestConsumer):
pass
@override_settings(CONSUMER_RECURSIVE=True, CONSUMER_POLLING=1, CONSUMER_POLLING_DELAY=3, CONSUMER_POLLING_RETRY_COUNT=20)
@override_settings(
CONSUMER_RECURSIVE=True,
CONSUMER_POLLING=1,
CONSUMER_POLLING_DELAY=3,
CONSUMER_POLLING_RETRY_COUNT=20,
)
class TestConsumerRecursivePolling(TestConsumer):
# just do all the tests with polling and recursive
pass
class TestConsumerTags(DirectoriesMixin, ConsumerMixin, TransactionTestCase):
@override_settings(CONSUMER_RECURSIVE=True)
@override_settings(CONSUMER_SUBDIRS_AS_TAGS=True)
def test_consume_file_with_path_tags(self):
tag_names = ("existingTag", "Space Tag")
# Create a Tag prior to consuming a file using it in path
tag_ids = [Tag.objects.create(name="existingtag").pk,]
tag_ids = [
Tag.objects.create(name="existingtag").pk,
]
self.t_start()
@@ -292,6 +318,8 @@ class TestConsumerTags(DirectoriesMixin, ConsumerMixin, TransactionTestCase):
# their order.
self.assertCountEqual(kwargs["override_tag_ids"], tag_ids)
@override_settings(CONSUMER_POLLING=1, CONSUMER_POLLING_DELAY=1, CONSUMER_POLLING_RETRY_COUNT=20)
@override_settings(
CONSUMER_POLLING=1, CONSUMER_POLLING_DELAY=1, CONSUMER_POLLING_RETRY_COUNT=20
)
def test_consume_file_with_path_tags_polling(self):
self.test_consume_file_with_path_tags()

190
src/documents/tests/test_management_exporter.py
View File

@@ -17,15 +17,41 @@ from documents.tests.utils import DirectoriesMixin, paperless_environment
class TestExportImport(DirectoriesMixin, TestCase):
def setUp(self) -> None:
self.target = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, self.target)
self.d1 = Document.objects.create(content="Content", checksum="42995833e01aea9b3edee44bbfdd7ce1", archive_checksum="62acb0bcbfbcaa62ca6ad3668e4e404b", title="wow1", filename="0000001.pdf", mime_type="application/pdf", archive_filename="0000001.pdf")
self.d2 = Document.objects.create(content="Content", checksum="9c9691e51741c1f4f41a20896af31770", title="wow2", filename="0000002.pdf", mime_type="application/pdf")
self.d3 = Document.objects.create(content="Content", checksum="d38d7ed02e988e072caf924e0f3fcb76", title="wow2", filename="0000003.pdf", mime_type="application/pdf")
self.d4 = Document.objects.create(content="Content", checksum="82186aaa94f0b98697d704b90fd1c072", title="wow_dec", filename="0000004.pdf.gpg", mime_type="application/pdf", storage_type=Document.STORAGE_TYPE_GPG)
self.d1 = Document.objects.create(
content="Content",
checksum="42995833e01aea9b3edee44bbfdd7ce1",
archive_checksum="62acb0bcbfbcaa62ca6ad3668e4e404b",
title="wow1",
filename="0000001.pdf",
mime_type="application/pdf",
archive_filename="0000001.pdf",
)
self.d2 = Document.objects.create(
content="Content",
checksum="9c9691e51741c1f4f41a20896af31770",
title="wow2",
filename="0000002.pdf",
mime_type="application/pdf",
)
self.d3 = Document.objects.create(
content="Content",
checksum="d38d7ed02e988e072caf924e0f3fcb76",
title="wow2",
filename="0000003.pdf",
mime_type="application/pdf",
)
self.d4 = Document.objects.create(
content="Content",
checksum="82186aaa94f0b98697d704b90fd1c072",
title="wow_dec",
filename="0000004.pdf.gpg",
mime_type="application/pdf",
storage_type=Document.STORAGE_TYPE_GPG,
)
self.t1 = Tag.objects.create(name="t")
self.dt1 = DocumentType.objects.create(name="dt")
@@ -38,17 +64,21 @@ class TestExportImport(DirectoriesMixin, TestCase):
super(TestExportImport, self).setUp()
def _get_document_from_manifest(self, manifest, id):
f = list(filter(lambda d: d['model'] == "documents.document" and d['pk'] == id, manifest))
f = list(
filter(
lambda d: d["model"] == "documents.document" and d["pk"] == id, manifest
)
)
if len(f) == 1:
return f[0]
else:
raise ValueError(f"document with id {id} does not exist in manifest")
@override_settings(
PASSPHRASE="test"
)
def _do_export(self, use_filename_format=False, compare_checksums=False, delete=False):
args = ['document_exporter', self.target]
@override_settings(PASSPHRASE="test")
def _do_export(
self, use_filename_format=False, compare_checksums=False, delete=False
):
args = ["document_exporter", self.target]
if use_filename_format:
args += ["--use-filename-format"]
if compare_checksums:
@@ -65,39 +95,69 @@ class TestExportImport(DirectoriesMixin, TestCase):
def test_exporter(self, use_filename_format=False):
shutil.rmtree(os.path.join(self.dirs.media_dir, "documents"))
shutil.copytree(os.path.join(os.path.dirname(__file__), "samples", "documents"), os.path.join(self.dirs.media_dir, "documents"))
shutil.copytree(
os.path.join(os.path.dirname(__file__), "samples", "documents"),
os.path.join(self.dirs.media_dir, "documents"),
)
manifest = self._do_export(use_filename_format=use_filename_format)
self.assertEqual(len(manifest), 8)
self.assertEqual(len(list(filter(lambda e: e['model'] == 'documents.document', manifest))), 4)
self.assertEqual(
len(list(filter(lambda e: e["model"] == "documents.document", manifest))), 4
)
self.assertTrue(os.path.exists(os.path.join(self.target, "manifest.json")))
self.assertEqual(self._get_document_from_manifest(manifest, self.d1.id)['fields']['title'], "wow1")
self.assertEqual(self._get_document_from_manifest(manifest, self.d2.id)['fields']['title'], "wow2")
self.assertEqual(self._get_document_from_manifest(manifest, self.d3.id)['fields']['title'], "wow2")
self.assertEqual(self._get_document_from_manifest(manifest, self.d4.id)['fields']['title'], "wow_dec")
self.assertEqual(
self._get_document_from_manifest(manifest, self.d1.id)["fields"]["title"],
"wow1",
)
self.assertEqual(
self._get_document_from_manifest(manifest, self.d2.id)["fields"]["title"],
"wow2",
)
self.assertEqual(
self._get_document_from_manifest(manifest, self.d3.id)["fields"]["title"],
"wow2",
)
self.assertEqual(
self._get_document_from_manifest(manifest, self.d4.id)["fields"]["title"],
"wow_dec",
)
for element in manifest:
if element['model'] == 'documents.document':
fname = os.path.join(self.target, element[document_exporter.EXPORTER_FILE_NAME])
if element["model"] == "documents.document":
fname = os.path.join(
self.target, element[document_exporter.EXPORTER_FILE_NAME]
)
self.assertTrue(os.path.exists(fname))
self.assertTrue(os.path.exists(os.path.join(self.target, element[document_exporter.EXPORTER_THUMBNAIL_NAME])))
self.assertTrue(
os.path.exists(
os.path.join(
self.target,
element[document_exporter.EXPORTER_THUMBNAIL_NAME],
)
)
)
with open(fname, "rb") as f:
checksum = hashlib.md5(f.read()).hexdigest()
self.assertEqual(checksum, element['fields']['checksum'])
self.assertEqual(checksum, element["fields"]["checksum"])
self.assertEqual(element['fields']['storage_type'], Document.STORAGE_TYPE_UNENCRYPTED)
self.assertEqual(
element["fields"]["storage_type"], Document.STORAGE_TYPE_UNENCRYPTED
)
if document_exporter.EXPORTER_ARCHIVE_NAME in element:
fname = os.path.join(self.target, element[document_exporter.EXPORTER_ARCHIVE_NAME])
fname = os.path.join(
self.target, element[document_exporter.EXPORTER_ARCHIVE_NAME]
)
self.assertTrue(os.path.exists(fname))
with open(fname, "rb") as f:
checksum = hashlib.md5(f.read()).hexdigest()
self.assertEqual(checksum, element['fields']['archive_checksum'])
self.assertEqual(checksum, element["fields"]["archive_checksum"])
with paperless_environment() as dirs:
self.assertEqual(Document.objects.count(), 4)
@@ -107,7 +167,7 @@ class TestExportImport(DirectoriesMixin, TestCase):
Tag.objects.all().delete()
self.assertEqual(Document.objects.count(), 0)
call_command('document_importer', self.target)
call_command("document_importer", self.target)
self.assertEqual(Document.objects.count(), 4)
self.assertEqual(Tag.objects.count(), 1)
self.assertEqual(Correspondent.objects.count(), 1)
@@ -122,21 +182,31 @@ class TestExportImport(DirectoriesMixin, TestCase):
def test_exporter_with_filename_format(self):
shutil.rmtree(os.path.join(self.dirs.media_dir, "documents"))
shutil.copytree(os.path.join(os.path.dirname(__file__), "samples", "documents"), os.path.join(self.dirs.media_dir, "documents"))
shutil.copytree(
os.path.join(os.path.dirname(__file__), "samples", "documents"),
os.path.join(self.dirs.media_dir, "documents"),
)
with override_settings(PAPERLESS_FILENAME_FORMAT="{created_year}/{correspondent}/{title}"):
with override_settings(
PAPERLESS_FILENAME_FORMAT="{created_year}/{correspondent}/{title}"
):
self.test_exporter(use_filename_format=True)
def test_update_export_changed_time(self):
shutil.rmtree(os.path.join(self.dirs.media_dir, "documents"))
shutil.copytree(os.path.join(os.path.dirname(__file__), "samples", "documents"), os.path.join(self.dirs.media_dir, "documents"))
shutil.copytree(
os.path.join(os.path.dirname(__file__), "samples", "documents"),
os.path.join(self.dirs.media_dir, "documents"),
)
self._do_export()
self.assertTrue(os.path.exists(os.path.join(self.target, "manifest.json")))
st_mtime_1 = os.stat(os.path.join(self.target, "manifest.json")).st_mtime
with mock.patch("documents.management.commands.document_exporter.shutil.copy2") as m:
with mock.patch(
"documents.management.commands.document_exporter.shutil.copy2"
) as m:
self._do_export()
m.assert_not_called()
@@ -145,7 +215,9 @@ class TestExportImport(DirectoriesMixin, TestCase):
Path(self.d1.source_path).touch()
with mock.patch("documents.management.commands.document_exporter.shutil.copy2") as m:
with mock.patch(
"documents.management.commands.document_exporter.shutil.copy2"
) as m:
self._do_export()
self.assertEqual(m.call_count, 1)
@@ -157,13 +229,18 @@ class TestExportImport(DirectoriesMixin, TestCase):
def test_update_export_changed_checksum(self):
shutil.rmtree(os.path.join(self.dirs.media_dir, "documents"))
shutil.copytree(os.path.join(os.path.dirname(__file__), "samples", "documents"), os.path.join(self.dirs.media_dir, "documents"))
shutil.copytree(
os.path.join(os.path.dirname(__file__), "samples", "documents"),
os.path.join(self.dirs.media_dir, "documents"),
)
self._do_export()
self.assertTrue(os.path.exists(os.path.join(self.target, "manifest.json")))
with mock.patch("documents.management.commands.document_exporter.shutil.copy2") as m:
with mock.patch(
"documents.management.commands.document_exporter.shutil.copy2"
) as m:
self._do_export()
m.assert_not_called()
@@ -172,7 +249,9 @@ class TestExportImport(DirectoriesMixin, TestCase):
self.d2.checksum = "asdfasdgf3"
self.d2.save()
with mock.patch("documents.management.commands.document_exporter.shutil.copy2") as m:
with mock.patch(
"documents.management.commands.document_exporter.shutil.copy2"
) as m:
self._do_export(compare_checksums=True)
self.assertEqual(m.call_count, 1)
@@ -180,28 +259,48 @@ class TestExportImport(DirectoriesMixin, TestCase):
def test_update_export_deleted_document(self):
shutil.rmtree(os.path.join(self.dirs.media_dir, "documents"))
shutil.copytree(os.path.join(os.path.dirname(__file__), "samples", "documents"), os.path.join(self.dirs.media_dir, "documents"))
shutil.copytree(
os.path.join(os.path.dirname(__file__), "samples", "documents"),
os.path.join(self.dirs.media_dir, "documents"),
)
manifest = self._do_export()
self.assertTrue(len(manifest), 7)
doc_from_manifest = self._get_document_from_manifest(manifest, self.d3.id)
self.assertTrue(os.path.isfile(os.path.join(self.target, doc_from_manifest[EXPORTER_FILE_NAME])))
self.assertTrue(
os.path.isfile(
os.path.join(self.target, doc_from_manifest[EXPORTER_FILE_NAME])
)
)
self.d3.delete()
manifest = self._do_export()
self.assertRaises(ValueError, self._get_document_from_manifest, manifest, self.d3.id)
self.assertTrue(os.path.isfile(os.path.join(self.target, doc_from_manifest[EXPORTER_FILE_NAME])))
self.assertRaises(
ValueError, self._get_document_from_manifest, manifest, self.d3.id
)
self.assertTrue(
os.path.isfile(
os.path.join(self.target, doc_from_manifest[EXPORTER_FILE_NAME])
)
)
manifest = self._do_export(delete=True)
self.assertFalse(os.path.isfile(os.path.join(self.target, doc_from_manifest[EXPORTER_FILE_NAME])))
self.assertFalse(
os.path.isfile(
os.path.join(self.target, doc_from_manifest[EXPORTER_FILE_NAME])
)
)
self.assertTrue(len(manifest), 6)
@override_settings(PAPERLESS_FILENAME_FORMAT="{title}/{correspondent}")
def test_update_export_changed_location(self):
shutil.rmtree(os.path.join(self.dirs.media_dir, "documents"))
shutil.copytree(os.path.join(os.path.dirname(__file__), "samples", "documents"), os.path.join(self.dirs.media_dir, "documents"))
shutil.copytree(
os.path.join(os.path.dirname(__file__), "samples", "documents"),
os.path.join(self.dirs.media_dir, "documents"),
)
m = self._do_export(use_filename_format=True)
self.assertTrue(os.path.isfile(os.path.join(self.target, "wow1", "c.pdf")))
@@ -216,11 +315,18 @@ class TestExportImport(DirectoriesMixin, TestCase):
self.assertTrue(os.path.isfile(os.path.join(self.target, "new_title", "c.pdf")))
self.assertTrue(os.path.exists(os.path.join(self.target, "manifest.json")))
self.assertTrue(os.path.isfile(os.path.join(self.target, "wow2", "none.pdf")))
self.assertTrue(os.path.isfile(os.path.join(self.target, "wow2", "none_01.pdf")))
self.assertTrue(
os.path.isfile(os.path.join(self.target, "wow2", "none_01.pdf"))
)
def test_export_missing_files(self):
target = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, target)
Document.objects.create(checksum="AAAAAAAAAAAAAAAAA", title="wow", filename="0000004.pdf", mime_type="application/pdf")
self.assertRaises(FileNotFoundError, call_command, 'document_exporter', target)
Document.objects.create(
checksum="AAAAAAAAAAAAAAAAA",
title="wow",
filename="0000004.pdf",
mime_type="application/pdf",
)
self.assertRaises(FileNotFoundError, call_command, "document_exporter", target)

95
src/documents/tests/test_management_retagger.py
View File

@@ -6,44 +6,64 @@ from documents.tests.utils import DirectoriesMixin
class TestRetagger(DirectoriesMixin, TestCase):
def make_models(self):
self.d1 = Document.objects.create(checksum="A", title="A", content="first document")
self.d2 = Document.objects.create(checksum="B", title="B", content="second document")
self.d3 = Document.objects.create(checksum="C", title="C", content="unrelated document")
self.d4 = Document.objects.create(checksum="D", title="D", content="auto document")
self.d1 = Document.objects.create(
checksum="A", title="A", content="first document"
)
self.d2 = Document.objects.create(
checksum="B", title="B", content="second document"
)
self.d3 = Document.objects.create(
checksum="C", title="C", content="unrelated document"
)
self.d4 = Document.objects.create(
checksum="D", title="D", content="auto document"
)
self.tag_first = Tag.objects.create(name="tag1", match="first", matching_algorithm=Tag.MATCH_ANY)
self.tag_second = Tag.objects.create(name="tag2", match="second", matching_algorithm=Tag.MATCH_ANY)
self.tag_first = Tag.objects.create(
name="tag1", match="first", matching_algorithm=Tag.MATCH_ANY
)
self.tag_second = Tag.objects.create(
name="tag2", match="second", matching_algorithm=Tag.MATCH_ANY
)
self.tag_inbox = Tag.objects.create(name="test", is_inbox_tag=True)
self.tag_no_match = Tag.objects.create(name="test2")
self.tag_auto = Tag.objects.create(name="tagauto", matching_algorithm=Tag.MATCH_AUTO)
self.tag_auto = Tag.objects.create(
name="tagauto", matching_algorithm=Tag.MATCH_AUTO
)
self.d3.tags.add(self.tag_inbox)
self.d3.tags.add(self.tag_no_match)
self.d4.tags.add(self.tag_auto)
self.correspondent_first = Correspondent.objects.create(
name="c1", match="first", matching_algorithm=Correspondent.MATCH_ANY)
name="c1", match="first", matching_algorithm=Correspondent.MATCH_ANY
)
self.correspondent_second = Correspondent.objects.create(
name="c2", match="second", matching_algorithm=Correspondent.MATCH_ANY)
name="c2", match="second", matching_algorithm=Correspondent.MATCH_ANY
)
self.doctype_first = DocumentType.objects.create(
name="dt1", match="first", matching_algorithm=DocumentType.MATCH_ANY)
name="dt1", match="first", matching_algorithm=DocumentType.MATCH_ANY
)
self.doctype_second = DocumentType.objects.create(
name="dt2", match="second", matching_algorithm=DocumentType.MATCH_ANY)
name="dt2", match="second", matching_algorithm=DocumentType.MATCH_ANY
)
def get_updated_docs(self):
return Document.objects.get(title="A"), Document.objects.get(title="B"), \
Document.objects.get(title="C"), Document.objects.get(title="D")
return (
Document.objects.get(title="A"),
Document.objects.get(title="B"),
Document.objects.get(title="C"),
Document.objects.get(title="D"),
)
def setUp(self) -> None:
super(TestRetagger, self).setUp()
self.make_models()
def test_add_tags(self):
call_command('document_retagger', '--tags')
call_command("document_retagger", "--tags")
d_first, d_second, d_unrelated, d_auto = self.get_updated_docs()
self.assertEqual(d_first.tags.count(), 1)
@@ -55,14 +75,14 @@ class TestRetagger(DirectoriesMixin, TestCase):
self.assertEqual(d_second.tags.first(), self.tag_second)
def test_add_type(self):
call_command('document_retagger', '--document_type')
call_command("document_retagger", "--document_type")
d_first, d_second, d_unrelated, d_auto = self.get_updated_docs()
self.assertEqual(d_first.document_type, self.doctype_first)
self.assertEqual(d_second.document_type, self.doctype_second)
def test_add_correspondent(self):
call_command('document_retagger', '--correspondent')
call_command("document_retagger", "--correspondent")
d_first, d_second, d_unrelated, d_auto = self.get_updated_docs()
self.assertEqual(d_first.correspondent, self.correspondent_first)
@@ -71,19 +91,26 @@ class TestRetagger(DirectoriesMixin, TestCase):
def test_overwrite_preserve_inbox(self):
self.d1.tags.add(self.tag_second)
call_command('document_retagger', '--tags', '--overwrite')
call_command("document_retagger", "--tags", "--overwrite")
d_first, d_second, d_unrelated, d_auto = self.get_updated_docs()
self.assertIsNotNone(Tag.objects.get(id=self.tag_second.id))
self.assertCountEqual([tag.id for tag in d_first.tags.all()], [self.tag_first.id])
self.assertCountEqual([tag.id for tag in d_second.tags.all()], [self.tag_second.id])
self.assertCountEqual([tag.id for tag in d_unrelated.tags.all()], [self.tag_inbox.id, self.tag_no_match.id])
self.assertCountEqual(
[tag.id for tag in d_first.tags.all()], [self.tag_first.id]
)
self.assertCountEqual(
[tag.id for tag in d_second.tags.all()], [self.tag_second.id]
)
self.assertCountEqual(
[tag.id for tag in d_unrelated.tags.all()],
[self.tag_inbox.id, self.tag_no_match.id],
)
self.assertEqual(d_auto.tags.count(), 0)
def test_add_tags_suggest(self):
call_command('document_retagger', '--tags', '--suggest')
call_command("document_retagger", "--tags", "--suggest")
d_first, d_second, d_unrelated, d_auto = self.get_updated_docs()
self.assertEqual(d_first.tags.count(), 0)
@@ -91,21 +118,23 @@ class TestRetagger(DirectoriesMixin, TestCase):
self.assertEqual(d_auto.tags.count(), 1)
def test_add_type_suggest(self):
call_command('document_retagger', '--document_type', '--suggest')
call_command("document_retagger", "--document_type", "--suggest")
d_first, d_second, d_unrelated, d_auto = self.get_updated_docs()
self.assertEqual(d_first.document_type, None)
self.assertEqual(d_second.document_type, None)
def test_add_correspondent_suggest(self):
call_command('document_retagger', '--correspondent', '--suggest')
call_command("document_retagger", "--correspondent", "--suggest")
d_first, d_second, d_unrelated, d_auto = self.get_updated_docs()
self.assertEqual(d_first.correspondent, None)
self.assertEqual(d_second.correspondent, None)
def test_add_tags_suggest_url(self):
call_command('document_retagger', '--tags', '--suggest', '--base-url=http://localhost')
call_command(
"document_retagger", "--tags", "--suggest", "--base-url=http://localhost"
)
d_first, d_second, d_unrelated, d_auto = self.get_updated_docs()
self.assertEqual(d_first.tags.count(), 0)
@@ -113,14 +142,24 @@ class TestRetagger(DirectoriesMixin, TestCase):
self.assertEqual(d_auto.tags.count(), 1)
def test_add_type_suggest_url(self):
call_command('document_retagger', '--document_type', '--suggest', '--base-url=http://localhost')
call_command(
"document_retagger",
"--document_type",
"--suggest",
"--base-url=http://localhost",
)
d_first, d_second, d_unrelated, d_auto = self.get_updated_docs()
self.assertEqual(d_first.document_type, None)
self.assertEqual(d_second.document_type, None)
def test_add_correspondent_suggest_url(self):
call_command('document_retagger', '--correspondent', '--suggest', '--base-url=http://localhost')
call_command(
"document_retagger",
"--correspondent",
"--suggest",
"--base-url=http://localhost",
)
d_first, d_second, d_unrelated, d_auto = self.get_updated_docs()
self.assertEqual(d_first.correspondent, None)

1
src/documents/tests/test_management_superuser.py
View File

@@ -12,7 +12,6 @@ from documents.tests.utils import DirectoriesMixin
class TestManageSuperUser(DirectoriesMixin, TestCase):
def reset_environment(self):
if "PAPERLESS_ADMIN_USER" in os.environ:
del os.environ["PAPERLESS_ADMIN_USER"]

31
src/documents/tests/test_management_thumbnails.py
View File

@@ -11,13 +11,30 @@ from documents.tests.utils import DirectoriesMixin
class TestMakeThumbnails(DirectoriesMixin, TestCase):
def make_models(self):
self.d1 = Document.objects.create(checksum="A", title="A", content="first document", mime_type="application/pdf", filename="test.pdf")
shutil.copy(os.path.join(os.path.dirname(__file__), "samples", "simple.pdf"), self.d1.source_path)
self.d1 = Document.objects.create(
checksum="A",
title="A",
content="first document",
mime_type="application/pdf",
filename="test.pdf",
)
shutil.copy(
os.path.join(os.path.dirname(__file__), "samples", "simple.pdf"),
self.d1.source_path,
)
self.d2 = Document.objects.create(checksum="Ass", title="A", content="first document", mime_type="application/pdf", filename="test2.pdf")
shutil.copy(os.path.join(os.path.dirname(__file__), "samples", "simple.pdf"), self.d2.source_path)
self.d2 = Document.objects.create(
checksum="Ass",
title="A",
content="first document",
mime_type="application/pdf",
filename="test2.pdf",
)
shutil.copy(
os.path.join(os.path.dirname(__file__), "samples", "simple.pdf"),
self.d2.source_path,
)
def setUp(self) -> None:
super(TestMakeThumbnails, self).setUp()
@@ -40,13 +57,13 @@ class TestMakeThumbnails(DirectoriesMixin, TestCase):
def test_command(self):
self.assertFalse(os.path.isfile(self.d1.thumbnail_path))
self.assertFalse(os.path.isfile(self.d2.thumbnail_path))
call_command('document_thumbnails')
call_command("document_thumbnails")
self.assertTrue(os.path.isfile(self.d1.thumbnail_path))
self.assertTrue(os.path.isfile(self.d2.thumbnail_path))
def test_command_documentid(self):
self.assertFalse(os.path.isfile(self.d1.thumbnail_path))
self.assertFalse(os.path.isfile(self.d2.thumbnail_path))
call_command('document_thumbnails', '-d', f"{self.d1.id}")
call_command("document_thumbnails", "-d", f"{self.d1.id}")
self.assertTrue(os.path.isfile(self.d1.thumbnail_path))
self.assertFalse(os.path.isfile(self.d2.thumbnail_path))

86
src/documents/tests/test_matchables.py
View File

@@ -12,25 +12,24 @@ from ..signals import document_consumption_finished
class TestMatching(TestCase):
def _test_matching(self, text, algorithm, true, false):
for klass in (Tag, Correspondent, DocumentType):
instance = klass.objects.create(
name=str(randint(10000, 99999)),
match=text,
matching_algorithm=getattr(klass, algorithm)
matching_algorithm=getattr(klass, algorithm),
)
for string in true:
doc = Document(content=string)
self.assertTrue(
matching.matches(instance, doc),
'"%s" should match "%s" but it does not' % (text, string)
'"%s" should match "%s" but it does not' % (text, string),
)
for string in false:
doc = Document(content=string)
self.assertFalse(
matching.matches(instance, doc),
'"%s" should not match "%s" but it does' % (text, string)
'"%s" should not match "%s" but it does' % (text, string),
)
def test_match_all(self):
@@ -47,15 +46,13 @@ class TestMatching(TestCase):
"I have alphas, charlie, and gamma in me",
"I have alphas in me",
"I have bravo in me",
)
),
)
self._test_matching(
"12 34 56",
"MATCH_ALL",
(
"I have 12 34, and 56 in me",
),
("I have 12 34, and 56 in me",),
(
"I have 12 in me",
"I have 34 in me",
@@ -64,7 +61,7 @@ class TestMatching(TestCase):
"I have 120, 34, and 56 in me",
"I have 123456 in me",
"I have 01234567 in me",
)
),
)
self._test_matching(
@@ -79,7 +76,7 @@ class TestMatching(TestCase):
"the quick brown wolf jumped over the lazy dogs",
"the quick brown fox jumped over the fat dogs",
"the quick brown fox jumped over the lazy... dogs",
)
),
)
def test_match_any(self):
@@ -97,7 +94,7 @@ class TestMatching(TestCase):
(
"I have alphas in me",
"I have bravo in me",
)
),
)
self._test_matching(
@@ -114,7 +111,7 @@ class TestMatching(TestCase):
(
"I have 123456 in me",
"I have 01234567 in me",
)
),
)
self._test_matching(
@@ -124,9 +121,7 @@ class TestMatching(TestCase):
"the quick brown fox",
"jumped over the lazy dogs.",
),
(
"the lazy fox jumped over the brown dogs",
)
("the lazy fox jumped over the brown dogs",),
)
def test_match_literal(self):
@@ -134,9 +129,7 @@ class TestMatching(TestCase):
self._test_matching(
"alpha charlie gamma",
"MATCH_LITERAL",
(
"I have 'alpha charlie gamma' in me",
),
("I have 'alpha charlie gamma' in me",),
(
"I have alpha in me",
"I have charlie in me",
@@ -146,15 +139,13 @@ class TestMatching(TestCase):
"I have alphas, charlie, and gamma in me",
"I have alphas in me",
"I have bravo in me",
)
),
)
self._test_matching(
"12 34 56",
"MATCH_LITERAL",
(
"I have 12 34 56 in me",
),
("I have 12 34 56 in me",),
(
"I have 12 in me",
"I have 34 in me",
@@ -165,7 +156,7 @@ class TestMatching(TestCase):
"I have 120, 340, and 560 in me",
"I have 123456 in me",
"I have 01234567 in me",
)
),
)
def test_match_regex(self):
@@ -186,18 +177,11 @@ class TestMatching(TestCase):
"I have alpha, charlie, and gamma in me",
"I have alphas, charlie, and gamma in me",
"I have alphas in me",
)
),
)
def test_tach_invalid_regex(self):
self._test_matching(
"[[",
"MATCH_REGEX",
[],
[
"Don't match this"
]
)
self._test_matching("[[", "MATCH_REGEX", [], ["Don't match this"])
def test_match_fuzzy(self):
@@ -210,9 +194,7 @@ class TestMatching(TestCase):
"1220 Main Street, Springfeld, Miss.",
"1220 Main Street Springfield Miss",
),
(
"1220 Main Street, Springfield, Mich.",
)
("1220 Main Street, Springfield, Mich.",),
)
@@ -225,9 +207,10 @@ class TestDocumentConsumptionFinishedSignal(TestCase):
def setUp(self):
TestCase.setUp(self)
User.objects.create_user(username='test_consumer', password='12345')
User.objects.create_user(username="test_consumer", password="12345")
self.doc_contains = Document.objects.create(
content="I contain the keyword.", mime_type="application/pdf")
content="I contain the keyword.", mime_type="application/pdf"
)
self.index_dir = tempfile.mkdtemp()
# TODO: we should not need the index here.
@@ -238,40 +221,43 @@ class TestDocumentConsumptionFinishedSignal(TestCase):
def test_tag_applied_any(self):
t1 = Tag.objects.create(
name="test", match="keyword", matching_algorithm=Tag.MATCH_ANY)
name="test", match="keyword", matching_algorithm=Tag.MATCH_ANY
)
document_consumption_finished.send(
sender=self.__class__, document=self.doc_contains)
sender=self.__class__, document=self.doc_contains
)
self.assertTrue(list(self.doc_contains.tags.all()) == [t1])
def test_tag_not_applied(self):
Tag.objects.create(
name="test", match="no-match", matching_algorithm=Tag.MATCH_ANY)
name="test", match="no-match", matching_algorithm=Tag.MATCH_ANY
)
document_consumption_finished.send(
sender=self.__class__, document=self.doc_contains)
sender=self.__class__, document=self.doc_contains
)
self.assertTrue(list(self.doc_contains.tags.all()) == [])
def test_correspondent_applied(self):
correspondent = Correspondent.objects.create(
name="test",
match="keyword",
matching_algorithm=Correspondent.MATCH_ANY
name="test", match="keyword", matching_algorithm=Correspondent.MATCH_ANY
)
document_consumption_finished.send(
sender=self.__class__, document=self.doc_contains)
sender=self.__class__, document=self.doc_contains
)
self.assertTrue(self.doc_contains.correspondent == correspondent)
def test_correspondent_not_applied(self):
Tag.objects.create(
name="test",
match="no-match",
matching_algorithm=Correspondent.MATCH_ANY
name="test", match="no-match", matching_algorithm=Correspondent.MATCH_ANY
)
document_consumption_finished.send(
sender=self.__class__, document=self.doc_contains)
sender=self.__class__, document=self.doc_contains
)
self.assertEqual(self.doc_contains.correspondent, None)
def test_logentry_created(self):
document_consumption_finished.send(
sender=self.__class__, document=self.doc_contains)
sender=self.__class__, document=self.doc_contains
)
self.assertEqual(LogEntry.objects.count(), 1)

352
src/documents/tests/test_migration_archive_files.py
View File

@@ -24,20 +24,14 @@ def archive_path_old(self):
else:
fname = "{:07}.pdf".format(self.pk)
return os.path.join(
settings.ARCHIVE_DIR,
fname
)
return os.path.join(settings.ARCHIVE_DIR, fname)
def archive_path_new(doc):
if doc.archive_filename is not None:
return os.path.join(
settings.ARCHIVE_DIR,
str(doc.archive_filename)
)
else:
return None
if doc.archive_filename is not None:
return os.path.join(settings.ARCHIVE_DIR, str(doc.archive_filename))
else:
return None
def source_path(doc):
@@ -48,10 +42,7 @@ def source_path(doc):
if doc.storage_type == STORAGE_TYPE_GPG:
fname += ".gpg" # pragma: no cover
return os.path.join(
settings.ORIGINALS_DIR,
fname
)
return os.path.join(settings.ORIGINALS_DIR, fname)
def thumbnail_path(doc):
@@ -59,13 +50,18 @@ def thumbnail_path(doc):
if doc.storage_type == STORAGE_TYPE_GPG:
file_name += ".gpg"
return os.path.join(
settings.THUMBNAIL_DIR,
file_name
)
return os.path.join(settings.THUMBNAIL_DIR, file_name)
def make_test_document(document_class, title: str, mime_type: str, original: str, original_filename: str, archive: str = None, archive_filename: str = None):
def make_test_document(
document_class,
title: str,
mime_type: str,
original: str,
original_filename: str,
archive: str = None,
archive_filename: str = None,
):
doc = document_class()
doc.filename = original_filename
doc.title = title
@@ -96,8 +92,12 @@ def make_test_document(document_class, title: str, mime_type: str, original: str
simple_jpg = os.path.join(os.path.dirname(__file__), "samples", "simple.jpg")
simple_pdf = os.path.join(os.path.dirname(__file__), "samples", "simple.pdf")
simple_pdf2 = os.path.join(os.path.dirname(__file__), "samples", "documents", "originals", "0000002.pdf")
simple_pdf3 = os.path.join(os.path.dirname(__file__), "samples", "documents", "originals", "0000003.pdf")
simple_pdf2 = os.path.join(
os.path.dirname(__file__), "samples", "documents", "originals", "0000002.pdf"
)
simple_pdf3 = os.path.join(
os.path.dirname(__file__), "samples", "documents", "originals", "0000003.pdf"
)
simple_txt = os.path.join(os.path.dirname(__file__), "samples", "simple.txt")
simple_png = os.path.join(os.path.dirname(__file__), "samples", "simple-noalpha.png")
simple_png2 = os.path.join(os.path.dirname(__file__), "examples", "no-text.png")
@@ -106,26 +106,52 @@ simple_png2 = os.path.join(os.path.dirname(__file__), "examples", "no-text.png")
@override_settings(PAPERLESS_FILENAME_FORMAT="")
class TestMigrateArchiveFiles(DirectoriesMixin, TestMigrations):
migrate_from = '1011_auto_20210101_2340'
migrate_to = '1012_fix_archive_files'
migrate_from = "1011_auto_20210101_2340"
migrate_to = "1012_fix_archive_files"
def setUpBeforeMigration(self, apps):
Document = apps.get_model("documents", "Document")
self.unrelated = make_test_document(Document, "unrelated", "application/pdf", simple_pdf3, "unrelated.pdf", simple_pdf)
self.no_text = make_test_document(Document, "no-text", "image/png", simple_png2, "no-text.png", simple_pdf)
self.doc_no_archive = make_test_document(Document, "no_archive", "text/plain", simple_txt, "no_archive.txt")
self.clash1 = make_test_document(Document, "clash", "application/pdf", simple_pdf, "clash.pdf", simple_pdf)
self.clash2 = make_test_document(Document, "clash", "image/jpeg", simple_jpg, "clash.jpg", simple_pdf)
self.clash3 = make_test_document(Document, "clash", "image/png", simple_png, "clash.png", simple_pdf)
self.clash4 = make_test_document(Document, "clash.png", "application/pdf", simple_pdf2, "clash.png.pdf", simple_pdf2)
self.unrelated = make_test_document(
Document,
"unrelated",
"application/pdf",
simple_pdf3,
"unrelated.pdf",
simple_pdf,
)
self.no_text = make_test_document(
Document, "no-text", "image/png", simple_png2, "no-text.png", simple_pdf
)
self.doc_no_archive = make_test_document(
Document, "no_archive", "text/plain", simple_txt, "no_archive.txt"
)
self.clash1 = make_test_document(
Document, "clash", "application/pdf", simple_pdf, "clash.pdf", simple_pdf
)
self.clash2 = make_test_document(
Document, "clash", "image/jpeg", simple_jpg, "clash.jpg", simple_pdf
)
self.clash3 = make_test_document(
Document, "clash", "image/png", simple_png, "clash.png", simple_pdf
)
self.clash4 = make_test_document(
Document,
"clash.png",
"application/pdf",
simple_pdf2,
"clash.png.pdf",
simple_pdf2,
)
self.assertEqual(archive_path_old(self.clash1), archive_path_old(self.clash2))
self.assertEqual(archive_path_old(self.clash1), archive_path_old(self.clash3))
self.assertNotEqual(archive_path_old(self.clash1), archive_path_old(self.clash4))
self.assertNotEqual(
archive_path_old(self.clash1), archive_path_old(self.clash4)
)
def testArchiveFilesMigrated(self):
Document = self.apps.get_model('documents', 'Document')
Document = self.apps.get_model("documents", "Document")
for doc in Document.objects.all():
if doc.archive_checksum:
@@ -144,31 +170,65 @@ class TestMigrateArchiveFiles(DirectoriesMixin, TestMigrations):
archive_checksum = hashlib.md5(f.read()).hexdigest()
self.assertEqual(archive_checksum, doc.archive_checksum)
self.assertEqual(Document.objects.filter(archive_checksum__isnull=False).count(), 6)
self.assertEqual(
Document.objects.filter(archive_checksum__isnull=False).count(), 6
)
def test_filenames(self):
Document = self.apps.get_model('documents', 'Document')
self.assertEqual(Document.objects.get(id=self.unrelated.id).archive_filename, "unrelated.pdf")
self.assertEqual(Document.objects.get(id=self.no_text.id).archive_filename, "no-text.pdf")
self.assertEqual(Document.objects.get(id=self.doc_no_archive.id).archive_filename, None)
self.assertEqual(Document.objects.get(id=self.clash1.id).archive_filename, f"{self.clash1.id:07}.pdf")
self.assertEqual(Document.objects.get(id=self.clash2.id).archive_filename, f"{self.clash2.id:07}.pdf")
self.assertEqual(Document.objects.get(id=self.clash3.id).archive_filename, f"{self.clash3.id:07}.pdf")
self.assertEqual(Document.objects.get(id=self.clash4.id).archive_filename, "clash.png.pdf")
Document = self.apps.get_model("documents", "Document")
self.assertEqual(
Document.objects.get(id=self.unrelated.id).archive_filename, "unrelated.pdf"
)
self.assertEqual(
Document.objects.get(id=self.no_text.id).archive_filename, "no-text.pdf"
)
self.assertEqual(
Document.objects.get(id=self.doc_no_archive.id).archive_filename, None
)
self.assertEqual(
Document.objects.get(id=self.clash1.id).archive_filename,
f"{self.clash1.id:07}.pdf",
)
self.assertEqual(
Document.objects.get(id=self.clash2.id).archive_filename,
f"{self.clash2.id:07}.pdf",
)
self.assertEqual(
Document.objects.get(id=self.clash3.id).archive_filename,
f"{self.clash3.id:07}.pdf",
)
self.assertEqual(
Document.objects.get(id=self.clash4.id).archive_filename, "clash.png.pdf"
)
@override_settings(PAPERLESS_FILENAME_FORMAT="{correspondent}/{title}")
class TestMigrateArchiveFilesWithFilenameFormat(TestMigrateArchiveFiles):
def test_filenames(self):
Document = self.apps.get_model('documents', 'Document')
self.assertEqual(Document.objects.get(id=self.unrelated.id).archive_filename, "unrelated.pdf")
self.assertEqual(Document.objects.get(id=self.no_text.id).archive_filename, "no-text.pdf")
self.assertEqual(Document.objects.get(id=self.doc_no_archive.id).archive_filename, None)
self.assertEqual(Document.objects.get(id=self.clash1.id).archive_filename, "none/clash.pdf")
self.assertEqual(Document.objects.get(id=self.clash2.id).archive_filename, "none/clash_01.pdf")
self.assertEqual(Document.objects.get(id=self.clash3.id).archive_filename, "none/clash_02.pdf")
self.assertEqual(Document.objects.get(id=self.clash4.id).archive_filename, "clash.png.pdf")
Document = self.apps.get_model("documents", "Document")
self.assertEqual(
Document.objects.get(id=self.unrelated.id).archive_filename, "unrelated.pdf"
)
self.assertEqual(
Document.objects.get(id=self.no_text.id).archive_filename, "no-text.pdf"
)
self.assertEqual(
Document.objects.get(id=self.doc_no_archive.id).archive_filename, None
)
self.assertEqual(
Document.objects.get(id=self.clash1.id).archive_filename, "none/clash.pdf"
)
self.assertEqual(
Document.objects.get(id=self.clash2.id).archive_filename,
"none/clash_01.pdf",
)
self.assertEqual(
Document.objects.get(id=self.clash3.id).archive_filename,
"none/clash_02.pdf",
)
self.assertEqual(
Document.objects.get(id=self.clash4.id).archive_filename, "clash.png.pdf"
)
def fake_parse_wrapper(parser, path, mime_type, file_name):
@@ -179,34 +239,63 @@ def fake_parse_wrapper(parser, path, mime_type, file_name):
@override_settings(PAPERLESS_FILENAME_FORMAT="")
class TestMigrateArchiveFilesErrors(DirectoriesMixin, TestMigrations):
migrate_from = '1011_auto_20210101_2340'
migrate_to = '1012_fix_archive_files'
migrate_from = "1011_auto_20210101_2340"
migrate_to = "1012_fix_archive_files"
auto_migrate = False
def test_archive_missing(self):
Document = self.apps.get_model("documents", "Document")
doc = make_test_document(Document, "clash", "application/pdf", simple_pdf, "clash.pdf", simple_pdf)
doc = make_test_document(
Document, "clash", "application/pdf", simple_pdf, "clash.pdf", simple_pdf
)
os.unlink(archive_path_old(doc))
self.assertRaisesMessage(ValueError, "does not exist at: ", self.performMigration)
self.assertRaisesMessage(
ValueError, "does not exist at: ", self.performMigration
)
def test_parser_missing(self):
Document = self.apps.get_model("documents", "Document")
doc1 = make_test_document(Document, "document", "invalid/typesss768", simple_png, "document.png", simple_pdf)
doc2 = make_test_document(Document, "document", "invalid/typesss768", simple_jpg, "document.jpg", simple_pdf)
doc1 = make_test_document(
Document,
"document",
"invalid/typesss768",
simple_png,
"document.png",
simple_pdf,
)
doc2 = make_test_document(
Document,
"document",
"invalid/typesss768",
simple_jpg,
"document.jpg",
simple_pdf,
)
self.assertRaisesMessage(ValueError, "no parsers are available", self.performMigration)
self.assertRaisesMessage(
ValueError, "no parsers are available", self.performMigration
)
@mock.patch("documents.migrations.1012_fix_archive_files.parse_wrapper")
def test_parser_error(self, m):
m.side_effect = ParseError()
Document = self.apps.get_model("documents", "Document")
doc1 = make_test_document(Document, "document", "image/png", simple_png, "document.png", simple_pdf)
doc2 = make_test_document(Document, "document", "application/pdf", simple_jpg, "document.jpg", simple_pdf)
doc1 = make_test_document(
Document, "document", "image/png", simple_png, "document.png", simple_pdf
)
doc2 = make_test_document(
Document,
"document",
"application/pdf",
simple_jpg,
"document.jpg",
simple_pdf,
)
self.assertIsNotNone(doc1.archive_checksum)
self.assertIsNotNone(doc2.archive_checksum)
@@ -217,12 +306,29 @@ class TestMigrateArchiveFilesErrors(DirectoriesMixin, TestMigrations):
self.assertEqual(m.call_count, 6)
self.assertEqual(
len(list(filter(lambda log: "Parse error, will try again in 5 seconds" in log, capture.output))),
4)
len(
list(
filter(
lambda log: "Parse error, will try again in 5 seconds" in log,
capture.output,
)
)
),
4,
)
self.assertEqual(
len(list(filter(lambda log: "Unable to regenerate archive document for ID:" in log, capture.output))),
2)
len(
list(
filter(
lambda log: "Unable to regenerate archive document for ID:"
in log,
capture.output,
)
)
),
2,
)
Document = self.apps.get_model("documents", "Document")
@@ -240,15 +346,33 @@ class TestMigrateArchiveFilesErrors(DirectoriesMixin, TestMigrations):
Document = self.apps.get_model("documents", "Document")
doc1 = make_test_document(Document, "document", "image/png", simple_png, "document.png", simple_pdf)
doc2 = make_test_document(Document, "document", "application/pdf", simple_jpg, "document.jpg", simple_pdf)
doc1 = make_test_document(
Document, "document", "image/png", simple_png, "document.png", simple_pdf
)
doc2 = make_test_document(
Document,
"document",
"application/pdf",
simple_jpg,
"document.jpg",
simple_pdf,
)
with self.assertLogs() as capture:
self.performMigration()
self.assertEqual(
len(list(filter(lambda log: "Parser did not return an archive document for document" in log, capture.output))),
2)
len(
list(
filter(
lambda log: "Parser did not return an archive document for document"
in log,
capture.output,
)
)
),
2,
)
Document = self.apps.get_model("documents", "Document")
@@ -264,19 +388,37 @@ class TestMigrateArchiveFilesErrors(DirectoriesMixin, TestMigrations):
@override_settings(PAPERLESS_FILENAME_FORMAT="")
class TestMigrateArchiveFilesBackwards(DirectoriesMixin, TestMigrations):
migrate_from = '1012_fix_archive_files'
migrate_to = '1011_auto_20210101_2340'
migrate_from = "1012_fix_archive_files"
migrate_to = "1011_auto_20210101_2340"
def setUpBeforeMigration(self, apps):
Document = apps.get_model("documents", "Document")
doc_unrelated = make_test_document(Document, "unrelated", "application/pdf", simple_pdf2, "unrelated.txt", simple_pdf2, "unrelated.pdf")
doc_no_archive = make_test_document(Document, "no_archive", "text/plain", simple_txt, "no_archive.txt")
clashB = make_test_document(Document, "clash", "image/jpeg", simple_jpg, "clash.jpg", simple_pdf, "clash_02.pdf")
doc_unrelated = make_test_document(
Document,
"unrelated",
"application/pdf",
simple_pdf2,
"unrelated.txt",
simple_pdf2,
"unrelated.pdf",
)
doc_no_archive = make_test_document(
Document, "no_archive", "text/plain", simple_txt, "no_archive.txt"
)
clashB = make_test_document(
Document,
"clash",
"image/jpeg",
simple_jpg,
"clash.jpg",
simple_pdf,
"clash_02.pdf",
)
def testArchiveFilesReverted(self):
Document = self.apps.get_model('documents', 'Document')
Document = self.apps.get_model("documents", "Document")
for doc in Document.objects.all():
if doc.archive_checksum:
@@ -291,35 +433,77 @@ class TestMigrateArchiveFilesBackwards(DirectoriesMixin, TestMigrations):
archive_checksum = hashlib.md5(f.read()).hexdigest()
self.assertEqual(archive_checksum, doc.archive_checksum)
self.assertEqual(Document.objects.filter(archive_checksum__isnull=False).count(), 2)
self.assertEqual(
Document.objects.filter(archive_checksum__isnull=False).count(), 2
)
@override_settings(PAPERLESS_FILENAME_FORMAT="{correspondent}/{title}")
class TestMigrateArchiveFilesBackwardsWithFilenameFormat(TestMigrateArchiveFilesBackwards):
class TestMigrateArchiveFilesBackwardsWithFilenameFormat(
TestMigrateArchiveFilesBackwards
):
pass
@override_settings(PAPERLESS_FILENAME_FORMAT="")
class TestMigrateArchiveFilesBackwardsErrors(DirectoriesMixin, TestMigrations):
migrate_from = '1012_fix_archive_files'
migrate_to = '1011_auto_20210101_2340'
migrate_from = "1012_fix_archive_files"
migrate_to = "1011_auto_20210101_2340"
auto_migrate = False
def test_filename_clash(self):
Document = self.apps.get_model("documents", "Document")
self.clashA = make_test_document(Document, "clash", "application/pdf", simple_pdf, "clash.pdf", simple_pdf, "clash_02.pdf")
self.clashB = make_test_document(Document, "clash", "image/jpeg", simple_jpg, "clash.jpg", simple_pdf, "clash_01.pdf")
self.clashA = make_test_document(
Document,
"clash",
"application/pdf",
simple_pdf,
"clash.pdf",
simple_pdf,
"clash_02.pdf",
)
self.clashB = make_test_document(
Document,
"clash",
"image/jpeg",
simple_jpg,
"clash.jpg",
simple_pdf,
"clash_01.pdf",
)
self.assertRaisesMessage(ValueError, "would clash with another archive filename", self.performMigration)
self.assertRaisesMessage(
ValueError,
"would clash with another archive filename",
self.performMigration,
)
def test_filename_exists(self):
Document = self.apps.get_model("documents", "Document")
self.clashA = make_test_document(Document, "clash", "application/pdf", simple_pdf, "clash.pdf", simple_pdf, "clash.pdf")
self.clashB = make_test_document(Document, "clash", "image/jpeg", simple_jpg, "clash.jpg", simple_pdf, "clash_01.pdf")
self.clashA = make_test_document(
Document,
"clash",
"application/pdf",
simple_pdf,
"clash.pdf",
simple_pdf,
"clash.pdf",
)
self.clashB = make_test_document(
Document,
"clash",
"image/jpeg",
simple_jpg,
"clash.jpg",
simple_pdf,
"clash_01.pdf",
)
self.assertRaisesMessage(ValueError, "file already exists.", self.performMigration)
self.assertRaisesMessage(
ValueError, "file already exists.", self.performMigration
)

55
src/documents/tests/test_migration_mime_type.py
View File

@@ -19,10 +19,7 @@ def source_path_before(self):
if self.storage_type == STORAGE_TYPE_GPG:
fname += ".gpg"
return os.path.join(
settings.ORIGINALS_DIR,
fname
)
return os.path.join(settings.ORIGINALS_DIR, fname)
def file_type_after(self):
@@ -37,30 +34,43 @@ def source_path_after(doc):
if doc.storage_type == STORAGE_TYPE_GPG:
fname += ".gpg" # pragma: no cover
return os.path.join(
settings.ORIGINALS_DIR,
fname
)
return os.path.join(settings.ORIGINALS_DIR, fname)
@override_settings(PASSPHRASE="test")
class TestMigrateMimeType(DirectoriesMixin, TestMigrations):
migrate_from = '1002_auto_20201111_1105'
migrate_to = '1003_mime_types'
migrate_from = "1002_auto_20201111_1105"
migrate_to = "1003_mime_types"
def setUpBeforeMigration(self, apps):
Document = apps.get_model("documents", "Document")
doc = Document.objects.create(title="test", file_type="pdf", filename="file1.pdf")
doc = Document.objects.create(
title="test", file_type="pdf", filename="file1.pdf"
)
self.doc_id = doc.id
shutil.copy(os.path.join(os.path.dirname(__file__), "samples", "simple.pdf"), source_path_before(doc))
shutil.copy(
os.path.join(os.path.dirname(__file__), "samples", "simple.pdf"),
source_path_before(doc),
)
doc2 = Document.objects.create(checksum="B", file_type="pdf", storage_type=STORAGE_TYPE_GPG)
doc2 = Document.objects.create(
checksum="B", file_type="pdf", storage_type=STORAGE_TYPE_GPG
)
self.doc2_id = doc2.id
shutil.copy(os.path.join(os.path.dirname(__file__), "samples", "documents", "originals", "0000004.pdf.gpg"), source_path_before(doc2))
shutil.copy(
os.path.join(
os.path.dirname(__file__),
"samples",
"documents",
"originals",
"0000004.pdf.gpg",
),
source_path_before(doc2),
)
def testMimeTypesMigrated(self):
Document = self.apps.get_model('documents', 'Document')
Document = self.apps.get_model("documents", "Document")
doc = Document.objects.get(id=self.doc_id)
self.assertEqual(doc.mime_type, "application/pdf")
@@ -72,17 +82,22 @@ class TestMigrateMimeType(DirectoriesMixin, TestMigrations):
@override_settings(PASSPHRASE="test")
class TestMigrateMimeTypeBackwards(DirectoriesMixin, TestMigrations):
migrate_from = '1003_mime_types'
migrate_to = '1002_auto_20201111_1105'
migrate_from = "1003_mime_types"
migrate_to = "1002_auto_20201111_1105"
def setUpBeforeMigration(self, apps):
Document = apps.get_model("documents", "Document")
doc = Document.objects.create(title="test", mime_type="application/pdf", filename="file1.pdf")
doc = Document.objects.create(
title="test", mime_type="application/pdf", filename="file1.pdf"
)
self.doc_id = doc.id
shutil.copy(os.path.join(os.path.dirname(__file__), "samples", "simple.pdf"), source_path_after(doc))
shutil.copy(
os.path.join(os.path.dirname(__file__), "samples", "simple.pdf"),
source_path_after(doc),
)
def testMimeTypesReverted(self):
Document = self.apps.get_model('documents', 'Document')
Document = self.apps.get_model("documents", "Document")
doc = Document.objects.get(id=self.doc_id)
self.assertEqual(doc.file_type, "pdf")

6
src/documents/tests/test_migration_remove_null_characters.py
View File

@@ -3,13 +3,13 @@ from documents.tests.utils import DirectoriesMixin, TestMigrations
class TestMigrateNullCharacters(DirectoriesMixin, TestMigrations):
migrate_from = '1014_auto_20210228_1614'
migrate_to = '1015_remove_null_characters'
migrate_from = "1014_auto_20210228_1614"
migrate_to = "1015_remove_null_characters"
def setUpBeforeMigration(self, apps):
Document = apps.get_model("documents", "Document")
self.doc = Document.objects.create(content="aaa\0bbb")
def testMimeTypesMigrated(self):
Document = self.apps.get_model('documents', 'Document')
Document = self.apps.get_model("documents", "Document")
self.assertNotIn("\0", Document.objects.get(id=self.doc.id).content)

12
src/documents/tests/test_migration_tag_colors.py
View File

@@ -3,8 +3,8 @@ from documents.tests.utils import DirectoriesMixin, TestMigrations
class TestMigrateTagColor(DirectoriesMixin, TestMigrations):
migrate_from = '1012_fix_archive_files'
migrate_to = '1013_migrate_tag_colour'
migrate_from = "1012_fix_archive_files"
migrate_to = "1013_migrate_tag_colour"
def setUpBeforeMigration(self, apps):
Tag = apps.get_model("documents", "Tag")
@@ -13,7 +13,7 @@ class TestMigrateTagColor(DirectoriesMixin, TestMigrations):
self.t3_id = Tag.objects.create(name="tag3", colour=5).id
def testMimeTypesMigrated(self):
Tag = self.apps.get_model('documents', 'Tag')
Tag = self.apps.get_model("documents", "Tag")
self.assertEqual(Tag.objects.get(id=self.t1_id).color, "#a6cee3")
self.assertEqual(Tag.objects.get(id=self.t2_id).color, "#a6cee3")
self.assertEqual(Tag.objects.get(id=self.t3_id).color, "#fb9a99")
@@ -21,8 +21,8 @@ class TestMigrateTagColor(DirectoriesMixin, TestMigrations):
class TestMigrateTagColorBackwards(DirectoriesMixin, TestMigrations):
migrate_from = '1013_migrate_tag_colour'
migrate_to = '1012_fix_archive_files'
migrate_from = "1013_migrate_tag_colour"
migrate_to = "1012_fix_archive_files"
def setUpBeforeMigration(self, apps):
Tag = apps.get_model("documents", "Tag")
@@ -31,7 +31,7 @@ class TestMigrateTagColorBackwards(DirectoriesMixin, TestMigrations):
self.t3_id = Tag.objects.create(name="tag3", color="#123456").id
def testMimeTypesReverted(self):
Tag = self.apps.get_model('documents', 'Tag')
Tag = self.apps.get_model("documents", "Tag")
self.assertEqual(Tag.objects.get(id=self.t1_id).colour, 1)
self.assertEqual(Tag.objects.get(id=self.t2_id).colour, 9)
self.assertEqual(Tag.objects.get(id=self.t3_id).colour, 1)

2
src/documents/tests/test_models.py
View File

@@ -5,7 +5,6 @@ from ..models import Document, Correspondent
class CorrespondentTestCase(TestCase):
def test___str__(self):
for s in ("test", "οχι", "test with fun_charÅc'\"terß"):
correspondent = CorrespondentFactory.create(name=s)
@@ -13,7 +12,6 @@ class CorrespondentTestCase(TestCase):
class DocumentTestCase(TestCase):
def test_correspondent_deletion_does_not_cascade(self):
self.assertEqual(Correspondent.objects.all().count(), 0)

91
src/documents/tests/test_parsers.py
View File

@@ -6,8 +6,14 @@ from unittest import mock
from django.test import TestCase, override_settings
from documents.parsers import get_parser_class, get_supported_file_extensions, get_default_file_extension, \
get_parser_class_for_mime_type, DocumentParser, is_file_ext_supported
from documents.parsers import (
get_parser_class,
get_supported_file_extensions,
get_default_file_extension,
get_parser_class_for_mime_type,
DocumentParser,
is_file_ext_supported,
)
from paperless_tesseract.parsers import RasterisedDocumentParser
from paperless_text.parsers import TextDocumentParser
@@ -25,24 +31,26 @@ def fake_magic_from_file(file, mime=False):
@mock.patch("documents.parsers.magic.from_file", fake_magic_from_file)
class TestParserDiscovery(TestCase):
@mock.patch("documents.parsers.document_consumer_declaration.send")
def test__get_parser_class_1_parser(self, m, *args):
class DummyParser(object):
pass
m.return_value = (
(None, {"weight": 0, "parser": DummyParser, "mime_types": {"application/pdf": ".pdf"}}),
(
None,
{
"weight": 0,
"parser": DummyParser,
"mime_types": {"application/pdf": ".pdf"},
},
),
)
self.assertEqual(
get_parser_class("doc.pdf"),
DummyParser
)
self.assertEqual(get_parser_class("doc.pdf"), DummyParser)
@mock.patch("documents.parsers.document_consumer_declaration.send")
def test__get_parser_class_n_parsers(self, m, *args):
class DummyParser1(object):
pass
@@ -50,22 +58,31 @@ class TestParserDiscovery(TestCase):
pass
m.return_value = (
(None, {"weight": 0, "parser": DummyParser1, "mime_types": {"application/pdf": ".pdf"}}),
(None, {"weight": 1, "parser": DummyParser2, "mime_types": {"application/pdf": ".pdf"}}),
(
None,
{
"weight": 0,
"parser": DummyParser1,
"mime_types": {"application/pdf": ".pdf"},
},
),
(
None,
{
"weight": 1,
"parser": DummyParser2,
"mime_types": {"application/pdf": ".pdf"},
},
),
)
self.assertEqual(
get_parser_class("doc.pdf"),
DummyParser2
)
self.assertEqual(get_parser_class("doc.pdf"), DummyParser2)
@mock.patch("documents.parsers.document_consumer_declaration.send")
def test__get_parser_class_0_parsers(self, m, *args):
m.return_value = []
with TemporaryDirectory() as tmpdir:
self.assertIsNone(
get_parser_class("doc.pdf")
)
self.assertIsNone(get_parser_class("doc.pdf"))
def fake_get_thumbnail(self, path, mimetype, file_name):
@@ -73,13 +90,10 @@ def fake_get_thumbnail(self, path, mimetype, file_name):
class TestBaseParser(TestCase):
def setUp(self) -> None:
self.scratch = tempfile.mkdtemp()
override_settings(
SCRATCH_DIR=self.scratch
).enable()
override_settings(SCRATCH_DIR=self.scratch).enable()
def tearDown(self) -> None:
shutil.rmtree(self.scratch)
@@ -101,23 +115,28 @@ class TestBaseParser(TestCase):
class TestParserAvailability(TestCase):
def test_file_extensions(self):
for ext in [".pdf", ".jpe", ".jpg", ".jpeg", ".txt", ".csv"]:
self.assertIn(ext, get_supported_file_extensions())
self.assertEqual(get_default_file_extension('application/pdf'), ".pdf")
self.assertEqual(get_default_file_extension('image/png'), ".png")
self.assertEqual(get_default_file_extension('image/jpeg'), ".jpg")
self.assertEqual(get_default_file_extension('text/plain'), ".txt")
self.assertEqual(get_default_file_extension('text/csv'), ".csv")
self.assertEqual(get_default_file_extension('application/zip'), ".zip")
self.assertEqual(get_default_file_extension('aasdasd/dgfgf'), "")
self.assertEqual(get_default_file_extension("application/pdf"), ".pdf")
self.assertEqual(get_default_file_extension("image/png"), ".png")
self.assertEqual(get_default_file_extension("image/jpeg"), ".jpg")
self.assertEqual(get_default_file_extension("text/plain"), ".txt")
self.assertEqual(get_default_file_extension("text/csv"), ".csv")
self.assertEqual(get_default_file_extension("application/zip"), ".zip")
self.assertEqual(get_default_file_extension("aasdasd/dgfgf"), "")
self.assertIsInstance(get_parser_class_for_mime_type('application/pdf')(logging_group=None), RasterisedDocumentParser)
self.assertIsInstance(get_parser_class_for_mime_type('text/plain')(logging_group=None), TextDocumentParser)
self.assertEqual(get_parser_class_for_mime_type('text/sdgsdf'), None)
self.assertIsInstance(
get_parser_class_for_mime_type("application/pdf")(logging_group=None),
RasterisedDocumentParser,
)
self.assertIsInstance(
get_parser_class_for_mime_type("text/plain")(logging_group=None),
TextDocumentParser,
)
self.assertEqual(get_parser_class_for_mime_type("text/sdgsdf"), None)
self.assertTrue(is_file_ext_supported('.pdf'))
self.assertFalse(is_file_ext_supported('.hsdfh'))
self.assertFalse(is_file_ext_supported(''))
self.assertTrue(is_file_ext_supported(".pdf"))
self.assertFalse(is_file_ext_supported(".hsdfh"))
self.assertFalse(is_file_ext_supported(""))

57
src/documents/tests/test_sanity_check.py
View File

@@ -13,7 +13,6 @@ from documents.tests.utils import DirectoriesMixin
class TestSanityCheckMessages(TestCase):
def test_no_messages(self):
messages = SanityCheckMessages()
self.assertEqual(len(messages), 0)
@@ -23,7 +22,9 @@ class TestSanityCheckMessages(TestCase):
messages.log_messages()
self.assertEqual(len(capture.output), 1)
self.assertEqual(capture.records[0].levelno, logging.INFO)
self.assertEqual(capture.records[0].message, "Sanity checker detected no issues.")
self.assertEqual(
capture.records[0].message, "Sanity checker detected no issues."
)
def test_info(self):
messages = SanityCheckMessages()
@@ -61,22 +62,58 @@ class TestSanityCheckMessages(TestCase):
self.assertEqual(capture.records[0].levelno, logging.ERROR)
self.assertEqual(capture.records[0].message, "Something is seriously wrong")
class TestSanityCheck(DirectoriesMixin, TestCase):
class TestSanityCheck(DirectoriesMixin, TestCase):
def make_test_data(self):
with filelock.FileLock(settings.MEDIA_LOCK):
# just make sure that the lockfile is present.
shutil.copy(os.path.join(os.path.dirname(__file__), "samples", "documents", "originals", "0000001.pdf"), os.path.join(self.dirs.originals_dir, "0000001.pdf"))
shutil.copy(os.path.join(os.path.dirname(__file__), "samples", "documents", "archive", "0000001.pdf"), os.path.join(self.dirs.archive_dir, "0000001.pdf"))
shutil.copy(os.path.join(os.path.dirname(__file__), "samples", "documents", "thumbnails", "0000001.png"), os.path.join(self.dirs.thumbnail_dir, "0000001.png"))
shutil.copy(
os.path.join(
os.path.dirname(__file__),
"samples",
"documents",
"originals",
"0000001.pdf",
),
os.path.join(self.dirs.originals_dir, "0000001.pdf"),
)
shutil.copy(
os.path.join(
os.path.dirname(__file__),
"samples",
"documents",
"archive",
"0000001.pdf",
),
os.path.join(self.dirs.archive_dir, "0000001.pdf"),
)
shutil.copy(
os.path.join(
os.path.dirname(__file__),
"samples",
"documents",
"thumbnails",
"0000001.png",
),
os.path.join(self.dirs.thumbnail_dir, "0000001.png"),
)
return Document.objects.create(title="test", checksum="42995833e01aea9b3edee44bbfdd7ce1", archive_checksum="62acb0bcbfbcaa62ca6ad3668e4e404b", content="test", pk=1, filename="0000001.pdf", mime_type="application/pdf", archive_filename="0000001.pdf")
return Document.objects.create(
title="test",
checksum="42995833e01aea9b3edee44bbfdd7ce1",
archive_checksum="62acb0bcbfbcaa62ca6ad3668e4e404b",
content="test",
pk=1,
filename="0000001.pdf",
mime_type="application/pdf",
archive_filename="0000001.pdf",
)
def assertSanityError(self, messageRegex):
messages = check_sanity()
self.assertTrue(messages.has_error())
self.assertRegex(messages[0]['message'], messageRegex)
self.assertRegex(messages[0]["message"], messageRegex)
def test_no_docs(self):
self.assertEqual(len(check_sanity()), 0)
@@ -138,7 +175,7 @@ class TestSanityCheck(DirectoriesMixin, TestCase):
self.assertFalse(messages.has_error())
self.assertFalse(messages.has_warning())
self.assertEqual(len(messages), 1)
self.assertRegex(messages[0]['message'], "Document .* has no content.")
self.assertRegex(messages[0]["message"], "Document .* has no content.")
def test_orphaned_file(self):
doc = self.make_test_data()
@@ -147,7 +184,7 @@ class TestSanityCheck(DirectoriesMixin, TestCase):
self.assertFalse(messages.has_error())
self.assertTrue(messages.has_warning())
self.assertEqual(len(messages), 1)
self.assertRegex(messages[0]['message'], "Orphaned file in media dir")
self.assertRegex(messages[0]["message"], "Orphaned file in media dir")
def test_archive_filename_no_checksum(self):
doc = self.make_test_data()

5
src/documents/tests/test_settings.py
View File

@@ -7,7 +7,6 @@ from paperless.settings import default_task_workers, default_threads_per_worker
class TestSettings(TestCase):
@mock.patch("paperless.settings.multiprocessing.cpu_count")
def test_single_core(self, cpu_count):
cpu_count.return_value = 1
@@ -21,7 +20,9 @@ class TestSettings(TestCase):
def test_workers_threads(self):
for i in range(1, 64):
with mock.patch("paperless.settings.multiprocessing.cpu_count") as cpu_count:
with mock.patch(
"paperless.settings.multiprocessing.cpu_count"
) as cpu_count:
cpu_count.return_value = i
default_workers = default_task_workers()

37
src/documents/tests/test_tasks.py
View File

@@ -12,14 +12,27 @@ from documents.tests.utils import DirectoriesMixin
class TestTasks(DirectoriesMixin, TestCase):
def test_index_reindex(self):
Document.objects.create(title="test", content="my document", checksum="wow", added=timezone.now(), created=timezone.now(), modified=timezone.now())
Document.objects.create(
title="test",
content="my document",
checksum="wow",
added=timezone.now(),
created=timezone.now(),
modified=timezone.now(),
)
tasks.index_reindex()
def test_index_optimize(self):
Document.objects.create(title="test", content="my document", checksum="wow", added=timezone.now(), created=timezone.now(), modified=timezone.now())
Document.objects.create(
title="test",
content="my document",
checksum="wow",
added=timezone.now(),
created=timezone.now(),
modified=timezone.now(),
)
tasks.index_optimize()
@@ -92,7 +105,9 @@ class TestTasks(DirectoriesMixin, TestCase):
messages = SanityCheckMessages()
messages.warning("Some warning")
m.return_value = messages
self.assertEqual(tasks.sanity_check(), "Sanity check exited with warnings. See log.")
self.assertEqual(
tasks.sanity_check(), "Sanity check exited with warnings. See log."
)
m.assert_called_once()
@mock.patch("documents.tasks.sanity_checker.check_sanity")
@@ -100,11 +115,19 @@ class TestTasks(DirectoriesMixin, TestCase):
messages = SanityCheckMessages()
messages.info("Some info")
m.return_value = messages
self.assertEqual(tasks.sanity_check(), "Sanity check exited with infos. See log.")
self.assertEqual(
tasks.sanity_check(), "Sanity check exited with infos. See log."
)
m.assert_called_once()
def test_bulk_update_documents(self):
doc1 = Document.objects.create(title="test", content="my document", checksum="wow", added=timezone.now(),
created=timezone.now(), modified=timezone.now())
doc1 = Document.objects.create(
title="test",
content="my document",
checksum="wow",
added=timezone.now(),
created=timezone.now(),
modified=timezone.now(),
)
tasks.bulk_update_documents([doc1.pk])

45
src/documents/tests/test_views.py
View File

@@ -4,27 +4,52 @@ from django.test import TestCase
class TestViews(TestCase):
def setUp(self) -> None:
self.user = User.objects.create_user("testuser")
def test_login_redirect(self):
response = self.client.get('/')
response = self.client.get("/")
self.assertEqual(response.status_code, 302)
self.assertEqual(response.url, "/accounts/login/?next=/")
def test_index(self):
self.client.force_login(self.user)
for (language_given, language_actual) in [("", "en-US"), ("en-US", "en-US"), ("de", "de-DE"), ("en", "en-US"), ("en-us", "en-US"), ("fr", "fr-FR"), ("jp", "en-US")]:
for (language_given, language_actual) in [
("", "en-US"),
("en-US", "en-US"),
("de", "de-DE"),
("en", "en-US"),
("en-us", "en-US"),
("fr", "fr-FR"),
("jp", "en-US"),
]:
if language_given:
self.client.cookies.load({settings.LANGUAGE_COOKIE_NAME: language_given})
self.client.cookies.load(
{settings.LANGUAGE_COOKIE_NAME: language_given}
)
elif settings.LANGUAGE_COOKIE_NAME in self.client.cookies.keys():
self.client.cookies.pop(settings.LANGUAGE_COOKIE_NAME)
response = self.client.get('/', )
response = self.client.get(
"/",
)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.context_data['webmanifest'], f"frontend/{language_actual}/manifest.webmanifest")
self.assertEqual(response.context_data['styles_css'], f"frontend/{language_actual}/styles.css")
self.assertEqual(response.context_data['runtime_js'], f"frontend/{language_actual}/runtime.js")
self.assertEqual(response.context_data['polyfills_js'], f"frontend/{language_actual}/polyfills.js")
self.assertEqual(response.context_data['main_js'], f"frontend/{language_actual}/main.js")
self.assertEqual(
response.context_data["webmanifest"],
f"frontend/{language_actual}/manifest.webmanifest",
)
self.assertEqual(
response.context_data["styles_css"],
f"frontend/{language_actual}/styles.css",
)
self.assertEqual(
response.context_data["runtime_js"],
f"frontend/{language_actual}/runtime.js",
)
self.assertEqual(
response.context_data["polyfills_js"],
f"frontend/{language_actual}/polyfills.js",
)
self.assertEqual(
response.context_data["main_js"], f"frontend/{language_actual}/main.js"
)

12
src/documents/tests/utils.py
View File

@@ -42,8 +42,7 @@ def setup_directories():
LOGGING_DIR=dirs.logging_dir,
INDEX_DIR=dirs.index_dir,
MODEL_FILE=os.path.join(dirs.data_dir, "classification_model.pickle"),
MEDIA_LOCK=os.path.join(dirs.media_dir, "media.lock")
MEDIA_LOCK=os.path.join(dirs.media_dir, "media.lock"),
)
dirs.settings_override.enable()
@@ -70,7 +69,6 @@ def paperless_environment():
class DirectoriesMixin:
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.dirs = None
@@ -85,7 +83,6 @@ class DirectoriesMixin:
class TestMigrations(TransactionTestCase):
@property
def app(self):
return apps.get_containing_app_config(type(self).__module__).name
@@ -97,8 +94,11 @@ class TestMigrations(TransactionTestCase):
def setUp(self):
super(TestMigrations, self).setUp()
assert self.migrate_from and self.migrate_to, \
"TestCase '{}' must define migrate_from and migrate_to properties".format(type(self).__name__)
assert (
self.migrate_from and self.migrate_to
), "TestCase '{}' must define migrate_from and migrate_to properties".format(
type(self).__name__
)
self.migrate_from = [(self.app, self.migrate_from)]
self.migrate_to = [(self.app, self.migrate_to)]
executor = MigrationExecutor(connection)