+from __future__ import annotations
+
from email import message_from_bytes
-from pathlib import Path
+from typing import TYPE_CHECKING
from django.conf import settings
from django.core.mail import EmailMessage
from filelock import FileLock
+if TYPE_CHECKING:
+ from documents.models import Document
+
def send_email(
subject: str,
body: str,
to: list[str],
- attachments: list[tuple[Path, str]],
+ attachments: list[Document],
+ *,
+ use_archive: bool,
) -> int:
"""
Send an email with attachments.
subject: Email subject
body: Email body text
to: List of recipient email addresses
- attachments: List of (path, mime_type) tuples for attachments (the list may be empty)
+ attachments: List of documents to attach (the list may be empty)
+ use_archive: Whether to attach archive versions when available
Returns:
Number of emails sent
to=to,
)
+ used_filenames: set[str] = set()
+
# Something could be renaming the file concurrently so it can't be attached
with FileLock(settings.MEDIA_LOCK):
- for attachment_path, mime_type in attachments:
+ for document in attachments:
+ attachment_path = (
+ document.archive_path
+ if use_archive and document.has_archive_version
+ else document.source_path
+ )
+
+ friendly_filename = _get_unique_filename(
+ document,
+ used_filenames,
+ archive=use_archive and document.has_archive_version,
+ )
+ used_filenames.add(friendly_filename)
+
with attachment_path.open("rb") as f:
content = f.read()
- if mime_type == "message/rfc822":
+ if document.mime_type == "message/rfc822":
# See https://forum.djangoproject.com/t/using-emailmessage-with-an-attached-email-file-crashes-due-to-non-ascii/37981
content = message_from_bytes(content)
email.attach(
- filename=attachment_path.name,
+ filename=friendly_filename,
content=content,
- mimetype=mime_type,
+ mimetype=document.mime_type,
)
return email.send()
+
+
+def _get_unique_filename(doc: Document, used_names: set[str], *, archive: bool) -> str:
+ """
+ Constructs a unique friendly filename for the given document.
+
+ The filename might not be unique enough, so a counter is appended if needed.
+ """
+ counter = 0
+ while True:
+ filename = doc.get_public_filename(archive=archive, counter=counter)
+ if filename not in used_names:
+ return filename
+ counter += 1
filename="test2.pdf",
)
- # Copy sample files to document paths
- shutil.copy(self.SAMPLE_DIR / "simple.pdf", self.doc1.archive_path)
- shutil.copy(self.SAMPLE_DIR / "simple.pdf", self.doc1.source_path)
- shutil.copy(self.SAMPLE_DIR / "simple.pdf", self.doc2.source_path)
+ # Copy sample files to document paths (using different files to distinguish versions)
+ shutil.copy(
+ self.SAMPLE_DIR / "documents" / "originals" / "0000001.pdf",
+ self.doc1.archive_path,
+ )
+ shutil.copy(
+ self.SAMPLE_DIR / "documents" / "originals" / "0000002.pdf",
+ self.doc1.source_path,
+ )
+ shutil.copy(
+ self.SAMPLE_DIR / "documents" / "originals" / "0000003.pdf",
+ self.doc2.source_path,
+ )
@override_settings(
EMAIL_ENABLED=True,
def test_email_success(self):
"""
GIVEN:
- - Multiple existing documents
+ - Multiple existing documents (doc1 with archive, doc2 without)
WHEN:
- API request is made to bulk email documents
THEN:
- Email is sent with all documents attached
+ - Archive version used by default for doc1
+ - Original version used for doc2 (no archive available)
"""
response = self.client.post(
self.ENDPOINT,
self.assertEqual(email.body, "Here are your documents")
self.assertEqual(len(email.attachments), 2)
- # Check attachment names (should default to archive version for doc1, original for doc2)
attachment_names = [att[0] for att in email.attachments]
- self.assertIn("archive1.pdf", attachment_names)
- self.assertIn("test2.pdf", attachment_names)
+ self.assertEqual(len(attachment_names), 2)
+ self.assertIn(f"{self.doc1!s}.pdf", attachment_names)
+ self.assertIn(f"{self.doc2!s}.pdf", attachment_names)
+
+ doc1_attachment = next(
+ att for att in email.attachments if att[0] == f"{self.doc1!s}.pdf"
+ )
+ archive_size = self.doc1.archive_path.stat().st_size
+ self.assertEqual(len(doc1_attachment[1]), archive_size)
+
+ doc2_attachment = next(
+ att for att in email.attachments if att[0] == f"{self.doc2!s}.pdf"
+ )
+ original_size = self.doc2.source_path.stat().st_size
+ self.assertEqual(len(doc2_attachment[1]), original_size)
@override_settings(
EMAIL_ENABLED=True,
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(mail.outbox), 1)
- self.assertEqual(mail.outbox[0].attachments[0][0], "test1.pdf")
+
+ attachment = mail.outbox[0].attachments[0]
+ self.assertEqual(attachment[0], f"{self.doc1!s}.pdf")
+
+ original_size = self.doc1.source_path.stat().st_size
+ self.assertEqual(len(attachment[1]), original_size)
def test_email_missing_required_fields(self):
"""
)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+ @override_settings(
+ EMAIL_ENABLED=True,
+ EMAIL_BACKEND="django.core.mail.backends.locmem.EmailBackend",
+ )
+ def test_email_duplicate_filenames(self):
+ """
+ GIVEN:
+ - Multiple documents with the same title
+ WHEN:
+ - API request is made to bulk email documents
+ THEN:
+ - Filenames are made unique with counters
+ """
+ doc3 = Document.objects.create(
+ title="test1",
+ mime_type="application/pdf",
+ content="this is document 3",
+ checksum="3",
+ filename="test3.pdf",
+ )
+ shutil.copy(self.SAMPLE_DIR / "simple.pdf", doc3.source_path)
+
+ doc4 = Document.objects.create(
+ title="test1",
+ mime_type="application/pdf",
+ content="this is document 4",
+ checksum="4",
+ filename="test4.pdf",
+ )
+ shutil.copy(self.SAMPLE_DIR / "simple.pdf", doc4.source_path)
+
+ response = self.client.post(
+ self.ENDPOINT,
+ json.dumps(
+ {
+ "documents": [self.doc1.pk, doc3.pk, doc4.pk],
+ "addresses": "test@example.com",
+ "subject": "Test",
+ "message": "Test message",
+ },
+ ),
+ content_type="application/json",
+ )
+
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertEqual(len(mail.outbox), 1)
+
+ attachment_names = [att[0] for att in mail.outbox[0].attachments]
+ self.assertEqual(len(attachment_names), 3)
+ self.assertIn(f"{self.doc1!s}.pdf", attachment_names)
+ self.assertIn(f"{doc3!s}_01.pdf", attachment_names)
+ self.assertIn(f"{doc3!s}_02.pdf", attachment_names)
+
@mock.patch(
"django.core.mail.message.EmailMessage.send",
side_effect=Exception("Email error"),
projects / thirdparty / paperless-ngx.git / commitdiff
