Cutlery/paperless-ngx
1
0
Fork 0
mirror of https://github.com/paperless-ngx/paperless-ngx.git synced 2026-04-21 01:08:48 -04:00
Code Issues Packages Projects Releases Wiki Activity

Update tests

Browse Source
This commit is contained in:
shamoon 2026-03-17 11:46:44 -07:00
parent 8cff99bef3
commit b8faae72ab
No known key found for this signature in database
1 changed files with 10 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -12,8 +12,6 @@ from django.test import override_settings
from django.utils import timezone
from documents import tasks
from documents.data_models import ConsumableDocument
from documents.data_models import DocumentSource
from documents.models import Correspondent
from documents.models import Document
from documents.models import DocumentType
@ -21,11 +19,8 @@ from documents.models import PaperlessTask
from documents.models import Tag
from documents.sanity_checker import SanityCheckFailedException
from documents.sanity_checker import SanityCheckMessages
from documents.signals.handlers import before_task_publish_handler
from documents.signals.handlers import task_failure_handler
from documents.tests.test_classifier import dummy_preprocess
from documents.tests.utils import DirectoriesMixin
from documents.tests.utils import DummyProgressManager
from documents.tests.utils import FileSystemAssertsMixin
from documents.tests.utils import SampleDirMixin
@ -246,44 +241,18 @@ class TestRetryConsumeTask(
TestCase,
):
def do_failed_task(self, test_file: Path) -> PaperlessTask:
temp_copy = self.dirs.scratch_dir / test_file.name
shutil.copy(test_file, temp_copy)
failed_file = settings.CONSUMPTION_FAILED_DIR / test_file.name
shutil.copy(test_file, failed_file)
headers = {
"id": str(uuid.uuid4()),
"task": "documents.tasks.consume_file",
}
body = (
# args
(
ConsumableDocument(
source=DocumentSource.ConsumeFolder,
original_file=str(temp_copy),
),
None,
),
# kwargs
{},
# celery metadata
{"callbacks": None, "errbacks": None, "chain": None, "chord": None},
task = PaperlessTask.objects.create(
type=PaperlessTask.TaskType.AUTO,
task_id=str(uuid.uuid4()),
task_file_name=failed_file.name,
task_name=PaperlessTask.TaskName.CONSUME_FILE,
status=states.FAILURE,
date_created=timezone.now(),
date_done=timezone.now(),
)
before_task_publish_handler(headers=headers, body=body)
with mock.patch("documents.tasks.ProgressManager", DummyProgressManager):
with self.assertRaises(Exception):
tasks.consume_file(
ConsumableDocument(
source=DocumentSource.ConsumeFolder,
original_file=temp_copy,
),
)
task_failure_handler(
task_id=headers["id"],
exception="Example failure",
)
task = PaperlessTask.objects.first()
self.assertIsFile(settings.CONSUMPTION_FAILED_DIR / task.task_file_name)
return task