diff --git a/chat/__init__.py b/chat/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/chat/apps.py b/chat/apps.py
new file mode 100644
index 00000000..41925b4e
--- /dev/null
+++ b/chat/apps.py
@@ -0,0 +1,28 @@
+from django.apps import AppConfig
+from django.conf import settings
+from django.dispatch import receiver
+from django.utils.translation import gettext_lazy as _
+
+from anymail.backends.base import AnymailBaseBackend
+from anymail.message import AnymailMessage
+from anymail.signals import pre_send
+
+
+class ChatConfig(AppConfig):
+ name = "chat"
+ verbose_name = _("Communicator")
+
+
+@receiver(pre_send)
+def enrich_envelope(sender: type[AnymailBaseBackend], message: AnymailMessage, **kwargs):
+ if getattr(message, 'mass_mail', False):
+ # Mass emails must be sent via the broadcast stream.
+ return
+ if message.subject.startswith('[[CHAT]]'):
+ extra = getattr(message, 'esp_extra', {})
+ extra['MessageStream'] = 'notifications-to-users'
+ message.esp_extra = extra
+ message.tags = ['notification:chat']
+ message.track_opens = True
+ message.metadata = {'env': settings.ENVIRONMENT}
+ message.subject = message.subject.removeprefix('[[CHAT]]').strip()
diff --git a/core/apps.py b/core/apps.py
index 3ac62640..3b340f4a 100644
--- a/core/apps.py
+++ b/core/apps.py
@@ -1,7 +1,13 @@
+import re
+
from django.apps import AppConfig
from django.conf import settings
+from django.dispatch import receiver
from django.utils.translation import gettext_lazy as _
+from anymail.backends.base import AnymailBaseBackend
+from anymail.message import AnymailMessage
+from anymail.signals import pre_send
from gql import Client as GQLClient, gql
from gql.transport.exceptions import TransportError, TransportQueryError
from gql.transport.requests import RequestsHTTPTransport as GQLHttpTransport
@@ -39,3 +45,30 @@ def ready(self):
FEEDBACK_TYPES[feedback_key] = (
feedback._replace(url=discussion['node']['url'])
)
+
+
+@receiver(pre_send)
+def enrich_envelope(sender: type[AnymailBaseBackend], message: AnymailMessage, **kwargs):
+ """
+ Add extra Anymail / Postmark information to an email message, for correct
+ processing. This is done via a `pre_send` signal because Django sends
+ emails in multiple manners, and customizing each one of them individually
+ is quite cumbersome.
+ `message` can be an EmailMessage, an EmailMultiAlternatives, or a derivative.
+ """
+ if getattr(message, 'mass_mail', False):
+ # Mass emails must be sent via the broadcast stream.
+ return
+ tags = {
+ '[[ACCOUNT]]': 'account',
+ '[[ACCOUNT-EMAIL]]': 'email',
+ '[[PLACE-DETAILS]]': 'authorized',
+ }
+ possible_prefixes_pattern = '|'.join(re.escape(prefix) for prefix in tags.keys())
+ if match := re.match(possible_prefixes_pattern, message.subject):
+ extra = getattr(message, 'esp_extra', {})
+ extra['MessageStream'] = 'notifications-to-users'
+ message.esp_extra = extra
+ message.tags = [f'notification:{tags[match.group()]}']
+ message.metadata = {'env': settings.ENVIRONMENT}
+ message.subject = message.subject.removeprefix(match.group()).strip()
diff --git a/core/forms.py b/core/forms.py
index 7eb95cdb..5ce3473d 100644
--- a/core/forms.py
+++ b/core/forms.py
@@ -184,6 +184,7 @@ def save(self, commit=True):
context = {
'site_name': config.site_name,
'ENV': settings.ENVIRONMENT,
+ 'RICH_ENVELOPE': getattr(settings, 'EMAIL_RICH_ENVELOPES', None),
'subject_prefix': settings.EMAIL_SUBJECT_PREFIX_FULL,
'url': url,
'url_first': url[:url.rindex('/')+1],
@@ -267,7 +268,11 @@ def send_mail(self,
args = [subject_template_name, email_template_name, context, *args]
kwargs.update(html_email_template_name=html_email_template_name)
- context.update({'ENV': settings.ENVIRONMENT, 'subject_prefix': settings.EMAIL_SUBJECT_PREFIX_FULL})
+ context.update({
+ 'ENV': settings.ENVIRONMENT,
+ 'RICH_ENVELOPE': getattr(settings, 'EMAIL_RICH_ENVELOPES', None),
+ 'subject_prefix': settings.EMAIL_SUBJECT_PREFIX_FULL,
+ })
super().send_mail(*args, **kwargs)
def save(self, **kwargs):
diff --git a/core/templates/email/new_email_subject.txt b/core/templates/email/new_email_subject.txt
index 2da3b2f3..dd5ad5a6 100644
--- a/core/templates/email/new_email_subject.txt
+++ b/core/templates/email/new_email_subject.txt
@@ -1,5 +1,7 @@
{% load i18n %}{% autoescape off %}
+{% if RICH_ENVELOPE %}[[ACCOUNT-EMAIL]]{% endif %}
+
{{ subject_prefix }}{% trans "Change of email address" context "Email subject" %}
{% endautoescape %}
diff --git a/core/templates/email/old_email_subject.txt b/core/templates/email/old_email_subject.txt
index 2da3b2f3..dd5ad5a6 100644
--- a/core/templates/email/old_email_subject.txt
+++ b/core/templates/email/old_email_subject.txt
@@ -1,5 +1,7 @@
{% load i18n %}{% autoescape off %}
+{% if RICH_ENVELOPE %}[[ACCOUNT-EMAIL]]{% endif %}
+
{{ subject_prefix }}{% trans "Change of email address" context "Email subject" %}
{% endautoescape %}
diff --git a/core/templates/email/password_reset_subject.txt b/core/templates/email/password_reset_subject.txt
index 99f6f5d6..dda3f0f3 100644
--- a/core/templates/email/password_reset_subject.txt
+++ b/core/templates/email/password_reset_subject.txt
@@ -1,5 +1,7 @@
{% load i18n %}{% autoescape off %}
+{% if RICH_ENVELOPE %}[[ACCOUNT]]{% endif %}
+
{{ subject_prefix }}{% trans "Password reset" context "Email subject" %}
{% endautoescape %}
diff --git a/core/templates/email/snippets/preview_header.html b/core/templates/email/snippets/preview_header.html
index 602e6c76..425a9c3c 100644
--- a/core/templates/email/snippets/preview_header.html
+++ b/core/templates/email/snippets/preview_header.html
@@ -1,9 +1,9 @@
- Pasporta Servo
+ Pasporta Servo
- -
+ –
-
\ No newline at end of file
+
diff --git a/core/templates/email/system-email_verify_subject.txt b/core/templates/email/system-email_verify_subject.txt
index b4f9e84b..54f100a3 100644
--- a/core/templates/email/system-email_verify_subject.txt
+++ b/core/templates/email/system-email_verify_subject.txt
@@ -1,5 +1,7 @@
{% load i18n %}{% autoescape off %}
+{% if RICH_ENVELOPE %}[[ACCOUNT-EMAIL]]{% endif %}
+
{{ subject_prefix }}{% trans "Is this your email address?" context "Email subject" %}
{% endautoescape %}
diff --git a/core/templates/email/username_remind_subject.txt b/core/templates/email/username_remind_subject.txt
index d2c7682e..1d52ecfc 100644
--- a/core/templates/email/username_remind_subject.txt
+++ b/core/templates/email/username_remind_subject.txt
@@ -1,5 +1,7 @@
{% load i18n %}{% autoescape off %}
+{% if RICH_ENVELOPE %}[[ACCOUNT]]{% endif %}
+
{{ subject_prefix }}{% trans "Username reminder" context "Email subject" %}
{% endautoescape %}
diff --git a/core/utils.py b/core/utils.py
index f747b7af..d8821f3a 100644
--- a/core/utils.py
+++ b/core/utils.py
@@ -3,9 +3,11 @@
import operator
import re
from functools import reduce
+from typing import Optional, Sequence, Tuple
from django.conf import settings
-from django.core.mail import EmailMultiAlternatives, get_connection
+from django.core.mail import EmailMessage, get_connection
+from django.core.mail.backends.base import BaseEmailBackend
from django.utils.functional import (
SimpleLazyObject, keep_lazy_text, lazy, new_method_proxy,
)
@@ -13,6 +15,7 @@
from django.utils.safestring import mark_safe
import requests
+from anymail.message import AnymailMessage
def getattr_(obj, path):
@@ -49,8 +52,12 @@ def _lazy_joiner(sep, items, item_to_string=str):
setattr(SimpleLazyObject, '__mul__', new_method_proxy(operator.mul))
-def send_mass_html_mail(datatuple, fail_silently=False, user=None, password=None,
- connection=None):
+def send_mass_html_mail(
+ datatuple: Sequence[Tuple[str, str, str, Optional[str], Sequence[str] | None]],
+ fail_silently: bool = False,
+ auth_user: Optional[str] = None, auth_password: Optional[str] = None,
+ connection: Optional[BaseEmailBackend] = None,
+) -> int:
"""
Given a datatuple of (subject, text_content, html_content, from_email,
recipient_list), sends each message to each recipient list. Returns the
@@ -62,16 +69,23 @@ def send_mass_html_mail(datatuple, fail_silently=False, user=None, password=None
If auth_password is None, the EMAIL_HOST_PASSWORD setting is used.
"""
connection = connection or get_connection(
- username=user, password=password, fail_silently=fail_silently)
- messages = []
+ username=auth_user, password=auth_password, fail_silently=fail_silently)
+ messages: Sequence[EmailMessage] = []
default_from = settings.DEFAULT_FROM_EMAIL
for subject, text, html, from_email, recipients in datatuple:
subject = ''.join(subject.splitlines())
- recipients = [r.strip() for r in recipients]
- message = EmailMultiAlternatives(
- subject, text, default_from, recipients,
+ recipients = [r.strip() for r in recipients] if recipients else []
+ message = AnymailMessage(
+ subject, text, from_email or default_from, recipients,
headers={'Reply-To': 'Pasporta Servo '})
message.attach_alternative(html, 'text/html')
+ # TODO: Implement custom one-click unsubscribe.
+ message.esp_extra = {'MessageStream': 'broadcast'}
+ if tag_match := re.match(r'\[\[([a-zA-Z0-9_-]+)\]\]', subject):
+ message.tags = [tag_match.group(1)]
+ message.subject = subject.removeprefix(tag_match.group()).strip()
+ message.merge_data = {} # Enable batch sending mode.
+ setattr(message, 'mass_mail', True)
messages.append(message)
return connection.send_messages(messages) or 0
diff --git a/core/views.py b/core/views.py
index 411c5824..8144e15e 100644
--- a/core/views.py
+++ b/core/views.py
@@ -349,7 +349,7 @@ class PasswordResetView(PasswordResetBuiltinView):
"""
This extension of Django's built-in view allows to send a different
email depending on whether the user is active (True) or no (False).
- See also the companion SystemPasswordResetRequestForm.
+ See also the companion `SystemPasswordResetRequestForm`.
"""
html_email_template_name = {True: 'email/password_reset.html', False: 'email/password_reset_activate.html'}
email_template_name = {True: 'email/password_reset.txt', False: 'email/password_reset_activate.txt'}
@@ -461,6 +461,7 @@ def post(self, request, *args, **kwargs):
context = {
'site_name': config.site_name,
'ENV': settings.ENVIRONMENT,
+ 'RICH_ENVELOPE': getattr(settings, 'EMAIL_RICH_ENVELOPES', None),
'subject_prefix': settings.EMAIL_SUBJECT_PREFIX_FULL,
'url': url,
'url_first': url[:url.rindex('/')+1],
@@ -781,11 +782,19 @@ class MassMailView(AuthMixin, generic.FormView):
form_class = MassMailForm
display_permission_denied = False
exact_role = AuthRole.ADMIN
+ # Keep the email address separate from the one used for transactional
+ # emails, for better email sender reputation.
+ mailing_address = 'anoncoj@pasportaservo.org'
def dispatch(self, request, *args, **kwargs):
kwargs['auth_base'] = None
return super().dispatch(request, *args, **kwargs)
+ def get_context_data(self, **kwargs):
+ context = super().get_context_data(**kwargs)
+ context['mailing_address'] = self.mailing_address
+ return context
+
def get_success_url(self):
return format_lazy(
'{success_url}?nb={sent}',
@@ -800,7 +809,7 @@ def form_valid(self, form):
preheader = form.cleaned_data['preheader']
heading = form.cleaned_data['heading']
category = form.cleaned_data['categories']
- default_from = settings.DEFAULT_FROM_EMAIL
+ default_from = f'Pasporta Servo <{self.mailing_address}>'
template = get_template('email/mass_email.html')
opening = make_aware(datetime(2014, 11, 24))
diff --git a/hosting/templates/email/new_authorization_subject.txt b/hosting/templates/email/new_authorization_subject.txt
index b340c78c..444d9922 100644
--- a/hosting/templates/email/new_authorization_subject.txt
+++ b/hosting/templates/email/new_authorization_subject.txt
@@ -1,5 +1,7 @@
{% load i18n %}{% autoescape off %}
+{% if RICH_ENVELOPE %}[[PLACE-DETAILS]]{% endif %}
+
{{ subject_prefix }}{% trans "You received an Authorization" context "Email subject" %}
{% endautoescape %}
diff --git a/hosting/views/places.py b/hosting/views/places.py
index 88b9ab97..1bdc4962 100644
--- a/hosting/views/places.py
+++ b/hosting/views/places.py
@@ -461,6 +461,7 @@ def send_email(self, user, place):
email_context = {
'site_name': config.site_name,
'ENV': settings.ENVIRONMENT,
+ 'RICH_ENVELOPE': getattr(settings, 'EMAIL_RICH_ENVELOPES', None),
'subject_prefix': settings.EMAIL_SUBJECT_PREFIX_FULL,
'user': user,
'place': place,
diff --git a/locale/eo/LC_MESSAGES/django.po b/locale/eo/LC_MESSAGES/django.po
index faaba72c..dc4406b2 100644
--- a/locale/eo/LC_MESSAGES/django.po
+++ b/locale/eo/LC_MESSAGES/django.po
@@ -113,6 +113,10 @@ msgstr "Aboni"
msgid "yes,no"
msgstr "jes,ne"
+#: chat/apps.py
+msgid "Communicator"
+msgstr "Komunikilo"
+
#: core/admin/admin.py core/models.py core/templates/core/base.html
#: core/views.py hosting/admin/admin.py
#: hosting/templates/hosting/phone_form.html
diff --git a/pasportaservo/settings/base.py b/pasportaservo/settings/base.py
index 8ea25d00..80cdbecc 100644
--- a/pasportaservo/settings/base.py
+++ b/pasportaservo/settings/base.py
@@ -67,6 +67,7 @@ def get_env_setting(setting):
'django.contrib.gis',
'fontawesomefree',
+ 'anymail',
'compressor',
'crispy_forms',
'django_extensions',
@@ -84,6 +85,7 @@ def get_env_setting(setting):
'blog',
'book',
+ 'chat',
'core',
'hosting',
'links',
@@ -199,6 +201,17 @@ def get_env_setting(setting):
'BACKEND': 'django.core.cache.backends.dummy.DummyCache',
}
}
+TEST_EMAIL_BACKENDS = {
+ 'dummy': {
+ 'EMAIL_BACKEND': 'anymail.backends.test.EmailBackend',
+ 'EMAIL_RICH_ENVELOPES': True,
+ },
+ 'live': {
+ 'EMAIL_BACKEND': 'anymail.backends.postmark.EmailBackend',
+ 'POSTMARK_SERVER_TOKEN': 'POSTMARK_API_TEST',
+ 'EMAIL_RICH_ENVELOPES': True,
+ }
+}
# Internationalization
# https://docs.djangoproject.com/en/stable/topics/i18n/
diff --git a/pasportaservo/settings/dev.py b/pasportaservo/settings/dev.py
index ed15ecde..9e19990a 100644
--- a/pasportaservo/settings/dev.py
+++ b/pasportaservo/settings/dev.py
@@ -67,6 +67,7 @@
EMAIL_HOST = '127.0.0.1'
EMAIL_PORT = '1025'
INTERNAL_IPS = ('127.0.0.1',)
+ANYMAIL_DEBUG_API_REQUESTS = True
EMAIL_SUBJECT_PREFIX = '[PS test] '
EMAIL_SUBJECT_PREFIX_FULL = '[Pasporta Servo][{}] '.format(ENVIRONMENT)
diff --git a/pasportaservo/settings/prod.py b/pasportaservo/settings/prod.py
index 469da81f..c1bd0b5e 100644
--- a/pasportaservo/settings/prod.py
+++ b/pasportaservo/settings/prod.py
@@ -20,8 +20,9 @@
}
}
-EMAIL_BACKEND = "sgbackend.SendGridBackend"
-SENDGRID_API_KEY = get_env_setting('SENDGRID_API_KEY')
+EMAIL_BACKEND = 'anymail.backends.postmark.EmailBackend'
+POSTMARK_SERVER_TOKEN = get_env_setting('POSTMARK_SERVER_TOKEN')
+EMAIL_RICH_ENVELOPES = True
EMAIL_SUBJECT_PREFIX = '[PS] '
EMAIL_SUBJECT_PREFIX_FULL = '[Pasporta Servo] '
diff --git a/pasportaservo/settings/staging.py b/pasportaservo/settings/staging.py
index e6894367..137763c1 100644
--- a/pasportaservo/settings/staging.py
+++ b/pasportaservo/settings/staging.py
@@ -22,8 +22,9 @@
}
}
-EMAIL_BACKEND = "sgbackend.SendGridBackend"
-SENDGRID_API_KEY = get_env_setting('SENDGRID_API_KEY')
+EMAIL_BACKEND = 'anymail.backends.postmark.EmailBackend'
+POSTMARK_SERVER_TOKEN = get_env_setting('POSTMARK_SERVER_TOKEN')
+EMAIL_RICH_ENVELOPES = True
EMAIL_SUBJECT_PREFIX = '[PS ido] '
EMAIL_SUBJECT_PREFIX_FULL = '[Pasporta Servo][{}] '.format(ENVIRONMENT)
diff --git a/pasportaservo/templates/postman/email_user_subject.txt b/pasportaservo/templates/postman/email_user_subject.txt
index b7f131a7..aecef8ec 100644
--- a/pasportaservo/templates/postman/email_user_subject.txt
+++ b/pasportaservo/templates/postman/email_user_subject.txt
@@ -1,4 +1,5 @@
{% load i18n utils %}
+[[CHAT]]
{% filter compact %}
{% autoescape off %}
{% blocktrans with object.obfuscated_sender as sender and object.subject as subject trimmed %}
diff --git a/requirements/base.txt b/requirements/base.txt
index bc688703..28331fe1 100644
--- a/requirements/base.txt
+++ b/requirements/base.txt
@@ -5,6 +5,7 @@ Pillow==10.3.0
awesome-slugify==1.6.5
commonmark==0.9.1
csscompressor==0.9.5
+django-anymail[postmark]==10.3
django-braces==1.14.0
djangocodemirror==2.1.0
django-compressor==2.4
@@ -34,7 +35,6 @@ pymemcache==3.5.2
phonenumberslite==8.12.19
requests==2.27.1
rstr==3.0.0
-sendgrid-django==4.2.0
sentry-sdk>=1.18
user_agents==2.2.0
diff --git a/tests/forms/test_auth_forms.py b/tests/forms/test_auth_forms.py
index dfa1352d..152444ab 100644
--- a/tests/forms/test_auth_forms.py
+++ b/tests/forms/test_auth_forms.py
@@ -1,4 +1,5 @@
import re
+from typing import Optional, cast
from unittest.mock import patch
from django.conf import settings
@@ -13,10 +14,10 @@
from django.utils.encoding import force_bytes
from django.utils.http import urlsafe_base64_encode
+from anymail.message import AnymailMessage
from django_webtest import WebTest
from factory import Faker
-from core.auth import auth_log
from core.forms import (
EmailStaffUpdateForm, EmailUpdateForm, SystemPasswordChangeForm,
SystemPasswordResetForm, SystemPasswordResetRequestForm,
@@ -26,12 +27,13 @@
from core.views import (
PasswordResetConfirmView, PasswordResetView, UsernameRemindView,
)
+from hosting.models import PasportaServoUser
from ..assertions import AdditionalAsserts
from ..factories import UserFactory
-def _snake_str(string):
+def _snake_str(string: str) -> str:
return ''.join([c if i % 2 else c.upper() for i, c in enumerate(string)])
@@ -265,7 +267,7 @@ def test_honeypot(self, mock_pwd_check):
self.assertFalse(form.is_valid())
self.assertIn(self.honeypot_field, form.errors)
self.assertEqual(form.errors[self.honeypot_field], [""])
- self.assertEqual(len(log.records), 1)
+ self.assertLength(log.records, 1)
self.assertEqual(
log.records[0].message,
"Registration failed, flies found in honeypot."
@@ -316,7 +318,7 @@ def test_form_submit(self, mock_pwd_check):
class UserAuthenticationFormTests(AdditionalAsserts, WebTest):
@classmethod
def setUpTestData(cls):
- cls.user = UserFactory()
+ cls.user = UserFactory.create()
def setUp(self):
self.dummy_request = HttpRequest()
@@ -384,7 +386,7 @@ def test_inactive_user_login(self):
self.assertIn('restore_request_id', self.dummy_request.session)
self.assertIs(type(self.dummy_request.session['restore_request_id']), tuple)
self.assertEqual(len(self.dummy_request.session['restore_request_id']), 2)
- self.assertEqual(len(log.records), 1)
+ self.assertLength(log.records, 1)
self.assertIn("the account is deactivated", log.output[0])
def test_active_user_login(self):
@@ -465,7 +467,7 @@ def test_form_submit_valid_credentials(self):
class UsernameUpdateFormTests(AdditionalAsserts, WebTest):
@classmethod
def setUpTestData(cls):
- cls.user = UserFactory()
+ cls.user = UserFactory.create()
def test_init(self):
form = UsernameUpdateForm(instance=self.user)
@@ -583,7 +585,7 @@ def test_case_modified_nonunique_username(self):
)
def test_nonunique_username(self):
- other_user = UserFactory()
+ other_user = UserFactory.create()
for new_username in (other_user.username,
other_user.username.capitalize(),
_snake_str(other_user.username)):
@@ -635,8 +637,8 @@ class EmailUpdateFormTests(AdditionalAsserts, WebTest):
@classmethod
def setUpTestData(cls):
- cls.user = UserFactory()
- cls.invalid_email_user = UserFactory(invalid_email=True)
+ cls.user = UserFactory.create()
+ cls.invalid_email_user = UserFactory.create(invalid_email=True)
def _init_form(self, data=None, instance=None):
return EmailUpdateForm(data=data, instance=instance)
@@ -772,7 +774,7 @@ def test_same_email(self):
self.assertTrue(form.is_valid())
form.save(commit=False)
# Since no change is done in the address, no email is expected to be sent.
- self.assertEqual(len(mail.outbox), 0)
+ self.assertLength(mail.outbox, 0)
form = self._init_form(
data={'email': self.invalid_email_user._clean_email},
@@ -780,7 +782,7 @@ def test_same_email(self):
self.assertTrue(form.is_valid())
form.save(commit=False)
# Since no change is done in the address, no email is expected to be sent.
- self.assertEqual(len(mail.outbox), 0)
+ self.assertLength(mail.outbox, 0)
def test_case_modified_email(self):
test_transforms = [
@@ -802,7 +804,7 @@ def test_case_modified_email(self):
self.assertTrue(form.is_valid(), msg=repr(form.errors))
def test_nonunique_email(self):
- normal_email_user = UserFactory()
+ normal_email_user = UserFactory.create()
test_transforms = [
lambda e: e,
lambda e: _snake_str(e),
@@ -882,13 +884,13 @@ def test_view_page(self):
self.assertIsInstance(page.context['form'], EmailUpdateForm)
@override_settings(EMAIL_SUBJECT_PREFIX_FULL="TEST ")
- def form_submission_tests(self, *, lang, obj=None):
+ def form_submission_tests(self, *, lang: str, obj: Optional[PasportaServoUser] = None):
obj = self.user if obj is None else obj
old_email = obj._clean_email
new_email = '{}@ps.org'.format(_snake_str(obj.username))
unchanged_email = obj.email
- with override_settings(LANGUAGE_CODE=lang):
+ def submit_form_and_assert():
page = self.app.get(reverse('email_update'), user=obj)
page.form['email'] = new_email
page = page.form.submit()
@@ -898,8 +900,11 @@ def form_submission_tests(self, *, lang, obj=None):
reverse('profile_edit', kwargs={
'pk': obj.profile.pk, 'slug': obj.profile.autoslug})
)
- self.assertEqual(obj.email, unchanged_email)
- self.assertEqual(len(mail.outbox), 2)
+ self.assertEqual(obj.email, unchanged_email)
+
+ with override_settings(LANGUAGE_CODE=lang):
+ submit_form_and_assert()
+ self.assertLength(mail.outbox, 2)
test_subject = {
'en': "TEST Change of email address",
'eo': "TEST Retpoŝtadreso ĉe retejo ŝanĝita",
@@ -925,6 +930,19 @@ def form_submission_tests(self, *, lang, obj=None):
for content in test_contents[recipient][lang]:
self.assertIn(content, mail.outbox[i].body)
+ with override_settings(
+ **settings.TEST_EMAIL_BACKENDS['dummy'],
+ LANGUAGE_CODE=lang,
+ ):
+ submit_form_and_assert()
+ self.assertLength(mail.outbox, 4)
+ for i, recipient in enumerate([old_email, new_email], start=2):
+ self.assertEqual(mail.outbox[i].subject, test_subject[lang])
+ self.assertEqual(mail.outbox[i].to, [recipient])
+ self.assertEqual(cast(AnymailMessage, mail.outbox[i]).tags, ['notification:email'])
+ self.assertFalse(mail.outbox[i].anymail_test_params.get('is_batch_send'))
+ self.assertFalse(mail.outbox[i].anymail_test_params.get('track_opens'))
+
def test_form_submit(self):
mail.outbox = []
self.form_submission_tests(lang='en')
@@ -972,7 +990,7 @@ def form_submission_tests(self, *, lang, obj=None):
'pk': obj.profile.pk, 'slug': obj.profile.autoslug})
)
self.assertEqual(obj.email, new_email)
- self.assertEqual(len(mail.outbox), 0)
+ self.assertLength(mail.outbox, 0)
@tag('forms', 'forms-auth', 'auth')
@@ -1123,7 +1141,7 @@ def test_active_user_request(self):
with override_settings(LANGUAGE_CODE=lang):
with self.subTest(tag=user_tag, lang=lang):
# No warnings are expected on the auth log.
- with self.assertLogs('PasportaServo.auth', level='WARNING') as log:
+ with self.assertNoLogs('PasportaServo.auth', level='WARNING'):
form = self._init_form({'email': user._clean_email})
self.assertTrue(form.is_valid())
form.save(
@@ -1131,13 +1149,9 @@ def test_active_user_request(self):
email_template_name=self._related_view.email_template_name,
html_email_template_name=self._related_view.html_email_template_name,
)
- # Workaround for lack of assertNotLogs.
- auth_log.warning("No warning emitted.")
- self.assertEqual(len(log.records), 1)
- self.assertEqual(log.records[0].message, "No warning emitted.")
# The email message is expected to describe the password reset procedure.
title, expected_content, not_expected_content = self._get_email_content(True, lang)
- self.assertEqual(len(mail.outbox), 1)
+ self.assertLength(mail.outbox, 1)
self.assertEqual(mail.outbox[0].subject, title)
self.assertEqual(mail.outbox[0].from_email, settings.DEFAULT_FROM_EMAIL)
self.assertEqual(mail.outbox[0].to, [user._clean_email])
@@ -1145,6 +1159,23 @@ def test_active_user_request(self):
self.assertIn(content, mail.outbox[0].body)
for content in not_expected_content:
self.assertNotIn(content, mail.outbox[0].body)
+
+ # Verify that when dispatched via an email backend, the email message's
+ # subject and ESP parameters are the expected ones.
+ with override_settings(**settings.TEST_EMAIL_BACKENDS['dummy']):
+ form.save(
+ subject_template_name=self._related_view.subject_template_name,
+ email_template_name=self._related_view.email_template_name,
+ html_email_template_name=self._related_view.html_email_template_name,
+ )
+ self.assertLength(mail.outbox, 2)
+ self.assertEqual(mail.outbox[1].subject, title)
+ self.assertEqual(
+ cast(AnymailMessage, mail.outbox[1]).tags,
+ ['notification:account'])
+ self.assertFalse(mail.outbox[1].anymail_test_params.get('is_batch_send'))
+ self.assertFalse(mail.outbox[1].anymail_test_params.get('track_opens'))
+
mail.outbox = []
@override_settings(EMAIL_SUBJECT_PREFIX_FULL="TEST ")
@@ -1157,6 +1188,9 @@ def test_inactive_user_request(self):
with override_settings(LANGUAGE_CODE=lang):
with self.subTest(tag=user_tag, lang=lang):
# A warning about a deactivated account is expected on the auth log.
+ # Note: AssertLogs Context Manager disables all existing handlers of
+ # the logger, resulting in no emails being dispatched to the
+ # admins, if configured.
with self.assertLogs('PasportaServo.auth', level='WARNING') as log:
form = self._init_form({'email': user._clean_email})
self.assertTrue(form.is_valid())
@@ -1165,15 +1199,16 @@ def test_inactive_user_request(self):
email_template_name=self._related_view.email_template_name,
html_email_template_name=self._related_view.html_email_template_name,
)
- self.assertEqual(len(log.records), 1)
+ self.assertLength(log.records, 1)
self.assertStartsWith(log.records[0].message, self._get_admin_message(user))
# The warning is expected to include a reference number.
code = re.search(r'\[([A-F0-9-]+)\]', log.records[0].message)
self.assertIsNotNone(code)
code = code.group(1)
+
# The email message is expected to describe the account reactivation procedure.
title, expected_content, not_expected_content = self._get_email_content(False, lang)
- self.assertEqual(len(mail.outbox), 1)
+ self.assertLength(mail.outbox, 1)
self.assertEqual(mail.outbox[0].subject, title)
self.assertEqual(mail.outbox[0].from_email, settings.DEFAULT_FROM_EMAIL)
self.assertEqual(mail.outbox[0].to, [user._clean_email])
@@ -1183,6 +1218,26 @@ def test_inactive_user_request(self):
self.assertNotIn(content, mail.outbox[0].body)
# The email message is expected to include the reference number.
self.assertIn(code, mail.outbox[0].body)
+
+ # Verify that when dispatched via an email backend, the email message's
+ # subject and ESP parameters are the expected ones.
+ # Note: AssertLogs Context Manager disables all existing handlers of
+ # the logger, resulting in no emails being dispatched to admins.
+ with override_settings(**settings.TEST_EMAIL_BACKENDS['dummy']):
+ with self.assertLogs('PasportaServo.auth', level='WARNING') as log:
+ form.save(
+ subject_template_name=self._related_view.subject_template_name,
+ email_template_name=self._related_view.email_template_name,
+ html_email_template_name=self._related_view.html_email_template_name,
+ )
+ self.assertLength(mail.outbox, 2)
+ self.assertEqual(mail.outbox[1].subject, title)
+ self.assertEqual(
+ cast(AnymailMessage, mail.outbox[1]).tags,
+ ['notification:account'])
+ self.assertFalse(mail.outbox[1].anymail_test_params.get('is_batch_send'))
+ self.assertFalse(mail.outbox[1].anymail_test_params.get('track_opens'))
+
mail.outbox = []
def test_view_page(self):
@@ -1282,7 +1337,7 @@ def setUpClass(cls):
@classmethod
def setUpTestData(cls):
- cls.user = UserFactory(invalid_email=True)
+ cls.user = UserFactory.create(invalid_email=True)
cls.user.profile.email = cls.user.email
cls.user.profile.save(update_fields=['email'])
diff --git a/tests/forms/test_chat_forms.py b/tests/forms/test_chat_forms.py
index bb546f5f..f5f293f7 100644
--- a/tests/forms/test_chat_forms.py
+++ b/tests/forms/test_chat_forms.py
@@ -1,10 +1,13 @@
+from typing import cast
from unittest import expectedFailure
+from django.conf import settings
from django.contrib.auth.models import AnonymousUser
from django.core import mail
from django.test import override_settings, tag
from django.urls import reverse
+from anymail.message import AnymailMessage
from django_webtest import WebTest
from factory import Faker
from postman.models import Message
@@ -88,7 +91,7 @@ def test_clean_recipients(self):
def test_view_page(self):
page = self.app.get(reverse('postman:write'), user=self.sender)
- self.assertEqual(page.status_int, 200)
+ self.assertEqual(page.status_code, 200)
self.assertEqual(len(page.forms), 1)
self.assertIsInstance(page.context['form'], CustomWriteForm)
@@ -102,7 +105,7 @@ def do_test_form_submit(self, recipient, deceased, lang, invalid_email=False):
page_result = page.form.submit()
if deceased:
- self.assertEqual(page_result.status_int, 200)
+ self.assertEqual(page_result.status_code, 200)
expected_form_errors = {
'en': "Cannot send the message: This user has passed away.",
'eo': "Ne eblas sendi la mesaĝon: Tiu ĉi uzanto forpasis.",
@@ -111,16 +114,25 @@ def do_test_form_submit(self, recipient, deceased, lang, invalid_email=False):
page_result,
'form', 'recipients',
expected_form_errors[lang])
- self.assertEqual(len(mail.outbox), 0)
+ self.assertLength(mail.outbox, 0)
else:
- self.assertEqual(page_result.status_int, 302)
+ self.assertEqual(page_result.status_code, 302)
self.assertRedirects(page_result, '/origin', fetch_redirect_response=False)
if invalid_email:
- self.assertEqual(len(mail.outbox), 0)
+ self.assertLength(mail.outbox, 0)
else:
- self.assertEqual(len(mail.outbox), 1)
+ self.assertLength(mail.outbox, 1)
self.assertEqual(mail.outbox[0].to, [self.sender.email])
self.assertEndsWith(mail.outbox[0].subject, page.form['subject'].value)
+ with override_settings(**settings.TEST_EMAIL_BACKENDS['dummy']):
+ page_result = page.form.submit()
+ self.assertEqual(page_result.status_code, 302)
+ self.assertLength(mail.outbox, 2)
+ self.assertEqual(
+ cast(AnymailMessage, mail.outbox[1]).tags,
+ ['notification:chat'])
+ self.assertFalse(mail.outbox[1].anymail_test_params.get('is_batch_send'))
+ self.assertTrue(mail.outbox[1].anymail_test_params.get('track_opens'))
def test_form_submit_living(self):
self.do_test_form_submit(recipient=self.sender, deceased=False, lang='en')
@@ -301,7 +313,7 @@ def test_view_page(self):
page = self.app.get(
reverse(view_name, kwargs={'message_id': self.message.pk}),
user=self.sender)
- self.assertEqual(page.status_int, 200)
+ self.assertEqual(page.status_code, 200)
self.assertEqual(len(page.forms), 1)
self.assertIsInstance(page.context['form'], form_class)
@@ -323,7 +335,7 @@ def do_test_form_submit(self, orig_message, deceased, lang, invalid_email=False)
page_result = page.form.submit()
if deceased:
- self.assertEqual(page_result.status_int, 200)
+ self.assertEqual(page_result.status_code, 200)
expected_form_errors = {
'en': "Cannot send the message: This user has passed away.",
'eo': "Ne eblas sendi la mesaĝon: Tiu ĉi uzanto forpasis.",
@@ -332,16 +344,25 @@ def do_test_form_submit(self, orig_message, deceased, lang, invalid_email=False)
page_result,
'form', None,
expected_form_errors[lang])
- self.assertEqual(len(mail.outbox), 0)
+ self.assertLength(mail.outbox, 0)
else:
- self.assertEqual(page_result.status_int, 302)
+ self.assertEqual(page_result.status_code, 302)
self.assertRedirects(page_result, '/origin', fetch_redirect_response=False)
if invalid_email:
- self.assertEqual(len(mail.outbox), 0)
+ self.assertLength(mail.outbox, 0)
else:
- self.assertEqual(len(mail.outbox), 1)
+ self.assertLength(mail.outbox, 1)
self.assertEqual(mail.outbox[0].to, [orig_message.sender.email])
self.assertEndsWith(mail.outbox[0].subject, page.form['subject'].value)
+ with override_settings(**settings.TEST_EMAIL_BACKENDS['dummy']):
+ page_result = page.form.submit()
+ self.assertEqual(page_result.status_code, 302)
+ self.assertLength(mail.outbox, 2)
+ self.assertEqual(
+ cast(AnymailMessage, mail.outbox[1]).tags,
+ ['notification:chat'])
+ self.assertFalse(mail.outbox[1].anymail_test_params.get('is_batch_send'))
+ self.assertTrue(mail.outbox[1].anymail_test_params.get('track_opens'))
def test_form_submit_living(self):
self.do_test_form_submit(orig_message=self.message_other, deceased=False, lang='en')
diff --git a/tests/integration/test_integration.py b/tests/integration/test_integration.py
index 8bfb0b55..9dd79bff 100644
--- a/tests/integration/test_integration.py
+++ b/tests/integration/test_integration.py
@@ -1,14 +1,17 @@
from random import randint
+from unittest import skipUnless
from django.conf import settings
from django.contrib.auth.models import AnonymousUser, Group
from django.core import serializers
from django.core.exceptions import ImproperlyConfigured
from django.http import JsonResponse
-from django.test import RequestFactory, TestCase, tag
+from django.test import RequestFactory, TestCase, override_settings, tag
from django.urls import reverse
from django.views.generic import CreateView, View
+from anymail.exceptions import AnymailAPIError, AnymailRecipientsRefused
+from anymail.message import AnymailMessage
from django_webtest import WebTest
from core.auth import AuthMixin, AuthRole
@@ -21,6 +24,51 @@
)
+@tag('integration', 'mailing')
+class MailingTests(AdditionalAsserts, TestCase):
+ """
+ Tests for the mailing functionalities and integration with an external
+ provider.
+ """
+
+ @tag('external')
+ @override_settings(**settings.TEST_EMAIL_BACKENDS['live'])
+ @skipUnless(settings.TEST_EXTERNAL_SERVICES, 'External services are tested only explicitly')
+ def test_mail_backend_integration_contract(self):
+ message = AnymailMessage(
+ "Single message test",
+ to=["abcd@example.org"],
+ body="Fusce felis lectus, dapibus ut velit non, pharetra molestie tellus.")
+ with self.assertNotRaises(AnymailAPIError):
+ message.send()
+ status = message.anymail_status
+ self.assertEqual(status.status, {'sent'})
+ self.assertIsNotNone(status.message_id)
+ self.assertEqual(list(status.recipients.keys()), ["abcd@example.org"])
+
+ message = AnymailMessage(
+ "Broadcast message test",
+ to=["efgh@example.org"],
+ body="Mauris purus sapien, aliquam id viverra ut, bibendum sed metus.")
+ message.esp_extra = {'MessageStream': 'broadcast'}
+ with self.assertNotRaises(AnymailAPIError):
+ message.send()
+ status = message.anymail_status
+ self.assertEqual(status.status, {'sent'})
+ self.assertIsNotNone(status.message_id)
+ self.assertEqual(list(status.recipients.keys()), ["efgh@example.org"])
+
+ message = AnymailMessage(
+ "Invalid recipient test",
+ to=["pqrs@localhost"],
+ body="Curabitur elit massa, elementum id consectetur at, semper a odio.")
+ with self.assertRaises(AnymailRecipientsRefused):
+ message.send()
+ status = message.anymail_status
+ self.assertEqual(status.status, {'invalid'})
+ self.assertIsNone(status.message_id)
+
+
@tag('integration')
class ModelSignalTests(AdditionalAsserts, TestCase):
"""
diff --git a/tests/test_utils.py b/tests/test_utils.py
index 5ba718e3..502a6af4 100644
--- a/tests/test_utils.py
+++ b/tests/test_utils.py
@@ -2,7 +2,7 @@
import logging
import operator
import random
-from typing import NamedTuple
+from typing import NamedTuple, cast
from unittest import skipUnless
from unittest.mock import patch
@@ -12,6 +12,8 @@
from django.test import TestCase, override_settings, tag
from django.utils.functional import SimpleLazyObject, lazy, lazystr
+from anymail.message import AnymailMessage
+from anymail.utils import UNSET
from factory import Faker
from geocoder.opencage import OpenCageQuery, OpenCageResult
from requests.exceptions import (
@@ -818,16 +820,25 @@ def test_empty_list(self):
self.assertEqual(send_mass_html_mail(tuple()), 0)
def test_mass_html_mail(self):
- test_data = list()
+ test_data: list[tuple[str, str, str, str | None, list[str]]] = []
+ test_subjects: list[tuple[str | None, str]] = []
faker = Faker._get_faker()
for i in range(random.randint(3, 7)):
+ test_subjects.append((
+ faker.optional_value(
+ 'pystr_format', ratio=0.2 if i else 1.0,
+ string_format='{{word}}-{{random_int}}'),
+ faker.sentence(),
+ ))
test_data.append((
# subject line
- faker.sentence(),
+ test_subjects[i][1]
+ if not test_subjects[i][0]
+ else f"[[{test_subjects[i][0]}]] \t {test_subjects[i][1]}",
# content: plain text & html
- faker.word(), "{}".format(faker.word()),
- # author email (ignored) & emails of recipients
- "test@ps", [],
+ faker.word(), f"
{faker.word()}",
+ # author email & emails of recipients
+ "test@ps" if i else None, [],
))
for _ in range(random.randint(1, 3)):
test_data[i][4].append(faker.company_email())
@@ -836,10 +847,27 @@ def test_mass_html_mail(self):
self.assertEqual(result, len(test_data))
self.assertLength(mail.outbox, len(test_data))
for i in range(len(test_data)):
- self.assertEqual(mail.outbox[i].subject, test_data[i][0])
- self.assertEqual(mail.outbox[i].from_email, settings.DEFAULT_FROM_EMAIL)
+ self.assertEqual(mail.outbox[i].subject, test_subjects[i][1])
+ if i == 0:
+ self.assertEqual(mail.outbox[i].from_email, settings.DEFAULT_FROM_EMAIL)
+ else:
+ self.assertEqual(mail.outbox[i].from_email, "test@ps")
self.assertEqual(mail.outbox[i].to, test_data[i][4])
+ mail.outbox = []
+ with override_settings(**settings.TEST_EMAIL_BACKENDS['dummy']):
+ result = send_mass_html_mail(test_data)
+ self.assertEqual(result, len(test_data))
+ self.assertLength(mail.outbox, len(test_data))
+ for i in range(len(test_data)):
+ outbox_item = cast(AnymailMessage, mail.outbox[i])
+ if test_subjects[i][0]:
+ self.assertEqual(outbox_item.tags, [test_subjects[i][0]])
+ else:
+ self.assertEqual(outbox_item.tags, UNSET)
+ self.assertTrue(outbox_item.anymail_test_params.get('is_batch_send'))
+ self.assertFalse(outbox_item.anymail_test_params.get('track_opens'))
+
def test_invalid_values(self):
faker = Faker._get_faker()
expected_subject = faker.sentence()