mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2025-12-31 13:58:04 -06:00
More backend coverage, rename
This commit is contained in:
shamoon
@@ -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())
|
|
||||||
51
src/documents/tests/test_migration_share_link_bundle.py
Normal file
51
src/documents/tests/test_migration_share_link_bundle.py
Normal file
@@ -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(),
|
||||||
|
)
|
||||||
478
src/documents/tests/test_share_link_bundles.py
Normal file
478
src/documents/tests/test_share_link_bundles.py
Normal file
@@ -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])
|
||||||
Reference in New Issue
Block a user