Update bulk_edit.py

src-ui/messages.xlf

@@ -1238,8 +1238,8 @@
           <context context-type="linenumber">82</context>
         </context-group>
       </trans-unit>
-      <trans-unit id="8035757452478567832" datatype="html">
-        <source>Update existing document</source>
+      <trans-unit id="7860582931776068318" datatype="html">
+        <source>Add document version</source>
         <context-group purpose="location">
           <context context-type="sourcefile">src/app/components/admin/settings/settings.component.html</context>
           <context context-type="linenumber">280</context>
@@ -8411,8 +8411,8 @@
           <context context-type="linenumber">832</context>
         </context-group>
       </trans-unit>
-      <trans-unit id="6390006284731990222" datatype="html">
-        <source>This operation will permanently rotate the original version of <x id="PH" equiv-text="this.list.selected.size"/> document(s).</source>
+      <trans-unit id="5203024009814367559" datatype="html">
+        <source>This operation will add rotated versions of the <x id="PH" equiv-text="this.list.selected.size"/> document(s).</source>
         <context-group purpose="location">
           <context context-type="sourcefile">src/app/components/document-list/bulk-editor/bulk-editor.component.ts</context>
           <context context-type="linenumber">833</context>

src-ui/src/app/components/admin/settings/settings.component.html

@@ -277,7 +277,7 @@
               <div class="col">
                 <select class="form-select" formControlName="pdfEditorDefaultEditMode">
                   <option [ngValue]="PdfEditorEditMode.Create" i18n>Create new document(s)</option>
-                  <option [ngValue]="PdfEditorEditMode.Update" i18n>Update existing document</option>
+                  <option [ngValue]="PdfEditorEditMode.Update" i18n>Add document version</option>
                 </select>
               </div>
             </div>

src-ui/src/app/components/common/pdf-editor/pdf-editor.component.html

@@ -84,7 +84,7 @@
       <input type="radio" class="btn-check" [(ngModel)]="editMode" [value]="PdfEditorEditMode.Update" id="editModeUpdate" name="editmode" [disabled]="hasSplit()">
       <label for="editModeUpdate" class="btn btn-outline-primary btn-sm">
         <i-bs name="pencil"></i-bs>
-        <span class="form-check-label ms-2" i18n>Update existing document</span>
+        <span class="form-check-label ms-2" i18n>Add document version</span>
       </label>
     </div>
     @if (editMode === PdfEditorEditMode.Create) {

src-ui/src/app/components/common/pdf-editor/pdf-editor.component.spec.ts

@@ -3,6 +3,7 @@ import { provideHttpClientTesting } from '@angular/common/http/testing'
 import { ComponentFixture, TestBed } from '@angular/core/testing'
 import { NgbActiveModal } from '@ng-bootstrap/ng-bootstrap'
 import { NgxBootstrapIconsModule, allIcons } from 'ngx-bootstrap-icons'
+import { DocumentService } from 'src/app/services/rest/document.service'
 import { PDFEditorComponent } from './pdf-editor.component'
 
 describe('PDFEditorComponent', () => {
@@ -139,4 +140,16 @@ describe('PDFEditorComponent', () => {
     expect(component.pages[1].page).toBe(2)
     expect(component.pages[2].page).toBe(3)
   })
+
+  it('should include selected version in preview source when provided', () => {
+    const documentService = TestBed.inject(DocumentService)
+    const previewSpy = jest
+      .spyOn(documentService, 'getPreviewUrl')
+      .mockReturnValue('preview-version')
+    component.documentID = 3
+    component.versionID = 10
+
+    expect(component.pdfSrc).toBe('preview-version')
+    expect(previewSpy).toHaveBeenCalledWith(3, false, 10)
+  })
 })

src-ui/src/app/components/common/pdf-editor/pdf-editor.component.ts

@@ -46,6 +46,7 @@ export class PDFEditorComponent extends ConfirmDialogComponent {
   activeModal: NgbActiveModal = inject(NgbActiveModal)
 
   documentID: number
+  versionID?: number
   pages: PageOperation[] = []
   totalPages = 0
   editMode: PdfEditorEditMode = this.settingsService.get(
@@ -55,7 +56,11 @@ export class PDFEditorComponent extends ConfirmDialogComponent {
   includeMetadata: boolean = true
 
   get pdfSrc(): string {
-    return this.documentService.getPreviewUrl(this.documentID)
+    return this.documentService.getPreviewUrl(
+      this.documentID,
+      false,
+      this.versionID
+    )
   }
 
   pdfLoaded(pdf: PngxPdfDocumentProxy) {

src-ui/src/app/components/document-detail/document-detail.component.spec.ts

@@ -1459,22 +1459,25 @@ describe('DocumentDetailComponent', () => {
     const closeSpy = jest.spyOn(openDocumentsService, 'closeDocument')
     const errorSpy = jest.spyOn(toastService, 'showError')
     initNormally()
+    component.selectedVersionId = 10
     component.editPdf()
     expect(modal).not.toBeUndefined()
     modal.componentInstance.documentID = doc.id
+    expect(modal.componentInstance.versionID).toBe(10)
     modal.componentInstance.pages = [{ page: 1, rotate: 0, splitAfter: false }]
     modal.componentInstance.confirm()
     let req = httpTestingController.expectOne(
       `${environment.apiBaseUrl}documents/bulk_edit/`
     )
     expect(req.request.body).toEqual({
-      documents: [doc.id],
+      documents: [10],
       method: 'edit_pdf',
       parameters: {
         operations: [{ page: 1, rotate: 0, doc: 0 }],
         delete_original: false,
         update_document: false,
         include_metadata: true,
+        source_mode: 'explicit_selection',
       },
     })
     req.error(new ErrorEvent('failed'))
@@ -1496,6 +1499,7 @@ describe('DocumentDetailComponent', () => {
     let modal: NgbModalRef
     modalService.activeInstances.subscribe((m) => (modal = m[0]))
     initNormally()
+    component.selectedVersionId = 10
     component.password = 'secret'
     component.removePassword()
     const dialog =
@@ -1508,13 +1512,14 @@ describe('DocumentDetailComponent', () => {
       `${environment.apiBaseUrl}documents/bulk_edit/`
     )
     expect(req.request.body).toEqual({
-      documents: [doc.id],
+      documents: [10],
       method: 'remove_password',
       parameters: {
         password: 'secret',
         update_document: false,
         include_metadata: false,
         delete_original: true,
+        source_mode: 'explicit_selection',
       },
     })
     req.flush(true)

src-ui/src/app/components/document-detail/document-detail.component.ts

@@ -73,7 +73,10 @@ import {
 import { CorrespondentService } from 'src/app/services/rest/correspondent.service'
 import { CustomFieldsService } from 'src/app/services/rest/custom-fields.service'
 import { DocumentTypeService } from 'src/app/services/rest/document-type.service'
-import { DocumentService } from 'src/app/services/rest/document.service'
+import {
+  BulkEditSourceMode,
+  DocumentService,
+} from 'src/app/services/rest/document.service'
 import { SavedViewService } from 'src/app/services/rest/saved-view.service'
 import { StoragePathService } from 'src/app/services/rest/storage-path.service'
 import { TagService } from 'src/app/services/rest/tag.service'
@@ -1626,20 +1629,23 @@ export class DocumentDetailComponent
       size: 'xl',
       scrollable: true,
     })
+    const sourceDocumentId = this.selectedVersionId ?? this.document.id
     modal.componentInstance.title = $localize`PDF Editor`
     modal.componentInstance.btnCaption = $localize`Proceed`
     modal.componentInstance.documentID = this.document.id
+    modal.componentInstance.versionID = sourceDocumentId
     modal.componentInstance.confirmClicked
       .pipe(takeUntil(this.unsubscribeNotifier))
       .subscribe(() => {
         modal.componentInstance.buttonsEnabled = false
         this.documentsService
-          .bulkEdit([this.document.id], 'edit_pdf', {
+          .bulkEdit([sourceDocumentId], 'edit_pdf', {
             operations: modal.componentInstance.getOperations(),
             delete_original: modal.componentInstance.deleteOriginal,
             update_document:
               modal.componentInstance.editMode == PdfEditorEditMode.Update,
             include_metadata: modal.componentInstance.includeMetadata,
+            source_mode: BulkEditSourceMode.EXPLICIT_SELECTION,
           })
           .pipe(first(), takeUntil(this.unsubscribeNotifier))
           .subscribe({
@@ -1685,16 +1691,18 @@ export class DocumentDetailComponent
     modal.componentInstance.confirmClicked
       .pipe(takeUntil(this.unsubscribeNotifier))
       .subscribe(() => {
+        const sourceDocumentId = this.selectedVersionId ?? this.document.id
         const dialog =
           modal.componentInstance as PasswordRemovalConfirmDialogComponent
         dialog.buttonsEnabled = false
         this.networkActive = true
         this.documentsService
-          .bulkEdit([this.document.id], 'remove_password', {
+          .bulkEdit([sourceDocumentId], 'remove_password', {
             password: this.password,
             update_document: dialog.updateDocument,
             include_metadata: dialog.includeMetadata,
             delete_original: dialog.deleteOriginal,
+            source_mode: BulkEditSourceMode.EXPLICIT_SELECTION,
           })
           .pipe(first(), takeUntil(this.unsubscribeNotifier))
           .subscribe({

src-ui/src/app/components/document-list/bulk-editor/bulk-editor.component.ts

@@ -830,7 +830,7 @@ export class BulkEditorComponent
     })
     const rotateDialog = modal.componentInstance as RotateConfirmDialogComponent
     rotateDialog.title = $localize`Rotate confirm`
-    rotateDialog.messageBold = $localize`This operation will permanently rotate the original version of ${this.list.selected.size} document(s).`
+    rotateDialog.messageBold = $localize`This operation will add rotated versions of the ${this.list.selected.size} document(s).`
     rotateDialog.btnClass = 'btn-danger'
     rotateDialog.btnCaption = $localize`Proceed`
     rotateDialog.documentID = Array.from(this.list.selected)[0]

src-ui/src/app/services/rest/document.service.ts

@@ -37,6 +37,11 @@ export interface SelectionData {
   selected_custom_fields: SelectionDataItem[]
 }
 
+export enum BulkEditSourceMode {
+  LATEST_VERSION = 'latest_version',
+  EXPLICIT_SELECTION = 'explicit_selection',
+}
+
 @Injectable({
   providedIn: 'root',
 })

src/documents/bulk_edit.py

@@ -29,12 +29,21 @@ from documents.plugins.helpers import DocumentsStatusManager
 from documents.tasks import bulk_update_documents
 from documents.tasks import consume_file
 from documents.tasks import update_document_content_maybe_archive_file
+from documents.versioning import get_latest_version_for_root
+from documents.versioning import get_root_document
 
 if TYPE_CHECKING:
     from django.contrib.auth.models import User
 
 logger: logging.Logger = logging.getLogger("paperless.bulk_edit")
 
+SourceMode = Literal["latest_version", "explicit_selection"]
+
+
+class SourceModeChoices:
+    LATEST_VERSION: SourceMode = "latest_version"
+    EXPLICIT_SELECTION: SourceMode = "explicit_selection"
+
 
 @shared_task(bind=True)
 def restore_archive_serial_numbers_task(
@@ -72,46 +81,21 @@ def restore_archive_serial_numbers(backup: dict[int, int | None]) -> None:
     logger.info(f"Restored archive serial numbers for documents {list(backup.keys())}")
 
 
-def _get_root_ids_by_doc_id(doc_ids: list[int]) -> dict[int, int]:
-    """
-    Resolve each provided document id to its root document id.
+def _resolve_root_and_source_doc(
+    doc: Document,
+    *,
+    source_mode: SourceMode = SourceModeChoices.LATEST_VERSION,
+) -> tuple[Document, Document]:
+    root_doc = get_root_document(doc)
 
-    - If the id is already a root document: root id is itself.
-    - If the id is a version document: root id is its `root_document_id`.
-    """
-    qs = Document.objects.filter(id__in=doc_ids).only("id", "root_document_id")
-    return {doc.id: doc.root_document_id or doc.id for doc in qs}
+    if source_mode == SourceModeChoices.EXPLICIT_SELECTION:
+        return root_doc, doc
 
+    # Version IDs are explicit by default, only a selected root resolves to latest
+    if doc.root_document_id is not None:
+        return root_doc, doc
 
-def _get_root_and_current_docs_by_root_id(
-    root_ids: set[int],
-) -> tuple[dict[int, Document], dict[int, Document]]:
-    """
-    Returns:
-      - root_docs: root_id -> root Document
-      - current_docs: root_id -> newest version Document (or root if none)
-    """
-    root_docs = {
-        doc.id: doc
-        for doc in Document.objects.filter(id__in=root_ids).select_related(
-            "owner",
-        )
-    }
-    latest_versions_by_root_id: dict[int, Document] = {}
-    for version_doc in Document.objects.filter(root_document_id__in=root_ids).order_by(
-        "root_document_id",
-        "-id",
-    ):
-        root_id = version_doc.root_document_id
-        if root_id is None:
-            continue
-        latest_versions_by_root_id.setdefault(root_id, version_doc)
-
-    current_docs: dict[int, Document] = {
-        root_id: latest_versions_by_root_id.get(root_id, root_docs[root_id])
-        for root_id in root_docs
-    }
-    return root_docs, current_docs
+    return root_doc, get_latest_version_for_root(root_doc)
 
 
 def set_correspondent(
@@ -421,21 +405,32 @@ def rotate(
     doc_ids: list[int],
     degrees: int,
     *,
+    source_mode: SourceMode = SourceModeChoices.LATEST_VERSION,
     user: User | None = None,
 ) -> Literal["OK"]:
     logger.info(
         f"Attempting to rotate {len(doc_ids)} documents by {degrees} degrees.",
     )
-    doc_to_root_id = _get_root_ids_by_doc_id(doc_ids)
-    root_ids = set(doc_to_root_id.values())
-    root_docs_by_id, current_docs_by_root_id = _get_root_and_current_docs_by_root_id(
-        root_ids,
-    )
+    docs_by_id = {
+        doc.id: doc
+        for doc in Document.objects.select_related("root_document").filter(
+            id__in=doc_ids,
+        )
+    }
+    docs_by_root_id: dict[int, tuple[Document, Document]] = {}
+    for doc_id in doc_ids:
+        doc = docs_by_id.get(doc_id)
+        if doc is None:
+            continue
+        root_doc, source_doc = _resolve_root_and_source_doc(
+            doc,
+            source_mode=source_mode,
+        )
+        docs_by_root_id.setdefault(root_doc.id, (root_doc, source_doc))
+
     import pikepdf
 
-    for root_id in root_ids:
-        root_doc = root_docs_by_id[root_id]
-        source_doc = current_docs_by_root_id[root_id]
+    for root_doc, source_doc in docs_by_root_id.values():
         if source_doc.mime_type != "application/pdf":
             logger.warning(
                 f"Document {root_doc.id} is not a PDF, skipping rotation.",
@@ -659,25 +654,17 @@ def delete_pages(
     doc_ids: list[int],
     pages: list[int],
     *,
+    source_mode: SourceMode = SourceModeChoices.LATEST_VERSION,
     user: User | None = None,
 ) -> Literal["OK"]:
     logger.info(
         f"Attempting to delete pages {pages} from {len(doc_ids)} documents",
     )
     doc = Document.objects.select_related("root_document").get(id=doc_ids[0])
-    root_doc: Document
-    if doc.root_document_id is None or doc.root_document is None:
-        root_doc = doc
-    else:
-        root_doc = doc.root_document
-
-    source_doc = (
-        Document.objects.filter(Q(id=root_doc.id) | Q(root_document=root_doc))
-        .order_by("-id")
-        .first()
+    root_doc, source_doc = _resolve_root_and_source_doc(
+        doc,
+        source_mode=source_mode,
     )
-    if source_doc is None:
-        source_doc = root_doc
     pages = sorted(pages)  # sort pages to avoid index issues
     import pikepdf
 
@@ -722,6 +709,7 @@ def edit_pdf(
     delete_original: bool = False,
     update_document: bool = False,
     include_metadata: bool = True,
+    source_mode: SourceMode = SourceModeChoices.LATEST_VERSION,
     user: User | None = None,
 ) -> Literal["OK"]:
     """
@@ -736,19 +724,10 @@ def edit_pdf(
         f"Editing PDF of document {doc_ids[0]} with {len(operations)} operations",
     )
     doc = Document.objects.select_related("root_document").get(id=doc_ids[0])
-    root_doc: Document
-    if doc.root_document_id is None or doc.root_document is None:
-        root_doc = doc
-    else:
-        root_doc = doc.root_document
-
-    source_doc = (
-        Document.objects.filter(Q(id=root_doc.id) | Q(root_document=root_doc))
-        .order_by("-id")
-        .first()
+    root_doc, source_doc = _resolve_root_and_source_doc(
+        doc,
+        source_mode=source_mode,
     )
-    if source_doc is None:
-        source_doc = root_doc
     import pikepdf
 
     pdf_docs: list[pikepdf.Pdf] = []
@@ -859,6 +838,7 @@ def remove_password(
     update_document: bool = False,
     delete_original: bool = False,
     include_metadata: bool = True,
+    source_mode: SourceMode = SourceModeChoices.LATEST_VERSION,
     user: User | None = None,
 ) -> Literal["OK"]:
     """
@@ -868,19 +848,10 @@ def remove_password(
 
     for doc_id in doc_ids:
         doc = Document.objects.select_related("root_document").get(id=doc_id)
-        root_doc: Document
-        if doc.root_document_id is None or doc.root_document is None:
-            root_doc = doc
-        else:
-            root_doc = doc.root_document
-
-        source_doc = (
-            Document.objects.filter(Q(id=root_doc.id) | Q(root_document=root_doc))
-            .order_by("-id")
-            .first()
+        root_doc, source_doc = _resolve_root_and_source_doc(
+            doc,
+            source_mode=source_mode,
         )
-        if source_doc is None:
-            source_doc = root_doc
         try:
             logger.info(
                 f"Attempting password removal from document {doc_ids[0]}",

src/documents/serialisers.py

@@ -1724,6 +1724,15 @@ class BulkEditSerializer(
         except ValueError:
             raise serializers.ValidationError("invalid rotation degrees")
 
+    def _validate_source_mode(self, parameters) -> None:
+        source_mode = parameters.get(
+            "source_mode",
+            bulk_edit.SourceModeChoices.LATEST_VERSION,
+        )
+        if source_mode not in bulk_edit.SourceModeChoices.__dict__.values():
+            raise serializers.ValidationError("Invalid source_mode")
+        parameters["source_mode"] = source_mode
+
     def _validate_parameters_split(self, parameters) -> None:
         if "pages" not in parameters:
             raise serializers.ValidationError("pages not specified")
@@ -1824,6 +1833,9 @@ class BulkEditSerializer(
         method = attrs["method"]
         parameters = attrs["parameters"]
 
+        if "source_mode" in parameters:
+            self._validate_source_mode(parameters)
+
         if method == bulk_edit.set_correspondent:
             self._validate_parameters_correspondent(parameters)
         elif method == bulk_edit.set_document_type:

src/documents/tests/test_api_bulk_edit.py

@@ -1395,7 +1395,10 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
                 {
                     "documents": [self.doc2.id],
                     "method": "edit_pdf",
-                    "parameters": {"operations": [{"page": 1}]},
+                    "parameters": {
+                        "operations": [{"page": 1}],
+                        "source_mode": "explicit_selection",
+                    },
                 },
             ),
             content_type="application/json",
@@ -1407,6 +1410,7 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
         args, kwargs = m.call_args
         self.assertCountEqual(args[0], [self.doc2.id])
         self.assertEqual(kwargs["operations"], [{"page": 1}])
+        self.assertEqual(kwargs["source_mode"], "explicit_selection")
         self.assertEqual(kwargs["user"], self.user)
 
     def test_edit_pdf_invalid_params(self) -> None:
@@ -1572,6 +1576,24 @@ class TestBulkEditAPI(DirectoriesMixin, APITestCase):
             response.content,
         )
 
+        # invalid source mode
+        response = self.client.post(
+            "/api/documents/bulk_edit/",
+            json.dumps(
+                {
+                    "documents": [self.doc2.id],
+                    "method": "edit_pdf",
+                    "parameters": {
+                        "operations": [{"page": 1}],
+                        "source_mode": "not_a_mode",
+                    },
+                },
+            ),
+            content_type="application/json",
+        )
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertIn(b"Invalid source_mode", response.content)
+
     @mock.patch("documents.serialisers.bulk_edit.edit_pdf")
     def test_edit_pdf_page_out_of_bounds(self, m) -> None:
         """

src/documents/tests/test_bulk_edit.py

@@ -405,7 +405,9 @@ class TestBulkEdit(DirectoriesMixin, TestCase):
         self.assertTrue(Document.objects.filter(id=self.doc1.id).exists())
         self.assertFalse(Document.objects.filter(id=version.id).exists())
 
-    def test_get_root_and_current_doc_mapping(self) -> None:
+    def test_resolve_root_and_source_doc_latest_version_prefers_newest_version(
+        self,
+    ) -> None:
         version1 = Document.objects.create(
             checksum="B-v1",
             title="B version 1",
@@ -417,18 +419,14 @@ class TestBulkEdit(DirectoriesMixin, TestCase):
             root_document=self.doc2,
         )
 
-        root_ids_by_doc_id = bulk_edit._get_root_ids_by_doc_id(
-            [self.doc2.id, version1.id, version2.id],
+        root_doc, source_doc = bulk_edit._resolve_root_and_source_doc(
+            self.doc2,
+            source_mode="latest_version",
         )
-        self.assertEqual(root_ids_by_doc_id[self.doc2.id], self.doc2.id)
-        self.assertEqual(root_ids_by_doc_id[version1.id], self.doc2.id)
-        self.assertEqual(root_ids_by_doc_id[version2.id], self.doc2.id)
 
-        root_docs, current_docs = bulk_edit._get_root_and_current_docs_by_root_id(
-            {self.doc2.id},
-        )
-        self.assertEqual(root_docs[self.doc2.id].id, self.doc2.id)
-        self.assertEqual(current_docs[self.doc2.id].id, version2.id)
+        self.assertEqual(root_doc.id, self.doc2.id)
+        self.assertEqual(source_doc.id, version2.id)
+        self.assertNotEqual(source_doc.id, version1.id)
 
     @mock.patch("documents.tasks.bulk_update_documents.delay")
     def test_set_permissions(self, m) -> None:
@@ -1041,6 +1039,34 @@ class TestPDFActions(DirectoriesMixin, TestCase):
             self.assertIsNotNone(overrides)
             self.assertEqual(result, "OK")
 
+    @mock.patch("documents.data_models.magic.from_file", return_value="application/pdf")
+    @mock.patch("documents.tasks.consume_file.delay")
+    @mock.patch("pikepdf.open")
+    def test_rotate_explicit_selection_uses_root_source_when_root_selected(
+        self,
+        mock_open,
+        mock_consume_delay,
+        mock_magic,
+    ):
+        Document.objects.create(
+            checksum="B-v1",
+            title="B version 1",
+            root_document=self.doc2,
+        )
+        fake_pdf = mock.MagicMock()
+        fake_pdf.pages = [mock.Mock()]
+        mock_open.return_value.__enter__.return_value = fake_pdf
+
+        result = bulk_edit.rotate(
+            [self.doc2.id],
+            90,
+            source_mode="explicit_selection",
+        )
+
+        self.assertEqual(result, "OK")
+        mock_open.assert_called_once_with(self.doc2.source_path)
+        mock_consume_delay.assert_called_once()
+
     @mock.patch("documents.tasks.consume_file.delay")
     @mock.patch("pikepdf.Pdf.save")
     @mock.patch("documents.data_models.magic.from_file", return_value="application/pdf")
@@ -1065,6 +1091,34 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         self.assertIsNotNone(overrides)
         self.assertEqual(result, "OK")
 
+    @mock.patch("documents.data_models.magic.from_file", return_value="application/pdf")
+    @mock.patch("documents.tasks.consume_file.delay")
+    @mock.patch("pikepdf.open")
+    def test_delete_pages_explicit_selection_uses_root_source_when_root_selected(
+        self,
+        mock_open,
+        mock_consume_delay,
+        mock_magic,
+    ):
+        Document.objects.create(
+            checksum="B-v1",
+            title="B version 1",
+            root_document=self.doc2,
+        )
+        fake_pdf = mock.MagicMock()
+        fake_pdf.pages = [mock.Mock(), mock.Mock()]
+        mock_open.return_value.__enter__.return_value = fake_pdf
+
+        result = bulk_edit.delete_pages(
+            [self.doc2.id],
+            [1],
+            source_mode="explicit_selection",
+        )
+
+        self.assertEqual(result, "OK")
+        mock_open.assert_called_once_with(self.doc2.source_path)
+        mock_consume_delay.assert_called_once()
+
     @mock.patch("documents.tasks.consume_file.delay")
     @mock.patch("pikepdf.Pdf.save")
     def test_delete_pages_with_error(self, mock_pdf_save, mock_consume_delay):
@@ -1213,6 +1267,40 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         self.assertTrue(str(consumable.original_file).endswith("_edited.pdf"))
         self.assertIsNotNone(overrides)
 
+    @mock.patch("documents.data_models.magic.from_file", return_value="application/pdf")
+    @mock.patch("documents.tasks.consume_file.delay")
+    @mock.patch("pikepdf.new")
+    @mock.patch("pikepdf.open")
+    def test_edit_pdf_explicit_selection_uses_root_source_when_root_selected(
+        self,
+        mock_open,
+        mock_new,
+        mock_consume_delay,
+        mock_magic,
+    ):
+        Document.objects.create(
+            checksum="B-v1",
+            title="B version 1",
+            root_document=self.doc2,
+        )
+        fake_pdf = mock.MagicMock()
+        fake_pdf.pages = [mock.Mock()]
+        mock_open.return_value.__enter__.return_value = fake_pdf
+        output_pdf = mock.MagicMock()
+        output_pdf.pages = []
+        mock_new.return_value = output_pdf
+
+        result = bulk_edit.edit_pdf(
+            [self.doc2.id],
+            operations=[{"page": 1}],
+            update_document=True,
+            source_mode="explicit_selection",
+        )
+
+        self.assertEqual(result, "OK")
+        mock_open.assert_called_once_with(self.doc2.source_path)
+        mock_consume_delay.assert_called_once()
+
     @mock.patch("documents.bulk_edit.group")
     @mock.patch("documents.tasks.consume_file.s")
     def test_edit_pdf_without_metadata(
@@ -1333,6 +1421,34 @@ class TestPDFActions(DirectoriesMixin, TestCase):
         self.assertEqual(consumable.root_document_id, doc.id)
         self.assertIsNotNone(overrides)
 
+    @mock.patch("documents.data_models.magic.from_file", return_value="application/pdf")
+    @mock.patch("documents.tasks.consume_file.delay")
+    @mock.patch("pikepdf.open")
+    def test_remove_password_explicit_selection_uses_root_source_when_root_selected(
+        self,
+        mock_open,
+        mock_consume_delay,
+        mock_magic,
+    ) -> None:
+        Document.objects.create(
+            checksum="A-v1",
+            title="A version 1",
+            root_document=self.doc1,
+        )
+        fake_pdf = mock.MagicMock()
+        mock_open.return_value.__enter__.return_value = fake_pdf
+
+        result = bulk_edit.remove_password(
+            [self.doc1.id],
+            password="secret",
+            update_document=True,
+            source_mode="explicit_selection",
+        )
+
+        self.assertEqual(result, "OK")
+        mock_open.assert_called_once_with(self.doc1.source_path, password="secret")
+        mock_consume_delay.assert_called_once()
+
     @mock.patch("documents.bulk_edit.chord")
     @mock.patch("documents.bulk_edit.group")
     @mock.patch("documents.tasks.consume_file.s")

src/paperless/tests/test_adapter.py

@@ -1,100 +1,107 @@
-import logging
+from unittest import mock
 
-import pytest
 from allauth.account.adapter import get_adapter
 from allauth.core import context
 from allauth.socialaccount.adapter import get_adapter as get_social_adapter
+from django.conf import settings
 from django.contrib.auth.models import AnonymousUser
 from django.contrib.auth.models import Group
 from django.contrib.auth.models import User
 from django.forms import ValidationError
 from django.http import HttpRequest
+from django.test import TestCase
+from django.test import override_settings
 from django.urls import reverse
-from pytest_django.fixtures import SettingsWrapper
-from pytest_mock import MockerFixture
 from rest_framework.authtoken.models import Token
 
 from paperless.adapter import DrfTokenStrategy
 
 
-@pytest.mark.django_db
-class TestCustomAccountAdapter:
-    def test_is_open_for_signup(self, settings: SettingsWrapper) -> None:
+class TestCustomAccountAdapter(TestCase):
+    def test_is_open_for_signup(self) -> None:
         adapter = get_adapter()
 
         # With no accounts, signups should be allowed
-        assert adapter.is_open_for_signup(None)
+        self.assertTrue(adapter.is_open_for_signup(None))
 
         User.objects.create_user("testuser")
 
+        # Test when ACCOUNT_ALLOW_SIGNUPS is True
         settings.ACCOUNT_ALLOW_SIGNUPS = True
-        assert adapter.is_open_for_signup(None)
+        self.assertTrue(adapter.is_open_for_signup(None))
 
+        # Test when ACCOUNT_ALLOW_SIGNUPS is False
         settings.ACCOUNT_ALLOW_SIGNUPS = False
-        assert not adapter.is_open_for_signup(None)
+        self.assertFalse(adapter.is_open_for_signup(None))
 
-    def test_is_safe_url(self, settings: SettingsWrapper) -> None:
+    def test_is_safe_url(self) -> None:
         request = HttpRequest()
-        request.get_host = lambda: "example.com"
+        request.get_host = mock.Mock(return_value="example.com")
         with context.request_context(request):
             adapter = get_adapter()
+            with override_settings(ALLOWED_HOSTS=["*"]):
+                # True because request host is same
+                url = "https://example.com"
+                self.assertTrue(adapter.is_safe_url(url))
 
-            settings.ALLOWED_HOSTS = ["*"]
-            # True because request host is same
-            assert adapter.is_safe_url("https://example.com")
+            url = "https://evil.com"
             # False despite wildcard because request host is different
-            assert not adapter.is_safe_url("https://evil.com")
+            self.assertFalse(adapter.is_safe_url(url))
 
             settings.ALLOWED_HOSTS = ["example.com"]
+            url = "https://example.com"
             # True because request host is same
-            assert adapter.is_safe_url("https://example.com")
+            self.assertTrue(adapter.is_safe_url(url))
 
             settings.ALLOWED_HOSTS = ["*", "example.com"]
+            url = "//evil.com"
             # False because request host is not in allowed hosts
-            assert not adapter.is_safe_url("//evil.com")
+            self.assertFalse(adapter.is_safe_url(url))
 
-    def test_pre_authenticate(
-        self,
-        settings: SettingsWrapper,
-        mocker: MockerFixture,
-    ) -> None:
-        mocker.patch("allauth.core.internal.ratelimit.consume", return_value=True)
+    @mock.patch("allauth.core.internal.ratelimit.consume", return_value=True)
+    def test_pre_authenticate(self, mock_consume) -> None:
         adapter = get_adapter()
         request = HttpRequest()
-        request.get_host = lambda: "example.com"
+        request.get_host = mock.Mock(return_value="example.com")
 
         settings.DISABLE_REGULAR_LOGIN = False
         adapter.pre_authenticate(request)
 
         settings.DISABLE_REGULAR_LOGIN = True
-        with pytest.raises(ValidationError):
+        with self.assertRaises(ValidationError):
             adapter.pre_authenticate(request)
 
-    def test_get_reset_password_from_key_url(self, settings: SettingsWrapper) -> None:
+    def test_get_reset_password_from_key_url(self) -> None:
         request = HttpRequest()
-        request.get_host = lambda: "foo.org"
+        request.get_host = mock.Mock(return_value="foo.org")
         with context.request_context(request):
             adapter = get_adapter()
 
-            settings.PAPERLESS_URL = None
-            settings.ACCOUNT_DEFAULT_HTTP_PROTOCOL = "https"
-            expected_url = f"https://foo.org{reverse('account_reset_password_from_key', kwargs={'uidb36': 'UID', 'key': 'KEY'})}"
-            assert adapter.get_reset_password_from_key_url("UID-KEY") == expected_url
+            # Test when PAPERLESS_URL is None
+            with override_settings(
+                PAPERLESS_URL=None,
+                ACCOUNT_DEFAULT_HTTP_PROTOCOL="https",
+            ):
+                expected_url = f"https://foo.org{reverse('account_reset_password_from_key', kwargs={'uidb36': 'UID', 'key': 'KEY'})}"
+                self.assertEqual(
+                    adapter.get_reset_password_from_key_url("UID-KEY"),
+                    expected_url,
+                )
 
-            settings.PAPERLESS_URL = "https://bar.com"
-            expected_url = f"https://bar.com{reverse('account_reset_password_from_key', kwargs={'uidb36': 'UID', 'key': 'KEY'})}"
-            assert adapter.get_reset_password_from_key_url("UID-KEY") == expected_url
+            # Test when PAPERLESS_URL is not None
+            with override_settings(PAPERLESS_URL="https://bar.com"):
+                expected_url = f"https://bar.com{reverse('account_reset_password_from_key', kwargs={'uidb36': 'UID', 'key': 'KEY'})}"
+                self.assertEqual(
+                    adapter.get_reset_password_from_key_url("UID-KEY"),
+                    expected_url,
+                )
 
-    def test_save_user_adds_groups(
-        self,
-        settings: SettingsWrapper,
-        mocker: MockerFixture,
-    ) -> None:
-        settings.ACCOUNT_DEFAULT_GROUPS = ["group1", "group2"]
+    @override_settings(ACCOUNT_DEFAULT_GROUPS=["group1", "group2"])
+    def test_save_user_adds_groups(self) -> None:
         Group.objects.create(name="group1")
         user = User.objects.create_user("testuser")
         adapter = get_adapter()
-        form = mocker.MagicMock(
+        form = mock.Mock(
             cleaned_data={
                 "username": "testuser",
                 "email": "user@example.com",
@@ -103,81 +110,88 @@ class TestCustomAccountAdapter:
 
         user = adapter.save_user(HttpRequest(), user, form, commit=True)
 
-        assert user.groups.count() == 1
-        assert user.groups.filter(name="group1").exists()
-        assert not user.groups.filter(name="group2").exists()
+        self.assertEqual(user.groups.count(), 1)
+        self.assertTrue(user.groups.filter(name="group1").exists())
+        self.assertFalse(user.groups.filter(name="group2").exists())
 
-    def test_fresh_install_save_creates_superuser(self, mocker: MockerFixture) -> None:
+    def test_fresh_install_save_creates_superuser(self) -> None:
         adapter = get_adapter()
-        form = mocker.MagicMock(
+        form = mock.Mock(
             cleaned_data={
                 "username": "testuser",
                 "email": "user@paperless-ngx.com",
             },
         )
         user = adapter.save_user(HttpRequest(), User(), form, commit=True)
-        assert user.is_superuser
+        self.assertTrue(user.is_superuser)
 
-        form = mocker.MagicMock(
+        # Next time, it should not create a superuser
+        form = mock.Mock(
             cleaned_data={
                 "username": "testuser2",
                 "email": "user2@paperless-ngx.com",
             },
         )
         user2 = adapter.save_user(HttpRequest(), User(), form, commit=True)
-        assert not user2.is_superuser
+        self.assertFalse(user2.is_superuser)
 
 
-class TestCustomSocialAccountAdapter:
-    @pytest.mark.django_db
-    def test_is_open_for_signup(self, settings: SettingsWrapper) -> None:
+class TestCustomSocialAccountAdapter(TestCase):
+    def test_is_open_for_signup(self) -> None:
         adapter = get_social_adapter()
 
+        # Test when SOCIALACCOUNT_ALLOW_SIGNUPS is True
         settings.SOCIALACCOUNT_ALLOW_SIGNUPS = True
-        assert adapter.is_open_for_signup(None, None)
+        self.assertTrue(adapter.is_open_for_signup(None, None))
 
+        # Test when SOCIALACCOUNT_ALLOW_SIGNUPS is False
         settings.SOCIALACCOUNT_ALLOW_SIGNUPS = False
-        assert not adapter.is_open_for_signup(None, None)
+        self.assertFalse(adapter.is_open_for_signup(None, None))
 
     def test_get_connect_redirect_url(self) -> None:
         adapter = get_social_adapter()
-        assert adapter.get_connect_redirect_url(None, None) == reverse("base")
+        request = None
+        socialaccount = None
 
-    @pytest.mark.django_db
-    def test_save_user_adds_groups(
-        self,
-        settings: SettingsWrapper,
-        mocker: MockerFixture,
-    ) -> None:
-        settings.SOCIAL_ACCOUNT_DEFAULT_GROUPS = ["group1", "group2"]
+        # Test the default URL
+        expected_url = reverse("base")
+        self.assertEqual(
+            adapter.get_connect_redirect_url(request, socialaccount),
+            expected_url,
+        )
+
+    @override_settings(SOCIAL_ACCOUNT_DEFAULT_GROUPS=["group1", "group2"])
+    def test_save_user_adds_groups(self) -> None:
         Group.objects.create(name="group1")
         adapter = get_social_adapter()
+        request = HttpRequest()
         user = User.objects.create_user("testuser")
-        sociallogin = mocker.MagicMock(user=user)
+        sociallogin = mock.Mock(
+            user=user,
+        )
 
-        user = adapter.save_user(HttpRequest(), sociallogin, None)
+        user = adapter.save_user(request, sociallogin, None)
 
-        assert user.groups.count() == 1
-        assert user.groups.filter(name="group1").exists()
-        assert not user.groups.filter(name="group2").exists()
+        self.assertEqual(user.groups.count(), 1)
+        self.assertTrue(user.groups.filter(name="group1").exists())
+        self.assertFalse(user.groups.filter(name="group2").exists())
 
-    def test_error_logged_on_authentication_error(
-        self,
-        caplog: pytest.LogCaptureFixture,
-    ) -> None:
+    def test_error_logged_on_authentication_error(self) -> None:
         adapter = get_social_adapter()
-        with caplog.at_level(logging.INFO, logger="paperless.auth"):
+        request = HttpRequest()
+        with self.assertLogs("paperless.auth", level="INFO") as log_cm:
             adapter.on_authentication_error(
-                HttpRequest(),
+                request,
                 provider="test-provider",
                 error="Error",
                 exception="Test authentication error",
             )
-        assert any("Test authentication error" in msg for msg in caplog.messages)
+        self.assertTrue(
+            any("Test authentication error" in message for message in log_cm.output),
+        )
 
 
-@pytest.mark.django_db
-class TestDrfTokenStrategy:
+class TestDrfTokenStrategy(TestCase):
     def test_create_access_token_creates_new_token(self) -> None:
         """
         GIVEN:
@@ -187,6 +201,7 @@ class TestDrfTokenStrategy:
         THEN:
             - A new token is created and its key is returned
         """
+
         user = User.objects.create_user("testuser")
         request = HttpRequest()
         request.user = user
@@ -194,9 +209,13 @@ class TestDrfTokenStrategy:
         strategy = DrfTokenStrategy()
         token_key = strategy.create_access_token(request)
 
-        assert token_key is not None
-        assert Token.objects.filter(user=user).exists()
-        assert token_key == Token.objects.get(user=user).key
+        # Verify a token was created
+        self.assertIsNotNone(token_key)
+        self.assertTrue(Token.objects.filter(user=user).exists())
+
+        # Verify the returned key matches the created token
+        token = Token.objects.get(user=user)
+        self.assertEqual(token_key, token.key)
 
     def test_create_access_token_returns_existing_token(self) -> None:
         """
@@ -207,6 +226,7 @@ class TestDrfTokenStrategy:
         THEN:
             - The same token key is returned (no new token created)
         """
+
         user = User.objects.create_user("testuser")
         existing_token = Token.objects.create(user=user)
 
@@ -216,8 +236,11 @@ class TestDrfTokenStrategy:
         strategy = DrfTokenStrategy()
         token_key = strategy.create_access_token(request)
 
-        assert token_key == existing_token.key
-        assert Token.objects.filter(user=user).count() == 1
+        # Verify the existing token key is returned
+        self.assertEqual(token_key, existing_token.key)
+
+        # Verify only one token exists (no duplicate created)
+        self.assertEqual(Token.objects.filter(user=user).count(), 1)
 
     def test_create_access_token_returns_none_for_unauthenticated_user(self) -> None:
         """
@@ -228,11 +251,12 @@ class TestDrfTokenStrategy:
         THEN:
             - None is returned and no token is created
         """
+
         request = HttpRequest()
         request.user = AnonymousUser()
 
         strategy = DrfTokenStrategy()
         token_key = strategy.create_access_token(request)
 
-        assert token_key is None
-        assert Token.objects.count() == 0
+        self.assertIsNone(token_key)
+        self.assertEqual(Token.objects.count(), 0)

src/paperless/tests/test_checks.py

@@ -1,12 +1,15 @@
 import os
-from dataclasses import dataclass
 from pathlib import Path
+from unittest import mock
 
 import pytest
 from django.core.checks import Warning
-from pytest_django.fixtures import SettingsWrapper
+from django.test import TestCase
+from django.test import override_settings
 from pytest_mock import MockerFixture
 
+from documents.tests.utils import DirectoriesMixin
+from documents.tests.utils import FileSystemAssertsMixin
 from paperless.checks import audit_log_check
 from paperless.checks import binaries_check
 from paperless.checks import check_deprecated_db_settings
@@ -15,84 +18,54 @@ from paperless.checks import paths_check
 from paperless.checks import settings_values_check
 
 
-@dataclass(frozen=True, slots=True)
-class PaperlessTestDirs:
-    data_dir: Path
-    media_dir: Path
-    consumption_dir: Path
-
-
-# TODO: consolidate with documents/tests/conftest.py PaperlessDirs/paperless_dirs
-#       once the paperless and documents test suites are ready to share fixtures.
-@pytest.fixture()
-def directories(tmp_path: Path, settings: SettingsWrapper) -> PaperlessTestDirs:
-    data_dir = tmp_path / "data"
-    media_dir = tmp_path / "media"
-    consumption_dir = tmp_path / "consumption"
-
-    for d in (data_dir, media_dir, consumption_dir):
-        d.mkdir()
-
-    settings.DATA_DIR = data_dir
-    settings.MEDIA_ROOT = media_dir
-    settings.CONSUMPTION_DIR = consumption_dir
-
-    return PaperlessTestDirs(
-        data_dir=data_dir,
-        media_dir=media_dir,
-        consumption_dir=consumption_dir,
-    )
-
-
-class TestChecks:
+class TestChecks(DirectoriesMixin, TestCase):
     def test_binaries(self) -> None:
-        assert binaries_check(None) == []
+        self.assertEqual(binaries_check(None), [])
 
-    def test_binaries_fail(self, settings: SettingsWrapper) -> None:
-        settings.CONVERT_BINARY = "uuuhh"
-        assert len(binaries_check(None)) == 1
+    @override_settings(CONVERT_BINARY="uuuhh")
+    def test_binaries_fail(self) -> None:
+        self.assertEqual(len(binaries_check(None)), 1)
 
-    @pytest.mark.usefixtures("directories")
     def test_paths_check(self) -> None:
-        assert paths_check(None) == []
+        self.assertEqual(paths_check(None), [])
 
-    def test_paths_check_dont_exist(self, settings: SettingsWrapper) -> None:
-        settings.MEDIA_ROOT = Path("uuh")
-        settings.DATA_DIR = Path("whatever")
-        settings.CONSUMPTION_DIR = Path("idontcare")
+    @override_settings(
+        MEDIA_ROOT=Path("uuh"),
+        DATA_DIR=Path("whatever"),
+        CONSUMPTION_DIR=Path("idontcare"),
+    )
+    def test_paths_check_dont_exist(self) -> None:
+        msgs = paths_check(None)
+        self.assertEqual(len(msgs), 3, str(msgs))
+
+        for msg in msgs:
+            self.assertTrue(msg.msg.endswith("is set but doesn't exist."))
+
+    def test_paths_check_no_access(self) -> None:
+        Path(self.dirs.data_dir).chmod(0o000)
+        Path(self.dirs.media_dir).chmod(0o000)
+        Path(self.dirs.consumption_dir).chmod(0o000)
+
+        self.addCleanup(os.chmod, self.dirs.data_dir, 0o777)
+        self.addCleanup(os.chmod, self.dirs.media_dir, 0o777)
+        self.addCleanup(os.chmod, self.dirs.consumption_dir, 0o777)
 
         msgs = paths_check(None)
+        self.assertEqual(len(msgs), 3)
 
-        assert len(msgs) == 3, str(msgs)
         for msg in msgs:
-            assert msg.msg.endswith("is set but doesn't exist.")
+            self.assertTrue(msg.msg.endswith("is not writeable"))
 
-    def test_paths_check_no_access(self, directories: PaperlessTestDirs) -> None:
-        directories.data_dir.chmod(0o000)
-        directories.media_dir.chmod(0o000)
-        directories.consumption_dir.chmod(0o000)
+    @override_settings(DEBUG=False)
+    def test_debug_disabled(self) -> None:
+        self.assertEqual(debug_mode_check(None), [])
 
-        try:
-            msgs = paths_check(None)
-        finally:
-            directories.data_dir.chmod(0o777)
-            directories.media_dir.chmod(0o777)
-            directories.consumption_dir.chmod(0o777)
-
-        assert len(msgs) == 3
-        for msg in msgs:
-            assert msg.msg.endswith("is not writeable")
-
-    def test_debug_disabled(self, settings: SettingsWrapper) -> None:
-        settings.DEBUG = False
-        assert debug_mode_check(None) == []
-
-    def test_debug_enabled(self, settings: SettingsWrapper) -> None:
-        settings.DEBUG = True
-        assert len(debug_mode_check(None)) == 1
+    @override_settings(DEBUG=True)
+    def test_debug_enabled(self) -> None:
+        self.assertEqual(len(debug_mode_check(None)), 1)
 
 
-class TestSettingsChecksAgainstDefaults:
+class TestSettingsChecksAgainstDefaults(DirectoriesMixin, TestCase):
     def test_all_valid(self) -> None:
         """
         GIVEN:
@@ -103,71 +76,104 @@ class TestSettingsChecksAgainstDefaults:
             - No system check errors reported
         """
         msgs = settings_values_check(None)
-        assert len(msgs) == 0
+        self.assertEqual(len(msgs), 0)
 
 
-class TestOcrSettingsChecks:
-    @pytest.mark.parametrize(
-        ("setting", "value", "expected_msg"),
-        [
-            pytest.param(
-                "OCR_OUTPUT_TYPE",
-                "notapdf",
-                'OCR output type "notapdf"',
-                id="invalid-output-type",
-            ),
-            pytest.param(
-                "OCR_MODE",
-                "makeitso",
-                'OCR output mode "makeitso"',
-                id="invalid-mode",
-            ),
-            pytest.param(
-                "OCR_MODE",
-                "skip_noarchive",
-                "deprecated",
-                id="deprecated-mode",
-            ),
-            pytest.param(
-                "OCR_SKIP_ARCHIVE_FILE",
-                "invalid",
-                'OCR_SKIP_ARCHIVE_FILE setting "invalid"',
-                id="invalid-skip-archive-file",
-            ),
-            pytest.param(
-                "OCR_CLEAN",
-                "cleanme",
-                'OCR clean mode "cleanme"',
-                id="invalid-clean",
-            ),
-        ],
-    )
-    def test_invalid_setting_produces_one_error(
-        self,
-        settings: SettingsWrapper,
-        setting: str,
-        value: str,
-        expected_msg: str,
-    ) -> None:
+class TestOcrSettingsChecks(DirectoriesMixin, TestCase):
+    @override_settings(OCR_OUTPUT_TYPE="notapdf")
+    def test_invalid_output_type(self) -> None:
         """
         GIVEN:
             - Default settings
-            - One OCR setting is set to an invalid value
+            - OCR output type is invalid
         WHEN:
             - Settings are validated
         THEN:
-            - Exactly one system check error is reported containing the expected message
+            - system check error reported for OCR output type
         """
-        setattr(settings, setting, value)
-
         msgs = settings_values_check(None)
+        self.assertEqual(len(msgs), 1)
 
-        assert len(msgs) == 1
-        assert expected_msg in msgs[0].msg
+        msg = msgs[0]
+
+        self.assertIn('OCR output type "notapdf"', msg.msg)
+
+    @override_settings(OCR_MODE="makeitso")
+    def test_invalid_ocr_type(self) -> None:
+        """
+        GIVEN:
+            - Default settings
+            - OCR type is invalid
+        WHEN:
+            - Settings are validated
+        THEN:
+            - system check error reported for OCR type
+        """
+        msgs = settings_values_check(None)
+        self.assertEqual(len(msgs), 1)
+
+        msg = msgs[0]
+
+        self.assertIn('OCR output mode "makeitso"', msg.msg)
+
+    @override_settings(OCR_MODE="skip_noarchive")
+    def test_deprecated_ocr_type(self) -> None:
+        """
+        GIVEN:
+            - Default settings
+            - OCR type is deprecated
+        WHEN:
+            - Settings are validated
+        THEN:
+            - deprecation warning reported for OCR type
+        """
+        msgs = settings_values_check(None)
+        self.assertEqual(len(msgs), 1)
+
+        msg = msgs[0]
+
+        self.assertIn("deprecated", msg.msg)
+
+    @override_settings(OCR_SKIP_ARCHIVE_FILE="invalid")
+    def test_invalid_ocr_skip_archive_file(self) -> None:
+        """
+        GIVEN:
+            - Default settings
+            - OCR_SKIP_ARCHIVE_FILE is invalid
+        WHEN:
+            - Settings are validated
+        THEN:
+            - system check error reported for OCR_SKIP_ARCHIVE_FILE
+        """
+        msgs = settings_values_check(None)
+        self.assertEqual(len(msgs), 1)
+
+        msg = msgs[0]
+
+        self.assertIn('OCR_SKIP_ARCHIVE_FILE setting "invalid"', msg.msg)
+
+    @override_settings(OCR_CLEAN="cleanme")
+    def test_invalid_ocr_clean(self) -> None:
+        """
+        GIVEN:
+            - Default settings
+            - OCR cleaning type is invalid
+        WHEN:
+            - Settings are validated
+        THEN:
+            - system check error reported for OCR cleaning type
+        """
+        msgs = settings_values_check(None)
+        self.assertEqual(len(msgs), 1)
+
+        msg = msgs[0]
+
+        self.assertIn('OCR clean mode "cleanme"', msg.msg)
 
 
-class TestTimezoneSettingsChecks:
-    def test_invalid_timezone(self, settings: SettingsWrapper) -> None:
+class TestTimezoneSettingsChecks(DirectoriesMixin, TestCase):
+    @override_settings(TIME_ZONE="TheMoon\\MyCrater")
+    def test_invalid_timezone(self) -> None:
         """
         GIVEN:
             - Default settings
@@ -177,16 +183,17 @@ class TestTimezoneSettingsChecks:
         THEN:
             - system check error reported for timezone
         """
-        settings.TIME_ZONE = "TheMoon\\MyCrater"
-
         msgs = settings_values_check(None)
+        self.assertEqual(len(msgs), 1)
 
-        assert len(msgs) == 1
-        assert 'Timezone "TheMoon\\MyCrater"' in msgs[0].msg
+        msg = msgs[0]
+
+        self.assertIn('Timezone "TheMoon\\MyCrater"', msg.msg)
 
 
-class TestEmailCertSettingsChecks:
-    def test_not_valid_file(self, settings: SettingsWrapper) -> None:
+class TestEmailCertSettingsChecks(DirectoriesMixin, FileSystemAssertsMixin, TestCase):
+    @override_settings(EMAIL_CERTIFICATE_FILE=Path("/tmp/not_actually_here.pem"))
+    def test_not_valid_file(self) -> None:
         """
         GIVEN:
             - Default settings
@@ -196,22 +203,19 @@ class TestEmailCertSettingsChecks:
         THEN:
             - system check error reported for email certificate
         """
-        cert_path = Path("/tmp/not_actually_here.pem")
-        assert not cert_path.is_file()
-        settings.EMAIL_CERTIFICATE_FILE = cert_path
+        self.assertIsNotFile("/tmp/not_actually_here.pem")
 
         msgs = settings_values_check(None)
 
-        assert len(msgs) == 1
-        assert "Email cert /tmp/not_actually_here.pem is not a file" in msgs[0].msg
+        self.assertEqual(len(msgs), 1)
+
+        msg = msgs[0]
+
+        self.assertIn("Email cert /tmp/not_actually_here.pem is not a file", msg.msg)
 
 
-class TestAuditLogChecks:
-    def test_was_enabled_once(
-        self,
-        settings: SettingsWrapper,
-        mocker: MockerFixture,
-    ) -> None:
+class TestAuditLogChecks(TestCase):
+    def test_was_enabled_once(self) -> None:
         """
         GIVEN:
             - Audit log is not enabled
@@ -220,18 +224,23 @@ class TestAuditLogChecks:
         THEN:
             - system check error reported for disabling audit log
         """
-        settings.AUDIT_LOG_ENABLED = False
-        introspect_mock = mocker.MagicMock()
+        introspect_mock = mock.MagicMock()
         introspect_mock.introspection.table_names.return_value = ["auditlog_logentry"]
-        mocker.patch.dict(
-            "paperless.checks.connections",
-            {"default": introspect_mock},
-        )
+        with override_settings(AUDIT_LOG_ENABLED=False):
+            with mock.patch.dict(
+                "paperless.checks.connections",
+                {"default": introspect_mock},
+            ):
+                msgs = audit_log_check(None)
 
-        msgs = audit_log_check(None)
+                self.assertEqual(len(msgs), 1)
 
-        assert len(msgs) == 1
-        assert "auditlog table was found but audit log is disabled." in msgs[0].msg
+                msg = msgs[0]
+
+                self.assertIn(
+                    ("auditlog table was found but audit log is disabled."),
+                    msg.msg,
+                )
 
 
 DEPRECATED_VARS: dict[str, str] = {
@@ -260,16 +269,20 @@ class TestDeprecatedDbSettings:
     @pytest.mark.parametrize(
         ("env_var", "db_option_key"),
         [
-            pytest.param("PAPERLESS_DB_TIMEOUT", "timeout", id="db-timeout"),
-            pytest.param(
-                "PAPERLESS_DB_POOLSIZE",
-                "pool.min_size / pool.max_size",
-                id="db-poolsize",
-            ),
-            pytest.param("PAPERLESS_DBSSLMODE", "sslmode", id="ssl-mode"),
-            pytest.param("PAPERLESS_DBSSLROOTCERT", "sslrootcert", id="ssl-rootcert"),
-            pytest.param("PAPERLESS_DBSSLCERT", "sslcert", id="ssl-cert"),
-            pytest.param("PAPERLESS_DBSSLKEY", "sslkey", id="ssl-key"),
+            ("PAPERLESS_DB_TIMEOUT", "timeout"),
+            ("PAPERLESS_DB_POOLSIZE", "pool.min_size / pool.max_size"),
+            ("PAPERLESS_DBSSLMODE", "sslmode"),
+            ("PAPERLESS_DBSSLROOTCERT", "sslrootcert"),
+            ("PAPERLESS_DBSSLCERT", "sslcert"),
+            ("PAPERLESS_DBSSLKEY", "sslkey"),
+        ],
+        ids=[
+            "db-timeout",
+            "db-poolsize",
+            "ssl-mode",
+            "ssl-rootcert",
+            "ssl-cert",
+            "ssl-key",
         ],
     )
     def test_single_deprecated_var_produces_one_warning(

src/paperless/tests/test_utils.py

@@ -9,50 +9,35 @@ from paperless.utils import ocr_to_dateparser_languages
 @pytest.mark.parametrize(
     ("ocr_language", "expected"),
     [
-        pytest.param("eng", ["en"], id="single-language"),
-        pytest.param("fra+ita+lao", ["fr", "it", "lo"], id="multiple-languages"),
-        pytest.param("fil", ["fil"], id="no-two-letter-equivalent"),
-        pytest.param(
-            "aze_cyrl+srp_latn",
-            ["az-Cyrl", "sr-Latn"],
-            id="script-supported-by-dateparser",
-        ),
-        pytest.param(
-            "deu_frak",
-            ["de"],
-            id="script-not-supported-falls-back-to-language",
-        ),
-        pytest.param(
-            "chi_tra+chi_sim",
-            ["zh"],
-            id="chinese-variants-collapse-to-general",
-        ),
-        pytest.param(
-            "eng+unsupported_language+por",
-            ["en", "pt"],
-            id="unsupported-language-skipped",
-        ),
-        pytest.param(
-            "unsupported1+unsupported2",
-            [],
-            id="all-unsupported-returns-empty",
-        ),
-        pytest.param("eng+eng", ["en"], id="duplicates-deduplicated"),
-        pytest.param(
-            "ita_unknownscript",
-            ["it"],
-            id="unknown-script-falls-back-to-language",
-        ),
+        # One language
+        ("eng", ["en"]),
+        # Multiple languages
+        ("fra+ita+lao", ["fr", "it", "lo"]),
+        # Languages that don't have a two-letter equivalent
+        ("fil", ["fil"]),
+        # Languages with a script part supported by dateparser
+        ("aze_cyrl+srp_latn", ["az-Cyrl", "sr-Latn"]),
+        # Languages with a script part not supported by dateparser
+        # In this case, default to the language without script
+        ("deu_frak", ["de"]),
+        # Traditional and simplified chinese don't have the same name in dateparser,
+        # so they're converted to the general chinese language
+        ("chi_tra+chi_sim", ["zh"]),
+        # If a language is not supported by dateparser, fallback to the supported ones
+        ("eng+unsupported_language+por", ["en", "pt"]),
+        # If no language is supported, fallback to default
+        ("unsupported1+unsupported2", []),
+        # Duplicate languages, should not duplicate in result
+        ("eng+eng", ["en"]),
+        # Language with script, but script is not mapped
+        ("ita_unknownscript", ["it"]),
     ],
 )
-def test_ocr_to_dateparser_languages(ocr_language: str, expected: list[str]) -> None:
+def test_ocr_to_dateparser_languages(ocr_language, expected):
     assert sorted(ocr_to_dateparser_languages(ocr_language)) == sorted(expected)
 
 
-def test_ocr_to_dateparser_languages_exception(
-    monkeypatch: pytest.MonkeyPatch,
-    caplog: pytest.LogCaptureFixture,
-) -> None:
+def test_ocr_to_dateparser_languages_exception(monkeypatch, caplog):
     # Patch LocaleDataLoader.get_locale_map to raise an exception
     class DummyLoader:
         def get_locale_map(self, locales=None):

src/paperless/tests/test_views.py

@@ -1,31 +1,24 @@
+import tempfile
 from pathlib import Path
 
-from django.test import Client
-from pytest_django.fixtures import SettingsWrapper
+from django.test import override_settings
 
 
-def test_favicon_view(
-    client: Client,
-    tmp_path: Path,
-    settings: SettingsWrapper,
-) -> None:
-    favicon_path = tmp_path / "paperless" / "img" / "favicon.ico"
-    favicon_path.parent.mkdir(parents=True)
-    favicon_path.write_bytes(b"FAKE ICON DATA")
+def test_favicon_view(client):
+    with tempfile.TemporaryDirectory() as tmpdir:
+        static_dir = Path(tmpdir)
+        favicon_path = static_dir / "paperless" / "img" / "favicon.ico"
+        favicon_path.parent.mkdir(parents=True, exist_ok=True)
+        favicon_path.write_bytes(b"FAKE ICON DATA")
 
-    settings.STATIC_ROOT = tmp_path
-
-    response = client.get("/favicon.ico")
-    assert response.status_code == 200
-    assert response["Content-Type"] == "image/x-icon"
-    assert b"".join(response.streaming_content) == b"FAKE ICON DATA"
+        with override_settings(STATIC_ROOT=static_dir):
+            response = client.get("/favicon.ico")
+            assert response.status_code == 200
+            assert response["Content-Type"] == "image/x-icon"
+            assert b"".join(response.streaming_content) == b"FAKE ICON DATA"
 
 
-def test_favicon_view_missing_file(
-    client: Client,
-    tmp_path: Path,
-    settings: SettingsWrapper,
-) -> None:
-    settings.STATIC_ROOT = tmp_path
-    response = client.get("/favicon.ico")
-    assert response.status_code == 404
+def test_favicon_view_missing_file(client):
+    with override_settings(STATIC_ROOT=Path(tempfile.mkdtemp())):
+        response = client.get("/favicon.ico")
+        assert response.status_code == 404