From 0e12a18aae212c00457735701086092cdca574dc Mon Sep 17 00:00:00 2001 From: shamoon <4887959+shamoon@users.noreply.github.com> Date: Tue, 30 Dec 2025 15:16:05 -0800 Subject: [PATCH] More backend coverage, rename --- .../tests/test_api_share_link_bundles.py | 182 ------- .../tests/test_migration_share_link_bundle.py | 51 ++ .../tests/test_share_link_bundles.py | 478 ++++++++++++++++++ 3 files changed, 529 insertions(+), 182 deletions(-) delete mode 100644 src/documents/tests/test_api_share_link_bundles.py create mode 100644 src/documents/tests/test_migration_share_link_bundle.py create mode 100644 src/documents/tests/test_share_link_bundles.py diff --git a/src/documents/tests/test_api_share_link_bundles.py b/src/documents/tests/test_api_share_link_bundles.py deleted file mode 100644 index a8b5a91fd..000000000 --- a/src/documents/tests/test_api_share_link_bundles.py +++ /dev/null @@ -1,182 +0,0 @@ -from __future__ import annotations - -from datetime import timedelta -from pathlib import Path -from unittest import mock - -from django.contrib.auth.models import User -from django.utils import timezone -from rest_framework import status -from rest_framework.test import APITestCase - -from documents.models import ShareLink -from documents.models import ShareLinkBundle -from documents.tasks import cleanup_expired_share_link_bundles -from documents.tests.factories import DocumentFactory -from documents.tests.utils import DirectoriesMixin - - -class ShareLinkBundleAPITests(DirectoriesMixin, APITestCase): - ENDPOINT = "/api/share_link_bundles/" - - def setUp(self): - super().setUp() - self.user = User.objects.create_superuser(username="bundle_admin") - self.client.force_authenticate(self.user) - self.document = DocumentFactory.create() - - @mock.patch("documents.views.build_share_link_bundle.delay") - def test_create_bundle_triggers_build_job(self, delay_mock): - 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_with(bundle.pk) - - def test_create_bundle_rejects_missing_documents(self): - 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) - - @mock.patch("documents.views.build_share_link_bundle.delay") - def test_rebuild_bundle_resets_state(self, delay_mock): - bundle = ShareLinkBundle.objects.create( - slug="rebuild-slug", - file_version=ShareLink.FileVersion.ARCHIVE, - status=ShareLinkBundle.Status.FAILED, - ) - bundle.documents.set([self.document]) - bundle.last_error = "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.assertEqual(bundle.last_error, "") - self.assertIsNone(bundle.size_bytes) - self.assertEqual(bundle.file_path, "") - delay_mock.assert_called_once_with(bundle.pk) - - def test_rebuild_bundle_rejects_processing_status(self): - 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_download_ready_bundle_streams_file(self): - bundle_file = Path(self.dirs.media_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), - ) - 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): - 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) - - @mock.patch("documents.views.build_share_link_bundle.delay") - def test_download_missing_file_triggers_rebuild(self, delay_mock): - bundle = ShareLinkBundle.objects.create( - slug="missingfileslug", - file_version=ShareLink.FileVersion.ARCHIVE, - status=ShareLinkBundle.Status.READY, - file_path=str(Path(self.dirs.media_dir) / "does-not-exist.zip"), - ) - 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) - bundle.refresh_from_db() - self.assertEqual(bundle.status, ShareLinkBundle.Status.PENDING) - delay_mock.assert_called_once_with(bundle.pk) - - -class ShareLinkBundleTaskTests(DirectoriesMixin, APITestCase): - def setUp(self): - super().setUp() - self.document = DocumentFactory.create() - - def test_cleanup_expired_share_link_bundles(self): - expired_path = Path(self.dirs.media_dir) / "expired.zip" - expired_path.parent.mkdir(parents=True, exist_ok=True) - expired_path.write_bytes(b"expired") - - active_path = Path(self.dirs.media_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=str(expired_path), - ) - 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=str(active_path), - ) - 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()) diff --git a/src/documents/tests/test_migration_share_link_bundle.py b/src/documents/tests/test_migration_share_link_bundle.py new file mode 100644 index 000000000..c3804ea5d --- /dev/null +++ b/src/documents/tests/test_migration_share_link_bundle.py @@ -0,0 +1,51 @@ +from documents.tests.utils import TestMigrations + + +class TestMigrateShareLinkBundlePermissions(TestMigrations): + migrate_from = "1074_workflowrun_deleted_at_workflowrun_restored_at_and_more" + migrate_to = "1075_sharelinkbundle" + + def setUpBeforeMigration(self, apps): + User = apps.get_model("auth", "User") + Group = apps.get_model("auth", "Group") + self.Permission = apps.get_model("auth", "Permission") + self.user = User.objects.create(username="user1") + self.group = Group.objects.create(name="group1") + add_document = self.Permission.objects.get(codename="add_document") + self.user.user_permissions.add(add_document.id) + self.group.permissions.add(add_document.id) + + def test_share_link_permissions_granted_to_add_document_holders(self): + share_perms = self.Permission.objects.filter( + codename__contains="sharelinkbundle", + ) + self.assertTrue(self.user.user_permissions.filter(pk__in=share_perms).exists()) + self.assertTrue(self.group.permissions.filter(pk__in=share_perms).exists()) + + +class TestReverseMigrateShareLinkBundlePermissions(TestMigrations): + migrate_from = "1075_sharelinkbundle" + migrate_to = "1074_workflowrun_deleted_at_workflowrun_restored_at_and_more" + + def setUpBeforeMigration(self, apps): + User = apps.get_model("auth", "User") + Group = apps.get_model("auth", "Group") + self.Permission = apps.get_model("auth", "Permission") + self.user = User.objects.create(username="user1") + self.group = Group.objects.create(name="group1") + add_document = self.Permission.objects.get(codename="add_document") + share_perms = self.Permission.objects.filter( + codename__contains="sharelinkbundle", + ) + self.share_perm_ids = list(share_perms.values_list("id", flat=True)) + + self.user.user_permissions.add(add_document.id, *self.share_perm_ids) + self.group.permissions.add(add_document.id, *self.share_perm_ids) + + def test_share_link_permissions_revoked_on_reverse(self): + self.assertFalse( + self.user.user_permissions.filter(pk__in=self.share_perm_ids).exists(), + ) + self.assertFalse( + self.group.permissions.filter(pk__in=self.share_perm_ids).exists(), + ) diff --git a/src/documents/tests/test_share_link_bundles.py b/src/documents/tests/test_share_link_bundles.py new file mode 100644 index 000000000..1d77e024f --- /dev/null +++ b/src/documents/tests/test_share_link_bundles.py @@ -0,0 +1,478 @@ +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.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 documents.tests.factories import DocumentFactory +from documents.tests.utils import DirectoriesMixin + + +class ShareLinkBundleAPITests(DirectoriesMixin, APITestCase): + ENDPOINT = "/api/share_link_bundles/" + + def setUp(self): + super().setUp() + self.user = User.objects.create_superuser(username="bundle_admin") + self.client.force_authenticate(self.user) + self.document = DocumentFactory.create() + + @mock.patch("documents.views.build_share_link_bundle.delay") + def test_create_bundle_triggers_build_job(self, delay_mock): + 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_with(bundle.pk) + + def test_create_bundle_rejects_missing_documents(self): + 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) + + @mock.patch("documents.views.build_share_link_bundle.delay") + def test_rebuild_bundle_resets_state(self, delay_mock): + bundle = ShareLinkBundle.objects.create( + slug="rebuild-slug", + file_version=ShareLink.FileVersion.ARCHIVE, + status=ShareLinkBundle.Status.FAILED, + ) + bundle.documents.set([self.document]) + bundle.last_error = "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.assertEqual(bundle.last_error, "") + self.assertIsNone(bundle.size_bytes) + self.assertEqual(bundle.file_path, "") + delay_mock.assert_called_once_with(bundle.pk) + + def test_rebuild_bundle_rejects_processing_status(self): + 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): + 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): + bundle_file = Path(self.dirs.media_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), + ) + 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): + 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) + + @mock.patch("documents.views.build_share_link_bundle.delay") + def test_download_failed_bundle_triggers_rebuild(self, delay_mock): + bundle_path = ( + Path(settings.MEDIA_ROOT) + / "documents" + / "share_link_bundles" + / "failed.zip" + ) + bundle_path.parent.mkdir(parents=True, exist_ok=True) + bundle_path.write_bytes(b"old-content") + + bundle = ShareLinkBundle.objects.create( + slug="failedslug", + file_version=ShareLink.FileVersion.ARCHIVE, + status=ShareLinkBundle.Status.FAILED, + file_path=str(bundle_path.relative_to(settings.MEDIA_ROOT)), + last_error="Boom", + size_bytes=10, + ) + 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) + bundle.refresh_from_db() + self.assertEqual(bundle.status, ShareLinkBundle.Status.PENDING) + self.assertEqual(bundle.last_error, "") + self.assertIsNone(bundle.size_bytes) + self.assertEqual(bundle.file_path, "") + delay_mock.assert_called_once_with(bundle.pk) + self.assertFalse(bundle_path.exists()) + + @mock.patch("documents.views.build_share_link_bundle.delay") + def test_download_missing_file_triggers_rebuild(self, delay_mock): + bundle = ShareLinkBundle.objects.create( + slug="missingfileslug", + file_version=ShareLink.FileVersion.ARCHIVE, + status=ShareLinkBundle.Status.READY, + file_path=str(Path(self.dirs.media_dir) / "does-not-exist.zip"), + ) + 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) + bundle.refresh_from_db() + self.assertEqual(bundle.status, ShareLinkBundle.Status.PENDING) + delay_mock.assert_called_once_with(bundle.pk) + + def test_expired_share_link_redirects(self): + 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): + 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"]) + + +class ShareLinkBundleTaskTests(DirectoriesMixin, APITestCase): + def setUp(self): + super().setUp() + self.document = DocumentFactory.create() + + def test_cleanup_expired_share_link_bundles(self): + expired_path = Path(self.dirs.media_dir) / "expired.zip" + expired_path.parent.mkdir(parents=True, exist_ok=True) + expired_path.write_bytes(b"expired") + + active_path = Path(self.dirs.media_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=str(expired_path), + ) + 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=str(active_path), + ) + 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()) + + +class ShareLinkBundleBuildTaskTests(DirectoriesMixin, APITestCase): + def setUp(self): + super().setUp() + self.document = DocumentFactory.create( + mime_type="application/pdf", + checksum="123", + ) + self.document.archive_checksum = "" + self.document.save() + self.addCleanup( + setattr, + settings, + "SHARE_LINK_BUNDLE_DIR", + settings.SHARE_LINK_BUNDLE_DIR, + ) + settings.SHARE_LINK_BUNDLE_DIR = ( + Path(settings.MEDIA_ROOT) / "documents" / "share_link_bundles" + ) + + 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 + 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): + 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.assertEqual(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) + 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_stores_absolute_path_outside_media_root(self): + settings.SHARE_LINK_BUNDLE_DIR = Path(settings.DATA_DIR) / "share_link_bundles" + self._write_document_file(archive=False, content=b"source") + bundle = ShareLinkBundle.objects.create( + slug="outside-media", + file_version=ShareLink.FileVersion.ORIGINAL, + ) + bundle.documents.set([self.document]) + + build_share_link_bundle(bundle.pk) + + bundle.refresh_from_db() + self.assertTrue(Path(bundle.file_path).is_absolute()) + self.assertEqual(bundle.status, ShareLinkBundle.Status.READY) + + def test_build_share_link_bundle_failure_marks_failed(self): + 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"), + ): + with self.assertRaises(RuntimeError): + build_share_link_bundle(bundle.pk) + + bundle.refresh_from_db() + self.assertEqual(bundle.status, ShareLinkBundle.Status.FAILED) + self.assertEqual(bundle.last_error, "zip failure") + scratch_zips = list(Path(settings.SCRATCH_DIR).glob("*.zip")) + self.assertFalse(scratch_zips) + + +class ShareLinkBundleFilterSetTests(DirectoriesMixin, APITestCase): + def setUp(self): + 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): + 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): + filterset = ShareLinkBundleFilterSet( + data={"documents": "invalid"}, + queryset=ShareLinkBundle.objects.all(), + ) + + self.assertFalse(filterset.qs.exists()) + + def test_filter_documents_filters_by_multiple_ids(self): + 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]) + + +class ShareLinkBundleModelTests(DirectoriesMixin, APITestCase): + def test_absolute_file_path_handles_relative_and_absolute(self): + relative_path = Path("documents/share_link_bundles/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, + (Path(settings.MEDIA_ROOT) / relative_path).resolve(), + ) + + absolute_path = Path(self.dirs.media_dir) / "absolute.zip" + bundle.file_path = str(absolute_path) + + self.assertEqual(bundle.absolute_file_path, absolute_path.resolve()) + + def test_remove_file_deletes_existing_file(self): + bundle_path = ( + Path(settings.MEDIA_ROOT) + / "documents" + / "share_link_bundles" + / "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.MEDIA_ROOT)), + ) + + bundle.remove_file() + + self.assertFalse(bundle_path.exists()) + + +class ShareLinkBundleSerializerTests(DirectoriesMixin, APITestCase): + def setUp(self): + super().setUp() + self.document = DocumentFactory.create() + + def test_validate_document_ids_rejects_duplicates(self): + 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): + 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): + 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])