mirror of https://github.com/zulip/zulip.git
669 lines
26 KiB
Python
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)
|