zulip/zerver/tests/test_email_notifications.py

669 lines
26 KiB
Python

import tempfile
from datetime import datetime, timedelta, timezone
from typing import Dict
from unittest.mock import patch
import ldap
import orjson
from django.core import mail
from django.core.mail.message import EmailMultiAlternatives
from django.test import override_settings
from django.utils.timezone import now as timezone_now
from django_auth_ldap.config import LDAPSearch
from zerver.lib.email_notifications import (
enqueue_welcome_emails,
get_onboarding_email_schedule,
send_account_registered_email,
)
from zerver.lib.send_email import (
deliver_scheduled_emails,
send_custom_email,
send_custom_server_email,
)
from zerver.lib.test_classes import ZulipTestCase
from zerver.models import Realm, ScheduledEmail, UserProfile
from zerver.models.realms import get_realm
from zilencer.models import RemoteZulipServer
class TestCustomEmails(ZulipTestCase):
def test_send_custom_email_argument(self) -> None:
hamlet = self.example_user("hamlet")
email_subject = "subject_test"
reply_to = "reply_to_test"
from_name = "from_name_test"
with tempfile.NamedTemporaryFile() as markdown_template:
markdown_template.write(b"# Some heading\n\nSome content\n{{ realm_name }}")
markdown_template.flush()
send_custom_email(
UserProfile.objects.filter(id=hamlet.id),
dry_run=False,
options={
"markdown_template_path": markdown_template.name,
"reply_to": reply_to,
"subject": email_subject,
"from_name": from_name,
},
)
self.assert_length(mail.outbox, 1)
msg = mail.outbox[0]
self.assertEqual(msg.subject, email_subject)
self.assert_length(msg.reply_to, 1)
self.assertEqual(msg.reply_to[0], reply_to)
self.assertNotIn("{% block content %}", msg.body)
self.assertIn("# Some heading", msg.body)
self.assertIn("Zulip Dev", msg.body)
self.assertNotIn("{{ realm_name }}", msg.body)
self.assertNotIn("</div>", msg.body)
assert isinstance(msg, EmailMultiAlternatives)
self.assertIn("Some heading</h1>", str(msg.alternatives[0][0]))
self.assertNotIn("{{ realm_name }}", str(msg.alternatives[0][0]))
def test_send_custom_email_remote_server(self) -> None:
email_subject = "subject_test"
reply_to = "reply_to_test"
from_name = "from_name_test"
markdown_template_path = "templates/corporate/policies/index.md"
send_custom_server_email(
remote_servers=RemoteZulipServer.objects.all(),
dry_run=False,
options={
"markdown_template_path": markdown_template_path,
"reply_to": reply_to,
"subject": email_subject,
"from_name": from_name,
},
)
self.assert_length(mail.outbox, 1)
msg = mail.outbox[0]
self.assertEqual(msg.subject, email_subject)
self.assertEqual(msg.to, ["remotezulipserver@zulip.com"])
self.assert_length(msg.reply_to, 1)
self.assertEqual(msg.reply_to[0], reply_to)
self.assertNotIn("{% block content %}", msg.body)
# Verify that the HTML version contains the footer.
assert isinstance(msg, EmailMultiAlternatives)
self.assertIn(
"You are receiving this email to update you about important changes to Zulip",
str(msg.alternatives[0][0]),
)
def test_send_custom_email_headers(self) -> None:
hamlet = self.example_user("hamlet")
markdown_template_path = (
"zerver/tests/fixtures/email/custom_emails/email_base_headers_test.md"
)
send_custom_email(
UserProfile.objects.filter(id=hamlet.id),
dry_run=False,
options={
"markdown_template_path": markdown_template_path,
},
)
self.assert_length(mail.outbox, 1)
msg = mail.outbox[0]
self.assertEqual(msg.subject, "Test subject")
self.assertFalse(msg.reply_to)
self.assertEqual("Test body", msg.body)
def test_send_custom_email_context(self) -> None:
hamlet = self.example_user("hamlet")
markdown_template_path = (
"zerver/tests/fixtures/email/custom_emails/email_base_headers_test.md"
)
send_custom_email(
UserProfile.objects.filter(id=hamlet.id),
dry_run=False,
options={
"markdown_template_path": markdown_template_path,
},
)
self.assert_length(mail.outbox, 1)
msg = mail.outbox[0]
# We default to not including an unsubscribe link in the headers
self.assertEqual(msg.extra_headers.get("X-Auto-Response-Suppress"), "All")
self.assertIsNone(msg.extra_headers.get("List-Unsubscribe"))
mail.outbox = []
markdown_template_path = (
"zerver/tests/fixtures/email/custom_emails/email_base_headers_custom_test.md"
)
def add_context(context: Dict[str, object], user: UserProfile) -> None:
context["unsubscribe_link"] = "some@email"
context["custom"] = str(user.id)
send_custom_email(
UserProfile.objects.filter(id=hamlet.id),
dry_run=False,
options={
"markdown_template_path": markdown_template_path,
},
add_context=add_context,
)
self.assert_length(mail.outbox, 1)
msg = mail.outbox[0]
self.assertEqual(msg.extra_headers.get("X-Auto-Response-Suppress"), "All")
self.assertEqual(msg.extra_headers.get("List-Unsubscribe"), "<some@email>")
self.assertIn(f"Test body with {hamlet.id} value", msg.body)
def test_send_custom_email_no_argument(self) -> None:
hamlet = self.example_user("hamlet")
from_name = "from_name_test"
email_subject = "subject_test"
markdown_template_path = (
"zerver/tests/fixtures/email/custom_emails/email_base_headers_no_headers_test.md"
)
from zerver.lib.send_email import NoEmailArgumentError
self.assertRaises(
NoEmailArgumentError,
send_custom_email,
UserProfile.objects.filter(id=hamlet.id),
dry_run=False,
options={
"markdown_template_path": markdown_template_path,
"from_name": from_name,
},
)
self.assertRaises(
NoEmailArgumentError,
send_custom_email,
UserProfile.objects.filter(id=hamlet.id),
dry_run=False,
options={
"markdown_template_path": markdown_template_path,
"subject": email_subject,
},
)
def test_send_custom_email_doubled_arguments(self) -> None:
hamlet = self.example_user("hamlet")
from_name = "from_name_test"
email_subject = "subject_test"
markdown_template_path = (
"zerver/tests/fixtures/email/custom_emails/email_base_headers_test.md"
)
from zerver.lib.send_email import DoubledEmailArgumentError
self.assertRaises(
DoubledEmailArgumentError,
send_custom_email,
UserProfile.objects.filter(id=hamlet.id),
dry_run=False,
options={
"markdown_template_path": markdown_template_path,
"subject": email_subject,
},
)
self.assertRaises(
DoubledEmailArgumentError,
send_custom_email,
UserProfile.objects.filter(id=hamlet.id),
dry_run=False,
options={
"markdown_template_path": markdown_template_path,
"from_name": from_name,
},
)
def test_send_custom_email_dry_run(self) -> None:
hamlet = self.example_user("hamlet")
email_subject = "subject_test"
reply_to = "reply_to_test"
from_name = "from_name_test"
markdown_template_path = "templates/zerver/tests/markdown/test_nested_code_blocks.md"
with patch("builtins.print") as _:
send_custom_email(
UserProfile.objects.filter(id=hamlet.id),
dry_run=True,
options={
"markdown_template_path": markdown_template_path,
"reply_to": reply_to,
"subject": email_subject,
"from_name": from_name,
},
)
self.assert_length(mail.outbox, 0)
class TestFollowupEmails(ZulipTestCase):
def test_account_registered_email_context(self) -> None:
hamlet = self.example_user("hamlet")
send_account_registered_email(hamlet)
scheduled_emails = ScheduledEmail.objects.filter(users=hamlet).order_by(
"scheduled_timestamp"
)
email_data = orjson.loads(scheduled_emails[0].data)
self.assertEqual(email_data["context"]["email"], self.example_email("hamlet"))
self.assertEqual(email_data["context"]["is_realm_admin"], False)
self.assertEqual(
email_data["context"]["getting_user_started_link"],
"http://zulip.testserver/help/getting-started-with-zulip",
)
self.assertNotIn("ldap_username", email_data["context"])
ScheduledEmail.objects.all().delete()
iago = self.example_user("iago")
send_account_registered_email(iago)
scheduled_emails = ScheduledEmail.objects.filter(users=iago).order_by("scheduled_timestamp")
email_data = orjson.loads(scheduled_emails[0].data)
self.assertEqual(email_data["context"]["email"], self.example_email("iago"))
self.assertEqual(email_data["context"]["is_realm_admin"], True)
self.assertEqual(
email_data["context"]["getting_organization_started_link"],
"http://zulip.testserver/help/getting-your-organization-started-with-zulip",
)
self.assertEqual(
email_data["context"]["getting_user_started_link"],
"http://zulip.testserver/help/getting-started-with-zulip",
)
self.assertNotIn("ldap_username", email_data["context"])
# See https://zulip.readthedocs.io/en/latest/production/authentication-methods.html#ldap-including-active-directory
# for case details.
@override_settings(
AUTHENTICATION_BACKENDS=(
"zproject.backends.ZulipLDAPAuthBackend",
"zproject.backends.ZulipDummyBackend",
),
# configure email search for email address in the uid attribute:
AUTH_LDAP_REVERSE_EMAIL_SEARCH=LDAPSearch(
"ou=users,dc=zulip,dc=com", ldap.SCOPE_ONELEVEL, "(uid=%(email)s)"
),
)
def test_account_registered_email_ldap_case_a_login_credentials(self) -> None:
self.init_default_ldap_database()
ldap_user_attr_map = {"full_name": "cn"}
with self.settings(AUTH_LDAP_USER_ATTR_MAP=ldap_user_attr_map):
self.login_with_return(
"newuser_email_as_uid@zulip.com",
self.ldap_password("newuser_email_as_uid@zulip.com"),
)
user = UserProfile.objects.get(delivery_email="newuser_email_as_uid@zulip.com")
scheduled_emails = ScheduledEmail.objects.filter(users=user).order_by(
"scheduled_timestamp"
)
self.assert_length(scheduled_emails, 3)
email_data = orjson.loads(scheduled_emails[0].data)
self.assertEqual(email_data["context"]["ldap"], True)
self.assertEqual(
email_data["context"]["ldap_username"], "newuser_email_as_uid@zulip.com"
)
@override_settings(
AUTHENTICATION_BACKENDS=(
"zproject.backends.ZulipLDAPAuthBackend",
"zproject.backends.ZulipDummyBackend",
)
)
def test_account_registered_email_ldap_case_b_login_credentials(self) -> None:
self.init_default_ldap_database()
ldap_user_attr_map = {"full_name": "cn"}
with self.settings(
LDAP_APPEND_DOMAIN="zulip.com",
AUTH_LDAP_USER_ATTR_MAP=ldap_user_attr_map,
):
self.login_with_return("newuser@zulip.com", self.ldap_password("newuser"))
user = UserProfile.objects.get(delivery_email="newuser@zulip.com")
scheduled_emails = ScheduledEmail.objects.filter(users=user).order_by(
"scheduled_timestamp"
)
self.assert_length(scheduled_emails, 3)
email_data = orjson.loads(scheduled_emails[0].data)
self.assertEqual(email_data["context"]["ldap"], True)
self.assertEqual(email_data["context"]["ldap_username"], "newuser")
@override_settings(
AUTHENTICATION_BACKENDS=(
"zproject.backends.ZulipLDAPAuthBackend",
"zproject.backends.ZulipDummyBackend",
)
)
def test_account_registered_email_ldap_case_c_login_credentials(self) -> None:
self.init_default_ldap_database()
ldap_user_attr_map = {"full_name": "cn"}
with self.settings(
LDAP_EMAIL_ATTR="mail",
AUTH_LDAP_USER_ATTR_MAP=ldap_user_attr_map,
):
self.login_with_return("newuser_with_email", self.ldap_password("newuser_with_email"))
user = UserProfile.objects.get(delivery_email="newuser_email@zulip.com")
scheduled_emails = ScheduledEmail.objects.filter(users=user).order_by(
"scheduled_timestamp"
)
self.assert_length(scheduled_emails, 3)
email_data = orjson.loads(scheduled_emails[0].data)
self.assertEqual(email_data["context"]["ldap"], True)
self.assertEqual(email_data["context"]["ldap_username"], "newuser_with_email")
def test_followup_emails_count(self) -> None:
hamlet = self.example_user("hamlet")
iago = self.example_user("iago")
cordelia = self.example_user("cordelia")
realm = get_realm("zulip")
# Hamlet has account only in Zulip realm so all onboarding emails should be sent
send_account_registered_email(self.example_user("hamlet"))
enqueue_welcome_emails(self.example_user("hamlet"))
scheduled_emails = ScheduledEmail.objects.filter(users=hamlet).order_by(
"scheduled_timestamp"
)
self.assert_length(scheduled_emails, 3)
self.assertEqual(
orjson.loads(scheduled_emails[0].data)["template_prefix"],
"zerver/emails/account_registered",
)
self.assertEqual(
orjson.loads(scheduled_emails[1].data)["template_prefix"],
"zerver/emails/onboarding_zulip_topics",
)
self.assertEqual(
orjson.loads(scheduled_emails[2].data)["template_prefix"],
"zerver/emails/onboarding_zulip_guide",
)
ScheduledEmail.objects.all().delete()
# The onboarding_zulip_guide email should not be sent to non-admin users in organizations
# that are sent the `/for/communities/` guide; see note in enqueue_welcome_emails.
realm.org_type = Realm.ORG_TYPES["community"]["id"]
realm.save()
# Hamlet is not an admin so the `/for/communities/` zulip_guide should not be sent
send_account_registered_email(self.example_user("hamlet"))
enqueue_welcome_emails(self.example_user("hamlet"))
scheduled_emails = ScheduledEmail.objects.filter(users=hamlet).order_by(
"scheduled_timestamp"
)
self.assert_length(scheduled_emails, 2)
self.assertEqual(
orjson.loads(scheduled_emails[0].data)["template_prefix"],
"zerver/emails/account_registered",
)
self.assertEqual(
orjson.loads(scheduled_emails[1].data)["template_prefix"],
"zerver/emails/onboarding_zulip_topics",
)
ScheduledEmail.objects.all().delete()
# Iago is an admin so the `/for/communities/` zulip_guide should be sent
send_account_registered_email(self.example_user("iago"))
enqueue_welcome_emails(self.example_user("iago"))
scheduled_emails = ScheduledEmail.objects.filter(users=iago).order_by("scheduled_timestamp")
self.assert_length(scheduled_emails, 3)
self.assertEqual(
orjson.loads(scheduled_emails[0].data)["template_prefix"],
"zerver/emails/account_registered",
)
self.assertEqual(
orjson.loads(scheduled_emails[1].data)["template_prefix"],
"zerver/emails/onboarding_zulip_topics",
)
self.assertEqual(
orjson.loads(scheduled_emails[2].data)["template_prefix"],
"zerver/emails/onboarding_zulip_guide",
)
ScheduledEmail.objects.all().delete()
# The organization_type context for "education_nonprofit" orgs is simplified to be "education"
realm.org_type = Realm.ORG_TYPES["education_nonprofit"]["id"]
realm.save()
# Cordelia has account in more than 1 realm so onboarding_zulip_topics email should not be sent
send_account_registered_email(self.example_user("cordelia"))
enqueue_welcome_emails(self.example_user("cordelia"))
scheduled_emails = ScheduledEmail.objects.filter(users=cordelia).order_by(
"scheduled_timestamp"
)
self.assert_length(scheduled_emails, 2)
self.assertEqual(
orjson.loads(scheduled_emails[0].data)["template_prefix"],
"zerver/emails/account_registered",
)
self.assertEqual(
orjson.loads(scheduled_emails[1].data)["template_prefix"],
"zerver/emails/onboarding_zulip_guide",
)
self.assertEqual(
orjson.loads(scheduled_emails[1].data)["context"]["organization_type"],
"education",
)
ScheduledEmail.objects.all().delete()
# Only a subset of Realm.ORG_TYPES are sent the zulip_guide_followup email
realm.org_type = Realm.ORG_TYPES["other"]["id"]
realm.save()
# In this case, Cordelia should only be sent the account_registered email
send_account_registered_email(self.example_user("cordelia"))
enqueue_welcome_emails(self.example_user("cordelia"))
scheduled_emails = ScheduledEmail.objects.filter(users=cordelia)
self.assert_length(scheduled_emails, 1)
self.assertEqual(
orjson.loads(scheduled_emails[0].data)["template_prefix"],
"zerver/emails/account_registered",
)
def test_followup_emails_for_regular_realms(self) -> None:
cordelia = self.example_user("cordelia")
send_account_registered_email(self.example_user("cordelia"), realm_creation=True)
enqueue_welcome_emails(self.example_user("cordelia"), realm_creation=True)
scheduled_emails = ScheduledEmail.objects.filter(users=cordelia).order_by(
"scheduled_timestamp"
)
assert scheduled_emails is not None
self.assert_length(scheduled_emails, 3)
self.assertEqual(
orjson.loads(scheduled_emails[0].data)["template_prefix"],
"zerver/emails/account_registered",
)
self.assertEqual(
orjson.loads(scheduled_emails[1].data)["template_prefix"],
"zerver/emails/onboarding_zulip_guide",
)
self.assertEqual(
orjson.loads(scheduled_emails[2].data)["template_prefix"],
"zerver/emails/onboarding_team_to_zulip",
)
deliver_scheduled_emails(scheduled_emails[0])
from django.core.mail import outbox
self.assert_length(outbox, 1)
message = outbox[0]
self.assertIn("you have created a new Zulip organization", message.body)
self.assertNotIn("demo org", message.body)
def test_followup_emails_for_demo_realms(self) -> None:
cordelia = self.example_user("cordelia")
cordelia.realm.demo_organization_scheduled_deletion_date = timezone_now() + timedelta(
days=30
)
cordelia.realm.save()
send_account_registered_email(self.example_user("cordelia"), realm_creation=True)
enqueue_welcome_emails(self.example_user("cordelia"), realm_creation=True)
scheduled_emails = ScheduledEmail.objects.filter(users=cordelia).order_by(
"scheduled_timestamp"
)
assert scheduled_emails is not None
self.assert_length(scheduled_emails, 3)
self.assertEqual(
orjson.loads(scheduled_emails[0].data)["template_prefix"],
"zerver/emails/account_registered",
)
self.assertEqual(
orjson.loads(scheduled_emails[1].data)["template_prefix"],
"zerver/emails/onboarding_zulip_guide",
)
self.assertEqual(
orjson.loads(scheduled_emails[2].data)["template_prefix"],
"zerver/emails/onboarding_team_to_zulip",
)
deliver_scheduled_emails(scheduled_emails[0])
from django.core.mail import outbox
self.assert_length(outbox, 1)
message = outbox[0]
self.assertIn("you have created a new demo Zulip organization", message.body)
def test_onboarding_zulip_guide_with_invalid_org_type(self) -> None:
cordelia = self.example_user("cordelia")
realm = get_realm("zulip")
invalid_org_type_id = 999
realm.org_type = invalid_org_type_id
realm.save()
with self.assertLogs(level="ERROR") as m:
enqueue_welcome_emails(self.example_user("cordelia"))
scheduled_emails = ScheduledEmail.objects.filter(users=cordelia)
self.assert_length(scheduled_emails, 0)
self.assertEqual(
m.output,
[f"ERROR:root:Unknown organization type '{invalid_org_type_id}'"],
)
class TestOnboardingEmailDelay(ZulipTestCase):
def verify_onboarding_email_schedule(
self,
user: UserProfile,
date_joined: str,
onboarding_zulip_topics: int,
onboarding_zulip_guide: int,
onboarding_team_to_zulip: int,
) -> None:
DAY_OF_WEEK = {
"Monday": datetime(2018, 1, 1, 1, 0, 0, 0, tzinfo=timezone.utc),
"Tuesday": datetime(2018, 1, 2, 1, 0, 0, 0, tzinfo=timezone.utc),
"Wednesday": datetime(2018, 1, 3, 1, 0, 0, 0, tzinfo=timezone.utc),
"Thursday": datetime(2018, 1, 4, 1, 0, 0, 0, tzinfo=timezone.utc),
"Friday": datetime(2018, 1, 5, 1, 0, 0, 0, tzinfo=timezone.utc),
"Saturday": datetime(2018, 1, 6, 1, 0, 0, 0, tzinfo=timezone.utc),
"Sunday": datetime(2018, 1, 7, 1, 0, 0, 0, tzinfo=timezone.utc),
}
WEEKEND = [6, 7]
user.date_joined = DAY_OF_WEEK[date_joined]
onboarding_email_schedule = get_onboarding_email_schedule(user)
# onboarding_zulip_topics
day_sent = (
DAY_OF_WEEK[date_joined] + onboarding_email_schedule["onboarding_zulip_topics"]
).isoweekday()
self.assertEqual(day_sent, onboarding_zulip_topics)
self.assertNotIn(day_sent, WEEKEND)
# onboarding_zulip_guide
day_sent = (
DAY_OF_WEEK[date_joined] + onboarding_email_schedule["onboarding_zulip_guide"]
).isoweekday()
self.assertEqual(day_sent, onboarding_zulip_guide)
self.assertNotIn(day_sent, WEEKEND)
# onboarding_team_to_zulip
day_sent = (
DAY_OF_WEEK[date_joined] + onboarding_email_schedule["onboarding_team_to_zulip"]
).isoweekday()
self.assertEqual(day_sent, onboarding_team_to_zulip)
self.assertNotIn(day_sent, WEEKEND)
def test_get_onboarding_email_schedule(self) -> None:
user_profile = self.example_user("hamlet")
# joined Monday: schedule = Wednesday:3, Friday:5, Tuesday:2
self.verify_onboarding_email_schedule(user_profile, "Monday", 3, 5, 2)
# joined Tuesday: schedule = Thursday:4, Monday:1, Wednesday:3
self.verify_onboarding_email_schedule(user_profile, "Tuesday", 4, 1, 3)
# joined Wednesday: schedule = Friday:5, Tuesday:2, Thursday:4
self.verify_onboarding_email_schedule(user_profile, "Wednesday", 5, 2, 4)
# joined Thursday: schedule = Monday:1, Wednesday:3, Friday:5
self.verify_onboarding_email_schedule(user_profile, "Thursday", 1, 3, 5)
# joined Friday: schedule = Tuesday:2, Thursday:4, Monday:1
self.verify_onboarding_email_schedule(user_profile, "Friday", 2, 4, 1)
# joined Saturday: schedule = Monday:1, Wednesday:3, Friday:5
self.verify_onboarding_email_schedule(user_profile, "Saturday", 1, 3, 5)
# joined Sunday: schedule = Tuesday:2, Thursday:4, Monday:1
self.verify_onboarding_email_schedule(user_profile, "Sunday", 2, 4, 1)
def test_time_offset_for_onboarding_email_schedule(self) -> None:
user_profile = self.example_user("hamlet")
days_delayed = {
"4": timedelta(days=4, hours=-1),
"6": timedelta(days=6, hours=-1),
"8": timedelta(days=8, hours=-1),
}
# Time offset of America/Phoenix is -07:00
user_profile.timezone = "America/Phoenix"
# Test date_joined == Friday in UTC, but Thursday in the user's time zone
user_profile.date_joined = datetime(2018, 1, 5, 1, 0, 0, 0, tzinfo=timezone.utc)
onboarding_email_schedule = get_onboarding_email_schedule(user_profile)
# onboarding_zulip_topics email sent on Monday
self.assertEqual(
onboarding_email_schedule["onboarding_zulip_topics"],
days_delayed["4"],
)
# onboarding_zulip_guide sent on Wednesday
self.assertEqual(
onboarding_email_schedule["onboarding_zulip_guide"],
days_delayed["6"],
)
# onboarding_team_to_zulip sent on Friday
self.assertEqual(
onboarding_email_schedule["onboarding_team_to_zulip"],
days_delayed["8"],
)
class TestCustomWelcomeEmailSender(ZulipTestCase):
def test_custom_welcome_email_sender(self) -> None:
name = "Nonreg Email"
email = self.nonreg_email("test")
with override_settings(
WELCOME_EMAIL_SENDER={
"name": name,
"email": email,
}
):
hamlet = self.example_user("hamlet")
enqueue_welcome_emails(hamlet)
scheduled_emails = ScheduledEmail.objects.filter(users=hamlet).order_by(
"scheduled_timestamp"
)
email_data = orjson.loads(scheduled_emails[0].data)
self.assertEqual(email_data["from_name"], name)
self.assertEqual(email_data["from_address"], email)