from __future__ import annotations import zipfile from datetime import timedelta from pathlib import Path from unittest import mock from django.conf import settings from django.contrib.auth.models import User from django.test import override_settings from django.utils import timezone from rest_framework import serializers from rest_framework import status from rest_framework.test import APITestCase from documents.filters import ShareLinkBundleFilterSet from documents.models import ShareLink from documents.models import ShareLinkBundle from documents.serialisers import ShareLinkBundleSerializer from documents.tasks import build_share_link_bundle from documents.tasks import cleanup_expired_share_link_bundles from paperless_testing.dirs import DirectoriesMixin from paperless_testing.factories import DocumentFactory from paperless_testing.factories import UserFactory from paperless_testing.permissions import grant_global from paperless_testing.permissions import grant_object class ShareLinkBundleAPITests(DirectoriesMixin, APITestCase): ENDPOINT = "/api/share_link_bundles/" def setUp(self) -> None: super().setUp() self.user = UserFactory(username="bundle_admin", superuser=True) self.client.force_authenticate(self.user) self.document = DocumentFactory.create() @mock.patch("documents.views.build_share_link_bundle.apply_async") def test_create_bundle_triggers_build_job(self, delay_mock) -> None: payload = { "document_ids": [self.document.pk], "file_version": ShareLink.FileVersion.ARCHIVE, "expiration_days": 7, } response = self.client.post(self.ENDPOINT, payload, format="json") self.assertEqual(response.status_code, status.HTTP_201_CREATED) bundle = ShareLinkBundle.objects.get(pk=response.data["id"]) self.assertEqual(bundle.documents.count(), 1) self.assertEqual(bundle.status, ShareLinkBundle.Status.PENDING) delay_mock.assert_called_once() self.assertEqual(delay_mock.call_args.kwargs["kwargs"]["bundle_id"], bundle.pk) @mock.patch("documents.views.build_share_link_bundle.apply_async") def test_create_bundle_requires_global_document_view_permission( self, delay_mock, ) -> None: owner = UserFactory(username="document_owner") requester = UserFactory(username="bundle_creator") grant_global(requester, "add_sharelinkbundle") document = DocumentFactory.create(owner=owner) grant_object(requester, document, "view_document") self.client.force_authenticate(requester) payload = { "document_ids": [document.pk], "file_version": ShareLink.FileVersion.ARCHIVE, "expiration_days": 7, } response = self.client.post(self.ENDPOINT, payload, format="json") self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) grant_global(requester, "view_document") requester = User.objects.get(pk=requester.pk) self.client.force_authenticate(requester) response = self.client.post(self.ENDPOINT, payload, format="json") self.assertEqual(response.status_code, status.HTTP_201_CREATED) delay_mock.assert_called_once() def test_create_bundle_rejects_missing_documents(self) -> None: payload = { "document_ids": [9999], "file_version": ShareLink.FileVersion.ARCHIVE, "expiration_days": 7, } response = self.client.post(self.ENDPOINT, payload, format="json") self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn("document_ids", response.data) def test_create_bundle_rejects_insufficient_permissions(self) -> None: requester = UserFactory(username="bundle_creator") grant_global(requester, "add_sharelinkbundle", "view_document") self.client.force_authenticate(requester) document = DocumentFactory(owner=UserFactory(username="document_owner")) payload = { "document_ids": [self.document.pk, document.pk], "file_version": ShareLink.FileVersion.ARCHIVE, "expiration_days": 7, } response = self.client.post(self.ENDPOINT, payload, format="json") self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn(str(document.pk), str(response.data["document_ids"])) self.assertFalse(ShareLinkBundle.objects.exists()) @mock.patch("documents.views.build_share_link_bundle.apply_async") def test_rebuild_bundle_resets_state(self, delay_mock) -> None: bundle = ShareLinkBundle.objects.create( slug="rebuild-slug", file_version=ShareLink.FileVersion.ARCHIVE, status=ShareLinkBundle.Status.FAILED, ) bundle.documents.set([self.document]) bundle.last_error = {"message": "Something went wrong"} bundle.size_bytes = 100 bundle.file_path = "path/to/file.zip" bundle.save() response = self.client.post(f"{self.ENDPOINT}{bundle.pk}/rebuild/") self.assertEqual(response.status_code, status.HTTP_200_OK) bundle.refresh_from_db() self.assertEqual(bundle.status, ShareLinkBundle.Status.PENDING) self.assertIsNone(bundle.last_error) self.assertIsNone(bundle.size_bytes) self.assertEqual(bundle.file_path, "") delay_mock.assert_called_once() self.assertEqual(delay_mock.call_args.kwargs["kwargs"]["bundle_id"], bundle.pk) def test_rebuild_bundle_rejects_processing_status(self) -> None: bundle = ShareLinkBundle.objects.create( slug="processing-slug", file_version=ShareLink.FileVersion.ARCHIVE, status=ShareLinkBundle.Status.PROCESSING, ) bundle.documents.set([self.document]) response = self.client.post(f"{self.ENDPOINT}{bundle.pk}/rebuild/") self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn("detail", response.data) def test_create_bundle_rejects_duplicate_documents(self) -> None: payload = { "document_ids": [self.document.pk, self.document.pk], "file_version": ShareLink.FileVersion.ARCHIVE, "expiration_days": 7, } response = self.client.post(self.ENDPOINT, payload, format="json") self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn("document_ids", response.data) def test_download_ready_bundle_streams_file(self) -> None: bundle_file = settings.SHARE_LINK_BUNDLE_DIR / "bundles" / "ready.zip" bundle_file.parent.mkdir(parents=True, exist_ok=True) bundle_file.write_bytes(b"binary-zip-content") bundle = ShareLinkBundle.objects.create( slug="readyslug", file_version=ShareLink.FileVersion.ARCHIVE, status=ShareLinkBundle.Status.READY, file_path=str(bundle_file.relative_to(settings.SHARE_LINK_BUNDLE_DIR)), ) bundle.documents.set([self.document]) self.client.logout() response = self.client.get(f"/share/{bundle.slug}/") content = b"".join(response.streaming_content) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response["Content-Type"], "application/zip") self.assertEqual(content, b"binary-zip-content") self.assertIn("attachment;", response["Content-Disposition"]) def test_download_pending_bundle_returns_202(self) -> None: bundle = ShareLinkBundle.objects.create( slug="pendingslug", file_version=ShareLink.FileVersion.ARCHIVE, status=ShareLinkBundle.Status.PENDING, ) bundle.documents.set([self.document]) self.client.logout() response = self.client.get(f"/share/{bundle.slug}/") self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED) def test_download_failed_bundle_returns_503(self) -> None: bundle = ShareLinkBundle.objects.create( slug="failedslug", file_version=ShareLink.FileVersion.ARCHIVE, status=ShareLinkBundle.Status.FAILED, ) bundle.documents.set([self.document]) self.client.logout() response = self.client.get(f"/share/{bundle.slug}/") self.assertEqual(response.status_code, status.HTTP_503_SERVICE_UNAVAILABLE) def test_expired_share_link_redirects(self) -> None: share_link = ShareLink.objects.create( slug="expiredlink", document=self.document, file_version=ShareLink.FileVersion.ORIGINAL, expiration=timezone.now() - timedelta(hours=1), ) self.client.logout() response = self.client.get(f"/share/{share_link.slug}/") self.assertEqual(response.status_code, status.HTTP_302_FOUND) self.assertIn("sharelink_expired=1", response["Location"]) def test_unknown_share_link_redirects(self) -> None: self.client.logout() response = self.client.get("/share/unknownsharelink/") self.assertEqual(response.status_code, status.HTTP_302_FOUND) self.assertIn("sharelink_notfound=1", response["Location"]) @override_settings(LOGIN_URL="/paperless/accounts/login/") def test_unknown_share_link_redirects_under_subpath(self) -> None: """ GIVEN: - Paperless is served under a subpath WHEN: - An unknown public share link is requested THEN: - The redirect to the login page keeps the subpath """ self.client.logout() response = self.client.get("/share/unknownsharelink/") self.assertEqual(response.status_code, status.HTTP_302_FOUND) self.assertEqual( response["Location"], "/paperless/accounts/login/?sharelink_notfound=1", ) def test_share_link_missing_file_redirects(self) -> None: """ GIVEN: - A share link whose document file is missing from disk WHEN: - The public share link is requested anonymously THEN: - The user is redirected to login instead of a 500 error """ doc = DocumentFactory.create(filename="missing-original.pdf") share_link = ShareLink.objects.create( slug="missingfilelink", document=doc, file_version=ShareLink.FileVersion.ORIGINAL, ) self.client.logout() response = self.client.get(f"/share/{share_link.slug}/") self.assertEqual(response.status_code, status.HTTP_302_FOUND) self.assertIn("sharelink_notfound=1", response["Location"]) def test_download_ready_bundle_missing_file_returns_503(self) -> None: """ GIVEN: - A READY bundle whose zip file is missing from disk WHEN: - The public share link is requested anonymously THEN: - A 503 is returned instead of a 500 error """ bundle = ShareLinkBundle.objects.create( slug="missingbundlefile", file_version=ShareLink.FileVersion.ARCHIVE, status=ShareLinkBundle.Status.READY, file_path="bundles/gone.zip", ) bundle.documents.set([self.document]) self.client.logout() response = self.client.get(f"/share/{bundle.slug}/") self.assertEqual(response.status_code, status.HTTP_503_SERVICE_UNAVAILABLE) class ShareLinkBundleTaskTests(DirectoriesMixin, APITestCase): def setUp(self) -> None: super().setUp() self.document = DocumentFactory.create() def test_cleanup_expired_share_link_bundles(self) -> None: expired_path = settings.SHARE_LINK_BUNDLE_DIR / "expired.zip" expired_path.parent.mkdir(parents=True, exist_ok=True) expired_path.write_bytes(b"expired") active_path = settings.SHARE_LINK_BUNDLE_DIR / "active.zip" active_path.write_bytes(b"active") expired_bundle = ShareLinkBundle.objects.create( slug="expired-bundle", file_version=ShareLink.FileVersion.ARCHIVE, status=ShareLinkBundle.Status.READY, expiration=timezone.now() - timedelta(days=1), file_path=expired_path.name, ) expired_bundle.documents.set([self.document]) active_bundle = ShareLinkBundle.objects.create( slug="active-bundle", file_version=ShareLink.FileVersion.ARCHIVE, status=ShareLinkBundle.Status.READY, expiration=timezone.now() + timedelta(days=1), file_path=active_path.name, ) active_bundle.documents.set([self.document]) cleanup_expired_share_link_bundles() self.assertFalse(ShareLinkBundle.objects.filter(pk=expired_bundle.pk).exists()) self.assertTrue(ShareLinkBundle.objects.filter(pk=active_bundle.pk).exists()) self.assertFalse(expired_path.exists()) self.assertTrue(active_path.exists()) def test_cleanup_expired_share_link_bundles_logs_on_failure(self) -> None: expired_bundle = ShareLinkBundle.objects.create( slug="expired-bundle", file_version=ShareLink.FileVersion.ARCHIVE, status=ShareLinkBundle.Status.READY, expiration=timezone.now() - timedelta(days=1), ) expired_bundle.documents.set([self.document]) with mock.patch.object( ShareLinkBundle, "delete", side_effect=RuntimeError("fail"), ): with self.assertLogs("paperless.tasks", level="WARNING") as logs: cleanup_expired_share_link_bundles() self.assertTrue( any( "Failed to delete expired share link bundle" in msg for msg in logs.output ), ) class ShareLinkBundleBuildTaskTests(DirectoriesMixin, APITestCase): def setUp(self) -> None: super().setUp() self.document = DocumentFactory.create( mime_type="application/pdf", checksum="123", ) self.document.archive_checksum = "" self.document.save() def _write_document_file(self, *, archive: bool, content: bytes) -> Path: if archive: self.document.archive_filename = f"{self.document.pk:07}.pdf" self.document.save() path = self.document.archive_path assert path is not None else: path = self.document.source_path path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(content) return path def test_build_share_link_bundle_creates_zip_and_sets_metadata(self) -> None: self._write_document_file(archive=False, content=b"source") archive_path = self._write_document_file(archive=True, content=b"archive") bundle = ShareLinkBundle.objects.create( slug="build-archive", file_version=ShareLink.FileVersion.ARCHIVE, ) bundle.documents.set([self.document]) build_share_link_bundle(bundle.pk) bundle.refresh_from_db() self.assertEqual(bundle.status, ShareLinkBundle.Status.READY) self.assertIsNone(bundle.last_error) self.assertIsNotNone(bundle.built_at) self.assertGreater(bundle.size_bytes or 0, 0) final_path = bundle.absolute_file_path self.assertIsNotNone(final_path) assert final_path is not None self.assertTrue(final_path.exists()) with zipfile.ZipFile(final_path) as zipf: names = zipf.namelist() self.assertEqual(len(names), 1) self.assertEqual(zipf.read(names[0]), archive_path.read_bytes()) def test_build_share_link_bundle_overwrites_existing_file(self) -> None: self._write_document_file(archive=False, content=b"source") bundle = ShareLinkBundle.objects.create( slug="overwrite", file_version=ShareLink.FileVersion.ORIGINAL, ) bundle.documents.set([self.document]) existing = settings.SHARE_LINK_BUNDLE_DIR / "overwrite.zip" existing.parent.mkdir(parents=True, exist_ok=True) existing.write_bytes(b"old") build_share_link_bundle(bundle.pk) bundle.refresh_from_db() final_path = bundle.absolute_file_path self.assertIsNotNone(final_path) assert final_path is not None self.assertTrue(final_path.exists()) self.assertNotEqual(final_path.read_bytes(), b"old") def test_build_share_link_bundle_failure_marks_failed(self) -> None: self._write_document_file(archive=False, content=b"source") bundle = ShareLinkBundle.objects.create( slug="fail-bundle", file_version=ShareLink.FileVersion.ORIGINAL, ) bundle.documents.set([self.document]) with ( mock.patch( "documents.tasks.OriginalsOnlyStrategy.add_document", side_effect=RuntimeError("zip failure"), ), mock.patch("pathlib.Path.unlink") as unlink_mock, ): unlink_mock.side_effect = [OSError("unlink"), OSError("unlink-finally")] + [ None, ] * 5 with self.assertRaises(RuntimeError): build_share_link_bundle(bundle.pk) bundle.refresh_from_db() self.assertEqual(bundle.status, ShareLinkBundle.Status.FAILED) self.assertIsInstance(bundle.last_error, dict) assert isinstance(bundle.last_error, dict) self.assertEqual(bundle.last_error.get("message"), "zip failure") self.assertEqual(bundle.last_error.get("exception_type"), "RuntimeError") scratch_zips = list(Path(settings.SCRATCH_DIR).glob("*.zip")) self.assertTrue(scratch_zips) for path in scratch_zips: path.unlink(missing_ok=True) def test_build_share_link_bundle_missing_bundle_noop(self) -> None: # Should not raise when bundle does not exist build_share_link_bundle(99999) class ShareLinkBundleFilterSetTests(DirectoriesMixin, APITestCase): def setUp(self) -> None: super().setUp() self.document = DocumentFactory.create() self.document.checksum = "doc1checksum" self.document.save() self.other_document = DocumentFactory.create() self.other_document.checksum = "doc2checksum" self.other_document.save() self.bundle_one = ShareLinkBundle.objects.create( slug="bundle-one", file_version=ShareLink.FileVersion.ORIGINAL, ) self.bundle_one.documents.set([self.document]) self.bundle_two = ShareLinkBundle.objects.create( slug="bundle-two", file_version=ShareLink.FileVersion.ORIGINAL, ) self.bundle_two.documents.set([self.other_document]) def test_filter_documents_returns_all_for_empty_value(self) -> None: filterset = ShareLinkBundleFilterSet( data={"documents": ""}, queryset=ShareLinkBundle.objects.all(), ) self.assertCountEqual(filterset.qs, [self.bundle_one, self.bundle_two]) def test_filter_documents_handles_invalid_input(self) -> None: filterset = ShareLinkBundleFilterSet( data={"documents": "invalid"}, queryset=ShareLinkBundle.objects.all(), ) self.assertFalse(filterset.qs.exists()) def test_filter_documents_filters_by_multiple_ids(self) -> None: filterset = ShareLinkBundleFilterSet( data={"documents": f"{self.document.pk},{self.other_document.pk}"}, queryset=ShareLinkBundle.objects.all(), ) self.assertCountEqual(filterset.qs, [self.bundle_one, self.bundle_two]) def test_filter_documents_returns_queryset_for_empty_ids(self) -> None: filterset = ShareLinkBundleFilterSet( data={"documents": ","}, queryset=ShareLinkBundle.objects.all(), ) self.assertCountEqual(filterset.qs, [self.bundle_one, self.bundle_two]) class ShareLinkBundleModelTests(DirectoriesMixin, APITestCase): def test_absolute_file_path_handles_relative_path(self) -> None: relative_path = Path("relative.zip") bundle = ShareLinkBundle.objects.create( slug="relative-bundle", file_version=ShareLink.FileVersion.ORIGINAL, file_path=str(relative_path), ) self.assertEqual( bundle.absolute_file_path, (settings.SHARE_LINK_BUNDLE_DIR / relative_path).resolve(), ) def test_absolute_file_path_rejects_absolute_path(self) -> None: bundle = ShareLinkBundle.objects.create( slug="absolute-bundle", file_version=ShareLink.FileVersion.ORIGINAL, file_path=str(Path(self.dirs.media_dir) / "absolute.zip"), ) self.assertIsNone(bundle.absolute_file_path) def test_absolute_file_path_rejects_traversal_outside_bundle_dir(self) -> None: bundle = ShareLinkBundle.objects.create( slug="traversal-bundle", file_version=ShareLink.FileVersion.ORIGINAL, file_path="../escaped.zip", ) self.assertIsNone(bundle.absolute_file_path) def test_str_returns_translated_slug(self) -> None: bundle = ShareLinkBundle.objects.create( slug="string-slug", file_version=ShareLink.FileVersion.ORIGINAL, ) self.assertIn("string-slug", str(bundle)) def test_remove_file_deletes_existing_file(self) -> None: bundle_path = settings.SHARE_LINK_BUNDLE_DIR / "remove.zip" bundle_path.parent.mkdir(parents=True, exist_ok=True) bundle_path.write_bytes(b"remove-me") bundle = ShareLinkBundle.objects.create( slug="remove-bundle", file_version=ShareLink.FileVersion.ORIGINAL, file_path=str(bundle_path.relative_to(settings.SHARE_LINK_BUNDLE_DIR)), ) bundle.remove_file() self.assertFalse(bundle_path.exists()) def test_remove_file_handles_oserror(self) -> None: bundle_path = settings.SHARE_LINK_BUNDLE_DIR / "remove-error.zip" bundle_path.parent.mkdir(parents=True, exist_ok=True) bundle_path.write_bytes(b"remove-me") bundle = ShareLinkBundle.objects.create( slug="remove-error", file_version=ShareLink.FileVersion.ORIGINAL, file_path=str(bundle_path.relative_to(settings.SHARE_LINK_BUNDLE_DIR)), ) with mock.patch("pathlib.Path.unlink", side_effect=OSError("fail")): bundle.remove_file() self.assertTrue(bundle_path.exists()) def test_delete_calls_remove_file(self) -> None: bundle_path = settings.SHARE_LINK_BUNDLE_DIR / "delete.zip" bundle_path.parent.mkdir(parents=True, exist_ok=True) bundle_path.write_bytes(b"remove-me") bundle = ShareLinkBundle.objects.create( slug="delete-bundle", file_version=ShareLink.FileVersion.ORIGINAL, file_path=str(bundle_path.relative_to(settings.SHARE_LINK_BUNDLE_DIR)), ) bundle.delete() self.assertFalse(bundle_path.exists()) class ShareLinkBundleSerializerTests(DirectoriesMixin, APITestCase): def setUp(self) -> None: super().setUp() self.document = DocumentFactory.create() def test_validate_document_ids_rejects_duplicates(self) -> None: serializer = ShareLinkBundleSerializer( data={ "document_ids": [self.document.pk, self.document.pk], "file_version": ShareLink.FileVersion.ORIGINAL, }, ) self.assertFalse(serializer.is_valid()) self.assertIn("document_ids", serializer.errors) def test_create_assigns_documents_and_expiration(self) -> None: serializer = ShareLinkBundleSerializer( data={ "document_ids": [self.document.pk], "file_version": ShareLink.FileVersion.ORIGINAL, "expiration_days": 3, }, ) self.assertTrue(serializer.is_valid(), serializer.errors) bundle = serializer.save() self.assertEqual(list(bundle.documents.all()), [self.document]) expected_expiration = timezone.now() + timedelta(days=3) self.assertAlmostEqual( bundle.expiration, expected_expiration, delta=timedelta(seconds=10), ) def test_create_raises_when_missing_documents(self) -> None: serializer = ShareLinkBundleSerializer( data={ "document_ids": [self.document.pk, 9999], "file_version": ShareLink.FileVersion.ORIGINAL, }, ) self.assertTrue(serializer.is_valid(), serializer.errors) with self.assertRaises(serializers.ValidationError): serializer.save(documents=[self.document])