amorfo77/paperless-ngx
1
0
Fork 0
mirror of https://github.com/paperless-ngx/paperless-ngx.git synced 2026-07-18 01:44:53 +00:00
Code Issues Projects Releases Wiki Activity

test(tasks): parametrize repetitive test cases in task test files

Browse Source 
test_api_tasks.py:
- Collapse six trigger_source->v9-type tests into one parametrized test,
  adding the previously untested API_UPLOAD case
- Collapse three task_name mapping tests (two remaps + pass-through)
  into one parametrized test
- Collapse two acknowledge_all status tests into one parametrized test
- Collapse two run-endpoint 400 tests into one parametrized test
- Update run/ assertions to use TriggerSource enum values

test_task_signals.py:
- Collapse three trigger_source header tests into one parametrized test
- Collapse two DocumentSource->TriggerSource mapping tests into one
  parametrized test
- Collapse two prerun ignore-invalid-id tests into one parametrized test

All parametrize cases use pytest.param with descriptive ids.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
stumpylog
2026-04-16 10:46:26 -07:00
co-authored by Claude Sonnet 4.6
parent 78430892d1
commit 4d9a26c161
2 changed files with 187 additions and 168 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/documents/tests/test_api_tasks.py
+122 -105
View File
@@ -207,14 +207,93 @@ class TestGetTasksV10:
@pytest.mark.django_db()
class TestGetTasksV9:
def test_task_name_equals_task_type_value(self, v9_client: APIClient) -> None:
"""task_name mirrors the task_type value for v9 backwards compatibility."""
PaperlessTaskFactory(task_type=PaperlessTask.TaskType.CONSUME_FILE)
@pytest.mark.parametrize(
("task_type", "expected_task_name"),
[
pytest.param(
PaperlessTask.TaskType.CONSUME_FILE,
"consume_file",
id="consume_file-passthrough",
),
pytest.param(
PaperlessTask.TaskType.SANITY_CHECK,
"check_sanity",
id="sanity_check-remapped",
),
pytest.param(
PaperlessTask.TaskType.LLM_INDEX,
"llmindex_update",
id="llm_index-remapped",
),
],
)
def test_task_name_mapping(
self,
v9_client: APIClient,
task_type: PaperlessTask.TaskType,
expected_task_name: str,
) -> None:
"""v9 task_name is either a direct pass-through or a legacy remap of task_type."""
PaperlessTaskFactory(task_type=task_type)
response = v9_client.get(ENDPOINT)
assert response.status_code == status.HTTP_200_OK
assert response.data[0]["task_name"] == "consume_file"
assert response.data[0]["task_name"] == expected_task_name
@pytest.mark.parametrize(
("trigger_source", "expected_type"),
[
pytest.param(
PaperlessTask.TriggerSource.SCHEDULED,
"scheduled_task",
id="scheduled",
),
pytest.param(
PaperlessTask.TriggerSource.SYSTEM,
"auto_task",
id="system",
),
pytest.param(
PaperlessTask.TriggerSource.EMAIL_CONSUME,
"auto_task",
id="email_consume",
),
pytest.param(
PaperlessTask.TriggerSource.FOLDER_CONSUME,
"auto_task",
id="folder_consume",
),
pytest.param(
PaperlessTask.TriggerSource.WEB_UI,
"manual_task",
id="web_ui",
),
pytest.param(
PaperlessTask.TriggerSource.MANUAL,
"manual_task",
id="manual",
),
pytest.param(
PaperlessTask.TriggerSource.API_UPLOAD,
"manual_task",
id="api_upload",
),
],
)
def test_trigger_source_maps_to_v9_type(
self,
v9_client: APIClient,
trigger_source: PaperlessTask.TriggerSource,
expected_type: str,
) -> None:
"""Every TriggerSource value maps to the correct v9 type string."""
PaperlessTaskFactory(trigger_source=trigger_source)
response = v9_client.get(ENDPOINT)
assert response.status_code == status.HTTP_200_OK
assert response.data[0]["type"] == expected_type
def test_task_file_name_from_input_data(self, v9_client: APIClient) -> None:
"""task_file_name is read from input_data['filename']."""
@@ -237,42 +316,6 @@ class TestGetTasksV9:
assert response.status_code == status.HTTP_200_OK
assert response.data[0]["task_file_name"] is None
def test_type_scheduled_maps_to_scheduled_task(self, v9_client: APIClient) -> None:
"""trigger_source=scheduled maps to type='scheduled_task' in v9."""
PaperlessTaskFactory(trigger_source=PaperlessTask.TriggerSource.SCHEDULED)
response = v9_client.get(ENDPOINT)
assert response.status_code == status.HTTP_200_OK
assert response.data[0]["type"] == "scheduled_task"
def test_type_system_maps_to_auto_task(self, v9_client: APIClient) -> None:
"""trigger_source=system maps to type='auto_task' in v9."""
PaperlessTaskFactory(trigger_source=PaperlessTask.TriggerSource.SYSTEM)
response = v9_client.get(ENDPOINT)
assert response.status_code == status.HTTP_200_OK
assert response.data[0]["type"] == "auto_task"
def test_type_web_ui_maps_to_manual_task(self, v9_client: APIClient) -> None:
"""trigger_source=web_ui maps to type='manual_task' in v9."""
PaperlessTaskFactory(trigger_source=PaperlessTask.TriggerSource.WEB_UI)
response = v9_client.get(ENDPOINT)
assert response.status_code == status.HTTP_200_OK
assert response.data[0]["type"] == "manual_task"
def test_type_manual_maps_to_manual_task(self, v9_client: APIClient) -> None:
"""trigger_source=manual maps to type='manual_task' in v9."""
PaperlessTaskFactory(trigger_source=PaperlessTask.TriggerSource.MANUAL)
response = v9_client.get(ENDPOINT)
assert response.status_code == status.HTTP_200_OK
assert response.data[0]["type"] == "manual_task"
def test_related_document_from_result_data_document_id(
self,
v9_client: APIClient,
@@ -341,42 +384,6 @@ class TestGetTasksV9:
statuses = {t["status"] for t in response.data}
assert statuses == {"SUCCESS", "PENDING", "FAILURE"}
def test_task_name_remap_sanity_check(self, v9_client: APIClient) -> None:
"""v9 remaps task_type=sanity_check to task_name=check_sanity."""
PaperlessTaskFactory(task_type=PaperlessTask.TaskType.SANITY_CHECK)
response = v9_client.get(ENDPOINT)
assert response.status_code == status.HTTP_200_OK
assert response.data[0]["task_name"] == "check_sanity"
def test_task_name_remap_llm_index(self, v9_client: APIClient) -> None:
"""v9 remaps task_type=llm_index to task_name=llmindex_update."""
PaperlessTaskFactory(task_type=PaperlessTask.TaskType.LLM_INDEX)
response = v9_client.get(ENDPOINT)
assert response.status_code == status.HTTP_200_OK
assert response.data[0]["task_name"] == "llmindex_update"
def test_type_email_consume_maps_to_auto_task(self, v9_client: APIClient) -> None:
"""trigger_source=email_consume maps to type='auto_task' in v9."""
PaperlessTaskFactory(trigger_source=PaperlessTask.TriggerSource.EMAIL_CONSUME)
response = v9_client.get(ENDPOINT)
assert response.status_code == status.HTTP_200_OK
assert response.data[0]["type"] == "auto_task"
def test_type_folder_consume_maps_to_auto_task(self, v9_client: APIClient) -> None:
"""trigger_source=folder_consume maps to type='auto_task' in v9."""
PaperlessTaskFactory(trigger_source=PaperlessTask.TriggerSource.FOLDER_CONSUME)
response = v9_client.get(ENDPOINT)
assert response.status_code == status.HTTP_200_OK
assert response.data[0]["type"] == "auto_task"
def test_filter_by_task_name_maps_old_value(self, v9_client: APIClient) -> None:
"""?task_name=check_sanity maps to task_type=sanity_check in v9."""
PaperlessTaskFactory(task_type=PaperlessTask.TaskType.SANITY_CHECK)
@@ -559,24 +566,27 @@ class TestAcknowledgeAll:
assert response.status_code == status.HTTP_200_OK
assert response.data == {"result": 1}
def test_skips_pending_and_started(self, admin_client: APIClient) -> None:
"""acknowledge_all/ does not touch PENDING or STARTED tasks."""
PaperlessTaskFactory(status=PaperlessTask.Status.PENDING)
PaperlessTaskFactory(status=PaperlessTask.Status.STARTED)
@pytest.mark.parametrize(
("task_status", "expected_count"),
[
pytest.param(PaperlessTask.Status.PENDING, 0, id="pending-excluded"),
pytest.param(PaperlessTask.Status.STARTED, 0, id="started-excluded"),
pytest.param(PaperlessTask.Status.REVOKED, 1, id="revoked-included"),
],
)
def test_acknowledge_all_by_status(
self,
admin_client: APIClient,
task_status: PaperlessTask.Status,
expected_count: int,
) -> None:
"""acknowledge_all/ ignores PENDING/STARTED and includes REVOKED."""
PaperlessTaskFactory(status=task_status, acknowledged=False)
response = admin_client.post(ENDPOINT + "acknowledge_all/")
assert response.status_code == status.HTTP_200_OK
assert response.data == {"result": 0}
def test_includes_revoked(self, admin_client: APIClient) -> None:
"""acknowledge_all/ marks REVOKED tasks as acknowledged."""
PaperlessTaskFactory(status=PaperlessTask.Status.REVOKED, acknowledged=False)
response = admin_client.post(ENDPOINT + "acknowledge_all/")
assert response.status_code == status.HTTP_200_OK
assert response.data == {"result": 1}
assert response.data == {"result": expected_count}
@pytest.mark.django_db()
@@ -679,24 +689,31 @@ class TestRun:
assert response.data == {"task_id": fake_task_id}
mock_apply_async.assert_called_once_with(
kwargs={},
headers={"trigger_source": "manual"},
headers={"trigger_source": PaperlessTask.TriggerSource.MANUAL},
)
def test_returns_400_for_consume_file(self, admin_client: APIClient) -> None:
"""consume_file cannot be manually triggered via the run endpoint."""
@pytest.mark.parametrize(
"task_type",
[
pytest.param(
PaperlessTask.TaskType.CONSUME_FILE,
id="consume_file-not-runnable",
),
pytest.param(
"not_a_real_type",
id="invalid-task-type",
),
],
)
def test_returns_400_for_non_runnable_task_type(
self,
admin_client: APIClient,
task_type: str,
) -> None:
"""run/ returns 400 for task types that cannot be manually triggered."""
response = admin_client.post(
ENDPOINT + "run/",
{"task_type": PaperlessTask.TaskType.CONSUME_FILE},
format="json",
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
def test_returns_400_for_invalid_task_type(self, admin_client: APIClient) -> None:
"""run/ returns 400 for an unrecognized task_type value."""
response = admin_client.post(
ENDPOINT + "run/",
{"task_type": "not_a_real_type"},
{"task_type": task_type},
format="json",
)
@@ -727,5 +744,5 @@ class TestRun:
assert response.data == {"task_id": fake_task_id}
mock_apply_async.assert_called_once_with(
kwargs={"raise_on_error": False},
headers={"trigger_source": "manual"},
headers={"trigger_source": PaperlessTask.TriggerSource.MANUAL},
)
src/documents/tests/test_task_signals.py
+65 -63
View File
@@ -1,13 +1,19 @@
import sys
import uuid
from unittest import mock
import pytest
import pytest_mock
from django.utils import timezone
from documents.data_models import ConsumableDocument
from documents.data_models import DocumentMetadataOverrides
from documents.data_models import DocumentSource
from documents.models import PaperlessTask
from documents.signals.handlers import before_task_publish_handler
from documents.signals.handlers import task_failure_handler
from documents.signals.handlers import task_postrun_handler
from documents.signals.handlers import task_prerun_handler
from documents.signals.handlers import task_revoked_handler
from documents.tests.factories import PaperlessTaskFactory
@@ -39,7 +45,6 @@ def send_publish(
kwargs: dict,
headers: dict | None = None,
) -> str:
from documents.signals.handlers import before_task_publish_handler
task_id = str(uuid.uuid4())
hdrs = {"task": task_name, "id": task_id, **(headers or {})}
@@ -89,97 +94,108 @@ class TestBeforeTaskPublishHandler:
task = PaperlessTask.objects.get(task_id=task_id)
assert task.input_data == {}
def test_scheduled_header_sets_trigger_source(self):
@pytest.mark.parametrize(
("header_value", "expected_trigger_source"),
[
pytest.param(
PaperlessTask.TriggerSource.SCHEDULED,
PaperlessTask.TriggerSource.SCHEDULED,
id="scheduled",
),
pytest.param(
PaperlessTask.TriggerSource.SYSTEM,
PaperlessTask.TriggerSource.SYSTEM,
id="system",
),
pytest.param(
"bogus_value",
PaperlessTask.TriggerSource.MANUAL,
id="invalid-falls-back-to-manual",
),
],
)
def test_trigger_source_header_resolution(
self,
header_value: str,
expected_trigger_source: PaperlessTask.TriggerSource,
) -> None:
"""trigger_source header maps to the expected TriggerSource; invalid values fall back to MANUAL."""
task_id = send_publish(
"documents.tasks.train_classifier",
(),
{},
headers={"trigger_source": PaperlessTask.TriggerSource.SCHEDULED},
headers={"trigger_source": header_value},
)
task = PaperlessTask.objects.get(task_id=task_id)
assert task.trigger_source == PaperlessTask.TriggerSource.SCHEDULED
def test_system_header_sets_trigger_source(self):
task_id = send_publish(
"documents.tasks.llmindex_index",
(),
{"rebuild": True},
headers={"trigger_source": PaperlessTask.TriggerSource.SYSTEM},
)
task = PaperlessTask.objects.get(task_id=task_id)
assert task.trigger_source == PaperlessTask.TriggerSource.SYSTEM
def test_invalid_header_falls_back_to_manual(self):
task_id = send_publish(
"documents.tasks.train_classifier",
(),
{},
headers={"trigger_source": "bogus_value"},
)
task = PaperlessTask.objects.get(task_id=task_id)
assert task.trigger_source == PaperlessTask.TriggerSource.MANUAL
assert task.trigger_source == expected_trigger_source
def test_ignores_untracked_task(self):
send_publish("documents.tasks.some_untracked_task", (), {})
assert PaperlessTask.objects.count() == 0
def test_ignores_none_headers(self):
from documents.signals.handlers import before_task_publish_handler
before_task_publish_handler(sender=None, headers=None, body=None)
assert PaperlessTask.objects.count() == 0
def test_consume_folder_source_maps_correctly(
@pytest.mark.parametrize(
("document_source", "expected_trigger_source"),
[
pytest.param(
DocumentSource.ConsumeFolder,
PaperlessTask.TriggerSource.FOLDER_CONSUME,
id="folder_consume",
),
pytest.param(
DocumentSource.MailFetch,
PaperlessTask.TriggerSource.EMAIL_CONSUME,
id="email_consume",
),
],
)
def test_consume_document_source_maps_to_trigger_source(
self,
consume_input_doc,
consume_overrides,
):
consume_input_doc.source = DocumentSource.ConsumeFolder
document_source: DocumentSource,
expected_trigger_source: PaperlessTask.TriggerSource,
) -> None:
"""DocumentSource on the input doc maps to the correct TriggerSource."""
consume_input_doc.source = document_source
task_id = send_publish(
"documents.tasks.consume_file",
(consume_input_doc, consume_overrides),
{},
)
task = PaperlessTask.objects.get(task_id=task_id)
assert task.trigger_source == PaperlessTask.TriggerSource.FOLDER_CONSUME
def test_email_source_maps_correctly(self, consume_input_doc, consume_overrides):
consume_input_doc.source = DocumentSource.MailFetch
task_id = send_publish(
"documents.tasks.consume_file",
(consume_input_doc, consume_overrides),
{},
)
task = PaperlessTask.objects.get(task_id=task_id)
assert task.trigger_source == PaperlessTask.TriggerSource.EMAIL_CONSUME
assert task.trigger_source == expected_trigger_source
@pytest.mark.django_db
class TestTaskPrerunHandler:
def test_marks_task_started(self):
task = PaperlessTaskFactory(status=PaperlessTask.Status.PENDING)
from documents.signals.handlers import task_prerun_handler
task_prerun_handler(task_id=task.task_id)
task.refresh_from_db()
assert task.status == PaperlessTask.Status.STARTED
assert task.date_started is not None
def test_ignores_unknown_task_id(self):
from documents.signals.handlers import task_prerun_handler
@pytest.mark.parametrize(
"task_id",
[
pytest.param("nonexistent-id", id="unknown"),
pytest.param(None, id="none"),
],
)
def test_ignores_invalid_task_id(self, task_id: str | None) -> None:
task_prerun_handler(task_id="nonexistent-id") # must not raise
def test_ignores_none_task_id(self):
from documents.signals.handlers import task_prerun_handler
task_prerun_handler(task_id=None) # must not raise
task_prerun_handler(task_id=task_id) # must not raise
@pytest.mark.django_db
class TestTaskPostrunHandler:
def _started_task(self) -> PaperlessTask:
from django.utils import timezone
return PaperlessTaskFactory(
task_type=PaperlessTask.TaskType.TRAIN_CLASSIFIER,
@@ -189,7 +205,6 @@ class TestTaskPostrunHandler:
def test_records_success_with_dict_result(self):
task = self._started_task()
from documents.signals.handlers import task_postrun_handler
task_postrun_handler(
task_id=task.task_id,
@@ -206,7 +221,6 @@ class TestTaskPostrunHandler:
def test_skips_failure_state(self):
"""postrun skips FAILURE; task_failure_handler owns that path."""
task = self._started_task()
from documents.signals.handlers import task_postrun_handler
task_postrun_handler(task_id=task.task_id, retval="some error", state="FAILURE")
task.refresh_from_db()
@@ -214,7 +228,6 @@ class TestTaskPostrunHandler:
def test_parses_legacy_new_document_string(self):
task = self._started_task()
from documents.signals.handlers import task_postrun_handler
task_postrun_handler(
task_id=task.task_id,
@@ -228,7 +241,6 @@ class TestTaskPostrunHandler:
def test_parses_duplicate_string(self):
"""Duplicate detection returns a string with SUCCESS state (StopConsumeTaskError is caught and returned, not raised)."""
task = self._started_task()
from documents.signals.handlers import task_postrun_handler
task_postrun_handler(
task_id=task.task_id,
@@ -240,7 +252,6 @@ class TestTaskPostrunHandler:
assert task.result_data["duplicate_in_trash"] is False
def test_ignores_unknown_task_id(self):
from documents.signals.handlers import task_postrun_handler
task_postrun_handler(
task_id="nonexistent",
@@ -250,7 +261,6 @@ class TestTaskPostrunHandler:
def test_records_revoked_state(self):
task = self._started_task()
from documents.signals.handlers import task_postrun_handler
task_postrun_handler(task_id=task.task_id, retval=None, state="REVOKED")
task.refresh_from_db()
@@ -260,14 +270,12 @@ class TestTaskPostrunHandler:
@pytest.mark.django_db
class TestTaskFailureHandler:
def test_records_failure_with_exception(self):
from django.utils import timezone
task = PaperlessTaskFactory(
task_type=PaperlessTask.TaskType.CONSUME_FILE,
status=PaperlessTask.Status.STARTED,
date_started=timezone.now(),
)
from documents.signals.handlers import task_failure_handler
task_failure_handler(
task_id=task.task_id,
@@ -281,9 +289,6 @@ class TestTaskFailureHandler:
assert task.date_done is not None
def test_records_traceback_when_provided(self):
import sys
from django.utils import timezone
task = PaperlessTaskFactory(
task_type=PaperlessTask.TaskType.CONSUME_FILE,
@@ -307,7 +312,6 @@ class TestTaskFailureHandler:
assert len(task.result_data["traceback"]) <= 5000
def test_computes_duration_and_wait_time(self):
from django.utils import timezone
now = timezone.now()
task = PaperlessTaskFactory(
@@ -316,7 +320,6 @@ class TestTaskFailureHandler:
date_created=now - timezone.timedelta(seconds=10),
date_started=now - timezone.timedelta(seconds=5),
)
from documents.signals.handlers import task_failure_handler
task_failure_handler(
task_id=task.task_id,
@@ -328,7 +331,6 @@ class TestTaskFailureHandler:
assert task.wait_time_seconds == pytest.approx(5.0, abs=1.0)
def test_ignores_none_task_id(self):
from documents.signals.handlers import task_failure_handler
task_failure_handler(task_id=None, exception=ValueError("x"), traceback=None)