mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2026-08-10 21:03:18 +00:00
Trenton HandGitHub
c28c532bef
ProcessedMailViewSet.bulk_delete checked permissions inside the delete loop, so an unpermitted id returned 403 only after the mails ahead of it had already been deleted. Resolve the permitted set once via permitted_object_ids and reject before deleting anything, which also drops the per-mail permission queries.
787 lines
29 KiB
Python
787 lines
29 KiB
Python
import json
|
|
from unittest import mock
|
|
|
|
from django.contrib.auth.models import Permission
|
|
from django.contrib.auth.models import User
|
|
from guardian.shortcuts import assign_perm
|
|
from rest_framework import status
|
|
from rest_framework.test import APITestCase
|
|
|
|
from documents.tests.factories import CorrespondentFactory
|
|
from documents.tests.factories import DocumentTypeFactory
|
|
from documents.tests.factories import TagFactory
|
|
from documents.tests.utils import DirectoriesMixin
|
|
from paperless_mail.models import MailAccount
|
|
from paperless_mail.models import MailRule
|
|
from paperless_mail.models import ProcessedMail
|
|
from paperless_mail.tests.factories import MailAccountFactory
|
|
from paperless_mail.tests.factories import MailRuleFactory
|
|
from paperless_mail.tests.factories import ProcessedMailFactory
|
|
from paperless_mail.tests.test_mail import BogusMailBox
|
|
|
|
|
|
class TestAPIMailAccounts(DirectoriesMixin, APITestCase):
|
|
ENDPOINT = "/api/mail_accounts/"
|
|
|
|
def setUp(self) -> None:
|
|
self.bogus_mailbox = BogusMailBox()
|
|
|
|
patcher = mock.patch("paperless_mail.mail.MailBox")
|
|
m = patcher.start()
|
|
m.return_value = self.bogus_mailbox
|
|
self.addCleanup(patcher.stop)
|
|
|
|
super().setUp()
|
|
|
|
self.user = User.objects.create_user(username="temp_admin")
|
|
self.user.user_permissions.add(*Permission.objects.all())
|
|
self.user.save()
|
|
self.client.force_authenticate(user=self.user)
|
|
|
|
def test_get_mail_accounts(self) -> None:
|
|
"""
|
|
GIVEN:
|
|
- Configured mail accounts
|
|
WHEN:
|
|
- API call is made to get mail accounts
|
|
THEN:
|
|
- Configured mail accounts are provided
|
|
"""
|
|
|
|
account1 = MailAccountFactory(
|
|
name="Email1",
|
|
username="username1",
|
|
password="password1",
|
|
imap_server="server.example.com",
|
|
imap_port=443,
|
|
)
|
|
|
|
response = self.client.get(self.ENDPOINT)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(response.data["count"], 1)
|
|
returned_account1 = response.data["results"][0]
|
|
|
|
self.assertEqual(returned_account1["name"], account1.name)
|
|
self.assertEqual(returned_account1["username"], account1.username)
|
|
self.assertEqual(
|
|
returned_account1["password"],
|
|
"**********",
|
|
)
|
|
self.assertEqual(returned_account1["imap_server"], account1.imap_server)
|
|
self.assertEqual(returned_account1["imap_port"], account1.imap_port)
|
|
self.assertEqual(returned_account1["imap_security"], account1.imap_security)
|
|
self.assertEqual(returned_account1["character_set"], account1.character_set)
|
|
|
|
def test_create_mail_account(self) -> None:
|
|
"""
|
|
WHEN:
|
|
- API request is made to add a mail account
|
|
THEN:
|
|
- A new mail account is created
|
|
"""
|
|
|
|
account1 = {
|
|
"name": "Email1",
|
|
"username": "username1",
|
|
"password": "password1",
|
|
"imap_server": "server.example.com",
|
|
"imap_port": 443,
|
|
"imap_security": MailAccount.ImapSecurity.SSL,
|
|
"character_set": "UTF-8",
|
|
}
|
|
|
|
response = self.client.post(
|
|
self.ENDPOINT,
|
|
data=account1,
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
|
|
|
returned_account1 = MailAccount.objects.get(name="Email1")
|
|
|
|
self.assertEqual(returned_account1.name, account1["name"])
|
|
self.assertEqual(returned_account1.username, account1["username"])
|
|
self.assertEqual(returned_account1.password, account1["password"])
|
|
self.assertEqual(returned_account1.imap_server, account1["imap_server"])
|
|
self.assertEqual(returned_account1.imap_port, account1["imap_port"])
|
|
self.assertEqual(returned_account1.imap_security, account1["imap_security"])
|
|
self.assertEqual(returned_account1.character_set, account1["character_set"])
|
|
|
|
def test_delete_mail_account(self) -> None:
|
|
"""
|
|
GIVEN:
|
|
- Existing mail account
|
|
WHEN:
|
|
- API request is made to delete a mail account
|
|
THEN:
|
|
- Account is deleted
|
|
"""
|
|
|
|
account1 = MailAccountFactory()
|
|
|
|
response = self.client.delete(
|
|
f"{self.ENDPOINT}{account1.pk}/",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
|
|
|
|
self.assertEqual(len(MailAccount.objects.all()), 0)
|
|
|
|
def test_update_mail_account(self) -> None:
|
|
"""
|
|
GIVEN:
|
|
- Existing mail accounts
|
|
WHEN:
|
|
- API request is made to update mail account
|
|
THEN:
|
|
- The mail account is updated, password only updated if not '****'
|
|
"""
|
|
|
|
account1 = MailAccountFactory()
|
|
|
|
response = self.client.patch(
|
|
f"{self.ENDPOINT}{account1.pk}/",
|
|
data={
|
|
"name": "Updated Name 1",
|
|
"password": "******",
|
|
},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
|
|
returned_account1 = MailAccount.objects.get(pk=account1.pk)
|
|
self.assertEqual(returned_account1.name, "Updated Name 1")
|
|
self.assertEqual(returned_account1.password, account1.password)
|
|
|
|
response = self.client.patch(
|
|
f"{self.ENDPOINT}{account1.pk}/",
|
|
data={
|
|
"name": "Updated Name 2",
|
|
"password": "123xyz",
|
|
},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
|
|
returned_account2 = MailAccount.objects.get(pk=account1.pk)
|
|
self.assertEqual(returned_account2.name, "Updated Name 2")
|
|
self.assertEqual(returned_account2.password, "123xyz")
|
|
|
|
def test_mail_account_test_fail(self) -> None:
|
|
"""
|
|
GIVEN:
|
|
- Errnoeous mail account details
|
|
WHEN:
|
|
- API call is made to test account
|
|
THEN:
|
|
- API returns 400 bad request
|
|
"""
|
|
|
|
response = self.client.post(
|
|
f"{self.ENDPOINT}test/",
|
|
json.dumps(
|
|
{
|
|
"imap_server": "server.example.com",
|
|
"imap_port": 443,
|
|
"imap_security": MailAccount.ImapSecurity.SSL,
|
|
"username": "admin",
|
|
"password": "notcorrect",
|
|
},
|
|
),
|
|
content_type="application/json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
|
|
|
def test_mail_account_test_success(self) -> None:
|
|
"""
|
|
GIVEN:
|
|
- Working mail account details
|
|
WHEN:
|
|
- API call is made to test account
|
|
THEN:
|
|
- API returns success
|
|
"""
|
|
|
|
response = self.client.post(
|
|
f"{self.ENDPOINT}test/",
|
|
json.dumps(
|
|
{
|
|
"imap_server": "server.example.com",
|
|
"imap_port": 443,
|
|
"imap_security": MailAccount.ImapSecurity.SSL,
|
|
"username": "admin",
|
|
"password": "secret",
|
|
},
|
|
),
|
|
content_type="application/json",
|
|
)
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(response.data["success"], True)
|
|
|
|
def test_mail_account_test_existing(self) -> None:
|
|
"""
|
|
GIVEN:
|
|
- Testing server details for an existing account with obfuscated password (***)
|
|
WHEN:
|
|
- API call is made to test account
|
|
THEN:
|
|
- API returns success
|
|
"""
|
|
account = MailAccountFactory(
|
|
username="admin",
|
|
password="secret",
|
|
imap_server="server.example.com",
|
|
imap_port=443,
|
|
)
|
|
|
|
response = self.client.post(
|
|
f"{self.ENDPOINT}test/",
|
|
json.dumps(
|
|
{
|
|
"id": account.pk,
|
|
"imap_server": "server.example.com",
|
|
"imap_port": 443,
|
|
"imap_security": MailAccount.ImapSecurity.SSL,
|
|
"username": "admin",
|
|
"password": "******",
|
|
},
|
|
),
|
|
content_type="application/json",
|
|
)
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(response.data["success"], True)
|
|
|
|
def test_mail_account_test_existing_nonexistent_id_forbidden(self) -> None:
|
|
response = self.client.post(
|
|
f"{self.ENDPOINT}test/",
|
|
json.dumps(
|
|
{
|
|
"id": 999999,
|
|
"imap_server": "server.example.com",
|
|
"imap_port": 443,
|
|
"imap_security": MailAccount.ImapSecurity.SSL,
|
|
"username": "admin",
|
|
"password": "******",
|
|
},
|
|
),
|
|
content_type="application/json",
|
|
)
|
|
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|
|
self.assertEqual(response.content.decode(), "Insufficient permissions")
|
|
|
|
def test_get_mail_accounts_owner_aware(self) -> None:
|
|
"""
|
|
GIVEN:
|
|
- Configured accounts with different users
|
|
WHEN:
|
|
- API call is made to get mail accounts
|
|
THEN:
|
|
- Only unowned, owned by user or granted accounts are provided
|
|
"""
|
|
|
|
user2 = User.objects.create_user(username="temp_admin2")
|
|
|
|
account1 = MailAccountFactory(name="Email1")
|
|
account2 = MailAccountFactory(name="Email2", owner=self.user)
|
|
_account3 = MailAccountFactory(name="Email3", owner=user2)
|
|
account4 = MailAccountFactory(name="Email4", owner=user2)
|
|
assign_perm("view_mailaccount", self.user, account4)
|
|
|
|
response = self.client.get(self.ENDPOINT)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(response.data["count"], 3)
|
|
self.assertEqual(response.data["results"][0]["name"], account1.name)
|
|
self.assertEqual(response.data["results"][1]["name"], account2.name)
|
|
self.assertEqual(response.data["results"][2]["name"], account4.name)
|
|
|
|
|
|
class TestAPIMailRules(DirectoriesMixin, APITestCase):
|
|
ENDPOINT = "/api/mail_rules/"
|
|
|
|
def setUp(self) -> None:
|
|
super().setUp()
|
|
|
|
self.user = User.objects.create_user(username="temp_admin")
|
|
self.user.user_permissions.add(*Permission.objects.all())
|
|
self.user.save()
|
|
self.client.force_authenticate(user=self.user)
|
|
|
|
def test_get_mail_rules(self) -> None:
|
|
"""
|
|
GIVEN:
|
|
- Configured mail accounts and rules
|
|
WHEN:
|
|
- API call is made to get mail rules
|
|
THEN:
|
|
- Configured mail rules are provided
|
|
"""
|
|
|
|
account1 = MailAccountFactory()
|
|
rule1 = MailRuleFactory(
|
|
name="Rule1",
|
|
account=account1,
|
|
filter_from="from@example.com",
|
|
filter_to="someone@somewhere.com",
|
|
filter_subject="subject",
|
|
filter_body="body",
|
|
filter_attachment_filename_include="file.pdf",
|
|
)
|
|
|
|
response = self.client.get(self.ENDPOINT)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(response.data["count"], 1)
|
|
returned_rule1 = response.data["results"][0]
|
|
|
|
self.assertEqual(returned_rule1["name"], rule1.name)
|
|
self.assertEqual(returned_rule1["account"], account1.pk)
|
|
self.assertEqual(returned_rule1["folder"], rule1.folder)
|
|
self.assertEqual(returned_rule1["filter_from"], rule1.filter_from)
|
|
self.assertEqual(returned_rule1["filter_to"], rule1.filter_to)
|
|
self.assertEqual(returned_rule1["filter_subject"], rule1.filter_subject)
|
|
self.assertEqual(returned_rule1["filter_body"], rule1.filter_body)
|
|
self.assertEqual(
|
|
returned_rule1["filter_attachment_filename_include"],
|
|
rule1.filter_attachment_filename_include,
|
|
)
|
|
self.assertEqual(returned_rule1["maximum_age"], rule1.maximum_age)
|
|
self.assertEqual(returned_rule1["action"], rule1.action)
|
|
self.assertEqual(returned_rule1["assign_title_from"], rule1.assign_title_from)
|
|
self.assertEqual(
|
|
returned_rule1["assign_correspondent_from"],
|
|
rule1.assign_correspondent_from,
|
|
)
|
|
self.assertEqual(returned_rule1["order"], rule1.order)
|
|
self.assertEqual(returned_rule1["attachment_type"], rule1.attachment_type)
|
|
|
|
def test_create_mail_rule(self) -> None:
|
|
"""
|
|
GIVEN:
|
|
- Configured mail account exists
|
|
WHEN:
|
|
- API request is made to add a mail rule
|
|
THEN:
|
|
- A new mail rule is created
|
|
"""
|
|
|
|
account1 = MailAccountFactory()
|
|
tag = TagFactory(name="t")
|
|
correspondent = CorrespondentFactory(name="c")
|
|
document_type = DocumentTypeFactory(name="dt")
|
|
|
|
rule1 = {
|
|
"name": "Rule1",
|
|
"account": account1.pk,
|
|
"folder": "INBOX",
|
|
"filter_from": "from@example.com",
|
|
"filter_to": "aperson@aplace.com",
|
|
"filter_subject": "subject",
|
|
"filter_body": "body",
|
|
"filter_attachment_filename_include": "file.pdf",
|
|
"maximum_age": 30,
|
|
"action": MailRule.MailAction.MARK_READ,
|
|
"assign_title_from": MailRule.TitleSource.FROM_SUBJECT,
|
|
"assign_correspondent_from": MailRule.CorrespondentSource.FROM_NOTHING,
|
|
"order": 0,
|
|
"attachment_type": MailRule.AttachmentProcessing.ATTACHMENTS_ONLY,
|
|
"action_parameter": "parameter",
|
|
"assign_tags": [tag.pk],
|
|
"assign_correspondent": correspondent.pk,
|
|
"assign_document_type": document_type.pk,
|
|
"assign_owner_from_rule": True,
|
|
}
|
|
|
|
response = self.client.post(
|
|
self.ENDPOINT,
|
|
data=rule1,
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
|
|
|
response = self.client.get(self.ENDPOINT)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(response.data["count"], 1)
|
|
returned_rule1 = response.data["results"][0]
|
|
|
|
self.assertEqual(returned_rule1["name"], rule1["name"])
|
|
self.assertEqual(returned_rule1["account"], account1.pk)
|
|
self.assertEqual(returned_rule1["folder"], rule1["folder"])
|
|
self.assertEqual(returned_rule1["filter_from"], rule1["filter_from"])
|
|
self.assertEqual(returned_rule1["filter_to"], rule1["filter_to"])
|
|
self.assertEqual(returned_rule1["filter_subject"], rule1["filter_subject"])
|
|
self.assertEqual(returned_rule1["filter_body"], rule1["filter_body"])
|
|
self.assertEqual(
|
|
returned_rule1["filter_attachment_filename_include"],
|
|
rule1["filter_attachment_filename_include"],
|
|
)
|
|
self.assertEqual(returned_rule1["maximum_age"], rule1["maximum_age"])
|
|
self.assertEqual(returned_rule1["action"], rule1["action"])
|
|
self.assertEqual(
|
|
returned_rule1["assign_title_from"],
|
|
rule1["assign_title_from"],
|
|
)
|
|
self.assertEqual(
|
|
returned_rule1["assign_correspondent_from"],
|
|
rule1["assign_correspondent_from"],
|
|
)
|
|
self.assertEqual(returned_rule1["order"], rule1["order"])
|
|
self.assertEqual(returned_rule1["attachment_type"], rule1["attachment_type"])
|
|
self.assertEqual(returned_rule1["action_parameter"], rule1["action_parameter"])
|
|
self.assertEqual(
|
|
returned_rule1["assign_correspondent"],
|
|
rule1["assign_correspondent"],
|
|
)
|
|
self.assertEqual(
|
|
returned_rule1["assign_document_type"],
|
|
rule1["assign_document_type"],
|
|
)
|
|
self.assertEqual(returned_rule1["assign_tags"], rule1["assign_tags"])
|
|
self.assertEqual(
|
|
returned_rule1["assign_owner_from_rule"],
|
|
rule1["assign_owner_from_rule"],
|
|
)
|
|
|
|
def test_delete_mail_rule(self) -> None:
|
|
"""
|
|
GIVEN:
|
|
- Existing mail rule
|
|
WHEN:
|
|
- API request is made to delete a mail rule
|
|
THEN:
|
|
- Rule is deleted
|
|
"""
|
|
|
|
account1 = MailAccountFactory()
|
|
rule1 = MailRuleFactory(account=account1)
|
|
|
|
response = self.client.delete(
|
|
f"{self.ENDPOINT}{rule1.pk}/",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
|
|
|
|
self.assertEqual(len(MailRule.objects.all()), 0)
|
|
|
|
def test_update_mail_rule(self) -> None:
|
|
"""
|
|
GIVEN:
|
|
- Existing mail rule
|
|
WHEN:
|
|
- API request is made to update mail rule
|
|
THEN:
|
|
- The mail rule is updated
|
|
"""
|
|
|
|
account1 = MailAccountFactory()
|
|
rule1 = MailRuleFactory(account=account1)
|
|
|
|
response = self.client.patch(
|
|
f"{self.ENDPOINT}{rule1.pk}/",
|
|
data={
|
|
"name": "Updated Name 1",
|
|
"action": MailRule.MailAction.DELETE,
|
|
},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
|
|
returned_rule1 = MailRule.objects.get(pk=rule1.pk)
|
|
self.assertEqual(returned_rule1.name, "Updated Name 1")
|
|
self.assertEqual(returned_rule1.action, MailRule.MailAction.DELETE)
|
|
|
|
def test_create_mail_rule_scopes_accounts(self) -> None:
|
|
other_user = User.objects.create_user(username="mail-owner")
|
|
foreign_account = MailAccountFactory(name="ForeignEmail", owner=other_user)
|
|
|
|
response = self.client.post(
|
|
self.ENDPOINT,
|
|
data={
|
|
"name": "Rule1",
|
|
"account": foreign_account.pk,
|
|
"folder": "INBOX",
|
|
"filter_from": "from@example.com",
|
|
"maximum_age": 30,
|
|
"action": MailRule.MailAction.MARK_READ,
|
|
"assign_title_from": MailRule.TitleSource.FROM_SUBJECT,
|
|
"assign_correspondent_from": MailRule.CorrespondentSource.FROM_NOTHING,
|
|
"order": 0,
|
|
"attachment_type": MailRule.AttachmentProcessing.ATTACHMENTS_ONLY,
|
|
},
|
|
)
|
|
missing_response = self.client.post(
|
|
self.ENDPOINT,
|
|
data={
|
|
"name": "Rule1",
|
|
"account": foreign_account.pk + 1000,
|
|
"folder": "INBOX",
|
|
"filter_from": "from@example.com",
|
|
"maximum_age": 30,
|
|
"action": MailRule.MailAction.MARK_READ,
|
|
"assign_title_from": MailRule.TitleSource.FROM_SUBJECT,
|
|
"assign_correspondent_from": MailRule.CorrespondentSource.FROM_NOTHING,
|
|
"order": 0,
|
|
"attachment_type": MailRule.AttachmentProcessing.ATTACHMENTS_ONLY,
|
|
},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
|
self.assertEqual(missing_response.status_code, status.HTTP_400_BAD_REQUEST)
|
|
self.assertEqual(response.data["account"][0].code, "does_not_exist")
|
|
self.assertEqual(missing_response.data["account"][0].code, "does_not_exist")
|
|
self.assertEqual(MailRule.objects.count(), 0)
|
|
|
|
def test_create_mail_rule_allowed_for_granted_account_change_permission(
|
|
self,
|
|
) -> None:
|
|
other_user = User.objects.create_user(username="mail-owner")
|
|
foreign_account = MailAccountFactory(name="ForeignEmail", owner=other_user)
|
|
assign_perm("change_mailaccount", self.user, foreign_account)
|
|
|
|
response = self.client.post(
|
|
self.ENDPOINT,
|
|
data={
|
|
"name": "Rule1",
|
|
"account": foreign_account.pk,
|
|
"folder": "INBOX",
|
|
"filter_from": "from@example.com",
|
|
"maximum_age": 30,
|
|
"action": MailRule.MailAction.MARK_READ,
|
|
"assign_title_from": MailRule.TitleSource.FROM_SUBJECT,
|
|
"assign_correspondent_from": MailRule.CorrespondentSource.FROM_NOTHING,
|
|
"order": 0,
|
|
"attachment_type": MailRule.AttachmentProcessing.ATTACHMENTS_ONLY,
|
|
},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
|
self.assertEqual(MailRule.objects.get().account, foreign_account)
|
|
|
|
def test_update_mail_rule_forbidden_for_unpermitted_account(self) -> None:
|
|
own_account = MailAccountFactory()
|
|
other_user = User.objects.create_user(username="mail-owner")
|
|
foreign_account = MailAccountFactory(owner=other_user)
|
|
rule1 = MailRuleFactory(account=own_account)
|
|
|
|
response = self.client.patch(
|
|
f"{self.ENDPOINT}{rule1.pk}/",
|
|
data={"account": foreign_account.pk},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
|
rule1.refresh_from_db()
|
|
self.assertEqual(rule1.account, own_account)
|
|
|
|
def test_get_mail_rules_owner_aware(self) -> None:
|
|
"""
|
|
GIVEN:
|
|
- Configured rules with different users
|
|
WHEN:
|
|
- API call is made to get mail rules
|
|
THEN:
|
|
- Only unowned, owned by user or granted mail rules are provided
|
|
"""
|
|
|
|
user2 = User.objects.create_user(username="temp_admin2")
|
|
account1 = MailAccountFactory()
|
|
rule1 = MailRuleFactory(account=account1, order=0)
|
|
rule2 = MailRuleFactory(account=account1, order=1, owner=self.user)
|
|
MailRuleFactory(account=account1, order=2, owner=user2)
|
|
rule4 = MailRuleFactory(account=account1, order=3, owner=user2)
|
|
assign_perm("view_mailrule", self.user, rule4)
|
|
|
|
response = self.client.get(self.ENDPOINT)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(response.data["count"], 3)
|
|
self.assertEqual(response.data["results"][0]["name"], rule1.name)
|
|
self.assertEqual(response.data["results"][1]["name"], rule2.name)
|
|
self.assertEqual(response.data["results"][2]["name"], rule4.name)
|
|
|
|
def test_mailrule_maxage_validation(self) -> None:
|
|
"""
|
|
GIVEN:
|
|
- An existing mail account
|
|
WHEN:
|
|
- The user submits a mail rule with an excessively large maximum_age
|
|
THEN:
|
|
- The API should reject the request
|
|
"""
|
|
account = MailAccountFactory()
|
|
|
|
rule_data = {
|
|
"name": "Rule1",
|
|
"account": account.pk,
|
|
"folder": "INBOX",
|
|
"filter_from": "from@example.com",
|
|
"filter_to": "aperson@aplace.com",
|
|
"filter_subject": "subject",
|
|
"filter_body": "body",
|
|
"filter_attachment_filename_include": "file.pdf",
|
|
"maximum_age": 9000000,
|
|
"action": MailRule.MailAction.MARK_READ,
|
|
"assign_title_from": MailRule.TitleSource.FROM_SUBJECT,
|
|
"assign_correspondent_from": MailRule.CorrespondentSource.FROM_NOTHING,
|
|
"order": 0,
|
|
"attachment_type": MailRule.AttachmentProcessing.ATTACHMENTS_ONLY,
|
|
}
|
|
|
|
response = self.client.post(self.ENDPOINT, data=rule_data, format="json")
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
|
self.assertIn("maximum_age", response.data)
|
|
|
|
|
|
class TestAPIProcessedMails(DirectoriesMixin, APITestCase):
|
|
ENDPOINT = "/api/processed_mail/"
|
|
|
|
def setUp(self) -> None:
|
|
super().setUp()
|
|
|
|
self.user = User.objects.create_user(username="temp_admin")
|
|
self.user.user_permissions.add(*Permission.objects.all())
|
|
self.user.save()
|
|
self.client.force_authenticate(user=self.user)
|
|
|
|
def test_get_processed_mails_owner_aware(self) -> None:
|
|
"""
|
|
GIVEN:
|
|
- Configured processed mails with different users
|
|
WHEN:
|
|
- API call is made to get processed mails
|
|
THEN:
|
|
- Only unowned, owned by user or granted processed mails are provided
|
|
"""
|
|
user2 = User.objects.create_user(username="temp_admin2")
|
|
rule = MailRuleFactory()
|
|
pm1 = ProcessedMailFactory(rule=rule)
|
|
pm2 = ProcessedMailFactory(
|
|
rule=rule,
|
|
status="FAILED",
|
|
error="err",
|
|
owner=self.user,
|
|
)
|
|
ProcessedMailFactory(rule=rule, owner=user2)
|
|
pm4 = ProcessedMailFactory(rule=rule, owner=user2)
|
|
assign_perm("view_processedmail", self.user, pm4)
|
|
|
|
response = self.client.get(self.ENDPOINT)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(response.data["count"], 3)
|
|
returned_ids = {r["id"] for r in response.data["results"]}
|
|
self.assertSetEqual(returned_ids, {pm1.id, pm2.id, pm4.id})
|
|
|
|
def test_get_processed_mails_filter_by_rule(self) -> None:
|
|
"""
|
|
GIVEN:
|
|
- Processed mails belonging to two different rules
|
|
WHEN:
|
|
- API call is made with rule filter
|
|
THEN:
|
|
- Only processed mails for that rule are returned
|
|
"""
|
|
account = MailAccountFactory()
|
|
rule1 = MailRuleFactory(account=account)
|
|
rule2 = MailRuleFactory(account=account)
|
|
pm1 = ProcessedMailFactory(rule=rule1, owner=self.user)
|
|
pm2 = ProcessedMailFactory(rule=rule1, status="FAILED", error="e")
|
|
ProcessedMailFactory(rule=rule2)
|
|
|
|
response = self.client.get(f"{self.ENDPOINT}?rule={rule1.pk}")
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
returned_ids = {r["id"] for r in response.data["results"]}
|
|
self.assertSetEqual(returned_ids, {pm1.id, pm2.id})
|
|
|
|
def test_bulk_delete_processed_mails(self) -> None:
|
|
"""
|
|
GIVEN:
|
|
- Processed mails belonging to two different rules and different users
|
|
WHEN:
|
|
- API call is made to bulk delete some of the processed mails
|
|
THEN:
|
|
- Only the specified processed mails are deleted, respecting ownership and permissions
|
|
"""
|
|
user2 = User.objects.create_user(username="temp_admin2")
|
|
rule = MailRuleFactory()
|
|
# unowned, owned by self, and one with explicit object perm
|
|
pm_unowned = ProcessedMailFactory(rule=rule)
|
|
pm_owned = ProcessedMailFactory(
|
|
rule=rule,
|
|
status="FAILED",
|
|
error="e",
|
|
owner=self.user,
|
|
)
|
|
pm_granted = ProcessedMailFactory(rule=rule, owner=user2)
|
|
assign_perm("delete_processedmail", self.user, pm_granted)
|
|
pm_forbidden = ProcessedMailFactory(rule=rule, owner=user2)
|
|
|
|
# Success for allowed items
|
|
response = self.client.post(
|
|
f"{self.ENDPOINT}bulk_delete/",
|
|
data={
|
|
"mail_ids": [pm_unowned.id, pm_owned.id, pm_granted.id],
|
|
},
|
|
format="json",
|
|
)
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(response.data["result"], "OK")
|
|
self.assertSetEqual(
|
|
set(response.data["deleted_mail_ids"]),
|
|
{pm_unowned.id, pm_owned.id, pm_granted.id},
|
|
)
|
|
self.assertFalse(ProcessedMail.objects.filter(id=pm_unowned.id).exists())
|
|
self.assertFalse(ProcessedMail.objects.filter(id=pm_owned.id).exists())
|
|
self.assertFalse(ProcessedMail.objects.filter(id=pm_granted.id).exists())
|
|
self.assertTrue(ProcessedMail.objects.filter(id=pm_forbidden.id).exists())
|
|
|
|
# 403 and not deleted
|
|
response = self.client.post(
|
|
f"{self.ENDPOINT}bulk_delete/",
|
|
data={
|
|
"mail_ids": [pm_forbidden.id],
|
|
},
|
|
format="json",
|
|
)
|
|
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|
|
self.assertTrue(ProcessedMail.objects.filter(id=pm_forbidden.id).exists())
|
|
|
|
# missing mail_ids
|
|
response = self.client.post(
|
|
f"{self.ENDPOINT}bulk_delete/",
|
|
data={"mail_ids": "not-a-list"},
|
|
format="json",
|
|
)
|
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
|
|
|
def test_bulk_delete_processed_mails_rejects_mixed_batch_atomically(self) -> None:
|
|
"""
|
|
GIVEN:
|
|
- A permitted processed mail and one the user may not delete
|
|
WHEN:
|
|
- API call bulk deletes both in a single request
|
|
THEN:
|
|
- The request is rejected and neither mail is deleted
|
|
"""
|
|
user2 = User.objects.create_user(username="temp_admin2")
|
|
rule = MailRuleFactory()
|
|
# Created first so it sorts ahead of the forbidden mail, i.e. the
|
|
# permission check has to cover the whole batch before deleting rather
|
|
# than rejecting only once it reaches the forbidden one.
|
|
pm_owned = ProcessedMailFactory(rule=rule, owner=self.user)
|
|
pm_forbidden = ProcessedMailFactory(rule=rule, owner=user2)
|
|
|
|
response = self.client.post(
|
|
f"{self.ENDPOINT}bulk_delete/",
|
|
data={"mail_ids": [pm_owned.id, pm_forbidden.id]},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|
|
self.assertTrue(ProcessedMail.objects.filter(id=pm_owned.id).exists())
|
|
self.assertTrue(ProcessedMail.objects.filter(id=pm_forbidden.id).exists())
|