mypy stuff

src/documents/plugins/helpers.py

@@ -110,7 +110,7 @@ class DocumentsStatusManager(BaseStatusManager):
         users_can_view: list[int] | None = None,
         groups_can_view: list[int] | None = None,
     ) -> None:
-        payload = {
+        payload: dict[str, str | int | None] = {
             "type": "document_updated",
             "data": {
                 "document_id": document_id,

src/documents/signals/handlers.py

@@ -4,6 +4,7 @@ import logging
 import shutil
 from pathlib import Path
 from typing import TYPE_CHECKING
+from typing import Any
 
 from celery import shared_task
 from celery import states
@@ -1027,7 +1028,11 @@ def add_or_update_document_in_llm_index(sender, document, **kwargs):
 
 
 @receiver(models.signals.post_delete, sender=Document)
-def delete_document_from_llm_index(sender, instance: Document, **kwargs):
+def delete_document_from_llm_index(
+    sender: Any,
+    instance: Document,
+    **kwargs: Any,
+) -> None:
     """
     Delete a document from the LLM index when it is deleted.
     """

src/documents/tests/test_api_documents.py

@@ -1206,7 +1206,11 @@ class TestDocumentApi(DirectoriesMixin, DocumentConsumeDelayMixin, APITestCase):
         input_doc, overrides = self.get_last_consume_delay_call_args()
 
         self.assertEqual(input_doc.original_file.name, "simple.pdf")
-        self.assertIn(Path(settings.SCRATCH_DIR), input_doc.original_file.parents)
+        self.assertTrue(
+            input_doc.original_file.resolve(strict=False).is_relative_to(
+                Path(settings.SCRATCH_DIR).resolve(strict=False),
+            ),
+        )
         self.assertIsNone(overrides.title)
         self.assertIsNone(overrides.correspondent_id)
         self.assertIsNone(overrides.document_type_id)
@@ -1255,7 +1259,11 @@ class TestDocumentApi(DirectoriesMixin, DocumentConsumeDelayMixin, APITestCase):
         input_doc, overrides = self.get_last_consume_delay_call_args()
 
         self.assertEqual(input_doc.original_file.name, "simple.pdf")
-        self.assertIn(Path(settings.SCRATCH_DIR), input_doc.original_file.parents)
+        self.assertTrue(
+            input_doc.original_file.resolve(strict=False).is_relative_to(
+                Path(settings.SCRATCH_DIR).resolve(strict=False),
+            ),
+        )
         self.assertIsNone(overrides.title)
         self.assertIsNone(overrides.correspondent_id)
         self.assertIsNone(overrides.document_type_id)

src/documents/tests/test_workflows.py

@@ -640,7 +640,9 @@ class TestWorkflows(
 
         expected_str = f"Document did not match {w}"
         self.assertIn(expected_str, cm.output[0])
-        expected_str = f"Document path {test_file} does not match"
+        expected_str = (
+            f"Document path {Path(test_file).resolve(strict=False)} does not match"
+        )
         self.assertIn(expected_str, cm.output[1])
 
     def test_workflow_no_match_mail_rule(self) -> None:

src/paperless/consumers.py

@@ -1,4 +1,5 @@
 import json
+from typing import Any
 
 from asgiref.sync import async_to_sync
 from channels.exceptions import AcceptConnection
@@ -53,7 +54,7 @@ class StatusConsumer(WebsocketConsumer):
         else:
             self.send(json.dumps(event))
 
-    def document_updated(self, event) -> None:
+    def document_updated(self, event: Any) -> None:
         if not self._authenticated():
             self.close()
         else:

src/paperless/tests/test_websockets.py

@@ -180,6 +180,7 @@ class TestWebSockets(TestCase):
         }
 
         channel_layer = get_channel_layer()
+        assert channel_layer is not None
         await channel_layer.group_send(
             "status_updates",
             message,
@@ -223,7 +224,10 @@ class TestWebSockets(TestCase):
         )
 
     @mock.patch("channels.layers.InMemoryChannelLayer.group_send")
-    def test_manager_send_documents_deleted(self, mock_group_send) -> None:
+    def test_manager_send_documents_deleted(
+        self,
+        mock_group_send: mock.MagicMock,
+    ) -> None:
         with DocumentsStatusManager() as manager:
             manager.send_documents_deleted([1, 2, 3])