mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2025-12-31 13:58:04 -06:00
440 lines
14 KiB
Python
440 lines
14 KiB
Python
import json
|
|
import shutil
|
|
from unittest import mock
|
|
|
|
from django.contrib.auth.models import Permission
|
|
from django.contrib.auth.models import User
|
|
from django.core import mail
|
|
from django.test import override_settings
|
|
from rest_framework import status
|
|
from rest_framework.test import APITestCase
|
|
|
|
from documents.models import Document
|
|
from documents.tests.utils import DirectoriesMixin
|
|
from documents.tests.utils import SampleDirMixin
|
|
|
|
|
|
class TestEmail(DirectoriesMixin, SampleDirMixin, APITestCase):
|
|
ENDPOINT = "/api/documents/email/"
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
|
|
self.user = User.objects.create_superuser(username="temp_admin")
|
|
self.client.force_authenticate(user=self.user)
|
|
|
|
self.doc1 = Document.objects.create(
|
|
title="test1",
|
|
mime_type="application/pdf",
|
|
content="this is document 1",
|
|
checksum="1",
|
|
filename="test1.pdf",
|
|
archive_checksum="A1",
|
|
archive_filename="archive1.pdf",
|
|
)
|
|
self.doc2 = Document.objects.create(
|
|
title="test2",
|
|
mime_type="application/pdf",
|
|
content="this is document 2",
|
|
checksum="2",
|
|
filename="test2.pdf",
|
|
)
|
|
|
|
# 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,
|
|
EMAIL_BACKEND="django.core.mail.backends.locmem.EmailBackend",
|
|
)
|
|
def test_email_success(self):
|
|
"""
|
|
GIVEN:
|
|
- 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,
|
|
json.dumps(
|
|
{
|
|
"documents": [self.doc1.pk, self.doc2.pk],
|
|
"addresses": "hello@paperless-ngx.com,test@example.com",
|
|
"subject": "Bulk email test",
|
|
"message": "Here are your documents",
|
|
},
|
|
),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(response.data["message"], "Email sent")
|
|
self.assertEqual(len(mail.outbox), 1)
|
|
|
|
email = mail.outbox[0]
|
|
self.assertEqual(email.to, ["hello@paperless-ngx.com", "test@example.com"])
|
|
self.assertEqual(email.subject, "Bulk email test")
|
|
self.assertEqual(email.body, "Here are your documents")
|
|
self.assertEqual(len(email.attachments), 2)
|
|
|
|
attachment_names = [att[0] for att in email.attachments]
|
|
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,
|
|
EMAIL_BACKEND="django.core.mail.backends.locmem.EmailBackend",
|
|
)
|
|
def test_email_use_original_version(self):
|
|
"""
|
|
GIVEN:
|
|
- Documents with archive versions
|
|
WHEN:
|
|
- API request is made to bulk email with use_archive_version=False
|
|
THEN:
|
|
- Original files are attached instead of archive versions
|
|
"""
|
|
response = self.client.post(
|
|
self.ENDPOINT,
|
|
json.dumps(
|
|
{
|
|
"documents": [self.doc1.pk],
|
|
"addresses": "test@example.com",
|
|
"subject": "Test",
|
|
"message": "Test message",
|
|
"use_archive_version": False,
|
|
},
|
|
),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(len(mail.outbox), 1)
|
|
|
|
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):
|
|
"""
|
|
GIVEN:
|
|
- Request with missing required fields
|
|
WHEN:
|
|
- API request is made to bulk email endpoint
|
|
THEN:
|
|
- Bad request response is returned
|
|
"""
|
|
# Missing addresses
|
|
response = self.client.post(
|
|
self.ENDPOINT,
|
|
json.dumps(
|
|
{
|
|
"documents": [self.doc1.pk],
|
|
"subject": "Test",
|
|
"message": "Test message",
|
|
},
|
|
),
|
|
content_type="application/json",
|
|
)
|
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
|
|
|
# Missing subject
|
|
response = self.client.post(
|
|
self.ENDPOINT,
|
|
json.dumps(
|
|
{
|
|
"documents": [self.doc1.pk],
|
|
"addresses": "test@example.com",
|
|
"message": "Test message",
|
|
},
|
|
),
|
|
content_type="application/json",
|
|
)
|
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
|
|
|
# Missing message
|
|
response = self.client.post(
|
|
self.ENDPOINT,
|
|
json.dumps(
|
|
{
|
|
"documents": [self.doc1.pk],
|
|
"addresses": "test@example.com",
|
|
"subject": "Test",
|
|
},
|
|
),
|
|
content_type="application/json",
|
|
)
|
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
|
|
|
# Missing documents
|
|
response = self.client.post(
|
|
self.ENDPOINT,
|
|
json.dumps(
|
|
{
|
|
"addresses": "test@example.com",
|
|
"subject": "Test",
|
|
"message": "Test message",
|
|
},
|
|
),
|
|
content_type="application/json",
|
|
)
|
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
|
|
|
def test_email_empty_document_list(self):
|
|
"""
|
|
GIVEN:
|
|
- Request with empty document list
|
|
WHEN:
|
|
- API request is made to bulk email endpoint
|
|
THEN:
|
|
- Bad request response is returned
|
|
"""
|
|
response = self.client.post(
|
|
self.ENDPOINT,
|
|
json.dumps(
|
|
{
|
|
"documents": [],
|
|
"addresses": "test@example.com",
|
|
"subject": "Test",
|
|
"message": "Test message",
|
|
},
|
|
),
|
|
content_type="application/json",
|
|
)
|
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
|
|
|
def test_email_invalid_document_id(self):
|
|
"""
|
|
GIVEN:
|
|
- Request with non-existent document ID
|
|
WHEN:
|
|
- API request is made to bulk email endpoint
|
|
THEN:
|
|
- Bad request response is returned
|
|
"""
|
|
response = self.client.post(
|
|
self.ENDPOINT,
|
|
json.dumps(
|
|
{
|
|
"documents": [999],
|
|
"addresses": "test@example.com",
|
|
"subject": "Test",
|
|
"message": "Test message",
|
|
},
|
|
),
|
|
content_type="application/json",
|
|
)
|
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
|
|
|
def test_email_invalid_email_address(self):
|
|
"""
|
|
GIVEN:
|
|
- Request with invalid email address
|
|
WHEN:
|
|
- API request is made to bulk email endpoint
|
|
THEN:
|
|
- Bad request response is returned
|
|
"""
|
|
response = self.client.post(
|
|
self.ENDPOINT,
|
|
json.dumps(
|
|
{
|
|
"documents": [self.doc1.pk],
|
|
"addresses": "invalid-email",
|
|
"subject": "Test",
|
|
"message": "Test message",
|
|
},
|
|
),
|
|
content_type="application/json",
|
|
)
|
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
|
|
|
# Test multiple addresses with one invalid
|
|
response = self.client.post(
|
|
self.ENDPOINT,
|
|
json.dumps(
|
|
{
|
|
"documents": [self.doc1.pk],
|
|
"addresses": "valid@example.com,invalid-email",
|
|
"subject": "Test",
|
|
"message": "Test message",
|
|
},
|
|
),
|
|
content_type="application/json",
|
|
)
|
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
|
|
|
def test_email_insufficient_permissions(self):
|
|
"""
|
|
GIVEN:
|
|
- User without permissions to view document
|
|
WHEN:
|
|
- API request is made to bulk email documents
|
|
THEN:
|
|
- Forbidden response is returned
|
|
"""
|
|
user1 = User.objects.create_user(username="test1")
|
|
user1.user_permissions.add(*Permission.objects.filter(codename="view_document"))
|
|
|
|
doc_owned = Document.objects.create(
|
|
title="owned_doc",
|
|
mime_type="application/pdf",
|
|
checksum="owned",
|
|
owner=self.user,
|
|
)
|
|
|
|
self.client.force_authenticate(user1)
|
|
|
|
response = self.client.post(
|
|
self.ENDPOINT,
|
|
json.dumps(
|
|
{
|
|
"documents": [self.doc1.pk, doc_owned.pk],
|
|
"addresses": "test@example.com",
|
|
"subject": "Test",
|
|
"message": "Test message",
|
|
},
|
|
),
|
|
content_type="application/json",
|
|
)
|
|
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|
|
|
|
def test_email_only_requires_view_permission(self):
|
|
"""
|
|
GIVEN:
|
|
- User having only view documents permission
|
|
WHEN:
|
|
- API request is made to bulk email documents
|
|
THEN:
|
|
- Request succeeds
|
|
"""
|
|
user1 = User.objects.create_user(username="test1")
|
|
user1.user_permissions.add(*Permission.objects.filter(codename="view_document"))
|
|
|
|
self.client.force_authenticate(user1)
|
|
|
|
response = self.client.post(
|
|
self.ENDPOINT,
|
|
json.dumps(
|
|
{
|
|
"documents": [self.doc1.pk],
|
|
"addresses": "test@example.com",
|
|
"subject": "Test",
|
|
"message": "Test message",
|
|
},
|
|
),
|
|
content_type="application/json",
|
|
)
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
|
|
@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"),
|
|
)
|
|
def test_email_send_error(self, mocked_send):
|
|
"""
|
|
GIVEN:
|
|
- Existing documents
|
|
WHEN:
|
|
- API request is made to bulk email and error occurs during email send
|
|
THEN:
|
|
- Server error response is returned
|
|
"""
|
|
response = self.client.post(
|
|
self.ENDPOINT,
|
|
json.dumps(
|
|
{
|
|
"documents": [self.doc1.pk],
|
|
"addresses": "test@example.com",
|
|
"subject": "Test",
|
|
"message": "Test message",
|
|
},
|
|
),
|
|
content_type="application/json",
|
|
)
|
|
self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
|
|
self.assertIn("Error emailing documents", response.content.decode())
|