return path
+ def update(self, instance, validated_data):
+ """
+ When a storage path is updated, see if documents
+ using it require a rename/move
+ """
+ doc_ids = [doc.id for doc in instance.documents.all()]
+ if len(doc_ids):
+ bulk_edit.bulk_update_documents.delay(doc_ids)
+
+ return super().update(instance, validated_data)
+
class UiSettingsViewSerializer(serializers.ModelSerializer):
class Meta:
from filelock import FileLock
from .. import matching
-from ..bulk_edit import bulk_update_documents
from ..file_handling import create_source_path_directory
from ..file_handling import delete_empty_directories
from ..file_handling import generate_unique_filename
from ..models import Document
from ..models import MatchingModel
from ..models import PaperlessTask
-from ..models import StoragePath
from ..models import Tag
logger = logging.getLogger("paperless.handlers")
)
-@receiver(models.signals.post_save, sender=StoragePath)
-def update_document_storage_path(sender, instance: StoragePath, **kwargs):
- """
- Triggers when a storage path is changed, running against any documents using
- the path, and checks to see if they need to be renamed
- """
- doc_ids = [doc.id for doc in instance.documents.all()]
- if len(doc_ids):
- bulk_update_documents.delay(doc_ids)
-
-
def set_log_entry(sender, document=None, logging_group=None, **kwargs):
ct = ContentType.objects.get(model="document")
- Existing storage paths are returned
"""
response = self.client.get(self.ENDPOINT, format="json")
- self.assertEqual(response.status_code, 200)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.data["count"], 1)
json.dumps(
{
"name": "Storage path with placeholders",
- "path": "{title}/{correspondent}/{document_type}/{created}/{created_year}/{created_year_short}/{created_month}/{created_month_name}/{created_month_name_short}/{created_day}/{added}/{added_year}/{added_year_short}/{added_month}/{added_month_name}/{added_month_name_short}/{added_day}/{asn}/{tags}/{tag_list}/",
+ "path": "{title}/{correspondent}/{document_type}/{created}/{created_year}"
+ "/{created_year_short}/{created_month}/{created_month_name}"
+ "/{created_month_name_short}/{created_day}/{added}/{added_year}"
+ "/{added_year_short}/{added_month}/{added_month_name}"
+ "/{added_month_name_short}/{added_day}/{asn}/{tags}"
+ "/{tag_list}/",
},
),
content_type="application/json",
self.assertEqual(response.status_code, 201)
self.assertEqual(StoragePath.objects.count(), 2)
+ @mock.patch("documents.bulk_edit.bulk_update_documents.delay")
+ def test_api_update_storage_path(self, bulk_update_mock):
+ """
+ GIVEN:
+ - API request to get all storage paths
+ WHEN:
+ - API is called
+ THEN:
+ - Existing storage paths are returned
+ """
+ document = Document.objects.create(
+ mime_type="application/pdf",
+ storage_path=self.sp1,
+ )
+ response = self.client.patch(
+ f"{self.ENDPOINT}{self.sp1.pk}/",
+ data={
+ "path": "somewhere/{created} - {title}",
+ },
+ )
+
+ self.assertEqual(response.status_code, 200)
+
+ bulk_update_mock.assert_called_once()
+
+ args, _ = bulk_update_mock.call_args
+
+ self.assertCountEqual([document.pk], args[0])
+
class TestTasks(DirectoriesMixin, APITestCase):
ENDPOINT = "/api/tasks/"
self.assertEqual(generate_filename(doc_a), "0000002.pdf")
self.assertEqual(generate_filename(doc_b), "SomeImportantNone/2020-07-25.pdf")
- @mock.patch("documents.bulk_edit.bulk_update_documents.delay")
- def test_document_storage_path_changed_with_doc(self, bulk_update_mock):
- """
- GIVEN:
- - Document with a defined storage path
- WHEN:
- - The storage path format is updated
- THEN:
- - Document is renamed to the new format
- """
- sp = StoragePath.objects.create(path="TestFolder/{created}")
- document = Document.objects.create(
- mime_type="application/pdf",
- created=timezone.make_aware(datetime.datetime(2020, 6, 25, 7, 36, 51, 153)),
- storage_path=sp,
- )
-
- # Ensure that filename is properly generated
- document.filename = generate_filename(document)
- self.assertEqual(
- document.source_path,
- os.path.join(settings.ORIGINALS_DIR, "TestFolder/2020-06-25.pdf"),
- )
-
- # Actually create the file
- create_source_path_directory(document.source_path)
- Path(document.source_path).touch()
- self.assertTrue(os.path.isfile(document.source_path))
- document.save()
-
- # Change format
- bulk_update_mock.side_effect = (
- bulk_update_documents # call through, no Redis needed
- )
- sp.path = "NewFolder/{created}"
- sp.save()
-
- document.refresh_from_db()
-
- self.assertEqual(
- document.source_path,
- os.path.join(settings.ORIGINALS_DIR, "NewFolder/2020-06-25.pdf"),
- )
- self.assertTrue(os.path.isfile(document.source_path))
-
- @mock.patch("documents.bulk_edit.bulk_update_documents.delay")
- def test_document_storage_path_changed_with_no_doc(self, bulk_update_mock):
- """
- GIVEN:
- - Document with no storage path
- - Storage path exist
- WHEN:
- - The storage path format is updated
- THEN:
- - No changes to document
- """
- sp = StoragePath.objects.create(path="TestFolder/{created}")
- document = Document.objects.create(
- mime_type="application/pdf",
- created=timezone.make_aware(datetime.datetime(2021, 6, 25, 7, 36, 51, 153)),
- )
-
- # Ensure that filename is properly generated
- document.filename = generate_filename(document)
- self.assertEqual(
- document.source_path,
- os.path.join(settings.ORIGINALS_DIR, f"{document.pk:07}.pdf"),
- )
-
- # Actually create the file
- create_source_path_directory(document.source_path)
- Path(document.source_path).touch()
- self.assertTrue(os.path.isfile(document.source_path))
- document.save()
-
- # Change format
- sp.path = "NewFolder/{created}"
- sp.save()
-
- bulk_update_mock.assert_not_called()
-
- document.refresh_from_db()
-
- self.assertEqual(
- document.source_path,
- os.path.join(settings.ORIGINALS_DIR, f"{document.pk:07}.pdf"),
- )
- self.assertTrue(os.path.isfile(document.source_path))
-
@override_settings(
FILENAME_FORMAT="{created_year_short}/{created_month_name_short}/{created_month_name}/{title}",
)