/
githubmirror
/
oppia
Обзор
Документация
Войти
/
githubmirror
/
oppia
Код
Запросы
0
Пакеты
0
Релизы
0
Аналитика
Безопасность
develop
core/domain/email_manager_test.py
9 212 строк
370 KB
ateeksh22
Fix #26623: Migrate datetime usage to utils functions (#26648)
05 авг 2026, 07:42
Не верифицирован
05 авг 2026, 07:42
1d13470
Код
Авторство
О чём код?
# Copyright 2014 The Oppia Authors. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS-IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Tests for methods relating to sending emails.""" from __future__ import annotations import datetime import logging import types from core import feconf, utils from core.constants import constants from core.domain import ( email_manager, email_services, exp_domain, html_cleaner, platform_parameter_domain, ) from core.domain import platform_parameter_list as param_list from core.domain import ( platform_parameter_registry, question_domain, rights_domain, story_domain, subscription_services, suggestion_registry, suggestion_services, taskqueue_services, translation_domain, user_services, ) from core.platform import models from core.tests import test_utils from typing import ( Callable, DefaultDict, Dict, Final, List, Optional, Sequence, Set, Tuple, Type, Union, ) MYPY = False if MYPY: # pragma: no cover from mypy_imports import email_models, secrets_services, suggestion_models email_models, suggestion_models = models.Registry.import_models( [models.Names.EMAIL, models.Names.SUGGESTION] ) secrets_services = models.Registry.import_secrets_services() EMAIL_FOOTER = ( 'You can change your email preferences via the <a href="http://loca' 'lhost:8181/preferences">Preferences</a> page.' ) DEV_OPPIA_SITE_URL = 'http://localhost:8181' class EmailToAdminTest(test_utils.EmailTestBase): """Test that emails are correctly sent to the admin.""" @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_SENDER_NAME, 'Name'), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'admin@system.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'dummy@system.com'), (param_list.ParamName.SYSTEM_EMAIL_NAME, 'DUMMY_SYSTEM_NAME'), ] ) def test_email_to_admin_is_sent_correctly(self) -> None: # Make sure there are no emails already sent. admin_email_address = 'admin@system.com' messages = self._get_sent_email_messages(admin_email_address) self.assertEqual(len(messages), 0) # Send an email to admin. email_manager.send_mail_to_admin('Dummy Subject', 'Dummy Body') # Make sure emails are sent. messages = self._get_sent_email_messages(admin_email_address) self.assertEqual(len(messages), 1) self.assertEqual( messages[0].sender, 'DUMMY_SYSTEM_NAME <dummy@system.com>' ) self.assertEqual(messages[0].to, ['admin@system.com']) self.assertEqual(messages[0].subject, 'Dummy Subject') self.assertIn('Dummy Body', messages[0].html) class DummyMailTest(test_utils.EmailTestBase): """Test that emails are correctly sent to the testing email id.""" @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_SENDER_NAME, 'Name'), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'admin@system.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'dummy@system.com'), (param_list.ParamName.SYSTEM_EMAIL_NAME, 'DUMMY_SYSTEM_NAME'), ] ) def test_sending_emails(self) -> None: dummy_system_name = 'DUMMY_SYSTEM_NAME' dummy_receiver_address = 'admin@system.com' # Make sure there are no emails already sent. messages = self._get_sent_email_messages(dummy_receiver_address) self.assertEqual(len(messages), 0) # Send an email. email_manager.send_dummy_mail_to_admin(dummy_system_name) # Make sure emails are sent. messages = self._get_sent_email_messages(dummy_receiver_address) self.assertEqual(len(messages), 1) self.assertEqual( messages[0].sender, 'DUMMY_SYSTEM_NAME <dummy@system.com>' ) self.assertEqual(messages[0].to, [dummy_receiver_address]) self.assertEqual(messages[0].subject, 'Test Mail') self.assertIn( 'This is a test mail from DUMMY_SYSTEM_NAME', messages[0].html ) class EmailRightsTest(test_utils.GenericTestBase): """Test that only certain users can send certain types of emails.""" def setUp(self) -> None: super().setUp() self.signup(self.EDITOR_EMAIL, self.EDITOR_USERNAME) self.editor_id = self.get_user_id_from_email(self.EDITOR_EMAIL) self.signup(self.MODERATOR_EMAIL, self.MODERATOR_USERNAME) self.moderator_id = self.get_user_id_from_email(self.MODERATOR_EMAIL) self.set_moderators([self.MODERATOR_USERNAME]) self.signup(self.CURRICULUM_ADMIN_EMAIL, self.CURRICULUM_ADMIN_USERNAME) self.admin_id = self.get_user_id_from_email(self.CURRICULUM_ADMIN_EMAIL) self.set_curriculum_admins([self.CURRICULUM_ADMIN_USERNAME]) def test_sender_id_validation(self) -> None: sender_ids_to_test = [ feconf.SYSTEM_COMMITTER_ID, self.admin_id, self.moderator_id, self.editor_id, ] # These are given in the order of user_ids_to_test. expected_validation_results = { feconf.EMAIL_INTENT_SIGNUP: (True, False, False, False), feconf.EMAIL_INTENT_DAILY_BATCH: (True, False, False, False), feconf.EMAIL_INTENT_MARKETING: (True, False, False, False), feconf.EMAIL_INTENT_UNPUBLISH_EXPLORATION: ( True, False, True, False, ), feconf.EMAIL_INTENT_DELETE_EXPLORATION: (True, False, True, False), } for intent, results in expected_validation_results.items(): for ind, sender_id in enumerate(sender_ids_to_test): if results[ind]: email_manager.require_sender_id_is_valid(intent, sender_id) else: with self.assertRaisesRegex(Exception, 'Invalid sender_id'): email_manager.require_sender_id_is_valid( intent, sender_id ) # Also test null and invalid intent strings. with self.assertRaisesRegex(Exception, 'Invalid email intent string'): email_manager.require_sender_id_is_valid( '', feconf.SYSTEM_COMMITTER_ID ) with self.assertRaisesRegex(Exception, 'Invalid email intent string'): email_manager.require_sender_id_is_valid('', self.admin_id) with self.assertRaisesRegex(Exception, 'Invalid email intent string'): email_manager.require_sender_id_is_valid( 'invalid_intent', feconf.SYSTEM_COMMITTER_ID ) with self.assertRaisesRegex(Exception, 'Invalid email intent string'): email_manager.require_sender_id_is_valid( 'invalid_intent', self.admin_id ) class ExplorationMembershipEmailTests(test_utils.EmailTestBase): """Tests that sending exploration membership email works as expected.""" EXPLORATION_TITLE: Final = 'Title' def setUp(self) -> None: super().setUp() self.signup(self.EDITOR_EMAIL, self.EDITOR_USERNAME) self.editor_id = self.get_user_id_from_email(self.EDITOR_EMAIL) self.signup(self.NEW_USER_EMAIL, self.NEW_USER_USERNAME) self.new_user_id = self.get_user_id_from_email(self.NEW_USER_EMAIL) self.exploration = self.save_new_default_exploration( 'A', self.editor_id, title=self.EXPLORATION_TITLE ) self.expected_email_subject = ( '%s - invitation to collaborate' ) % self.EXPLORATION_TITLE self.can_send_editor_role_email_ctx = self.swap( feconf, 'CAN_SEND_TRANSACTIONAL_EMAILS', True ) self.can_not_send_editor_role_email_ctx = self.swap( feconf, 'CAN_SEND_TRANSACTIONAL_EMAILS', False ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_SENDER_NAME, 'Name'), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ] ) def test_role_email_is_sent_when_editor_assigns_role(self) -> None: with self.can_send_editor_role_email_ctx: self.login(self.EDITOR_EMAIL) csrf_token = self.get_new_csrf_token() self.put_json( '%s/%s' % (feconf.EXPLORATION_RIGHTS_PREFIX, self.exploration.id), { 'version': self.exploration.version, 'new_member_username': self.NEW_USER_USERNAME, 'new_member_role': rights_domain.ROLE_EDITOR, }, csrf_token=csrf_token, ) messages = self._get_sent_email_messages(self.NEW_USER_EMAIL) self.assertEqual(len(messages), 1) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), ] ) def test_email_is_not_sent_if_recipient_has_declined_such_emails( self, ) -> None: user_services.update_email_preferences( self.new_user_id, True, False, False, False, feconf.DEFAULT_CONTRIBUTOR_DASHBOARD_EMAIL_PREFERENCE, ) with self.can_send_editor_role_email_ctx: email_manager.send_role_notification_email( self.editor_id, self.new_user_id, rights_domain.ROLE_OWNER, self.exploration.id, self.exploration.title, ) messages = self._get_sent_email_messages(self.NEW_USER_EMAIL) self.assertEqual(len(messages), 0) @test_utils.set_platform_parameters( [(param_list.ParamName.SERVER_CAN_SEND_EMAILS, False)] ) def test_email_not_sent_if_server_can_send_emails_is_false(self) -> None: email_manager.send_role_notification_email( self.editor_id, self.new_user_id, rights_domain.ROLE_OWNER, self.exploration.id, self.exploration.title, ) messages = self._get_sent_email_messages(self.NEW_USER_EMAIL) self.assertEqual(len(messages), 0) @test_utils.set_platform_parameters( [(param_list.ParamName.SERVER_CAN_SEND_EMAILS, True)] ) def test_that_email_not_sent_if_can_send_transactional_emails_is_false( self, ) -> None: with self.can_not_send_editor_role_email_ctx: email_manager.send_role_notification_email( self.editor_id, self.new_user_id, rights_domain.ROLE_EDITOR, self.exploration.id, self.exploration.title, ) messages = self._get_sent_email_messages(self.NEW_USER_EMAIL) self.assertEqual(len(messages), 0) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_SENDER_NAME, 'Name'), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ] ) def test_role_emails_sent_are_correct(self) -> None: with self.can_send_editor_role_email_ctx: email_manager.send_role_notification_email( self.editor_id, self.new_user_id, rights_domain.ROLE_VIEWER, self.exploration.id, self.exploration.title, ) messages = self._get_sent_email_messages(self.NEW_USER_EMAIL) self.assertEqual(len(messages), 1) all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) self.assertEqual(len(all_models), 1) sent_email_model = all_models[0] # Check that email details are correct. self.assertEqual(sent_email_model.recipient_id, self.new_user_id) self.assertEqual( sent_email_model.recipient_email, self.NEW_USER_EMAIL ) self.assertEqual( sent_email_model.sender_id, feconf.SYSTEM_COMMITTER_ID ) self.assertEqual( sent_email_model.sender_email, '%s <%s>' % (self.EDITOR_USERNAME, 'noreply@example.com'), ) self.assertEqual( sent_email_model.intent, feconf.EMAIL_INTENT_EDITOR_ROLE_NOTIFICATION, ) self.assertEqual( sent_email_model.subject, self.expected_email_subject ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_SENDER_NAME, 'Name'), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ] ) def test_correct_rights_are_written_in_manager_role_email_body( self, ) -> None: expected_email_html_body = ( 'Hi newuser,<br>' '<br>' '<b>editor</b> has granted you manager rights to their ' 'exploration, ' '"<a href="https://www.oppia.org/create/A">Title</a>", ' 'on Oppia.org.<br>' '<br>' 'This allows you to:<br>' '<ul>' '<li>Change the exploration permissions</li><br>' '<li>Edit the exploration</li><br>' '<li>View and playtest the exploration</li><br>' '</ul>' 'You can find the exploration ' '<a href="https://www.oppia.org/create/A">here</a>.<br>' '<br>' 'Thanks, and happy collaborating!<br>' '<br>' 'Best wishes,<br>' 'The Oppia Team<br>' '<br>' 'You can change your email preferences via the ' '<a href="http://localhost:8181/preferences">Preferences</a> page.' ) expected_email_text_body = ( 'Hi newuser,\n' '\n' 'editor has granted you manager rights to their ' 'exploration, "Title", on Oppia.org.\n' '\n' 'This allows you to:\n' '- Change the exploration permissions\n' '- Edit the exploration\n' '- View and playtest the exploration\n' 'You can find the exploration here.\n' '\n' 'Thanks, and happy collaborating!\n' '\n' 'Best wishes,\n' 'The Oppia Team\n' '\n' 'You can change your email preferences via the Preferences page.' ) with self.can_send_editor_role_email_ctx: # Check that correct email content is sent for Manager. email_manager.send_role_notification_email( self.editor_id, self.new_user_id, rights_domain.ROLE_OWNER, self.exploration.id, self.exploration.title, ) messages = self._get_sent_email_messages(self.NEW_USER_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) self.assertEqual(messages[0].body, expected_email_text_body) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_SENDER_NAME, 'Name'), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ] ) def test_correct_rights_are_written_in_editor_role_email_body(self) -> None: expected_email_html_body = ( 'Hi newuser,<br>' '<br>' '<b>editor</b> has granted you editor rights to their ' 'exploration, ' '"<a href="https://www.oppia.org/create/A">Title</a>"' ', on Oppia.org.<br>' '<br>' 'This allows you to:<br>' '<ul>' '<li>Edit the exploration</li><br>' '<li>View and playtest the exploration</li><br>' '</ul>' 'You can find the exploration ' '<a href="https://www.oppia.org/create/A">here</a>.<br>' '<br>' 'Thanks, and happy collaborating!<br>' '<br>' 'Best wishes,<br>' 'The Oppia Team<br>' '<br>' 'You can change your email preferences via the ' '<a href="http://localhost:8181/preferences">Preferences</a> page.' ) expected_email_text_body = ( 'Hi newuser,\n' '\n' 'editor has granted you editor rights to their ' 'exploration, "Title", on Oppia.org.\n' '\n' 'This allows you to:\n' '- Edit the exploration\n' '- View and playtest the exploration\n' 'You can find the exploration here.\n' '\n' 'Thanks, and happy collaborating!\n' '\n' 'Best wishes,\n' 'The Oppia Team\n' '\n' 'You can change your email preferences via the Preferences page.' ) with self.can_send_editor_role_email_ctx: # Check that correct email content is sent for Editor. email_manager.send_role_notification_email( self.editor_id, self.new_user_id, rights_domain.ROLE_EDITOR, self.exploration.id, self.exploration.title, ) messages = self._get_sent_email_messages(self.NEW_USER_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) self.assertEqual(messages[0].body, expected_email_text_body) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_SENDER_NAME, 'Name'), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ] ) def test_correct_rights_are_written_in_playtester_role_email_body( self, ) -> None: expected_email_html_body = ( 'Hi newuser,<br>' '<br>' '<b>editor</b> has granted you playtest access to their ' 'exploration, ' '"<a href="https://www.oppia.org/create/A">Title</a>"' ', on Oppia.org.<br>' '<br>' 'This allows you to:<br>' '<ul>' '<li>View and playtest the exploration</li><br>' '</ul>' 'You can find the exploration ' '<a href="https://www.oppia.org/create/A">here</a>.<br>' '<br>' 'Thanks, and happy collaborating!<br>' '<br>' 'Best wishes,<br>' 'The Oppia Team<br>' '<br>' 'You can change your email preferences via the ' '<a href="http://localhost:8181/preferences">Preferences</a> page.' ) expected_email_text_body = ( 'Hi newuser,\n' '\n' 'editor has granted you playtest access to their ' 'exploration, "Title", on Oppia.org.\n' '\n' 'This allows you to:\n' '- View and playtest the exploration\n' 'You can find the exploration here.\n' '\n' 'Thanks, and happy collaborating!\n' '\n' 'Best wishes,\n' 'The Oppia Team\n' '\n' 'You can change your email preferences via the Preferences page.' ) with self.can_send_editor_role_email_ctx: # Check that correct email content is sent for Playtester. email_manager.send_role_notification_email( self.editor_id, self.new_user_id, rights_domain.ROLE_VIEWER, self.exploration.id, self.exploration.title, ) messages = self._get_sent_email_messages(self.NEW_USER_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) self.assertEqual(messages[0].body, expected_email_text_body) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_SENDER_NAME, 'Name'), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), ] ) def test_correct_undefined_role_raises_an_exception(self) -> None: with self.can_send_editor_role_email_ctx: # Check that an exception is raised when an invalid # role is supplied. with self.assertRaisesRegex(Exception, 'Invalid role'): email_manager.send_role_notification_email( self.editor_id, self.new_user_id, rights_domain.ROLE_NONE, self.exploration.id, self.exploration.title, ) class SignupEmailTests(test_utils.EmailTestBase): """Test that signup-email sending functionality works as expected.""" PLACEHOLDER_SUBJECT: Final = 'THIS IS A PLACEHOLDER.' PLACEHOLDER_HTML_BODY: Final = ( 'THIS IS A <b>PLACEHOLDER</b> AND SHOULD BE REPLACED.' ) EMAIL_BODY: Final = ( 'Here is some HTML text.<br>' 'With a <b>bold</b> bit and an <i>italic</i> bit.<br>' ) NEW_FOOTER: Final = ( 'Unsubscribe from emails at your ' '<a href="https://www.site.com/prefs">Preferences page</a>.' ) def setUp(self) -> None: super().setUp() self.signup(self.CURRICULUM_ADMIN_EMAIL, self.CURRICULUM_ADMIN_USERNAME) self.admin_id = self.get_user_id_from_email(self.CURRICULUM_ADMIN_EMAIL) self.set_curriculum_admins([self.CURRICULUM_ADMIN_USERNAME]) self.new_footer = ( 'Unsubscribe from emails at your ' '<a href="https://www.site.com/prefs">Preferences page</a>.' ) self.new_email_subject_content = 'Welcome!' self.new_email_body_content = ( 'Here is some HTML text.<br>' 'With a <b>bold</b> bit and an <i>italic</i> bit.<br>' ) self.expected_text_email_content = ( 'Hi editor,\n' '\n' 'Here is some HTML text.\n' 'With a bold bit and an italic bit.\n' '\n' '\n' 'Unsubscribe from emails at your Preferences page.' ) self.expected_html_email_content = ( 'Hi editor,<br>' '<br>' 'Here is some HTML text.<br>' 'With a <b>bold</b> bit and an <i>italic</i> bit.<br>' '<br>' '<br>' 'Unsubscribe from emails at your ' '<a href="https://www.site.com/prefs">Preferences page</a>.' ) def _set_signup_email_content_platform_parameter( self, new_email_subject_content: str, new_email_body_content: str ) -> None: """Sets email content platform parameters. Args: new_email_subject_content: str. The email subject. new_email_body_content: str. The email body. """ platform_parameter_registry.Registry.update_platform_parameter( param_list.ParamName.SIGNUP_EMAIL_SUBJECT_CONTENT.value, self.admin_id, 'Updating email subject.', [ platform_parameter_domain.PlatformParameterRule.from_dict( { 'filters': [ { 'type': 'platform_type', 'conditions': [['=', 'Web']], } ], 'value_when_matched': new_email_subject_content, } ) ], platform_parameter_registry.Registry.get_platform_parameter( param_list.ParamName.SIGNUP_EMAIL_SUBJECT_CONTENT.value ).default_value, ) platform_parameter_registry.Registry.update_platform_parameter( param_list.ParamName.SIGNUP_EMAIL_BODY_CONTENT.value, self.admin_id, 'Updating email body.', [ platform_parameter_domain.PlatformParameterRule.from_dict( { 'filters': [ { 'type': 'platform_type', 'conditions': [['=', 'Web']], } ], 'value_when_matched': new_email_body_content, } ) ], platform_parameter_registry.Registry.get_platform_parameter( param_list.ParamName.SIGNUP_EMAIL_BODY_CONTENT.value ).default_value, ) def _reset_signup_email_content_platform_parameters(self) -> None: """Resets email content platform parameters.""" platform_parameter_registry.Registry.update_platform_parameter( param_list.ParamName.SIGNUP_EMAIL_SUBJECT_CONTENT.value, self.admin_id, 'Resetting email subject.', [], platform_parameter_registry.Registry.get_platform_parameter( param_list.ParamName.SIGNUP_EMAIL_SUBJECT_CONTENT.value ).default_value, ) platform_parameter_registry.Registry.update_platform_parameter( param_list.ParamName.SIGNUP_EMAIL_BODY_CONTENT.value, self.admin_id, 'Resetting email body.', [], platform_parameter_registry.Registry.get_platform_parameter( param_list.ParamName.SIGNUP_EMAIL_BODY_CONTENT.value ).default_value, ) def _reset_the_email_platform_params_value(self) -> None: """Resets the email name and footer platform parameters.""" platform_parameter_registry.Registry.update_platform_parameter( param_list.ParamName.EMAIL_SENDER_NAME.value, self.admin_id, 'Reset the sender name to default', [], platform_parameter_registry.Registry.get_platform_parameter( param_list.ParamName.EMAIL_SENDER_NAME.value ).default_value, ) platform_parameter_registry.Registry.update_platform_parameter( param_list.ParamName.EMAIL_FOOTER.value, self.admin_id, 'Reset the email footer to default', [], platform_parameter_registry.Registry.get_platform_parameter( param_list.ParamName.EMAIL_FOOTER.value ).default_value, ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, False), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), ] ) def test_email_not_sent_if_config_does_not_permit_it(self) -> None: self._set_signup_email_content_platform_parameter( self.new_email_subject_content, self.new_email_body_content ) self.login(self.EDITOR_EMAIL) self.get_html_response('%s?return_url=/' % feconf.SIGNUP_URL) csrf_token = self.get_new_csrf_token() self.post_json( feconf.SIGNUP_DATA_URL, { 'agreed_to_terms': True, 'username': self.EDITOR_USERNAME, 'default_dashboard': constants.DASHBOARD_TYPE_LEARNER, 'can_receive_email_updates': ( feconf.DEFAULT_EMAIL_UPDATES_PREFERENCE ), }, csrf_token=csrf_token, ) # Check that no email was sent. messages = self._get_sent_email_messages(self.EDITOR_EMAIL) self.assertEqual(0, len(messages)) self._reset_signup_email_content_platform_parameters() @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_SENDER_NAME, 'Name'), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), ( param_list.ParamName.SIGNUP_EMAIL_SUBJECT_CONTENT, PLACEHOLDER_SUBJECT, ), (param_list.ParamName.SIGNUP_EMAIL_BODY_CONTENT, ''), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ] ) def test_email_not_sent_if_content_parameter_is_not_modified(self) -> None: log_new_error_counter = test_utils.CallCounter(logging.error) log_new_error_ctx = self.swap(logging, 'error', log_new_error_counter) with self.capture_logging(min_level=logging.ERROR) as logs: with log_new_error_ctx: self.assertEqual(log_new_error_counter.times_called, 0) self.login(self.EDITOR_EMAIL) self.get_html_response('%s?return_url=/' % feconf.SIGNUP_URL) csrf_token = self.get_new_csrf_token() # No user-facing error should surface. self.post_json( feconf.SIGNUP_DATA_URL, { 'agreed_to_terms': True, 'username': self.EDITOR_USERNAME, 'default_dashboard': constants.DASHBOARD_TYPE_LEARNER, 'can_receive_email_updates': ( feconf.DEFAULT_EMAIL_UPDATES_PREFERENCE ), }, csrf_token=csrf_token, ) # However, an error should be recorded in the logs. self.assertEqual(log_new_error_counter.times_called, 1) self.assertEqual( logs[0], 'Please ensure that the value for the admin platform ' 'property SIGNUP_EMAIL_SUBJECT_CONTENT is set, before ' 'allowing post-signup emails to be sent.', ) # Check that no email was sent. messages = self._get_sent_email_messages(self.EDITOR_EMAIL) self.assertEqual(0, len(messages)) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_SENDER_NAME, 'Name'), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), ( param_list.ParamName.SIGNUP_EMAIL_SUBJECT_CONTENT, 'dummy subject', ), ( param_list.ParamName.SIGNUP_EMAIL_BODY_CONTENT, PLACEHOLDER_HTML_BODY, ), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), ] ) def test_email_not_sent_if_content_config_is_partially_modified( self, ) -> None: platform_parameter_registry.Registry.update_platform_parameter( param_list.ParamName.SIGNUP_EMAIL_SUBJECT_CONTENT.value, self.admin_id, 'Updating email subject.', [ platform_parameter_domain.PlatformParameterRule.from_dict( { 'filters': [ { 'type': 'platform_type', 'conditions': [['=', 'Web']], } ], 'value_when_matched': self.new_email_subject_content, } ) ], platform_parameter_registry.Registry.get_platform_parameter( param_list.ParamName.SIGNUP_EMAIL_SUBJECT_CONTENT.value ).default_value, ) log_new_error_counter = test_utils.CallCounter(logging.error) log_new_error_ctx = self.swap(logging, 'error', log_new_error_counter) with self.capture_logging(min_level=logging.ERROR) as logs: with log_new_error_ctx: self.assertEqual(log_new_error_counter.times_called, 0) self.login(self.EDITOR_EMAIL) self.get_html_response('%s?return_url=/' % feconf.SIGNUP_URL) csrf_token = self.get_new_csrf_token() # No user-facing error should surface. self.post_json( feconf.SIGNUP_DATA_URL, { 'agreed_to_terms': True, 'username': self.EDITOR_USERNAME, 'default_dashboard': constants.DASHBOARD_TYPE_LEARNER, 'can_receive_email_updates': ( feconf.DEFAULT_EMAIL_UPDATES_PREFERENCE ), }, csrf_token=csrf_token, ) # However, an error should be recorded in the logs. self.assertEqual(log_new_error_counter.times_called, 1) self.assertEqual( logs[0], 'Please ensure that the value for the admin platform ' 'property SIGNUP_EMAIL_BODY_CONTENT is set, before ' 'allowing post-signup emails to be sent.', ) # Check that no email was sent. messages = self._get_sent_email_messages(self.EDITOR_EMAIL) self.assertEqual(0, len(messages)) self._reset_signup_email_content_platform_parameters() @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_SENDER_NAME, 'Name'), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), ( param_list.ParamName.SIGNUP_EMAIL_SUBJECT_CONTENT, 'dummy subject', ), ( param_list.ParamName.SIGNUP_EMAIL_BODY_CONTENT, 'New HTML body.<script>alert(3);</script>', ), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ] ) def test_email_with_bad_content_is_not_sent(self) -> None: log_new_error_counter = test_utils.CallCounter(logging.error) log_new_error_ctx = self.swap(logging, 'error', log_new_error_counter) with self.capture_logging(min_level=logging.ERROR) as logs: with log_new_error_ctx: self.assertEqual(log_new_error_counter.times_called, 0) self.login(self.EDITOR_EMAIL) self.get_html_response('%s?return_url=/' % feconf.SIGNUP_URL) csrf_token = self.get_new_csrf_token() # No user-facing error should surface. self.post_json( feconf.SIGNUP_DATA_URL, { 'agreed_to_terms': True, 'username': self.EDITOR_USERNAME, 'default_dashboard': constants.DASHBOARD_TYPE_LEARNER, 'can_receive_email_updates': ( feconf.DEFAULT_EMAIL_UPDATES_PREFERENCE ), }, csrf_token=csrf_token, ) # However, an error should be recorded in the logs. self.assertEqual(log_new_error_counter.times_called, 1) self.assertTrue( logs[0].startswith( 'Original email HTML body does not match cleaned HTML body' ) ) # Check that no email was sent. messages = self._get_sent_email_messages(self.EDITOR_EMAIL) self.assertEqual(0, len(messages)) self._reset_signup_email_content_platform_parameters() @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_SENDER_NAME, 'Email Sender'), (param_list.ParamName.EMAIL_FOOTER, NEW_FOOTER), (param_list.ParamName.SIGNUP_EMAIL_SUBJECT_CONTENT, 'Welcome!'), (param_list.ParamName.SIGNUP_EMAIL_BODY_CONTENT, EMAIL_BODY), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ] ) def test_contents_of_signup_email_are_correct(self) -> None: platform_parameter_registry.Registry.update_platform_parameter( param_list.ParamName.EMAIL_SENDER_NAME.value, self.admin_id, 'Update sender name', [ platform_parameter_domain.PlatformParameterRule.from_dict( { 'filters': [ { 'type': 'platform_type', 'conditions': [['=', 'Web']], } ], 'value_when_matched': 'Email Sender', } ) ], platform_parameter_registry.Registry.get_platform_parameter( param_list.ParamName.EMAIL_SENDER_NAME.value ).default_value, ) platform_parameter_registry.Registry.update_platform_parameter( param_list.ParamName.EMAIL_FOOTER.value, self.admin_id, 'Update email footer', [ platform_parameter_domain.PlatformParameterRule.from_dict( { 'filters': [ { 'type': 'platform_type', 'conditions': [['=', 'Web']], } ], 'value_when_matched': self.new_footer, } ) ], platform_parameter_registry.Registry.get_platform_parameter( param_list.ParamName.EMAIL_FOOTER.value ).default_value, ) self._set_signup_email_content_platform_parameter( self.new_email_subject_content, self.new_email_body_content ) self.login(self.EDITOR_EMAIL) self.get_html_response('%s?return_url=/' % feconf.SIGNUP_URL) csrf_token = self.get_new_csrf_token() self.post_json( feconf.SIGNUP_DATA_URL, { 'agreed_to_terms': True, 'username': self.EDITOR_USERNAME, 'default_dashboard': constants.DASHBOARD_TYPE_LEARNER, 'can_receive_email_updates': ( feconf.DEFAULT_EMAIL_UPDATES_PREFERENCE ), }, csrf_token=csrf_token, ) # Check that an email was sent with the correct content. messages = self._get_sent_email_messages(self.EDITOR_EMAIL) self.assertEqual(1, len(messages)) self.assertEqual( messages[0].sender, 'Email Sender <noreply@example.com>' ) self.assertEqual(messages[0].to, [self.EDITOR_EMAIL]) self.assertEqual(messages[0].subject, 'Welcome!') self.assertEqual(messages[0].body, self.expected_text_email_content) self.assertEqual(messages[0].html, self.expected_html_email_content) self._reset_the_email_platform_params_value() self._reset_signup_email_content_platform_parameters() @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_SENDER_NAME, 'Email Sender'), (param_list.ParamName.EMAIL_FOOTER, NEW_FOOTER), (param_list.ParamName.SIGNUP_EMAIL_SUBJECT_CONTENT, 'Welcome!'), (param_list.ParamName.SIGNUP_EMAIL_BODY_CONTENT, EMAIL_BODY), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ] ) def test_email_only_sent_once_for_repeated_signups_by_same_user( self, ) -> None: self._set_signup_email_content_platform_parameter( self.new_email_subject_content, self.new_email_body_content ) self.login(self.EDITOR_EMAIL) self.get_html_response('%s?return_url=/' % feconf.SIGNUP_URL) csrf_token = self.get_new_csrf_token() self.post_json( feconf.SIGNUP_DATA_URL, { 'agreed_to_terms': True, 'username': self.EDITOR_USERNAME, 'default_dashboard': constants.DASHBOARD_TYPE_LEARNER, 'can_receive_email_updates': ( feconf.DEFAULT_EMAIL_UPDATES_PREFERENCE ), }, csrf_token=csrf_token, ) # Check that an email was sent. messages = self._get_sent_email_messages(self.EDITOR_EMAIL) self.assertEqual(1, len(messages)) # Send a second POST request. self.post_json( feconf.SIGNUP_DATA_URL, { 'agreed_to_terms': True, 'username': self.EDITOR_USERNAME, 'default_dashboard': constants.DASHBOARD_TYPE_LEARNER, 'can_receive_email_updates': ( feconf.DEFAULT_EMAIL_UPDATES_PREFERENCE ), }, csrf_token=csrf_token, ) # Check that no new email was sent. messages = self._get_sent_email_messages(self.EDITOR_EMAIL) self.assertEqual(1, len(messages)) self._reset_signup_email_content_platform_parameters() @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_SENDER_NAME, 'Email Sender'), (param_list.ParamName.EMAIL_FOOTER, NEW_FOOTER), (param_list.ParamName.SIGNUP_EMAIL_SUBJECT_CONTENT, 'Welcome!'), (param_list.ParamName.SIGNUP_EMAIL_BODY_CONTENT, EMAIL_BODY), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ] ) def test_email_only_sent_if_signup_was_successful(self) -> None: self._set_signup_email_content_platform_parameter( self.new_email_subject_content, self.new_email_body_content ) self.login(self.EDITOR_EMAIL) self.get_html_response('%s?return_url=/' % feconf.SIGNUP_URL) csrf_token = self.get_new_csrf_token() self.post_json( feconf.SIGNUP_DATA_URL, { 'agreed_to_terms': True, 'username': 'BadUsername!!!', 'default_dashboard': constants.DASHBOARD_TYPE_LEARNER, 'can_receive_email_updates': ( feconf.DEFAULT_EMAIL_UPDATES_PREFERENCE ), }, csrf_token=csrf_token, expected_status_int=400, ) # Check that no email was sent. messages = self._get_sent_email_messages(self.EDITOR_EMAIL) self.assertEqual(0, len(messages)) # Redo the signup process with a good username. self.post_json( feconf.SIGNUP_DATA_URL, { 'agreed_to_terms': True, 'username': self.EDITOR_USERNAME, 'default_dashboard': constants.DASHBOARD_TYPE_LEARNER, 'can_receive_email_updates': ( feconf.DEFAULT_EMAIL_UPDATES_PREFERENCE ), }, csrf_token=csrf_token, ) # Check that a new email was sent. messages = self._get_sent_email_messages(self.EDITOR_EMAIL) self.assertEqual(1, len(messages)) self._reset_signup_email_content_platform_parameters() @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_SENDER_NAME, 'Email Sender'), (param_list.ParamName.EMAIL_FOOTER, NEW_FOOTER), (param_list.ParamName.SIGNUP_EMAIL_SUBJECT_CONTENT, 'Welcome!'), (param_list.ParamName.SIGNUP_EMAIL_BODY_CONTENT, EMAIL_BODY), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ] ) def test_record_of_sent_email_is_written_to_datastore(self) -> None: platform_parameter_registry.Registry.update_platform_parameter( param_list.ParamName.EMAIL_SENDER_NAME.value, self.admin_id, 'Update sender name', [ platform_parameter_domain.PlatformParameterRule.from_dict( { 'filters': [ { 'type': 'platform_type', 'conditions': [['=', 'Web']], } ], 'value_when_matched': 'Email Sender', } ) ], platform_parameter_registry.Registry.get_platform_parameter( param_list.ParamName.EMAIL_SENDER_NAME.value ).default_value, ) platform_parameter_registry.Registry.update_platform_parameter( param_list.ParamName.EMAIL_FOOTER.value, self.admin_id, 'Update email footer', [ platform_parameter_domain.PlatformParameterRule.from_dict( { 'filters': [ { 'type': 'platform_type', 'conditions': [['=', 'Web']], } ], 'value_when_matched': self.new_footer, } ) ], platform_parameter_registry.Registry.get_platform_parameter( param_list.ParamName.EMAIL_FOOTER.value ).default_value, ) self._set_signup_email_content_platform_parameter( self.new_email_subject_content, self.new_email_body_content ) all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) self.assertEqual(len(all_models), 0) self.login(self.EDITOR_EMAIL) self.get_html_response('%s?return_url=/' % feconf.SIGNUP_URL) csrf_token = self.get_new_csrf_token() self.post_json( feconf.SIGNUP_DATA_URL, { 'agreed_to_terms': True, 'username': self.EDITOR_USERNAME, 'default_dashboard': constants.DASHBOARD_TYPE_LEARNER, 'can_receive_email_updates': ( feconf.DEFAULT_EMAIL_UPDATES_PREFERENCE ), }, csrf_token=csrf_token, ) # Check that a new email was sent. messages = self._get_sent_email_messages(self.EDITOR_EMAIL) self.assertEqual(1, len(messages)) # Check that the content of this email was recorded in # SentEmailModel. all_models = email_models.SentEmailModel.get_all().fetch() self.assertEqual(len(all_models), 1) # Check that the contents of the model are correct. sent_email_model = all_models[0] self.assertEqual( sent_email_model.recipient_id, self.get_user_id_from_email(self.EDITOR_EMAIL), ) self.assertEqual(sent_email_model.recipient_email, self.EDITOR_EMAIL) self.assertEqual(sent_email_model.sender_id, feconf.SYSTEM_COMMITTER_ID) self.assertEqual( sent_email_model.sender_email, 'Email Sender <noreply@example.com>' ) self.assertEqual(sent_email_model.intent, feconf.EMAIL_INTENT_SIGNUP) self.assertEqual(sent_email_model.subject, 'Welcome!') self.assertEqual( sent_email_model.html_body, self.expected_html_email_content ) self._reset_the_email_platform_params_value() self._reset_signup_email_content_platform_parameters() class DuplicateEmailTests(test_utils.EmailTestBase): """Test that duplicate emails are not sent.""" NEW_FOOTER = ( 'You can change your email preferences via the ' '<a href="http://localhost:8181/preferences">Preferences</a> page.' ) NEW_EMAIL_SUBJECT = 'THIS IS A PLACEHOLDER.' PLACEHOLDER_HTML_BODY: Final = ( 'THIS IS A <b>PLACEHOLDER</b> AND SHOULD BE REPLACED.' ) def setUp(self) -> None: super().setUp() self.signup(self.NEW_USER_EMAIL, self.NEW_USER_USERNAME) self.new_user_id = self.get_user_id_from_email(self.NEW_USER_EMAIL) self.signup(self.CURRICULUM_ADMIN_EMAIL, self.CURRICULUM_ADMIN_USERNAME) self.admin_id = self.get_user_id_from_email(self.CURRICULUM_ADMIN_EMAIL) self.set_curriculum_admins([self.CURRICULUM_ADMIN_USERNAME]) self.new_footer = ( 'You can change your email preferences via the ' '<a href="http://localhost:8181/preferences">Preferences</a> page.' ) self.new_email_subject = 'THIS IS A PLACEHOLDER.' self.new_email_html_body = 'Hi %s,<br><br>%s<br><br>%s' % ( self.NEW_USER_USERNAME, 'THIS IS A <b>PLACEHOLDER</b> AND SHOULD BE REPLACED.', self.new_footer, ) self.system_email_address = 'system@example.com' def _generate_hash_for_tests( unused_cls: Type[test_utils.TestBase], unused_recipient_id: str, unused_email_subject: str, unused_email_body: str, ) -> str: """Returns the generated hash for tests.""" return 'Email Hash' self.generate_hash_ctx = self.swap( email_models.SentEmailModel, '_generate_hash', types.MethodType( _generate_hash_for_tests, email_models.SentEmailModel ), ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_SENDER_NAME, 'Name'), (param_list.ParamName.EMAIL_FOOTER, NEW_FOOTER), ( param_list.ParamName.SIGNUP_EMAIL_SUBJECT_CONTENT, NEW_EMAIL_SUBJECT, ), ( param_list.ParamName.SIGNUP_EMAIL_BODY_CONTENT, PLACEHOLDER_HTML_BODY, ), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ] ) def test_send_email_does_not_resend_if_same_hash_exists(self) -> None: duplicate_email_ctx = self.swap( feconf, 'DUPLICATE_EMAIL_INTERVAL_MINS', 1000 ) log_new_error_counter = test_utils.CallCounter(logging.error) log_new_error_ctx = self.swap(logging, 'error', log_new_error_counter) with self.capture_logging(min_level=logging.ERROR) as logs: with duplicate_email_ctx, log_new_error_ctx: all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) self.assertEqual(len(all_models), 0) cleaned_html_body = html_cleaner.clean(self.new_email_html_body) raw_plaintext_body = ( cleaned_html_body.replace('<br/>', '\n') .replace('<br>', '\n') .replace('<li>', '<li>- ') .replace('</p><p>', '</p>\n<p>') ) cleaned_plaintext_body = html_cleaner.strip_html_tags( raw_plaintext_body ) email_models.SentEmailModel.create( self.new_user_id, self.NEW_USER_EMAIL, feconf.SYSTEM_COMMITTER_ID, self.system_email_address, feconf.EMAIL_INTENT_SIGNUP, self.new_email_subject, cleaned_plaintext_body, utils.get_current_utc_datetime(), ) # Check that the content of this email was recorded in # SentEmailModel. all_models = email_models.SentEmailModel.get_all().fetch() self.assertEqual(len(all_models), 1) email_manager.send_post_signup_email( self.new_user_id, test_for_duplicate_email=True ) # An error should be recorded in the logs. self.assertEqual(log_new_error_counter.times_called, 1) self.assertRegex(logs[0], 'Duplicate email') # Check that a new email was not sent. messages = self._get_sent_email_messages(self.NEW_USER_EMAIL) self.assertEqual(0, len(messages)) # Check that the content of this email was not recorded in # SentEmailModel. all_models = email_models.SentEmailModel.get_all().fetch() self.assertEqual(len(all_models), 1) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_SENDER_NAME, 'Name'), (param_list.ParamName.EMAIL_FOOTER, NEW_FOOTER), ( param_list.ParamName.SIGNUP_EMAIL_SUBJECT_CONTENT, NEW_EMAIL_SUBJECT, ), ( param_list.ParamName.SIGNUP_EMAIL_BODY_CONTENT, PLACEHOLDER_HTML_BODY, ), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), ] ) def test_send_email_does_not_resend_within_duplicate_interval(self) -> None: duplicate_email_ctx = self.swap( feconf, 'DUPLICATE_EMAIL_INTERVAL_MINS', 2 ) log_new_error_counter = test_utils.CallCounter(logging.error) log_new_error_ctx = self.swap(logging, 'error', log_new_error_counter) with self.capture_logging(min_level=logging.ERROR) as logs: with duplicate_email_ctx, log_new_error_ctx: all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) self.assertEqual(len(all_models), 0) email_manager._send_email( # pylint: disable=protected-access self.new_user_id, feconf.SYSTEM_COMMITTER_ID, feconf.EMAIL_INTENT_SIGNUP, 'Email Subject', 'Email Body', self.system_email_address, ) # Check that a new email was sent. messages = self._get_sent_email_messages(self.NEW_USER_EMAIL) self.assertEqual(1, len(messages)) # Check that the content of this email was recorded in # SentEmailModel. all_models = email_models.SentEmailModel.get_all().fetch() self.assertEqual(len(all_models), 1) # No error should be recorded in the logs. self.assertEqual(log_new_error_counter.times_called, 0) email_manager._send_email( # pylint: disable=protected-access self.new_user_id, feconf.SYSTEM_COMMITTER_ID, feconf.EMAIL_INTENT_SIGNUP, 'Email Subject', 'Email Body', self.system_email_address, ) # An error should be recorded in the logs. self.assertEqual(log_new_error_counter.times_called, 1) self.assertRegex(logs[0], 'Duplicate email') # Check that a new email was not sent. messages = self._get_sent_email_messages(self.NEW_USER_EMAIL) self.assertEqual(1, len(messages)) # Check that the content of this email was not recorded in # SentEmailModel. all_models = email_models.SentEmailModel.get_all().fetch() self.assertEqual(len(all_models), 1) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_SENDER_NAME, 'Name'), (param_list.ParamName.EMAIL_FOOTER, NEW_FOOTER), ( param_list.ParamName.SIGNUP_EMAIL_SUBJECT_CONTENT, NEW_EMAIL_SUBJECT, ), ( param_list.ParamName.SIGNUP_EMAIL_BODY_CONTENT, PLACEHOLDER_HTML_BODY, ), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ] ) def test_sending_email_with_different_recipient_but_same_hash(self) -> None: """Hash for both messages is same but recipients are different.""" duplicate_email_ctx = self.swap( feconf, 'DUPLICATE_EMAIL_INTERVAL_MINS', 2 ) with duplicate_email_ctx, self.generate_hash_ctx: all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) self.assertEqual(len(all_models), 0) email_models.SentEmailModel.create( 'recipient_id', self.NEW_USER_EMAIL, feconf.SYSTEM_COMMITTER_ID, self.system_email_address, feconf.EMAIL_INTENT_SIGNUP, self.new_email_subject, self.new_email_html_body, utils.get_current_utc_datetime(), ) # Check that the content of this email was recorded in # SentEmailModel. all_models = email_models.SentEmailModel.get_all().fetch() self.assertEqual(len(all_models), 1) email_manager.send_post_signup_email( self.new_user_id, test_for_duplicate_email=True ) # Check that a new email was sent. messages = self._get_sent_email_messages(self.NEW_USER_EMAIL) self.assertEqual(1, len(messages)) # Check that the content of this email was recorded in # SentEmailModel. all_models = email_models.SentEmailModel.get_all().fetch() self.assertEqual(len(all_models), 2) # Check that the contents of the model are correct. sent_email_model1 = all_models[0] sent_email_model2 = all_models[1] self.assertEqual( sent_email_model1.email_hash, sent_email_model2.email_hash ) self.assertNotEqual( sent_email_model1.recipient_id, sent_email_model2.recipient_id ) self.assertEqual( sent_email_model1.subject, sent_email_model2.subject ) self.assertEqual( sent_email_model1.html_body, sent_email_model2.html_body ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_SENDER_NAME, 'Name'), (param_list.ParamName.EMAIL_FOOTER, NEW_FOOTER), ( param_list.ParamName.SIGNUP_EMAIL_SUBJECT_CONTENT, NEW_EMAIL_SUBJECT, ), ( param_list.ParamName.SIGNUP_EMAIL_BODY_CONTENT, PLACEHOLDER_HTML_BODY, ), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ] ) def test_sending_email_with_different_subject_but_same_hash(self) -> None: """Hash for both messages is same but subjects are different.""" duplicate_email_ctx = self.swap( feconf, 'DUPLICATE_EMAIL_INTERVAL_MINS', 2 ) with duplicate_email_ctx, self.generate_hash_ctx: all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) self.assertEqual(len(all_models), 0) email_models.SentEmailModel.create( self.new_user_id, self.NEW_USER_EMAIL, feconf.SYSTEM_COMMITTER_ID, self.system_email_address, feconf.EMAIL_INTENT_SIGNUP, '%s%s' % (self.new_email_subject, 1), self.new_email_html_body, utils.get_current_utc_datetime(), ) # Check that the content of this email was recorded in # SentEmailModel. all_models = email_models.SentEmailModel.get_all().fetch() self.assertEqual(len(all_models), 1) email_manager.send_post_signup_email( self.new_user_id, test_for_duplicate_email=True ) # Check that a new email was sent. messages = self._get_sent_email_messages(self.NEW_USER_EMAIL) self.assertEqual(1, len(messages)) # Check that the content of this email was recorded in # SentEmailModel. all_models = email_models.SentEmailModel.get_all().fetch() self.assertEqual(len(all_models), 2) # Check that the contents of the model are correct. sent_email_model1 = all_models[0] sent_email_model2 = all_models[1] self.assertEqual( sent_email_model1.email_hash, sent_email_model2.email_hash ) self.assertEqual( sent_email_model1.recipient_id, sent_email_model2.recipient_id ) self.assertNotEqual( sent_email_model1.subject, sent_email_model2.subject ) self.assertEqual( sent_email_model1.html_body, sent_email_model2.html_body ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_SENDER_NAME, 'Name'), (param_list.ParamName.EMAIL_FOOTER, NEW_FOOTER), ( param_list.ParamName.SIGNUP_EMAIL_SUBJECT_CONTENT, NEW_EMAIL_SUBJECT, ), ( param_list.ParamName.SIGNUP_EMAIL_BODY_CONTENT, PLACEHOLDER_HTML_BODY, ), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ] ) def test_sending_email_with_different_body_but_same_hash(self) -> None: """Hash for both messages is same but body is different.""" duplicate_email_ctx = self.swap( feconf, 'DUPLICATE_EMAIL_INTERVAL_MINS', 2 ) with duplicate_email_ctx, self.generate_hash_ctx: all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) self.assertEqual(len(all_models), 0) email_models.SentEmailModel.create( self.new_user_id, self.NEW_USER_EMAIL, feconf.SYSTEM_COMMITTER_ID, self.system_email_address, feconf.EMAIL_INTENT_SIGNUP, self.new_email_subject, '%s%s' % (self.new_email_html_body, 1), utils.get_current_utc_datetime(), ) # Check that the content of this email was recorded in # SentEmailModel. all_models = email_models.SentEmailModel.get_all().fetch() self.assertEqual(len(all_models), 1) email_manager.send_post_signup_email( self.new_user_id, test_for_duplicate_email=True ) # Check that a new email was sent. messages = self._get_sent_email_messages(self.NEW_USER_EMAIL) self.assertEqual(1, len(messages)) # Check that the content of this email was recorded in # SentEmailModel. all_models = email_models.SentEmailModel.get_all().fetch() self.assertEqual(len(all_models), 2) # Check that the contents of the model are correct. sent_email_model1 = all_models[0] sent_email_model2 = all_models[1] self.assertEqual( sent_email_model1.email_hash, sent_email_model2.email_hash ) self.assertEqual( sent_email_model1.recipient_id, sent_email_model2.recipient_id ) self.assertEqual( sent_email_model1.subject, sent_email_model2.subject ) self.assertNotEqual( sent_email_model1.html_body, sent_email_model2.html_body ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_SENDER_NAME, 'Name'), (param_list.ParamName.EMAIL_FOOTER, NEW_FOOTER), ( param_list.ParamName.SIGNUP_EMAIL_SUBJECT_CONTENT, NEW_EMAIL_SUBJECT, ), ( param_list.ParamName.SIGNUP_EMAIL_BODY_CONTENT, PLACEHOLDER_HTML_BODY, ), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ] ) def test_duplicate_emails_are_sent_after_some_time_has_elapsed( self, ) -> None: duplicate_email_ctx = self.swap( feconf, 'DUPLICATE_EMAIL_INTERVAL_MINS', 2 ) with duplicate_email_ctx: all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) self.assertEqual(len(all_models), 0) email_sent_time = ( utils.get_current_utc_datetime() - datetime.timedelta(minutes=4) ) email_models.SentEmailModel.create( self.new_user_id, self.NEW_USER_EMAIL, feconf.SYSTEM_COMMITTER_ID, self.system_email_address, feconf.EMAIL_INTENT_SIGNUP, self.new_email_subject, self.new_email_html_body, email_sent_time, ) # Check that the content of this email was recorded in # SentEmailModel. all_models = email_models.SentEmailModel.get_all().fetch() self.assertEqual(len(all_models), 1) email_sent_time = ( utils.get_current_utc_datetime() - datetime.timedelta(minutes=2) ) email_models.SentEmailModel.create( self.new_user_id, self.NEW_USER_EMAIL, feconf.SYSTEM_COMMITTER_ID, self.system_email_address, feconf.EMAIL_INTENT_SIGNUP, self.new_email_subject, self.new_email_html_body, email_sent_time, ) # Check that the content of this email was recorded in # SentEmailModel. all_models = email_models.SentEmailModel.get_all().fetch() self.assertEqual(len(all_models), 2) email_manager.send_post_signup_email( self.new_user_id, test_for_duplicate_email=True ) # Check that a new email was sent. messages = self._get_sent_email_messages(self.NEW_USER_EMAIL) self.assertEqual(1, len(messages)) # Check that the content of this email was recorded in # SentEmailModel. all_models = email_models.SentEmailModel.get_all().fetch() self.assertEqual(len(all_models), 3) # Check that the contents of the model are correct. sent_email_model1 = all_models[0] sent_email_model2 = all_models[1] sent_email_model3 = all_models[2] self.assertEqual( sent_email_model1.email_hash, sent_email_model2.email_hash ) self.assertEqual( sent_email_model1.email_hash, sent_email_model3.email_hash ) class FeedbackMessageBatchEmailTests(test_utils.EmailTestBase): def setUp(self) -> None: super().setUp() self.signup(self.EDITOR_EMAIL, self.EDITOR_USERNAME) self.editor_id = self.get_user_id_from_email(self.EDITOR_EMAIL) self.exploration = self.save_new_default_exploration( 'A', self.editor_id, title='Title' ) self.expected_email_subject = ( 'You\'ve received 3 new messages on your explorations' ) self.can_send_feedback_email_ctx = self.swap( feconf, 'CAN_SEND_TRANSACTIONAL_EMAILS', True ) self.can_not_send_feedback_email_ctx = self.swap( feconf, 'CAN_SEND_TRANSACTIONAL_EMAILS', False ) @test_utils.set_platform_parameters( [(param_list.ParamName.SERVER_CAN_SEND_EMAILS, False)] ) def test_email_not_sent_if_server_can_send_emails_is_false(self) -> None: feedback_messages: Dict[str, email_manager.FeedbackMessagesDict] = { self.exploration.id: { 'title': self.exploration.title, 'messages': ['Message 1.1', 'Message 1.2', 'Message 1.3'], } } email_manager.send_feedback_message_email( self.editor_id, feedback_messages ) # Check that email is not sent. messages = self._get_sent_email_messages(self.EDITOR_EMAIL) self.assertEqual(len(messages), 0) @test_utils.set_platform_parameters( [(param_list.ParamName.SERVER_CAN_SEND_EMAILS, True)] ) def test_email_not_sent_if_can_send_transactional_emails_is_false( self, ) -> None: feedback_messages: Dict[str, email_manager.FeedbackMessagesDict] = { self.exploration.id: { 'title': self.exploration.title, 'messages': ['Message 1.1', 'Message 1.2', 'Message 1.3'], } } with self.can_not_send_feedback_email_ctx: email_manager.send_feedback_message_email( self.editor_id, feedback_messages ) # Check that email is not sent. messages = self._get_sent_email_messages(self.EDITOR_EMAIL) self.assertEqual(len(messages), 0) @test_utils.set_platform_parameters( [(param_list.ParamName.SERVER_CAN_SEND_EMAILS, True)] ) def test_that_email_not_sent_if_feedback_messages_are_empty(self) -> None: feedback_messages: Dict[str, email_manager.FeedbackMessagesDict] = {} with self.can_send_feedback_email_ctx: email_manager.send_feedback_message_email( self.editor_id, feedback_messages ) # Check that email is not sent. messages = self._get_sent_email_messages(self.EDITOR_EMAIL) self.assertEqual(len(messages), 0) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ] ) def test_correct_email_body_is_sent(self) -> None: expected_email_html_body = ( 'Hi editor,<br>' '<br>' 'You\'ve received 3 new messages on your Oppia explorations:<br>' '<ul>' '<li><a href="https://www.oppia.org/create/A#/feedback">Title</a>:' '<br>' '<ul><li>Message 1.1<br></li>' '<li>Message 1.2<br></li>' '<li>Message 1.3<br></li>' '</ul></li></ul>' 'You can view and reply to your messages from your ' '<a href="https://www.oppia.org/creator-dashboard">dashboard</a>.' '<br>' '<br>Thanks, and happy teaching!<br>' '<br>' 'Best wishes,<br>' 'The Oppia Team<br>' '<br>' 'You can change your email preferences via the ' '<a href="http://localhost:8181/preferences">Preferences</a> page.' ) expected_email_text_body = ( 'Hi editor,\n' '\n' 'You\'ve received 3 new messages on your Oppia explorations:\n' '- Title:\n' '- Message 1.1\n' '- Message 1.2\n' '- Message 1.3\n' 'You can view and reply to your messages from your dashboard.\n' '\n' 'Thanks, and happy teaching!\n' '\n' 'Best wishes,\n' 'The Oppia Team\n' '\n' 'You can change your email preferences via the Preferences page.' ) feedback_messages: Dict[str, email_manager.FeedbackMessagesDict] = { self.exploration.id: { 'title': self.exploration.title, 'messages': ['Message 1.1', 'Message 1.2', 'Message 1.3'], } } with self.can_send_feedback_email_ctx: email_manager.send_feedback_message_email( self.editor_id, feedback_messages ) # Check that email body is correct. messages = self._get_sent_email_messages(self.EDITOR_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) self.assertEqual(messages[0].body, expected_email_text_body) # Check that email model is correct. all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) self.assertEqual(len(all_models), 1) sent_email_model = all_models[0] self.assertEqual(sent_email_model.recipient_id, self.editor_id) self.assertEqual( sent_email_model.recipient_email, self.EDITOR_EMAIL ) self.assertEqual( sent_email_model.sender_id, feconf.SYSTEM_COMMITTER_ID ) self.assertEqual( sent_email_model.sender_email, 'Site Admin <noreply@example.com>', ) self.assertEqual( sent_email_model.intent, feconf.EMAIL_INTENT_FEEDBACK_MESSAGE_NOTIFICATION, ) self.assertEqual( sent_email_model.subject, self.expected_email_subject ) class SuggestionEmailTests(test_utils.EmailTestBase): def setUp(self) -> None: super().setUp() self.signup(self.EDITOR_EMAIL, self.EDITOR_USERNAME) self.editor_id = self.get_user_id_from_email(self.EDITOR_EMAIL) self.signup(self.NEW_USER_EMAIL, self.NEW_USER_USERNAME) self.new_user_id = self.get_user_id_from_email(self.NEW_USER_EMAIL) self.exploration = self.save_new_default_exploration( 'A', self.editor_id, title='Title' ) self.recipient_list = [self.editor_id] self.can_send_feedback_email_ctx = self.swap( feconf, 'CAN_SEND_TRANSACTIONAL_EMAILS', True ) self.can_not_send_feedback_email_ctx = self.swap( feconf, 'CAN_SEND_TRANSACTIONAL_EMAILS', False ) @test_utils.set_platform_parameters( [(param_list.ParamName.SERVER_CAN_SEND_EMAILS, False)] ) def test_email_not_sent_if_can_send_emails_is_false(self) -> None: email_manager.send_suggestion_email( self.exploration.title, self.exploration.id, self.new_user_id, self.recipient_list, ) # Check that email is not sent. messages = self._get_sent_email_messages(self.EDITOR_EMAIL) self.assertEqual(len(messages), 0) @test_utils.set_platform_parameters( [(param_list.ParamName.SERVER_CAN_SEND_EMAILS, True)] ) def test_email_not_sent_if_can_send_transactional_emails_is_false( self, ) -> None: with self.can_not_send_feedback_email_ctx: email_manager.send_suggestion_email( self.exploration.title, self.exploration.id, self.new_user_id, self.recipient_list, ) # Check that email is not sent. messages = self._get_sent_email_messages(self.EDITOR_EMAIL) self.assertEqual(len(messages), 0) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ] ) def test_that_suggestion_emails_are_correct(self) -> None: expected_email_subject = 'New suggestion for "Title"' expected_email_html_body = ( 'Hi editor,<br>' 'newuser has submitted a new suggestion for your Oppia ' 'exploration, ' '<a href="https://www.oppia.org/create/A">"Title"</a>.<br>' 'You can accept or reject this suggestion by visiting the ' '<a href="https://www.oppia.org/create/A#/feedback">' 'feedback page</a> ' 'for your exploration.<br>' '<br>' 'Thanks!<br>' '- The Oppia Team<br>' '<br>' 'You can change your email preferences via the ' '<a href="http://localhost:8181/preferences">Preferences</a> page.' ) expected_email_text_body = ( 'Hi editor,\n' 'newuser has submitted a new suggestion for your Oppia ' 'exploration, "Title".\n' 'You can accept or reject this suggestion by visiting the ' 'feedback page for your exploration.\n' '\n' 'Thanks!\n' '- The Oppia Team\n' '\n' 'You can change your email preferences via the Preferences page.' ) with self.can_send_feedback_email_ctx: email_manager.send_suggestion_email( self.exploration.title, self.exploration.id, self.new_user_id, self.recipient_list, ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.EDITOR_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) self.assertEqual(messages[0].body, expected_email_text_body) # Make sure correct email model is stored. all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) sent_email_model = all_models[0] self.assertEqual(sent_email_model.subject, expected_email_subject) self.assertEqual(sent_email_model.recipient_id, self.editor_id) self.assertEqual( sent_email_model.recipient_email, self.EDITOR_EMAIL ) self.assertEqual( sent_email_model.sender_id, feconf.SYSTEM_COMMITTER_ID ) self.assertEqual( sent_email_model.sender_email, 'Site Admin <noreply@example.com>', ) self.assertEqual( sent_email_model.intent, feconf.EMAIL_INTENT_SUGGESTION_NOTIFICATION, ) class SubscriptionEmailTests(test_utils.EmailTestBase): def setUp(self) -> None: super().setUp() self.signup(self.EDITOR_EMAIL, self.EDITOR_USERNAME) self.editor_id = self.get_user_id_from_email(self.EDITOR_EMAIL) self.signup(self.NEW_USER_EMAIL, self.NEW_USER_USERNAME) self.new_user_id = self.get_user_id_from_email(self.NEW_USER_EMAIL) self.exploration = self.save_new_default_exploration( 'A', self.editor_id, title='Title' ) subscription_services.subscribe_to_creator( self.new_user_id, self.editor_id ) self.can_send_subscription_email_ctx = self.swap( feconf, 'CAN_SEND_TRANSACTIONAL_EMAILS', True ) self.can_not_send_subscription_email_ctx = self.swap( feconf, 'CAN_SEND_TRANSACTIONAL_EMAILS', False ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, False), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), ] ) def test_email_not_sent_if_server_can_send_emails_is_false(self) -> None: email_manager.send_emails_to_subscribers( self.editor_id, self.exploration.id, self.exploration.title ) messages = self._get_sent_email_messages(self.NEW_USER_EMAIL) self.assertEqual(len(messages), 0) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), ] ) def test_that_email_not_sent_if_can_send_transactional_emails_is_false( self, ) -> None: with self.can_not_send_subscription_email_ctx: email_manager.send_emails_to_subscribers( self.editor_id, self.exploration.id, self.exploration.title ) messages = self._get_sent_email_messages(self.NEW_USER_EMAIL) self.assertEqual(len(messages), 0) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ] ) def test_that_subscription_emails_are_correct(self) -> None: expected_email_subject = 'editor has published a new exploration!' expected_email_html_body = ( 'Hi newuser,<br>' '<br>' 'editor has published a new exploration! You can play it here: ' '<a href="https://www.oppia.org/explore/A">Title</a><br>' '<br>' 'Thanks, and happy learning!<br>' '<br>' 'Best wishes,<br>' '- The Oppia Team<br>' '<br>' 'You can change your email preferences via the ' '<a href="http://localhost:8181/preferences">Preferences</a> page.' ) expected_email_text_body = ( 'Hi newuser,\n' '\n' 'editor has published a new exploration! You can play it here: ' 'Title\n' '\n' 'Thanks, and happy learning!\n' '\n' 'Best wishes,\n' '- The Oppia Team\n' '\n' 'You can change your email preferences via the Preferences page.' ) with self.can_send_subscription_email_ctx: email_manager.send_emails_to_subscribers( self.editor_id, self.exploration.id, self.exploration.title ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.NEW_USER_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) self.assertEqual(messages[0].body, expected_email_text_body) # Make sure correct email model is stored. all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) sent_email_model = all_models[0] self.assertEqual(sent_email_model.subject, expected_email_subject) self.assertEqual(sent_email_model.recipient_id, self.new_user_id) self.assertEqual( sent_email_model.recipient_email, self.NEW_USER_EMAIL ) self.assertEqual( sent_email_model.sender_id, feconf.SYSTEM_COMMITTER_ID ) self.assertEqual( sent_email_model.sender_email, 'Site Admin <noreply@example.com>', ) self.assertEqual( sent_email_model.intent, feconf.EMAIL_INTENT_SUBSCRIPTION_NOTIFICATION, ) class FeedbackMessageInstantEmailTests(test_utils.EmailTestBase): def setUp(self) -> None: super().setUp() self.signup(self.EDITOR_EMAIL, self.EDITOR_USERNAME) self.editor_id = self.get_user_id_from_email(self.EDITOR_EMAIL) self.signup(self.NEW_USER_EMAIL, self.NEW_USER_USERNAME) self.new_user_id = self.get_user_id_from_email(self.NEW_USER_EMAIL) self.exploration = self.save_new_default_exploration( 'A', self.editor_id, title='Title' ) self.recipient_list = [self.editor_id] self.can_send_feedback_email_ctx = self.swap( feconf, 'CAN_SEND_TRANSACTIONAL_EMAILS', True ) self.can_not_send_feedback_email_ctx = self.swap( feconf, 'CAN_SEND_TRANSACTIONAL_EMAILS', False ) @test_utils.set_platform_parameters( [(param_list.ParamName.SERVER_CAN_SEND_EMAILS, False)] ) def test_email_not_sent_if_server_can_send_emails_is_false(self) -> None: email_manager.send_instant_feedback_message_email( self.new_user_id, self.editor_id, 'editor message', 'New Oppia message in "a subject"', self.exploration.title, self.exploration.id, 'a subject', ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.NEW_USER_EMAIL) self.assertEqual(len(messages), 0) @test_utils.set_platform_parameters( [(param_list.ParamName.SERVER_CAN_SEND_EMAILS, True)] ) def test_email_not_sent_if_can_send_transactional_emails_is_false( self, ) -> None: with self.can_not_send_feedback_email_ctx: email_manager.send_instant_feedback_message_email( self.new_user_id, self.editor_id, 'editor message', 'New Oppia message in "a subject"', self.exploration.title, self.exploration.id, 'a subject', ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.NEW_USER_EMAIL) self.assertEqual(len(messages), 0) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ] ) def test_that_feedback_message_emails_are_correct(self) -> None: expected_email_subject = 'New Oppia message in "a subject"' expected_email_html_body = ( 'Hi newuser,<br><br>' 'New update to thread "a subject" on ' '<a href="https://www.oppia.org/create/A#/feedback">Title</a>:<br>' '<ul><li>editor: editor message<br></li></ul>' '(You received this message because you are a ' 'participant in this thread.)<br><br>' 'Best wishes,<br>' 'The Oppia team<br>' '<br>' 'You can change your email preferences via the ' '<a href="http://localhost:8181/preferences">Preferences</a> page.' ) expected_email_text_body = ( 'Hi newuser,\n' '\n' 'New update to thread "a subject" on Title:\n' '- editor: editor message\n' '(You received this message because you are a' ' participant in this thread.)\n' '\n' 'Best wishes,\n' 'The Oppia team\n' '\n' 'You can change your email preferences via the Preferences page.' ) with self.can_send_feedback_email_ctx: email_manager.send_instant_feedback_message_email( self.new_user_id, self.editor_id, 'editor message', 'New Oppia message in "a subject"', self.exploration.title, self.exploration.id, 'a subject', ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.NEW_USER_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) self.assertEqual(messages[0].body, expected_email_text_body) # Make sure correct email model is stored. all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) sent_email_model = all_models[0] self.assertEqual(sent_email_model.subject, expected_email_subject) self.assertEqual(sent_email_model.recipient_id, self.new_user_id) self.assertEqual( sent_email_model.recipient_email, self.NEW_USER_EMAIL ) self.assertEqual( sent_email_model.sender_id, feconf.SYSTEM_COMMITTER_ID ) self.assertEqual( sent_email_model.sender_email, 'Site Admin <noreply@example.com>', ) self.assertEqual( sent_email_model.intent, feconf.EMAIL_INTENT_FEEDBACK_MESSAGE_NOTIFICATION, ) class FlagExplorationEmailTest(test_utils.EmailTestBase): """Test that emails are sent to moderators when explorations are flagged.""" def setUp(self) -> None: super().setUp() self.signup(self.EDITOR_EMAIL, self.EDITOR_USERNAME) self.editor_id = self.get_user_id_from_email(self.EDITOR_EMAIL) self.signup(self.NEW_USER_EMAIL, self.NEW_USER_USERNAME) self.new_user_id = self.get_user_id_from_email(self.NEW_USER_EMAIL) self.signup(self.MODERATOR_EMAIL, self.MODERATOR_USERNAME) self.moderator_id = self.get_user_id_from_email(self.MODERATOR_EMAIL) self.moderator2_email = 'moderator2@example.com' self.moderator2_username = 'moderator2' self.signup(self.moderator2_email, self.moderator2_username) self.moderator2_id = self.get_user_id_from_email(self.moderator2_email) self.set_moderators([self.moderator2_username, self.MODERATOR_USERNAME]) self.exploration = self.save_new_default_exploration( 'A', self.editor_id, title='Title' ) self.owner_ids = [self.editor_id] self.report_text = 'AD' @test_utils.set_platform_parameters( [(param_list.ParamName.SERVER_CAN_SEND_EMAILS, False)] ) def test_email_not_sent_if_server_can_send_emails_is_false(self) -> None: email_manager.send_flag_exploration_email( self.exploration.title, self.exploration.id, self.new_user_id, self.report_text, ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.MODERATOR_EMAIL) self.assertEqual(len(messages), 0) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ] ) def test_that_flag_exploration_emails_are_correct(self) -> None: expected_email_subject = 'Exploration flagged by user: "Title"' expected_email_html_body = ( 'Hello Moderator,<br>' 'newuser has flagged exploration "Title" on the following ' 'grounds: <br>' 'AD .<br>' 'You can modify the exploration by clicking ' '<a href="https://www.oppia.org/create/A">' 'here</a>.<br>' '<br>' 'Thanks!<br>' '- The Oppia Team<br>' '<br>' 'You can change your email preferences via the ' '<a href="http://localhost:8181/preferences">Preferences</a> page.' ) expected_email_text_body = ( 'Hello Moderator,\n' 'newuser has flagged exploration "Title" on the following ' 'grounds: \n' 'AD .\n' 'You can modify the exploration by clicking here.\n' '\n' 'Thanks!\n' '- The Oppia Team\n' '\n' 'You can change your email preferences via the Preferences page.' ) email_manager.send_flag_exploration_email( self.exploration.title, self.exploration.id, self.new_user_id, self.report_text, ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.MODERATOR_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) self.assertEqual(messages[0].body, expected_email_text_body) # Make sure correct email is sent to multiple moderators. messages = self._get_sent_email_messages(self.moderator2_email) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) self.assertEqual(messages[0].body, expected_email_text_body) # Make sure correct email models are stored. all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) sent_email_model = next( m for m in all_models if m.recipient_id == self.moderator_id ) self.assertEqual(sent_email_model.subject, expected_email_subject) self.assertEqual(sent_email_model.recipient_email, self.MODERATOR_EMAIL) self.assertEqual(sent_email_model.sender_id, feconf.SYSTEM_COMMITTER_ID) noreply_email_address = 'noreply@example.com' self.assertEqual( sent_email_model.sender_email, 'Site Admin <%s>' % noreply_email_address, ) self.assertEqual( sent_email_model.intent, feconf.EMAIL_INTENT_REPORT_BAD_CONTENT ) sent_email_model = next( m for m in all_models if m.recipient_id == self.moderator2_id ) self.assertEqual(sent_email_model.subject, expected_email_subject) self.assertEqual( sent_email_model.recipient_email, self.moderator2_email ) self.assertEqual(sent_email_model.sender_id, feconf.SYSTEM_COMMITTER_ID) self.assertEqual( sent_email_model.sender_email, 'Site Admin <%s>' % noreply_email_address, ) self.assertEqual( sent_email_model.intent, feconf.EMAIL_INTENT_REPORT_BAD_CONTENT ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), ( param_list.ParamName.EMAIL_FOOTER, 'You can change your email preferences via the ' '<a href="LINK_TO_PREFERENCES_PAGE">' 'Preferences</a> page.', ), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com', ), ( param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com', ), ( param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com', ), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, 'https://www.oppia.org', ), ] ) def test_flag_exploration_email_replaces_preferences_link(self) -> None: """Tests that the email footer preferences link is rendered.""" email_manager.send_flag_exploration_email( self.exploration.title, self.exploration.id, self.new_user_id, self.report_text, ) messages = self._get_sent_email_messages(self.MODERATOR_EMAIL) self.assertEqual(len(messages), 1) self.assertIn( 'href="https://www.oppia.org/preferences"', messages[0].html, ) self.assertNotIn( 'LINK_TO_PREFERENCES_PAGE', messages[0].html, ) class RenderedEmailFooterTest(test_utils.GenericTestBase): """Tests the email-footer rendering helper.""" @test_utils.set_platform_parameters( [ ( param_list.ParamName.EMAIL_FOOTER, 'You can change your email preferences via the ' '<a href="https://custom.example/account/preferences">' 'Preferences</a> page.', ), ] ) def test_preserves_custom_footer_without_placeholder(self) -> None: """Tests that a custom footer is returned unchanged.""" footer = email_manager.get_rendered_email_footer() self.assertEqual( footer, 'You can change your email preferences via the ' '<a href="https://custom.example/account/preferences">' 'Preferences</a> page.', ) class OnboardingReviewerInstantEmailTests(test_utils.EmailTestBase): """Test that correct email is sent while onboarding reviewers.""" REVIEWER_USERNAME: Final = 'reviewer' REVIEWER_EMAIL: Final = 'reviewer@example.com' def setUp(self) -> None: super().setUp() self.signup(self.REVIEWER_EMAIL, self.REVIEWER_USERNAME) self.reviewer_id = self.get_user_id_from_email(self.REVIEWER_EMAIL) user_services.update_email_preferences( self.reviewer_id, True, False, False, False, feconf.DEFAULT_CONTRIBUTOR_DASHBOARD_EMAIL_PREFERENCE, ) @test_utils.set_platform_parameters( [(param_list.ParamName.SERVER_CAN_SEND_EMAILS, False)] ) def test_email_not_sent_if_server_can_send_emails_is_false(self) -> None: email_manager.send_mail_to_onboard_new_reviewers( self.reviewer_id, 'Algebra' ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.REVIEWER_EMAIL) self.assertEqual(len(messages), 0) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ] ) def test_that_correct_completion_email_is_sent(self) -> None: expected_email_subject = 'Invitation to review suggestions' expected_email_html_body = ( 'Hi reviewer,<br><br>' 'Thank you for actively contributing high-quality suggestions for ' 'Oppia\'s lessons in Algebra, and for helping to make these lessons' ' better for students around the world!<br><br>' 'In recognition of your contributions, we would like to invite you' ' to become one of Oppia\'s reviewers. As a reviewer, you will be ' 'able to review suggestions in Algebra, and contribute to helping ' 'ensure that any edits made to lessons preserve the lessons\' ' 'quality and are beneficial for students.<br><br>' 'If you\'d like to help out as a reviewer, please visit your ' '<a href="https://www.oppia.org/creator-dashboard/">dashboard</a>. ' 'and set your review preferences accordingly. Note that, if you ' 'accept,you will receive occasional emails inviting you to review ' 'incoming suggestions by others.<br><br>' 'Again, thank you for your contributions to the Oppia ' 'community!<br>' '- The Oppia Team<br>' '<br>' 'You can change your email preferences via the ' '<a href="http://localhost:8181/preferences">Preferences</a> page.' ) email_manager.send_mail_to_onboard_new_reviewers( self.reviewer_id, 'Algebra' ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.REVIEWER_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) sent_email_model = all_models[0] self.assertEqual(sent_email_model.subject, expected_email_subject) self.assertEqual(sent_email_model.recipient_id, self.reviewer_id) self.assertEqual(sent_email_model.recipient_email, self.REVIEWER_EMAIL) self.assertEqual(sent_email_model.sender_id, feconf.SYSTEM_COMMITTER_ID) self.assertEqual( sent_email_model.sender_email, 'Site Admin <noreply@example.com>' ) self.assertEqual( sent_email_model.intent, feconf.EMAIL_INTENT_ONBOARD_CD_USER ) class NotifyReviewerInstantEmailTests(test_utils.EmailTestBase): """Test that correct email is sent while notifying reviewers.""" REVIEWER_USERNAME: Final = 'reviewer' REVIEWER_EMAIL: Final = 'reviewer@example.com' def setUp(self) -> None: super().setUp() self.signup(self.REVIEWER_EMAIL, self.REVIEWER_USERNAME) self.reviewer_id = self.get_user_id_from_email(self.REVIEWER_EMAIL) user_services.update_email_preferences( self.reviewer_id, True, False, False, False, feconf.DEFAULT_CONTRIBUTOR_DASHBOARD_EMAIL_PREFERENCE, ) @test_utils.set_platform_parameters( [(param_list.ParamName.SERVER_CAN_SEND_EMAILS, False)] ) def test_email_not_sent_if_server_can_send_emails_is_false(self) -> None: email_manager.send_mail_to_notify_users_to_review( self.reviewer_id, 'Algebra' ) messages = self._get_sent_email_messages(self.REVIEWER_EMAIL) self.assertEqual(len(messages), 0) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ] ) def test_that_correct_completion_email_is_sent(self) -> None: expected_email_subject = 'Notification to review suggestions' expected_email_html_body = ( 'Hi reviewer,<br><br>' 'Just a heads-up that there are new suggestions to ' 'review in Algebra, which you are registered as a reviewer for.' '<br><br>Please take a look at and accept/reject these suggestions ' 'at your earliest convenience. You can visit your ' '<a href="https://www.oppia.org/creator-dashboard/">dashboard</a> ' 'to view the list of suggestions that need a review.<br><br>' 'Thank you for helping improve Oppia\'s lessons!' '- The Oppia Team<br>' '<br>' 'You can change your email preferences via the ' '<a href="http://localhost:8181/preferences">Preferences</a> page.' ) email_manager.send_mail_to_notify_users_to_review( self.reviewer_id, 'Algebra' ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.REVIEWER_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) sent_email_model = all_models[0] self.assertEqual(sent_email_model.subject, expected_email_subject) self.assertEqual(sent_email_model.recipient_id, self.reviewer_id) self.assertEqual(sent_email_model.recipient_email, self.REVIEWER_EMAIL) self.assertEqual(sent_email_model.sender_id, feconf.SYSTEM_COMMITTER_ID) self.assertEqual( sent_email_model.sender_email, 'Site Admin <noreply@example.com>' ) self.assertEqual( sent_email_model.intent, feconf.EMAIL_INTENT_REVIEW_CREATOR_DASHBOARD_SUGGESTIONS, ) class NotifyContributionAchievementEmailTests(test_utils.EmailTestBase): """Test that correct email is sent when notifying contributor achievements.""" USERNAME: Final = 'user' USER_EMAIL: Final = 'user@example.com' def setUp(self) -> None: super().setUp() self.signup(self.USER_EMAIL, self.USERNAME) self.user_id = self.get_user_id_from_email(self.USER_EMAIL) user_services.update_email_preferences( self.user_id, True, False, False, False, feconf.DEFAULT_CONTRIBUTOR_DASHBOARD_EMAIL_PREFERENCE, ) @test_utils.set_platform_parameters( [(param_list.ParamName.SERVER_CAN_SEND_EMAILS, False)] ) def test_email_not_sent_if_server_can_send_emails_is_false(self) -> None: contributor_ranking_email_info = ( suggestion_registry.ContributorMilestoneEmailInfo( self.user_id, 'question', 'edit', None, 'Initial Contributor' ) ) email_manager.send_mail_to_notify_contributor_ranking_achievement( contributor_ranking_email_info ) messages = self._get_sent_email_messages(self.USER_EMAIL) self.assertEqual(len(messages), 0) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_that_email_not_sent_if_user_can_not_receive_emails(self) -> None: user_services.update_email_preferences( self.user_id, False, False, False, False, feconf.DEFAULT_CONTRIBUTOR_DASHBOARD_EMAIL_PREFERENCE, ) contributor_ranking_email_info = ( suggestion_registry.ContributorMilestoneEmailInfo( self.user_id, 'question', 'edit', None, 'Initial Contributor' ) ) email_manager.send_mail_to_notify_contributor_ranking_achievement( contributor_ranking_email_info ) messages = self._get_sent_email_messages(self.USER_EMAIL) self.assertEqual(len(messages), 0) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_that_translation_submitter_acceptance_ranking_email_is_sent( self, ) -> None: expected_email_subject = 'Oppia Translator Rank Achievement!' expected_email_html_body = ( 'Hi user,<br><br>' 'This is to let you know that you have successfully achieved the ' 'Initial Contributor rank for submitting translations in हिन्दी ' '(Hindi). Your efforts help Oppia grow better every day and ' 'support students around the world.<br><br>' 'You can check all the achievements you earned in the ' '<a href="http://localhost:8181/contributor-dashboard">' 'Contributor Dashboard</a>.<br><br>' 'Best wishes and we hope you can continue to contribute!<br><br>' 'The Oppia Contributor Dashboard Team' ) contributor_ranking_email_info = ( suggestion_registry.ContributorMilestoneEmailInfo( self.user_id, 'translation', 'acceptance', 'hi', 'Initial Contributor', ) ) email_manager.send_mail_to_notify_contributor_ranking_achievement( contributor_ranking_email_info ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.USER_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) sent_email_model = all_models[0] self.assertEqual(sent_email_model.subject, expected_email_subject) self.assertEqual(sent_email_model.recipient_id, self.user_id) self.assertEqual(sent_email_model.recipient_email, self.USER_EMAIL) self.assertEqual(sent_email_model.sender_id, feconf.SYSTEM_COMMITTER_ID) self.assertEqual( sent_email_model.sender_email, 'Site Admin <noreply@example.com>' ) self.assertEqual( sent_email_model.intent, feconf.EMAIL_INTENT_NOTIFY_CONTRIBUTOR_DASHBOARD_ACHIEVEMENTS, ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_that_question_submitter_acceptance_ranking_email_is_sent( self, ) -> None: expected_email_subject = 'Oppia Question Submitter Rank Achievement!' expected_email_html_body = ( 'Hi user,<br><br>' 'This is to let you know that you have successfully achieved the ' 'Initial Contributor rank for submitting practice questions. Your ' 'efforts help Oppia grow better every day and support students ' 'around the world.<br><br>' 'You can check all the achievements you earned in the ' '<a href="http://localhost:8181/contributor-dashboard">' 'Contributor Dashboard</a>.<br><br>' 'Best wishes and we hope you can continue to contribute!<br><br>' 'The Oppia Contributor Dashboard Team' ) contributor_ranking_email_info = ( suggestion_registry.ContributorMilestoneEmailInfo( self.user_id, 'question', 'acceptance', None, 'Initial Contributor', ) ) email_manager.send_mail_to_notify_contributor_ranking_achievement( contributor_ranking_email_info ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.USER_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) sent_email_model = all_models[0] self.assertEqual(sent_email_model.subject, expected_email_subject) self.assertEqual(sent_email_model.recipient_id, self.user_id) self.assertEqual(sent_email_model.recipient_email, self.USER_EMAIL) self.assertEqual(sent_email_model.sender_id, feconf.SYSTEM_COMMITTER_ID) self.assertEqual( sent_email_model.sender_email, 'Site Admin <noreply@example.com>' ) self.assertEqual( sent_email_model.intent, feconf.EMAIL_INTENT_NOTIFY_CONTRIBUTOR_DASHBOARD_ACHIEVEMENTS, ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_that_translation_reviewer_review_ranking_email_is_sent( self, ) -> None: expected_email_subject = 'Oppia Translation Reviewer Rank Achievement!' expected_email_html_body = ( 'Hi user,<br><br>' 'This is to let you know that you have successfully achieved the ' 'Initial Contributor rank for reviewing translations in हिन्दी ' '(Hindi). Your efforts help Oppia grow better every day and ' 'support students around the world.<br><br>' 'You can check all the achievements you earned in the ' '<a href="http://localhost:8181/contributor-dashboard">' 'Contributor Dashboard</a>.<br><br>' 'Best wishes and we hope you can continue to contribute!<br><br>' 'The Oppia Contributor Dashboard Team' ) contributor_ranking_email_info = ( suggestion_registry.ContributorMilestoneEmailInfo( self.user_id, 'translation', 'review', 'hi', 'Initial Contributor', ) ) email_manager.send_mail_to_notify_contributor_ranking_achievement( contributor_ranking_email_info ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.USER_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) sent_email_model = all_models[0] self.assertEqual(sent_email_model.subject, expected_email_subject) self.assertEqual(sent_email_model.recipient_id, self.user_id) self.assertEqual(sent_email_model.recipient_email, self.USER_EMAIL) self.assertEqual(sent_email_model.sender_id, feconf.SYSTEM_COMMITTER_ID) self.assertEqual( sent_email_model.sender_email, 'Site Admin <noreply@example.com>' ) self.assertEqual( sent_email_model.intent, feconf.EMAIL_INTENT_NOTIFY_CONTRIBUTOR_DASHBOARD_ACHIEVEMENTS, ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_that_question_reviewer_review_ranking_email_is_sent(self) -> None: expected_email_subject = 'Oppia Question Reviewer Rank Achievement!' expected_email_html_body = ( 'Hi user,<br><br>' 'This is to let you know that you have successfully achieved the ' 'Initial Contributor rank for reviewing practice questions. Your ' 'efforts help Oppia grow better every day and support students ' 'around the world.<br><br>' 'You can check all the achievements you earned in the ' '<a href="http://localhost:8181/contributor-dashboard">' 'Contributor Dashboard</a>.<br><br>' 'Best wishes and we hope you can continue to contribute!<br><br>' 'The Oppia Contributor Dashboard Team' ) contributor_ranking_email_info = ( suggestion_registry.ContributorMilestoneEmailInfo( self.user_id, 'question', 'review', None, 'Initial Contributor' ) ) email_manager.send_mail_to_notify_contributor_ranking_achievement( contributor_ranking_email_info ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.USER_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) sent_email_model = all_models[0] self.assertEqual(sent_email_model.subject, expected_email_subject) self.assertEqual(sent_email_model.recipient_id, self.user_id) self.assertEqual(sent_email_model.recipient_email, self.USER_EMAIL) self.assertEqual(sent_email_model.sender_id, feconf.SYSTEM_COMMITTER_ID) self.assertEqual( sent_email_model.sender_email, 'Site Admin <noreply@example.com>' ) self.assertEqual( sent_email_model.intent, feconf.EMAIL_INTENT_NOTIFY_CONTRIBUTOR_DASHBOARD_ACHIEVEMENTS, ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_that_translation_reviewer_edit_ranking_email_is_sent(self) -> None: expected_email_subject = 'Oppia Translation Reviewer Rank Achievement!' expected_email_html_body = ( 'Hi user,<br><br>' 'This is to let you know that you have successfully achieved the ' 'Initial Contributor rank for correcting translations in हिन्दी ' '(Hindi). Your efforts help Oppia grow better every day and ' 'support students around the world.<br><br>' 'You can check all the achievements you earned in the ' '<a href="http://localhost:8181/contributor-dashboard">' 'Contributor Dashboard</a>.<br><br>' 'Best wishes and we hope you can continue to contribute!<br><br>' 'The Oppia Contributor Dashboard Team' ) contributor_ranking_email_info = ( suggestion_registry.ContributorMilestoneEmailInfo( self.user_id, 'translation', 'edit', 'hi', 'Initial Contributor' ) ) email_manager.send_mail_to_notify_contributor_ranking_achievement( contributor_ranking_email_info ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.USER_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) sent_email_model = all_models[0] self.assertEqual(sent_email_model.subject, expected_email_subject) self.assertEqual(sent_email_model.recipient_id, self.user_id) self.assertEqual(sent_email_model.recipient_email, self.USER_EMAIL) self.assertEqual(sent_email_model.sender_id, feconf.SYSTEM_COMMITTER_ID) self.assertEqual( sent_email_model.sender_email, 'Site Admin <noreply@example.com>' ) self.assertEqual( sent_email_model.intent, feconf.EMAIL_INTENT_NOTIFY_CONTRIBUTOR_DASHBOARD_ACHIEVEMENTS, ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_that_question_reviewer_edit_ranking_email_is_sent(self) -> None: expected_email_subject = 'Oppia Question Reviewer Rank Achievement!' expected_email_html_body = ( 'Hi user,<br><br>' 'This is to let you know that you have successfully achieved the ' 'Initial Contributor rank for correcting practice questions. ' 'Your efforts help Oppia grow better every day and support ' 'students around the world.<br><br>' 'You can check all the achievements you earned in the ' '<a href="http://localhost:8181/contributor-dashboard">' 'Contributor Dashboard</a>.<br><br>' 'Best wishes and we hope you can continue to contribute!<br><br>' 'The Oppia Contributor Dashboard Team' ) contributor_ranking_email_info = ( suggestion_registry.ContributorMilestoneEmailInfo( self.user_id, 'question', 'edit', None, 'Initial Contributor' ) ) email_manager.send_mail_to_notify_contributor_ranking_achievement( contributor_ranking_email_info ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.USER_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) sent_email_model = all_models[0] self.assertEqual(sent_email_model.subject, expected_email_subject) self.assertEqual(sent_email_model.recipient_id, self.user_id) self.assertEqual(sent_email_model.recipient_email, self.USER_EMAIL) self.assertEqual(sent_email_model.sender_id, feconf.SYSTEM_COMMITTER_ID) self.assertEqual( sent_email_model.sender_email, 'Site Admin <noreply@example.com>' ) self.assertEqual( sent_email_model.intent, feconf.EMAIL_INTENT_NOTIFY_CONTRIBUTOR_DASHBOARD_ACHIEVEMENTS, ) class NotifyContributionDashboardReviewersEmailTests(test_utils.EmailTestBase): """Tests the send_mail_to_notify_contributor_dashboard_reviewers method, which sends an email to reviewers with information regarding the suggestions that have waited the longest for review. """ target_id: str = 'exp1' skill_id: str = 'skill_123456' mocked_review_submission_datetime: datetime.datetime = datetime.datetime( 2020, 6, 15, 5 ) AUTHOR_USERNAME: Final = 'author' AUTHOR_EMAIL: Final = 'author@example.com' REVIEWER_1_USERNAME: Final = 'reviewer1' REVIEWER_1_EMAIL: Final = 'reviewer1@community.org' REVIEWER_2_USERNAME: Final = 'reviewer2' REVIEWER_2_EMAIL: Final = 'reviewer2@community.org' EMAIL_FOOTER: Final = ( 'You can change your email preferences via the ' '<a href="http://localhost:8181/preferences">Preferences</a> page.' ) def _create_translation_suggestion_in_lang_with_html_and_datetime( self, language_code: str, translation_html: str, submission_datetime: datetime.datetime, ) -> suggestion_registry.BaseSuggestion: """Creates a translation suggestion in the given language_code with the given translation html and submission datetime. """ add_translation_change_dict = { 'cmd': exp_domain.CMD_ADD_WRITTEN_TRANSLATION, 'state_name': feconf.DEFAULT_INIT_STATE_NAME, 'content_id': 'content_0', 'language_code': language_code, 'content_html': feconf.DEFAULT_STATE_CONTENT_STR, 'translation_html': translation_html, 'data_format': 'html', } translation_suggestion = suggestion_services.create_suggestion( feconf.SUGGESTION_TYPE_TRANSLATE_CONTENT, feconf.ENTITY_TYPE_EXPLORATION, self.target_id, feconf.CURRENT_STATE_SCHEMA_VERSION, self.author_id, add_translation_change_dict, 'test description', ) translation_suggestion.last_updated = submission_datetime return translation_suggestion def _create_question_suggestion_with_question_html_and_datetime( self, question_html: str, submission_datetime: datetime.datetime ) -> suggestion_registry.BaseSuggestion: """Creates a question suggestion with the given question html and submission datetime. """ with self.swap(feconf, 'DEFAULT_STATE_CONTENT_STR', question_html): content_id_generator = translation_domain.ContentIdGenerator() add_question_change_dict: Dict[ str, Union[str, float, question_domain.QuestionDict] ] = { 'cmd': ( question_domain.CMD_CREATE_NEW_FULLY_SPECIFIED_QUESTION ), 'question_dict': { 'id': 'test_id', 'version': 12, 'question_state_data': self._create_valid_question_data( 'default_state', content_id_generator ).to_dict(), 'language_code': constants.DEFAULT_LANGUAGE_CODE, 'question_state_data_schema_version': ( feconf.CURRENT_STATE_SCHEMA_VERSION ), 'linked_skill_ids': ['skill_1'], 'inapplicable_skill_misconception_ids': ['skillid12345-1'], 'next_content_id_index': ( content_id_generator.next_content_id_index ), }, 'skill_id': self.skill_id, 'skill_difficulty': 0.3, } question_suggestion = suggestion_services.create_suggestion( feconf.SUGGESTION_TYPE_ADD_QUESTION, feconf.ENTITY_TYPE_SKILL, self.skill_id, feconf.CURRENT_STATE_SCHEMA_VERSION, self.author_id, add_question_change_dict, 'test description', ) question_suggestion.last_updated = submission_datetime return question_suggestion def _create_reviewable_suggestion_email_infos_from_suggestions( self, suggestions: List[suggestion_registry.BaseSuggestion] ) -> List[suggestion_registry.ReviewableSuggestionEmailInfo]: """Creates a list of ReviewableSuggestionEmailInfo objects from the given suggestions. """ return [ ( suggestion_services.create_reviewable_suggestion_email_info_from_suggestion( suggestion ) ) for suggestion in suggestions ] def _assert_email_data_stored_in_sent_email_model_is_correct( self, expected_email_html_body: str, reviewer_id: Optional[str], reviewer_email: str, ) -> None: """Asserts that the created sent email model from the sent email contains the right information. """ sent_email_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all() .filter(email_models.SentEmailModel.recipient_id == reviewer_id) .fetch() ) self.assertEqual(len(sent_email_models), 1) sent_email_model = sent_email_models[0] self.assertEqual( sent_email_model.subject, email_manager.CONTRIBUTOR_DASHBOARD_REVIEWER_NOTIFICATION_EMAIL_DATA[ 'email_subject' ], ) self.assertEqual(sent_email_model.recipient_id, reviewer_id) self.assertEqual(sent_email_model.recipient_email, reviewer_email) self.assertEqual(sent_email_model.html_body, expected_email_html_body) self.assertEqual(sent_email_model.sender_id, feconf.SYSTEM_COMMITTER_ID) self.assertEqual( sent_email_model.sender_email, 'Site Admin <noreply@example.com>' ) self.assertEqual( sent_email_model.intent, feconf.EMAIL_INTENT_REVIEW_CONTRIBUTOR_DASHBOARD_SUGGESTIONS, ) def _mock_logging_info(self, msg: str, *args: str) -> None: """Mocks logging.info() by appending the log message to the logged info list. """ self.logged_info.append(msg % args) def setUp(self) -> None: super().setUp() self.signup(self.AUTHOR_EMAIL, self.AUTHOR_USERNAME) self.author_id = self.get_user_id_from_email(self.AUTHOR_EMAIL) self.signup(self.REVIEWER_1_EMAIL, self.REVIEWER_1_USERNAME) self.reviewer_1_id = self.get_user_id_from_email(self.REVIEWER_1_EMAIL) user_services.update_email_preferences( self.reviewer_1_id, True, False, False, False, feconf.DEFAULT_CONTRIBUTOR_DASHBOARD_EMAIL_PREFERENCE, ) self.signup(self.REVIEWER_2_EMAIL, self.REVIEWER_2_USERNAME) self.reviewer_2_id = self.get_user_id_from_email(self.REVIEWER_2_EMAIL) user_services.update_email_preferences( self.reviewer_2_id, True, False, False, False, feconf.DEFAULT_CONTRIBUTOR_DASHBOARD_EMAIL_PREFERENCE, ) self.log_new_error_counter = test_utils.CallCounter(logging.error) self.log_new_error_ctx = self.swap( logging, 'error', self.log_new_error_counter ) self.logged_info: List[str] = [] self.log_new_info_ctx = self.swap( logging, 'info', self._mock_logging_info ) self.save_new_valid_exploration(self.target_id, self.author_id) self.save_new_skill(self.skill_id, self.author_id) question_suggestion = ( self._create_question_suggestion_with_question_html_and_datetime( '<p>What is the meaning of life?</p>', self.mocked_review_submission_datetime, ) ) self.reviewable_suggestion_email_info = suggestion_services.create_reviewable_suggestion_email_info_from_suggestion( question_suggestion ) @test_utils.set_platform_parameters( [(param_list.ParamName.SERVER_CAN_SEND_EMAILS, False)] ) def test_email_not_sent_if_server_can_send_emails_is_false(self) -> None: with self.capture_logging(min_level=logging.ERROR) as logs: with self.log_new_error_ctx: email_manager.send_mail_to_notify_contributor_dashboard_reviewers( # pylint: disable=line-too-long [self.reviewer_1_id], [[self.reviewable_suggestion_email_info]], ) messages = self._get_all_sent_email_messages() self.assertEqual(len(messages), 0) self.assertEqual(self.log_new_error_counter.times_called, 1) self.assertEqual(logs[0], 'This app cannot send emails to users.') @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.CONTRIBUTOR_DASHBOARD_REVIEWER_EMAILS_IS_ENABLED, False, ), # pylint: disable=line-too-long ] ) def test_email_not_sent_if_reviewer_notifications_is_disabled(self) -> None: with self.capture_logging(min_level=logging.ERROR) as logs: with self.log_new_error_ctx: email_manager.send_mail_to_notify_contributor_dashboard_reviewers( # pylint: disable=line-too-long [self.reviewer_1_id], [[self.reviewable_suggestion_email_info]], ) messages = self._get_all_sent_email_messages() self.assertEqual(len(messages), 0) self.assertEqual(self.log_new_error_counter.times_called, 1) self.assertEqual( logs[0], 'The "contributor_dashboard_reviewer_emails_is_enabled" ' 'property must be enabled on the platform parameters tab on ' 'the admin page in order to send reviewers the emails.', ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.CONTRIBUTOR_DASHBOARD_REVIEWER_EMAILS_IS_ENABLED, True, ), # pylint: disable=line-too-long (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_not_sent_if_reviewer_email_does_not_exist(self) -> None: with self.capture_logging(min_level=logging.ERROR) as logs: with self.log_new_error_ctx: email_manager.send_mail_to_notify_contributor_dashboard_reviewers( # pylint: disable=line-too-long ['reviewer_id_with_no_email'], [[self.reviewable_suggestion_email_info]], ) messages = self._get_all_sent_email_messages() self.assertEqual(len(messages), 0) self.assertEqual(self.log_new_error_counter.times_called, 1) self.assertEqual( logs[0], 'There was no email for the given reviewer id: ' 'reviewer_id_with_no_email.', ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.CONTRIBUTOR_DASHBOARD_REVIEWER_EMAILS_IS_ENABLED, True, ), # pylint: disable=line-too-long ] ) def test_email_not_sent_if_no_reviewers_to_notify(self) -> None: with self.capture_logging(min_level=logging.ERROR) as logs: with self.log_new_error_ctx: email_manager.send_mail_to_notify_contributor_dashboard_reviewers( # pylint: disable=line-too-long [], [[self.reviewable_suggestion_email_info]] ) messages = self._get_all_sent_email_messages() self.assertEqual(len(messages), 0) self.assertEqual(self.log_new_error_counter.times_called, 1) self.assertEqual( logs[0], 'No Contributor Dashboard reviewers to notify.' ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.CONTRIBUTOR_DASHBOARD_REVIEWER_EMAILS_IS_ENABLED, True, ), # pylint: disable=line-too-long (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_not_sent_if_no_suggestions_to_notify_the_reviewer_about( self, ) -> None: with self.log_new_info_ctx: email_manager.send_mail_to_notify_contributor_dashboard_reviewers( # pylint: disable=line-too-long [self.reviewer_1_id], [[]] ) messages = self._get_all_sent_email_messages() self.assertEqual(len(messages), 0) self.assertEqual( self.logged_info[0], 'There were no suggestions to recommend to the reviewer with user ' 'id: %s.' % self.reviewer_1_id, ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.CONTRIBUTOR_DASHBOARD_REVIEWER_EMAILS_IS_ENABLED, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_sent_to_reviewer_with_question_waiting_a_day_for_review( self, ) -> None: question_suggestion = ( self._create_question_suggestion_with_question_html_and_datetime( '<p>What is the meaning of life?</p>', self.mocked_review_submission_datetime, ) ) reviewable_suggestion_email_info = suggestion_services.create_reviewable_suggestion_email_info_from_suggestion( question_suggestion ) review_wait_time = 1 mocked_current_time = ( reviewable_suggestion_email_info.submission_datetime + datetime.timedelta(days=review_wait_time) ) expected_email_html_body = ( 'Hi reviewer1,' '<br><br>' 'There are new review opportunities that we think you might be ' 'interested in on the <a href="%s%s">Contributor Dashboard</a>. ' 'Here are some examples of contributions that have been waiting ' 'the longest for review:<br><br>' '<ul>' '<li>The following question suggestion was submitted for review 1 ' 'day ago:' '<br>What is the meaning of life?</li><br>' '</ul><br>' 'Please take some time to review any of the above contributions ' '(if they still need a review) or any other contributions on the ' 'dashboard. We appreciate your help!' '<br><br>' 'Thanks again, and happy reviewing!<br>' '- The Oppia Contributor Dashboard Team' '<br><br>%s' % ( DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, EMAIL_FOOTER, ) ) with self.log_new_error_ctx: with self.swap( utils, 'get_current_utc_datetime', lambda: mocked_current_time ): ( email_manager.send_mail_to_notify_contributor_dashboard_reviewers( [self.reviewer_1_id], [[reviewable_suggestion_email_info]], ) ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.REVIEWER_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body, self.reviewer_1_id, self.REVIEWER_1_EMAIL ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.CONTRIBUTOR_DASHBOARD_REVIEWER_EMAILS_IS_ENABLED, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_sent_to_reviewer_with_question_waiting_days_for_review( self, ) -> None: question_suggestion = ( self._create_question_suggestion_with_question_html_and_datetime( '<p>What is the meaning of life?</p>', self.mocked_review_submission_datetime, ) ) reviewable_suggestion_email_info = suggestion_services.create_reviewable_suggestion_email_info_from_suggestion( question_suggestion ) review_wait_time = 5 mocked_current_time = ( reviewable_suggestion_email_info.submission_datetime + datetime.timedelta(days=review_wait_time) ) expected_email_html_body = ( 'Hi reviewer1,' '<br><br>' 'There are new review opportunities that we think you might be ' 'interested in on the <a href="%s%s">Contributor Dashboard</a>. ' 'Here are some examples of contributions that have been waiting ' 'the longest for review:<br><br>' '<ul>' '<li>The following question suggestion was submitted for review 5 ' 'days ago:' '<br>What is the meaning of life?</li><br>' '</ul><br>' 'Please take some time to review any of the above contributions ' '(if they still need a review) or any other contributions on the ' 'dashboard. We appreciate your help!' '<br><br>' 'Thanks again, and happy reviewing!<br>' '- The Oppia Contributor Dashboard Team' '<br><br>%s' % ( DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, EMAIL_FOOTER, ) ) with self.log_new_error_ctx: with self.swap( utils, 'get_current_utc_datetime', lambda: mocked_current_time ): ( email_manager.send_mail_to_notify_contributor_dashboard_reviewers( [self.reviewer_1_id], [[reviewable_suggestion_email_info]], ) ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.REVIEWER_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body, self.reviewer_1_id, self.REVIEWER_1_EMAIL ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.CONTRIBUTOR_DASHBOARD_REVIEWER_EMAILS_IS_ENABLED, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_sent_to_reviewer_with_question_waiting_an_hour_for_review( self, ) -> None: question_suggestion = ( self._create_question_suggestion_with_question_html_and_datetime( '<p>What is the meaning of life?</p>', self.mocked_review_submission_datetime, ) ) reviewable_suggestion_email_info = suggestion_services.create_reviewable_suggestion_email_info_from_suggestion( question_suggestion ) review_wait_time = 1 mocked_current_time = ( reviewable_suggestion_email_info.submission_datetime + datetime.timedelta(hours=review_wait_time) ) expected_email_html_body = ( 'Hi reviewer1,' '<br><br>' 'There are new review opportunities that we think you might be ' 'interested in on the <a href="%s%s">Contributor Dashboard</a>. ' 'Here are some examples of contributions that have been waiting ' 'the longest for review:<br><br>' '<ul>' '<li>The following question suggestion was submitted for review 1 ' 'hour ago:' '<br>What is the meaning of life?</li><br>' '</ul><br>' 'Please take some time to review any of the above contributions ' '(if they still need a review) or any other contributions on the ' 'dashboard. We appreciate your help!' '<br><br>' 'Thanks again, and happy reviewing!<br>' '- The Oppia Contributor Dashboard Team' '<br><br>%s' % ( DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, EMAIL_FOOTER, ) ) with self.log_new_error_ctx: with self.swap( utils, 'get_current_utc_datetime', lambda: mocked_current_time ): ( email_manager.send_mail_to_notify_contributor_dashboard_reviewers( [self.reviewer_1_id], [[reviewable_suggestion_email_info]], ) ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.REVIEWER_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body, self.reviewer_1_id, self.REVIEWER_1_EMAIL ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.CONTRIBUTOR_DASHBOARD_REVIEWER_EMAILS_IS_ENABLED, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_sent_to_reviewer_with_question_waiting_hours_for_review( self, ) -> None: question_suggestion = ( self._create_question_suggestion_with_question_html_and_datetime( '<p>What is the meaning of life?</p>', self.mocked_review_submission_datetime, ) ) reviewable_suggestion_email_info = suggestion_services.create_reviewable_suggestion_email_info_from_suggestion( question_suggestion ) review_wait_time = 5 mocked_current_time = ( reviewable_suggestion_email_info.submission_datetime + datetime.timedelta(hours=review_wait_time) ) expected_email_html_body = ( 'Hi reviewer1,' '<br><br>' 'There are new review opportunities that we think you might be ' 'interested in on the <a href="%s%s">Contributor Dashboard</a>. ' 'Here are some examples of contributions that have been waiting ' 'the longest for review:<br><br>' '<ul>' '<li>The following question suggestion was submitted for review 5 ' 'hours ago:' '<br>What is the meaning of life?</li><br>' '</ul><br>' 'Please take some time to review any of the above contributions ' '(if they still need a review) or any other contributions on the ' 'dashboard. We appreciate your help!' '<br><br>' 'Thanks again, and happy reviewing!<br>' '- The Oppia Contributor Dashboard Team' '<br><br>%s' % ( DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, EMAIL_FOOTER, ) ) with self.log_new_error_ctx: with self.swap( utils, 'get_current_utc_datetime', lambda: mocked_current_time ): ( email_manager.send_mail_to_notify_contributor_dashboard_reviewers( [self.reviewer_1_id], [[reviewable_suggestion_email_info]], ) ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.REVIEWER_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body, self.reviewer_1_id, self.REVIEWER_1_EMAIL ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.CONTRIBUTOR_DASHBOARD_REVIEWER_EMAILS_IS_ENABLED, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_sent_to_reviewer_with_question_waiting_a_minute_for_review( self, ) -> None: question_suggestion = ( self._create_question_suggestion_with_question_html_and_datetime( '<p>What is the meaning of life?</p>', self.mocked_review_submission_datetime, ) ) reviewable_suggestion_email_info = suggestion_services.create_reviewable_suggestion_email_info_from_suggestion( question_suggestion ) review_wait_time = 1 mocked_current_time = ( reviewable_suggestion_email_info.submission_datetime + datetime.timedelta(minutes=review_wait_time) ) expected_email_html_body = ( 'Hi reviewer1,' '<br><br>' 'There are new review opportunities that we think you might be ' 'interested in on the <a href="%s%s">Contributor Dashboard</a>. ' 'Here are some examples of contributions that have been waiting ' 'the longest for review:<br><br>' '<ul>' '<li>The following question suggestion was submitted for review 1 ' 'minute ago:' '<br>What is the meaning of life?</li><br>' '</ul><br>' 'Please take some time to review any of the above contributions ' '(if they still need a review) or any other contributions on the ' 'dashboard. We appreciate your help!' '<br><br>' 'Thanks again, and happy reviewing!<br>' '- The Oppia Contributor Dashboard Team' '<br><br>%s' % ( DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, EMAIL_FOOTER, ) ) with self.log_new_error_ctx: with self.swap( utils, 'get_current_utc_datetime', lambda: mocked_current_time ): ( email_manager.send_mail_to_notify_contributor_dashboard_reviewers( [self.reviewer_1_id], [[reviewable_suggestion_email_info]], ) ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.REVIEWER_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body, self.reviewer_1_id, self.REVIEWER_1_EMAIL ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.CONTRIBUTOR_DASHBOARD_REVIEWER_EMAILS_IS_ENABLED, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_sent_to_reviewer_with_question_waiting_minutes_for_review( self, ) -> None: question_suggestion = ( self._create_question_suggestion_with_question_html_and_datetime( '<p>What is the meaning of life?</p>', self.mocked_review_submission_datetime, ) ) reviewable_suggestion_email_info = suggestion_services.create_reviewable_suggestion_email_info_from_suggestion( question_suggestion ) review_wait_time = 5 mocked_current_time = ( reviewable_suggestion_email_info.submission_datetime + datetime.timedelta(minutes=review_wait_time) ) expected_email_html_body = ( 'Hi reviewer1,' '<br><br>' 'There are new review opportunities that we think you might be ' 'interested in on the <a href="%s%s">Contributor Dashboard</a>. ' 'Here are some examples of contributions that have been waiting ' 'the longest for review:<br><br>' '<ul>' '<li>The following question suggestion was submitted for review 5 ' 'minutes ago:' '<br>What is the meaning of life?</li><br>' '</ul><br>' 'Please take some time to review any of the above contributions ' '(if they still need a review) or any other contributions on the ' 'dashboard. We appreciate your help!' '<br><br>' 'Thanks again, and happy reviewing!<br>' '- The Oppia Contributor Dashboard Team' '<br><br>%s' % ( DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, EMAIL_FOOTER, ) ) with self.log_new_error_ctx: with self.swap( utils, 'get_current_utc_datetime', lambda: mocked_current_time ): ( email_manager.send_mail_to_notify_contributor_dashboard_reviewers( [self.reviewer_1_id], [[reviewable_suggestion_email_info]], ) ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.REVIEWER_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body, self.reviewer_1_id, self.REVIEWER_1_EMAIL ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.CONTRIBUTOR_DASHBOARD_REVIEWER_EMAILS_IS_ENABLED, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_sent_to_reviewer_with_question_waiting_seconds_for_review( self, ) -> None: question_suggestion = ( self._create_question_suggestion_with_question_html_and_datetime( '<p>What is the meaning of life?</p>', self.mocked_review_submission_datetime, ) ) reviewable_suggestion_email_info = suggestion_services.create_reviewable_suggestion_email_info_from_suggestion( question_suggestion ) review_wait_time = 5 mocked_current_time = ( reviewable_suggestion_email_info.submission_datetime + datetime.timedelta(seconds=review_wait_time) ) expected_email_html_body = ( 'Hi reviewer1,' '<br><br>' 'There are new review opportunities that we think you might be ' 'interested in on the <a href="%s%s">Contributor Dashboard</a>. ' 'Here are some examples of contributions that have been waiting ' 'the longest for review:<br><br>' '<ul>' '<li>The following question suggestion was submitted for review 1 ' 'minute ago:' '<br>What is the meaning of life?</li><br>' '</ul><br>' 'Please take some time to review any of the above contributions ' '(if they still need a review) or any other contributions on the ' 'dashboard. We appreciate your help!' '<br><br>' 'Thanks again, and happy reviewing!<br>' '- The Oppia Contributor Dashboard Team' '<br><br>%s' % ( DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, EMAIL_FOOTER, ) ) with self.log_new_error_ctx: with self.swap( utils, 'get_current_utc_datetime', lambda: mocked_current_time ): ( email_manager.send_mail_to_notify_contributor_dashboard_reviewers( [self.reviewer_1_id], [[reviewable_suggestion_email_info]], ) ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.REVIEWER_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body, self.reviewer_1_id, self.REVIEWER_1_EMAIL ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.CONTRIBUTOR_DASHBOARD_REVIEWER_EMAILS_IS_ENABLED, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_sent_to_reviewer_with_multi_questions_waiting_for_a_review( self, ) -> None: mocked_current_time = ( self.mocked_review_submission_datetime + datetime.timedelta(days=1, hours=1) ) # Question suggestion 1 has waited 1 day for review. question_suggestion_1 = ( self._create_question_suggestion_with_question_html_and_datetime( '<p>Question 1</p>', self.mocked_review_submission_datetime + datetime.timedelta(hours=1), ) ) # Question suggestion 2 has waited 1 hour for review. question_suggestion_2 = ( self._create_question_suggestion_with_question_html_and_datetime( '<p>Question 2</p>', self.mocked_review_submission_datetime + datetime.timedelta(days=1), ) ) reviewable_suggestion_email_infos = ( self._create_reviewable_suggestion_email_infos_from_suggestions( [question_suggestion_1, question_suggestion_2] ) ) expected_email_html_body = ( 'Hi reviewer1,' '<br><br>' 'There are new review opportunities that we think you might be ' 'interested in on the <a href="%s%s">Contributor Dashboard</a>. ' 'Here are some examples of contributions that have been waiting ' 'the longest for review:<br><br>' '<ul>' '<li>The following question suggestion was submitted for review 1 ' 'day ago:' '<br>Question 1</li><br>' '<li>The following question suggestion was submitted for review 1 ' 'hour ago:' '<br>Question 2</li><br>' '</ul><br>' 'Please take some time to review any of the above contributions ' '(if they still need a review) or any other contributions on the ' 'dashboard. We appreciate your help!' '<br><br>' 'Thanks again, and happy reviewing!<br>' '- The Oppia Contributor Dashboard Team' '<br><br>%s' % ( DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, EMAIL_FOOTER, ) ) with self.log_new_error_ctx: with self.swap( utils, 'get_current_utc_datetime', lambda: mocked_current_time ): ( email_manager.send_mail_to_notify_contributor_dashboard_reviewers( [self.reviewer_1_id], [reviewable_suggestion_email_infos], ) ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.REVIEWER_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body, self.reviewer_1_id, self.REVIEWER_1_EMAIL ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.CONTRIBUTOR_DASHBOARD_REVIEWER_EMAILS_IS_ENABLED, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_sent_to_multi_reviewers_with_multi_question_suggestions( self, ) -> None: mocked_current_time = ( self.mocked_review_submission_datetime + datetime.timedelta(days=1, hours=1, minutes=1) ) # Question suggestion 1 has waited 1 day for review. question_suggestion_1 = ( self._create_question_suggestion_with_question_html_and_datetime( '<p>Question 1 for reviewer 1</p>', self.mocked_review_submission_datetime + datetime.timedelta(hours=1, minutes=1), ) ) # Question suggestion 2 has waited 1 hour for review. question_suggestion_2 = ( self._create_question_suggestion_with_question_html_and_datetime( '<p>Question 2 for reviewer 1</p>', self.mocked_review_submission_datetime + datetime.timedelta(days=1, minutes=1), ) ) # Question suggestion 3 has waited 1 minute for review. question_suggestion_3 = ( self._create_question_suggestion_with_question_html_and_datetime( '<p>Question 1 for reviewer 2</p>', self.mocked_review_submission_datetime + datetime.timedelta(days=1, hours=1), ) ) # Question suggestion 4 has waited 1 minute for review. question_suggestion_4 = ( self._create_question_suggestion_with_question_html_and_datetime( '<p>Question 2 for reviewer 2</p>', self.mocked_review_submission_datetime + datetime.timedelta(days=1, hours=1), ) ) reviewer_1_suggestion_email_infos = ( self._create_reviewable_suggestion_email_infos_from_suggestions( [question_suggestion_1, question_suggestion_2] ) ) reviewer_2_suggestion_email_infos = ( self._create_reviewable_suggestion_email_infos_from_suggestions( [question_suggestion_3, question_suggestion_4] ) ) expected_email_html_body_reviewer_1 = ( 'Hi reviewer1,' '<br><br>' 'There are new review opportunities that we think you might be ' 'interested in on the <a href="%s%s">Contributor Dashboard</a>. ' 'Here are some examples of contributions that have been waiting ' 'the longest for review:<br><br>' '<ul>' '<li>The following question suggestion was submitted for review 1 ' 'day ago:' '<br>Question 1 for reviewer 1</li><br>' '<li>The following question suggestion was submitted for review 1 ' 'hour ago:' '<br>Question 2 for reviewer 1</li><br>' '</ul><br>' 'Please take some time to review any of the above contributions ' '(if they still need a review) or any other contributions on the ' 'dashboard. We appreciate your help!' '<br><br>' 'Thanks again, and happy reviewing!<br>' '- The Oppia Contributor Dashboard Team' '<br><br>%s' % ( DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, EMAIL_FOOTER, ) ) expected_email_html_body_reviewer_2 = ( 'Hi reviewer2,' '<br><br>' 'There are new review opportunities that we think you might be ' 'interested in on the <a href="%s%s">Contributor Dashboard</a>. ' 'Here are some examples of contributions that have been waiting ' 'the longest for review:<br><br>' '<ul>' '<li>The following question suggestion was submitted for review 1 ' 'minute ago:' '<br>Question 1 for reviewer 2</li><br>' '<li>The following question suggestion was submitted for review 1 ' 'minute ago:' '<br>Question 2 for reviewer 2</li><br>' '</ul><br>' 'Please take some time to review any of the above contributions ' '(if they still need a review) or any other contributions on the ' 'dashboard. We appreciate your help!' '<br><br>' 'Thanks again, and happy reviewing!<br>' '- The Oppia Contributor Dashboard Team' '<br><br>%s' % ( DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, EMAIL_FOOTER, ) ) with self.log_new_error_ctx: with self.swap( utils, 'get_current_utc_datetime', lambda: mocked_current_time ): ( email_manager.send_mail_to_notify_contributor_dashboard_reviewers( [self.reviewer_1_id, self.reviewer_2_id], [ reviewer_1_suggestion_email_infos, reviewer_2_suggestion_email_infos, ], ) ) # Make sure correct emails are sent. messages = self._get_sent_email_messages(self.REVIEWER_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body_reviewer_1) messages = self._get_sent_email_messages(self.REVIEWER_2_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body_reviewer_2) # Make sure correct email models are stored. self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body_reviewer_1, self.reviewer_1_id, self.REVIEWER_1_EMAIL, ) self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body_reviewer_2, self.reviewer_2_id, self.REVIEWER_2_EMAIL, ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.CONTRIBUTOR_DASHBOARD_REVIEWER_EMAILS_IS_ENABLED, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_sent_to_reviewer_with_translation_waiting_a_day_for_review( self, ) -> None: translation_suggestion = ( self._create_translation_suggestion_in_lang_with_html_and_datetime( 'hi', '<p>Sample translation</p>', self.mocked_review_submission_datetime, ) ) reviewable_suggestion_email_info = suggestion_services.create_reviewable_suggestion_email_info_from_suggestion( translation_suggestion ) review_wait_time = 1 reviewable_suggestion_email_info.submission_datetime = ( self.mocked_review_submission_datetime ) mocked_current_time = ( reviewable_suggestion_email_info.submission_datetime + datetime.timedelta(days=review_wait_time) ) expected_email_html_body = ( 'Hi reviewer1,' '<br><br>' 'There are new review opportunities that we think you might be ' 'interested in on the <a href="%s%s">Contributor Dashboard</a>. ' 'Here are some examples of contributions that have been waiting ' 'the longest for review:<br><br>' '<ul>' '<li>The following हिन्दी (Hindi) translation suggestion was ' 'submitted for review 1 day ago:' '<br>Sample translation</li><br>' '</ul><br>' 'Please take some time to review any of the above contributions ' '(if they still need a review) or any other contributions on the ' 'dashboard. We appreciate your help!' '<br><br>' 'Thanks again, and happy reviewing!<br>' '- The Oppia Contributor Dashboard Team' '<br><br>%s' % ( DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, EMAIL_FOOTER, ) ) with self.log_new_error_ctx: with self.swap( utils, 'get_current_utc_datetime', lambda: mocked_current_time ): ( email_manager.send_mail_to_notify_contributor_dashboard_reviewers( [self.reviewer_1_id], [[reviewable_suggestion_email_info]], ) ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.REVIEWER_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body, self.reviewer_1_id, self.REVIEWER_1_EMAIL ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.CONTRIBUTOR_DASHBOARD_REVIEWER_EMAILS_IS_ENABLED, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_sent_to_reviewer_with_translation_waiting_days_for_review( self, ) -> None: translation_suggestion = ( self._create_translation_suggestion_in_lang_with_html_and_datetime( 'hi', '<p>Sample translation</p>', self.mocked_review_submission_datetime, ) ) reviewable_suggestion_email_info = suggestion_services.create_reviewable_suggestion_email_info_from_suggestion( translation_suggestion ) review_wait_time = 5 mocked_current_time = ( reviewable_suggestion_email_info.submission_datetime + datetime.timedelta(days=review_wait_time) ) expected_email_html_body = ( 'Hi reviewer1,' '<br><br>' 'There are new review opportunities that we think you might be ' 'interested in on the <a href="%s%s">Contributor Dashboard</a>. ' 'Here are some examples of contributions that have been waiting ' 'the longest for review:<br><br>' '<ul>' '<li>The following हिन्दी (Hindi) translation suggestion was ' 'submitted for review 5 days ago:' '<br>Sample translation</li><br>' '</ul><br>' 'Please take some time to review any of the above contributions ' '(if they still need a review) or any other contributions on the ' 'dashboard. We appreciate your help!' '<br><br>' 'Thanks again, and happy reviewing!<br>' '- The Oppia Contributor Dashboard Team' '<br><br>%s' % ( DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, EMAIL_FOOTER, ) ) with self.log_new_error_ctx: with self.swap( utils, 'get_current_utc_datetime', lambda: mocked_current_time ): ( email_manager.send_mail_to_notify_contributor_dashboard_reviewers( [self.reviewer_1_id], [[reviewable_suggestion_email_info]], ) ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.REVIEWER_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body, self.reviewer_1_id, self.REVIEWER_1_EMAIL ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.CONTRIBUTOR_DASHBOARD_REVIEWER_EMAILS_IS_ENABLED, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_sent_to_reviewer_with_translation_waiting_an_hour_for_review( self, ) -> None: translation_suggestion = ( self._create_translation_suggestion_in_lang_with_html_and_datetime( 'hi', '<p>Sample translation</p>', self.mocked_review_submission_datetime, ) ) reviewable_suggestion_email_info = suggestion_services.create_reviewable_suggestion_email_info_from_suggestion( translation_suggestion ) review_wait_time = 1 reviewable_suggestion_email_info.submission_datetime = ( self.mocked_review_submission_datetime ) mocked_current_time = ( reviewable_suggestion_email_info.submission_datetime + datetime.timedelta(hours=review_wait_time) ) expected_email_html_body = ( 'Hi reviewer1,' '<br><br>' 'There are new review opportunities that we think you might be ' 'interested in on the <a href="%s%s">Contributor Dashboard</a>. ' 'Here are some examples of contributions that have been waiting ' 'the longest for review:<br><br>' '<ul>' '<li>The following हिन्दी (Hindi) translation suggestion was ' 'submitted for review 1 hour ago:' '<br>Sample translation</li><br>' '</ul><br>' 'Please take some time to review any of the above contributions ' '(if they still need a review) or any other contributions on the ' 'dashboard. We appreciate your help!' '<br><br>' 'Thanks again, and happy reviewing!<br>' '- The Oppia Contributor Dashboard Team' '<br><br>%s' % ( DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, EMAIL_FOOTER, ) ) with self.log_new_error_ctx: with self.swap( utils, 'get_current_utc_datetime', lambda: mocked_current_time ): ( email_manager.send_mail_to_notify_contributor_dashboard_reviewers( [self.reviewer_1_id], [[reviewable_suggestion_email_info]], ) ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.REVIEWER_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body, self.reviewer_1_id, self.REVIEWER_1_EMAIL ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.CONTRIBUTOR_DASHBOARD_REVIEWER_EMAILS_IS_ENABLED, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_sent_to_reviewer_with_translation_waiting_hours_for_review( self, ) -> None: translation_suggestion = ( self._create_translation_suggestion_in_lang_with_html_and_datetime( 'hi', '<p>Sample translation</p>', self.mocked_review_submission_datetime, ) ) reviewable_suggestion_email_info = suggestion_services.create_reviewable_suggestion_email_info_from_suggestion( translation_suggestion ) review_wait_time = 5 reviewable_suggestion_email_info.submission_datetime = ( self.mocked_review_submission_datetime ) mocked_current_time = ( reviewable_suggestion_email_info.submission_datetime + datetime.timedelta(hours=review_wait_time) ) expected_email_html_body = ( 'Hi reviewer1,' '<br><br>' 'There are new review opportunities that we think you might be ' 'interested in on the <a href="%s%s">Contributor Dashboard</a>. ' 'Here are some examples of contributions that have been waiting ' 'the longest for review:<br><br>' '<ul>' '<li>The following हिन्दी (Hindi) translation suggestion was ' 'submitted for review 5 hours ago:' '<br>Sample translation</li><br>' '</ul><br>' 'Please take some time to review any of the above contributions ' '(if they still need a review) or any other contributions on the ' 'dashboard. We appreciate your help!' '<br><br>' 'Thanks again, and happy reviewing!<br>' '- The Oppia Contributor Dashboard Team' '<br><br>%s' % ( DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, EMAIL_FOOTER, ) ) with self.log_new_error_ctx: with self.swap( utils, 'get_current_utc_datetime', lambda: mocked_current_time ): ( email_manager.send_mail_to_notify_contributor_dashboard_reviewers( [self.reviewer_1_id], [[reviewable_suggestion_email_info]], ) ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.REVIEWER_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body, self.reviewer_1_id, self.REVIEWER_1_EMAIL ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.CONTRIBUTOR_DASHBOARD_REVIEWER_EMAILS_IS_ENABLED, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_sent_to_reviewer_with_translation_waiting_a_min_for_review( self, ) -> None: translation_suggestion = ( self._create_translation_suggestion_in_lang_with_html_and_datetime( 'hi', '<p>Sample translation</p>', self.mocked_review_submission_datetime, ) ) reviewable_suggestion_email_info = suggestion_services.create_reviewable_suggestion_email_info_from_suggestion( translation_suggestion ) review_wait_time = 1 reviewable_suggestion_email_info.submission_datetime = ( self.mocked_review_submission_datetime ) mocked_current_time = ( reviewable_suggestion_email_info.submission_datetime + datetime.timedelta(minutes=review_wait_time) ) expected_email_html_body = ( 'Hi reviewer1,' '<br><br>' 'There are new review opportunities that we think you might be ' 'interested in on the <a href="%s%s">Contributor Dashboard</a>. ' 'Here are some examples of contributions that have been waiting ' 'the longest for review:<br><br>' '<ul>' '<li>The following हिन्दी (Hindi) translation suggestion was ' 'submitted for review 1 minute ago:' '<br>Sample translation</li><br>' '</ul><br>' 'Please take some time to review any of the above contributions ' '(if they still need a review) or any other contributions on the ' 'dashboard. We appreciate your help!' '<br><br>' 'Thanks again, and happy reviewing!<br>' '- The Oppia Contributor Dashboard Team' '<br><br>%s' % ( DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, EMAIL_FOOTER, ) ) with self.log_new_error_ctx: with self.swap( utils, 'get_current_utc_datetime', lambda: mocked_current_time ): ( email_manager.send_mail_to_notify_contributor_dashboard_reviewers( [self.reviewer_1_id], [[reviewable_suggestion_email_info]], ) ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.REVIEWER_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body, self.reviewer_1_id, self.REVIEWER_1_EMAIL ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.CONTRIBUTOR_DASHBOARD_REVIEWER_EMAILS_IS_ENABLED, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_sent_to_reviewer_with_translation_waiting_mins_for_review( self, ) -> None: translation_suggestion = ( self._create_translation_suggestion_in_lang_with_html_and_datetime( 'hi', '<p>Sample translation</p>', self.mocked_review_submission_datetime, ) ) reviewable_suggestion_email_info = suggestion_services.create_reviewable_suggestion_email_info_from_suggestion( translation_suggestion ) review_wait_time = 5 reviewable_suggestion_email_info.submission_datetime = ( self.mocked_review_submission_datetime ) mocked_current_time = ( reviewable_suggestion_email_info.submission_datetime + datetime.timedelta(minutes=review_wait_time) ) expected_email_html_body = ( 'Hi reviewer1,' '<br><br>' 'There are new review opportunities that we think you might be ' 'interested in on the <a href="%s%s">Contributor Dashboard</a>. ' 'Here are some examples of contributions that have been waiting ' 'the longest for review:<br><br>' '<ul>' '<li>The following हिन्दी (Hindi) translation suggestion was ' 'submitted for review 5 minutes ago:' '<br>Sample translation</li><br>' '</ul><br>' 'Please take some time to review any of the above contributions ' '(if they still need a review) or any other contributions on the ' 'dashboard. We appreciate your help!' '<br><br>' 'Thanks again, and happy reviewing!<br>' '- The Oppia Contributor Dashboard Team' '<br><br>%s' % ( DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, EMAIL_FOOTER, ) ) with self.log_new_error_ctx: with self.swap( utils, 'get_current_utc_datetime', lambda: mocked_current_time ): ( email_manager.send_mail_to_notify_contributor_dashboard_reviewers( [self.reviewer_1_id], [[reviewable_suggestion_email_info]], ) ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.REVIEWER_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body, self.reviewer_1_id, self.REVIEWER_1_EMAIL ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.CONTRIBUTOR_DASHBOARD_REVIEWER_EMAILS_IS_ENABLED, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_sent_to_reviewer_with_translation_waiting_secs_for_review( self, ) -> None: translation_suggestion = ( self._create_translation_suggestion_in_lang_with_html_and_datetime( 'hi', '<p>Sample translation</p>', self.mocked_review_submission_datetime, ) ) reviewable_suggestion_email_info = suggestion_services.create_reviewable_suggestion_email_info_from_suggestion( translation_suggestion ) review_wait_time = 1 reviewable_suggestion_email_info.submission_datetime = ( self.mocked_review_submission_datetime ) mocked_current_time = ( reviewable_suggestion_email_info.submission_datetime + datetime.timedelta(seconds=review_wait_time) ) expected_email_html_body = ( 'Hi reviewer1,' '<br><br>' 'There are new review opportunities that we think you might be ' 'interested in on the <a href="%s%s">Contributor Dashboard</a>. ' 'Here are some examples of contributions that have been waiting ' 'the longest for review:<br><br>' '<ul>' '<li>The following हिन्दी (Hindi) translation suggestion was ' 'submitted for review 1 minute ago:' '<br>Sample translation</li><br>' '</ul><br>' 'Please take some time to review any of the above contributions ' '(if they still need a review) or any other contributions on the ' 'dashboard. We appreciate your help!' '<br><br>' 'Thanks again, and happy reviewing!<br>' '- The Oppia Contributor Dashboard Team' '<br><br>%s' % ( DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, EMAIL_FOOTER, ) ) with self.log_new_error_ctx: with self.swap( utils, 'get_current_utc_datetime', lambda: mocked_current_time ): ( email_manager.send_mail_to_notify_contributor_dashboard_reviewers( [self.reviewer_1_id], [[reviewable_suggestion_email_info]], ) ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.REVIEWER_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body, self.reviewer_1_id, self.REVIEWER_1_EMAIL ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.CONTRIBUTOR_DASHBOARD_REVIEWER_EMAILS_IS_ENABLED, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_sent_to_reviewer_with_multi_translation_waiting_for_review( self, ) -> None: mocked_current_time = ( self.mocked_review_submission_datetime + datetime.timedelta(days=1, hours=1) ) # Translation suggestion 1 has waited 1 day for review. translation_suggestion_1 = ( self._create_translation_suggestion_in_lang_with_html_and_datetime( 'en', '<p>Translation 1</p>', self.mocked_review_submission_datetime + datetime.timedelta(hours=1), ) ) # Translation suggestion 2 has waited 1 hour for review. translation_suggestion_2 = ( self._create_translation_suggestion_in_lang_with_html_and_datetime( 'fr', '<p>Translation 2</p>', self.mocked_review_submission_datetime + datetime.timedelta(days=1), ) ) reviewable_suggestion_email_infos = ( self._create_reviewable_suggestion_email_infos_from_suggestions( [translation_suggestion_1, translation_suggestion_2] ) ) expected_email_html_body = ( 'Hi reviewer1,' '<br><br>' 'There are new review opportunities that we think you might be ' 'interested in on the <a href="%s%s">Contributor Dashboard</a>. ' 'Here are some examples of contributions that have been waiting ' 'the longest for review:<br><br>' '<ul>' '<li>The following English translation suggestion was submitted ' 'for review 1 day ago:' '<br>Translation 1</li><br>' '<li>The following français (French) translation suggestion was ' 'submitted for review 1 hour ago:' '<br>Translation 2</li><br>' '</ul><br>' 'Please take some time to review any of the above contributions ' '(if they still need a review) or any other contributions on the ' 'dashboard. We appreciate your help!' '<br><br>' 'Thanks again, and happy reviewing!<br>' '- The Oppia Contributor Dashboard Team' '<br><br>%s' % ( DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, EMAIL_FOOTER, ) ) with self.log_new_error_ctx: with self.swap( utils, 'get_current_utc_datetime', lambda: mocked_current_time ): ( email_manager.send_mail_to_notify_contributor_dashboard_reviewers( [self.reviewer_1_id], [reviewable_suggestion_email_infos], ) ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.REVIEWER_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body, self.reviewer_1_id, self.REVIEWER_1_EMAIL ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.CONTRIBUTOR_DASHBOARD_REVIEWER_EMAILS_IS_ENABLED, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_sent_to_multi_reviewers_with_multi_translations_suggestions( self, ) -> None: mocked_current_time = ( self.mocked_review_submission_datetime + datetime.timedelta(days=1, hours=1, minutes=1) ) # Translation suggestion 1 has waited 1 day for review. translation_suggestion_1 = ( self._create_translation_suggestion_in_lang_with_html_and_datetime( 'en', '<p>Translation 1 for reviewer 1</p>', self.mocked_review_submission_datetime + datetime.timedelta(hours=1, minutes=1), ) ) # Translation suggestion 2 has waited 1 hour for review. translation_suggestion_2 = ( self._create_translation_suggestion_in_lang_with_html_and_datetime( 'fr', '<p>Translation 2 for reviewer 1</p>', self.mocked_review_submission_datetime + datetime.timedelta(days=1, minutes=1), ) ) # Translation suggestion 3 has waited 1 minute for review. translation_suggestion_3 = ( self._create_translation_suggestion_in_lang_with_html_and_datetime( 'hi', '<p>Translation 1 for reviewer 2</p>', self.mocked_review_submission_datetime + datetime.timedelta(days=1, hours=1), ) ) # Translation suggestion 4 has waited 1 minute for review. translation_suggestion_4 = ( self._create_translation_suggestion_in_lang_with_html_and_datetime( 'de', '<p>Translation 2 for reviewer 2</p>', self.mocked_review_submission_datetime + datetime.timedelta(days=1, hours=1), ) ) reviewer_1_suggestion_email_infos = ( self._create_reviewable_suggestion_email_infos_from_suggestions( [translation_suggestion_1, translation_suggestion_2] ) ) reviewer_2_suggestion_email_infos = ( self._create_reviewable_suggestion_email_infos_from_suggestions( [translation_suggestion_3, translation_suggestion_4] ) ) expected_email_html_body_reviewer_1 = ( 'Hi reviewer1,' '<br><br>' 'There are new review opportunities that we think you might be ' 'interested in on the <a href="%s%s">Contributor Dashboard</a>. ' 'Here are some examples of contributions that have been waiting ' 'the longest for review:<br><br>' '<ul>' '<li>The following English translation suggestion was submitted ' 'for review 1 day ago:' '<br>Translation 1 for reviewer 1</li><br>' '<li>The following français (French) translation suggestion was ' 'submitted for review 1 hour ago:' '<br>Translation 2 for reviewer 1</li><br>' '</ul><br>' 'Please take some time to review any of the above contributions ' '(if they still need a review) or any other contributions on the ' 'dashboard. We appreciate your help!' '<br><br>' 'Thanks again, and happy reviewing!<br>' '- The Oppia Contributor Dashboard Team' '<br><br>%s' % ( DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, EMAIL_FOOTER, ) ) expected_email_html_body_reviewer_2 = ( 'Hi reviewer2,' '<br><br>' 'There are new review opportunities that we think you might be ' 'interested in on the <a href="%s%s">Contributor Dashboard</a>. ' 'Here are some examples of contributions that have been waiting ' 'the longest for review:<br><br>' '<ul>' '<li>The following हिन्दी (Hindi) translation suggestion was ' 'submitted for review 1 minute ago:' '<br>Translation 1 for reviewer 2</li><br>' '<li>The following Deutsch (German) translation suggestion was ' 'submitted for review 1 minute ago:' '<br>Translation 2 for reviewer 2</li><br>' '</ul><br>' 'Please take some time to review any of the above contributions ' '(if they still need a review) or any other contributions on the ' 'dashboard. We appreciate your help!' '<br><br>' 'Thanks again, and happy reviewing!<br>' '- The Oppia Contributor Dashboard Team' '<br><br>%s' % ( DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, EMAIL_FOOTER, ) ) with self.log_new_error_ctx: with self.swap( utils, 'get_current_utc_datetime', lambda: mocked_current_time ): ( email_manager.send_mail_to_notify_contributor_dashboard_reviewers( [self.reviewer_1_id, self.reviewer_2_id], [ reviewer_1_suggestion_email_infos, reviewer_2_suggestion_email_infos, ], ) ) # Make sure correct emails are sent. messages = self._get_sent_email_messages(self.REVIEWER_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body_reviewer_1) messages = self._get_sent_email_messages(self.REVIEWER_2_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body_reviewer_2) # Make sure correct email models are stored. self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body_reviewer_1, self.reviewer_1_id, self.REVIEWER_1_EMAIL, ) self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body_reviewer_2, self.reviewer_2_id, self.REVIEWER_2_EMAIL, ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.CONTRIBUTOR_DASHBOARD_REVIEWER_EMAILS_IS_ENABLED, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_sent_to_multi_reviewers_with_multi_suggestions_waiting( self, ) -> None: mocked_current_time = ( self.mocked_review_submission_datetime + datetime.timedelta(days=1, hours=1, minutes=1) ) # Suggestion 1 has waited 1 day for review. suggestion_1 = ( self._create_translation_suggestion_in_lang_with_html_and_datetime( 'en', '<p>Translation 1</p>', self.mocked_review_submission_datetime + datetime.timedelta(hours=1, minutes=1), ) ) # Suggestion 2 has waited 1 hour for review. suggestion_2 = ( self._create_question_suggestion_with_question_html_and_datetime( '<p>Question 1</p>', self.mocked_review_submission_datetime + datetime.timedelta(days=1, minutes=1), ) ) # Suggestion 3 has waited 1 minute for review. suggestion_3 = ( self._create_translation_suggestion_in_lang_with_html_and_datetime( 'fr', '<p>Translation 2</p>', self.mocked_review_submission_datetime + datetime.timedelta(days=1, hours=1), ) ) # Suggestion 4 has waited 1 minute for review. suggestion_4 = ( self._create_question_suggestion_with_question_html_and_datetime( '<p>Question 2</p>', self.mocked_review_submission_datetime + datetime.timedelta(days=1, hours=1), ) ) reviewer_1_suggestion_email_infos = ( self._create_reviewable_suggestion_email_infos_from_suggestions( [suggestion_1, suggestion_2] ) ) reviewer_2_suggestion_email_infos = ( self._create_reviewable_suggestion_email_infos_from_suggestions( [suggestion_3, suggestion_4] ) ) expected_email_html_body_reviewer_1 = ( 'Hi reviewer1,' '<br><br>' 'There are new review opportunities that we think you might be ' 'interested in on the <a href="%s%s">Contributor Dashboard</a>. ' 'Here are some examples of contributions that have been waiting ' 'the longest for review:<br><br>' '<ul>' '<li>The following English translation suggestion was submitted ' 'for review 1 day ago:' '<br>Translation 1</li><br>' '<li>The following question suggestion was submitted for ' 'review 1 hour ago:' '<br>Question 1</li><br>' '</ul><br>' 'Please take some time to review any of the above contributions ' '(if they still need a review) or any other contributions on the ' 'dashboard. We appreciate your help!' '<br><br>' 'Thanks again, and happy reviewing!<br>' '- The Oppia Contributor Dashboard Team' '<br><br>%s' % ( DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, EMAIL_FOOTER, ) ) expected_email_html_body_reviewer_2 = ( 'Hi reviewer2,' '<br><br>' 'There are new review opportunities that we think you might be ' 'interested in on the <a href="%s%s">Contributor Dashboard</a>. ' 'Here are some examples of contributions that have been waiting ' 'the longest for review:<br><br>' '<ul>' '<li>The following français (French) translation suggestion was ' 'submitted for review 1 minute ago:' '<br>Translation 2</li><br>' '<li>The following question suggestion was submitted for ' 'review 1 minute ago:' '<br>Question 2</li><br>' '</ul><br>' 'Please take some time to review any of the above contributions ' '(if they still need a review) or any other contributions on the ' 'dashboard. We appreciate your help!' '<br><br>' 'Thanks again, and happy reviewing!<br>' '- The Oppia Contributor Dashboard Team' '<br><br>%s' % ( DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, EMAIL_FOOTER, ) ) with self.log_new_error_ctx: with self.swap( utils, 'get_current_utc_datetime', lambda: mocked_current_time ): ( email_manager.send_mail_to_notify_contributor_dashboard_reviewers( [self.reviewer_1_id, self.reviewer_2_id], [ reviewer_1_suggestion_email_infos, reviewer_2_suggestion_email_infos, ], ) ) # Make sure correct emails are sent. messages = self._get_sent_email_messages(self.REVIEWER_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body_reviewer_1) messages = self._get_sent_email_messages(self.REVIEWER_2_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body_reviewer_2) # Make sure correct email models are stored. self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body_reviewer_1, self.reviewer_1_id, self.REVIEWER_1_EMAIL, ) self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body_reviewer_2, self.reviewer_2_id, self.REVIEWER_2_EMAIL, ) class NotifyAdminsSuggestionsWaitingTooLongForReviewEmailTests( test_utils.EmailTestBase ): """Tests the send_mail_to_notify_admins_suggestions_waiting_long method, which sends an email to admins with information regarding the suggestions that have waited longer than suggestion_models.SUGGESTION_REVIEW_WAIT_TIME_THRESHOLD_IN_DAYS days for review on the Contributor Dashboard. """ target_id: str = 'exp1' skill_id: str = 'skill_123456' mocked_review_submission_datetime: datetime.datetime = datetime.datetime( 2020, 6, 15, 5 ) AUTHOR_USERNAME: Final = 'author' AUTHOR_EMAIL: Final = 'author@example.com' CURRICULUM_ADMIN_1_USERNAME: Final = 'user1' CURRICULUM_ADMIN_1_EMAIL: Final = 'user1@community.org' CURRICULUM_ADMIN_2_USERNAME: Final = 'user2' CURRICULUM_ADMIN_2_EMAIL: Final = 'user2@community.org' def _create_translation_suggestion_in_lang_with_html_and_datetime( self, language_code: str, translation_html: str, submission_datetime: datetime.datetime, ) -> suggestion_registry.BaseSuggestion: """Creates a translation suggestion in the given language_code with the given translation html and submission datetime. """ add_translation_change_dict = { 'cmd': exp_domain.CMD_ADD_WRITTEN_TRANSLATION, 'state_name': feconf.DEFAULT_INIT_STATE_NAME, 'content_id': 'content_0', 'language_code': language_code, 'content_html': feconf.DEFAULT_STATE_CONTENT_STR, 'translation_html': translation_html, 'data_format': 'html', } with self.swap( utils, 'get_current_utc_datetime', lambda: submission_datetime ): translation_suggestion = suggestion_services.create_suggestion( feconf.SUGGESTION_TYPE_TRANSLATE_CONTENT, feconf.ENTITY_TYPE_EXPLORATION, self.target_id, feconf.CURRENT_STATE_SCHEMA_VERSION, self.author_id, add_translation_change_dict, 'test description', ) return translation_suggestion def _create_question_suggestion_with_question_html_and_datetime( self, question_html: str, submission_datetime: datetime.datetime ) -> suggestion_registry.BaseSuggestion: """Creates a question suggestion with the given question html and submission datetime. """ with self.swap(feconf, 'DEFAULT_STATE_CONTENT_STR', question_html): content_id_generator = translation_domain.ContentIdGenerator() add_question_change_dict: Dict[ str, Union[str, float, question_domain.QuestionDict] ] = { 'cmd': ( question_domain.CMD_CREATE_NEW_FULLY_SPECIFIED_QUESTION ), 'question_dict': { 'id': 'test_id', 'version': 12, 'question_state_data': self._create_valid_question_data( 'default_state', content_id_generator ).to_dict(), 'language_code': constants.DEFAULT_LANGUAGE_CODE, 'question_state_data_schema_version': ( feconf.CURRENT_STATE_SCHEMA_VERSION ), 'linked_skill_ids': ['skill_1'], 'inapplicable_skill_misconception_ids': ['skillid12345-1'], 'next_content_id_index': ( content_id_generator.next_content_id_index ), }, 'skill_id': self.skill_id, 'skill_difficulty': 0.3, } with self.swap( utils, 'get_current_utc_datetime', lambda: submission_datetime ): question_suggestion = suggestion_services.create_suggestion( feconf.SUGGESTION_TYPE_ADD_QUESTION, feconf.ENTITY_TYPE_SKILL, self.skill_id, feconf.CURRENT_STATE_SCHEMA_VERSION, self.author_id, add_question_change_dict, 'test description', ) return question_suggestion def _create_reviewable_suggestion_email_infos_from_suggestions( self, suggestions: List[suggestion_registry.BaseSuggestion] ) -> List[suggestion_registry.ReviewableSuggestionEmailInfo]: """Creates a list of ReviewableSuggestionEmailInfo objects from the given suggestions. """ return [ ( suggestion_services.create_reviewable_suggestion_email_info_from_suggestion( suggestion ) ) for suggestion in suggestions ] def _assert_email_data_stored_in_sent_email_model_is_correct( self, expected_email_html_body: str, admin_id: Optional[str], admin_email: str, ) -> None: """Asserts that the created sent email model from the sent email contains the right information. """ sent_email_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all() .filter(email_models.SentEmailModel.recipient_id == admin_id) .fetch() ) self.assertEqual(len(sent_email_models), 1) sent_email_model = sent_email_models[0] self.assertEqual( sent_email_model.subject, ( email_manager.ADMIN_NOTIFICATION_FOR_SUGGESTIONS_NEEDING_REVIEW_EMAIL_DATA[ 'email_subject' ] ), ) self.assertEqual(sent_email_model.recipient_id, admin_id) self.assertEqual(sent_email_model.recipient_email, admin_email) self.assertEqual(sent_email_model.html_body, expected_email_html_body) self.assertEqual(sent_email_model.sender_id, feconf.SYSTEM_COMMITTER_ID) self.assertEqual( sent_email_model.sender_email, 'Site Admin <noreply@example.com>' ) self.assertEqual( sent_email_model.intent, feconf.EMAIL_INTENT_ADDRESS_CONTRIBUTOR_DASHBOARD_SUGGESTIONS, ) def _mock_logging_info(self, msg: str, *args: str) -> None: """Mocks logging.info() by appending the log message to the logged info list. """ self.logged_info.append(msg % args) def setUp(self) -> None: super().setUp() self.signup(self.AUTHOR_EMAIL, self.AUTHOR_USERNAME) self.author_id = self.get_user_id_from_email(self.AUTHOR_EMAIL) self.signup( self.CURRICULUM_ADMIN_1_EMAIL, self.CURRICULUM_ADMIN_1_USERNAME ) self.admin_1_id = self.get_user_id_from_email( self.CURRICULUM_ADMIN_1_EMAIL ) self.signup( self.CURRICULUM_ADMIN_2_EMAIL, self.CURRICULUM_ADMIN_2_USERNAME ) self.admin_2_id = self.get_user_id_from_email( self.CURRICULUM_ADMIN_2_EMAIL ) self.log_new_error_counter = test_utils.CallCounter(logging.error) self.log_new_error_ctx = self.swap( logging, 'error', self.log_new_error_counter ) self.logged_info: List[str] = [] self.log_new_info_ctx = self.swap( logging, 'info', self._mock_logging_info ) self.save_new_valid_exploration(self.target_id, self.author_id) self.save_new_skill(self.skill_id, self.author_id) question_suggestion = ( self._create_question_suggestion_with_question_html_and_datetime( '<p>What is the meaning of life?</p>', self.mocked_review_submission_datetime, ) ) self.reviewable_suggestion_email_info = suggestion_services.create_reviewable_suggestion_email_info_from_suggestion( question_suggestion ) @test_utils.set_platform_parameters( [(param_list.ParamName.SERVER_CAN_SEND_EMAILS, False)] ) def test_email_not_sent_if_server_can_send_emails_is_false(self) -> None: with self.capture_logging(min_level=logging.ERROR) as logs: with self.log_new_error_ctx: with self.swap( suggestion_models, 'SUGGESTION_REVIEW_WAIT_TIME_THRESHOLD_IN_DAYS', 0, ): ( email_manager.send_mail_to_notify_admins_suggestions_waiting_long( [self.admin_1_id], [], [], [self.reviewable_suggestion_email_info], ) ) messages = self._get_all_sent_email_messages() self.assertEqual(len(messages), 0) self.assertEqual(self.log_new_error_counter.times_called, 1) self.assertEqual(logs[0], 'This app cannot send emails to users.') @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), ( param_list.ParamName.ENABLE_ADMIN_NOTIFICATIONS_FOR_SUGGESTIONS_NEEDING_REVIEW, False, ), # pylint: disable=line-too-long ] ) def test_email_not_sent_if_notifying_admins_about_suggestions_is_disabled( self, ) -> None: with self.capture_logging(min_level=logging.ERROR) as logs: with self.log_new_error_ctx: with self.swap( suggestion_models, 'SUGGESTION_REVIEW_WAIT_TIME_THRESHOLD_IN_DAYS', 0, ): ( email_manager.send_mail_to_notify_admins_suggestions_waiting_long( [self.admin_1_id], [], [], [self.reviewable_suggestion_email_info], ) ) messages = self._get_all_sent_email_messages() self.assertEqual(len(messages), 0) self.assertEqual(self.log_new_error_counter.times_called, 1) self.assertEqual( logs[0], 'The "notify_admins_suggestions_waiting_too_long" property ' 'must be enabled on the admin config page in order to send ' 'admins the emails.', ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), ( param_list.ParamName.ENABLE_ADMIN_NOTIFICATIONS_FOR_SUGGESTIONS_NEEDING_REVIEW, True, ), # pylint: disable=line-too-long (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_not_sent_if_admin_email_does_not_exist(self) -> None: with self.capture_logging(min_level=logging.ERROR) as logs: with self.log_new_error_ctx: with self.swap( suggestion_models, 'SUGGESTION_REVIEW_WAIT_TIME_THRESHOLD_IN_DAYS', 0, ): ( email_manager.send_mail_to_notify_admins_suggestions_waiting_long( ['admin_id_without_email'], [], [], [self.reviewable_suggestion_email_info], ) ) messages = self._get_all_sent_email_messages() self.assertEqual(len(messages), 0) self.assertEqual(self.log_new_error_counter.times_called, 1) self.assertEqual( logs[0], 'There was no email for the given admin id: ' 'admin_id_without_email.', ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), ( param_list.ParamName.ENABLE_ADMIN_NOTIFICATIONS_FOR_SUGGESTIONS_NEEDING_REVIEW, True, ), # pylint: disable=line-too-long ] ) def test_email_not_sent_if_no_admins_to_notify(self) -> None: with self.capture_logging(min_level=logging.ERROR) as logs: with self.log_new_error_ctx: with self.swap( suggestion_models, 'SUGGESTION_REVIEW_WAIT_TIME_THRESHOLD_IN_DAYS', 0, ): ( email_manager.send_mail_to_notify_admins_suggestions_waiting_long( [], [], [], [self.reviewable_suggestion_email_info] ) ) messages = self._get_all_sent_email_messages() self.assertEqual(len(messages), 0) self.assertEqual(self.log_new_error_counter.times_called, 1) self.assertEqual(logs[0], 'There were no admins to notify.') @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.ENABLE_ADMIN_NOTIFICATIONS_FOR_SUGGESTIONS_NEEDING_REVIEW, True, ), # pylint: disable=line-too-long ] ) def test_email_not_sent_if_no_suggestions_to_notify_the_admin_about( self, ) -> None: with self.log_new_info_ctx: with self.swap( suggestion_models, 'SUGGESTION_REVIEW_WAIT_TIME_THRESHOLD_IN_DAYS', 0, ): ( email_manager.send_mail_to_notify_admins_suggestions_waiting_long( [self.admin_1_id], [], [], [] ) ) messages = self._get_all_sent_email_messages() self.assertEqual(len(messages), 0) self.assertEqual( self.logged_info[0], 'There were no Contributor Dashboard suggestions that were waiting ' 'too long for a review.', ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.ENABLE_ADMIN_NOTIFICATIONS_FOR_SUGGESTIONS_NEEDING_REVIEW, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_sent_to_admin_if_question_has_waited_too_long_for_a_review( self, ) -> None: question_suggestion = ( self._create_question_suggestion_with_question_html_and_datetime( '<p>What is the meaning of life?</p>', self.mocked_review_submission_datetime, ) ) reviewable_suggestion_email_info = suggestion_services.create_reviewable_suggestion_email_info_from_suggestion( question_suggestion ) review_wait_time = 5 mocked_current_time = ( reviewable_suggestion_email_info.submission_datetime + datetime.timedelta(days=review_wait_time) ) expected_email_html_body = ( 'Hi user1,' '<br><br>' 'There are suggestions on the ' '<a href="%s%s">Contributor Dashboard</a> that have been waiting ' 'for more than 0 days for review. Please take a look at the ' 'suggestions mentioned below and help them get reviewed by going ' 'to the <a href="%s%s">admin roles page</a> and either:' '<br><br><ul>' '<li>Add more reviewers to the suggestion types that have ' 'suggestions waiting too long for a review</li><br>' '<li>Find the existing reviewers and email reviewers directly ' 'about the suggestions waiting for a review</li><br>' '</ul><br>' 'Here are the suggestions that have been waiting too long for a ' 'review:' '<br><br>' '<ul>' '<li>The following question suggestion was submitted for review 5 ' 'days ago:' '<br>What is the meaning of life?</li><br>' '</ul><br>' 'Thanks so much - we appreciate your help!<br>' 'Best Wishes!<br><br>' '- The Oppia Contributor Dashboard Team' % ( DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_ADMIN_URL, ) ) with self.log_new_error_ctx: with self.swap( suggestion_models, 'SUGGESTION_REVIEW_WAIT_TIME_THRESHOLD_IN_DAYS', 0, ): with self.swap( utils, 'get_current_utc_datetime', lambda: mocked_current_time, ): ( email_manager.send_mail_to_notify_admins_suggestions_waiting_long( [self.admin_1_id], [], [], [reviewable_suggestion_email_info], ) ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.CURRICULUM_ADMIN_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body, self.admin_1_id, self.CURRICULUM_ADMIN_1_EMAIL, ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.ENABLE_ADMIN_NOTIFICATIONS_FOR_SUGGESTIONS_NEEDING_REVIEW, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_sent_to_admin_if_multiple_questions_have_waited_for_review( self, ) -> None: mocked_current_time = ( self.mocked_review_submission_datetime + datetime.timedelta(days=2, hours=1) ) question_suggestion_1 = ( self._create_question_suggestion_with_question_html_and_datetime( '<p>Question 1</p>', self.mocked_review_submission_datetime ) ) # Question suggestion 2 has waited slighlty less time than question # suggestion 1 so that the question suggestions are not # indistinguishable in terms of review wait time. question_suggestion_2 = ( self._create_question_suggestion_with_question_html_and_datetime( '<p>Question 2</p>', self.mocked_review_submission_datetime + datetime.timedelta(hours=1), ) ) reviewable_suggestion_email_infos = ( self._create_reviewable_suggestion_email_infos_from_suggestions( [question_suggestion_1, question_suggestion_2] ) ) expected_email_html_body = ( 'Hi user1,' '<br><br>' 'There are suggestions on the ' '<a href="%s%s">Contributor Dashboard</a> that have been waiting ' 'for more than 0 days for review. Please take a look at the ' 'suggestions mentioned below and help them get reviewed by going ' 'to the <a href="%s%s">admin roles page</a> and either:' '<br><br><ul>' '<li>Add more reviewers to the suggestion types that have ' 'suggestions waiting too long for a review</li><br>' '<li>Find the existing reviewers and email reviewers directly ' 'about the suggestions waiting for a review</li><br>' '</ul><br>' 'Here are the suggestions that have been waiting too long for a ' 'review:' '<br><br>' '<ul>' '<li>The following question suggestion was submitted for review 2 ' 'days ago:' '<br>Question 1</li><br>' '<li>The following question suggestion was submitted for review 2 ' 'days ago:' '<br>Question 2</li><br>' '</ul><br>' 'Thanks so much - we appreciate your help!<br>' 'Best Wishes!<br><br>' '- The Oppia Contributor Dashboard Team' % ( DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_ADMIN_URL, ) ) with self.log_new_error_ctx: with self.swap( suggestion_models, 'SUGGESTION_REVIEW_WAIT_TIME_THRESHOLD_IN_DAYS', 0, ): with self.swap( utils, 'get_current_utc_datetime', lambda: mocked_current_time, ): ( email_manager.send_mail_to_notify_admins_suggestions_waiting_long( [self.admin_1_id], [], [], reviewable_suggestion_email_infos, ) ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.CURRICULUM_ADMIN_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body, self.admin_1_id, self.CURRICULUM_ADMIN_1_EMAIL, ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.ENABLE_ADMIN_NOTIFICATIONS_FOR_SUGGESTIONS_NEEDING_REVIEW, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_sent_to_admin_if_translation_has_waited_too_long_for_review( self, ) -> None: translation_suggestion = ( self._create_translation_suggestion_in_lang_with_html_and_datetime( 'hi', '<p>Sample translation</p>', self.mocked_review_submission_datetime, ) ) reviewable_suggestion_email_info = suggestion_services.create_reviewable_suggestion_email_info_from_suggestion( translation_suggestion ) review_wait_time = 5 mocked_current_time = ( reviewable_suggestion_email_info.submission_datetime + datetime.timedelta(days=review_wait_time) ) expected_email_html_body = ( 'Hi user1,' '<br><br>' 'There are suggestions on the ' '<a href="%s%s">Contributor Dashboard</a> that have been waiting ' 'for more than 0 days for review. Please take a look at the ' 'suggestions mentioned below and help them get reviewed by going ' 'to the <a href="%s%s">admin roles page</a> and either:' '<br><br><ul>' '<li>Add more reviewers to the suggestion types that have ' 'suggestions waiting too long for a review</li><br>' '<li>Find the existing reviewers and email reviewers directly ' 'about the suggestions waiting for a review</li><br>' '</ul><br>' 'Here are the suggestions that have been waiting too long for a ' 'review:' '<br><br>' '<ul>' '<li>The following हिन्दी (Hindi) translation suggestion was ' 'submitted for review 5 days ago:' '<br>Sample translation</li><br>' '</ul><br>' 'Thanks so much - we appreciate your help!<br>' 'Best Wishes!<br><br>' '- The Oppia Contributor Dashboard Team' % ( DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_ADMIN_URL, ) ) with self.log_new_error_ctx: with self.swap( suggestion_models, 'SUGGESTION_REVIEW_WAIT_TIME_THRESHOLD_IN_DAYS', 0, ): with self.swap( utils, 'get_current_utc_datetime', lambda: mocked_current_time, ): ( email_manager.send_mail_to_notify_admins_suggestions_waiting_long( [self.admin_1_id], [], [], [reviewable_suggestion_email_info], ) ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.CURRICULUM_ADMIN_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body, self.admin_1_id, self.CURRICULUM_ADMIN_1_EMAIL, ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.ENABLE_ADMIN_NOTIFICATIONS_FOR_SUGGESTIONS_NEEDING_REVIEW, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_sent_to_admin_if_multi_translations_have_waited_for_review( self, ) -> None: mocked_current_time = ( self.mocked_review_submission_datetime + datetime.timedelta(days=2, hours=1) ) translation_suggestion_1 = ( self._create_translation_suggestion_in_lang_with_html_and_datetime( 'en', '<p>Translation 1</p>', self.mocked_review_submission_datetime, ) ) # Translation suggestion 2 has waited slighlty less time than # translation suggestion 1 so that the translation suggestions are not # indistinguishable in terms of review wait time. translation_suggestion_2 = ( self._create_translation_suggestion_in_lang_with_html_and_datetime( 'fr', '<p>Translation 2</p>', self.mocked_review_submission_datetime + datetime.timedelta(hours=1), ) ) reviewable_suggestion_email_infos = ( self._create_reviewable_suggestion_email_infos_from_suggestions( [translation_suggestion_1, translation_suggestion_2] ) ) expected_email_html_body = ( 'Hi user1,' '<br><br>' 'There are suggestions on the ' '<a href="%s%s">Contributor Dashboard</a> that have been waiting ' 'for more than 0 days for review. Please take a look at the ' 'suggestions mentioned below and help them get reviewed by going ' 'to the <a href="%s%s">admin roles page</a> and either:' '<br><br><ul>' '<li>Add more reviewers to the suggestion types that have ' 'suggestions waiting too long for a review</li><br>' '<li>Find the existing reviewers and email reviewers directly ' 'about the suggestions waiting for a review</li><br>' '</ul><br>' 'Here are the suggestions that have been waiting too long for a ' 'review:' '<br><br>' '<ul>' '<li>The following English translation suggestion was submitted ' 'for review 2 days ago:' '<br>Translation 1</li><br>' '<li>The following français (French) translation suggestion was ' 'submitted for review 2 days ago:' '<br>Translation 2</li><br>' '</ul><br>' 'Thanks so much - we appreciate your help!<br>' 'Best Wishes!<br><br>' '- The Oppia Contributor Dashboard Team' % ( DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_ADMIN_URL, ) ) with self.log_new_error_ctx: with self.swap( suggestion_models, 'SUGGESTION_REVIEW_WAIT_TIME_THRESHOLD_IN_DAYS', 0, ): with self.swap( utils, 'get_current_utc_datetime', lambda: mocked_current_time, ): ( email_manager.send_mail_to_notify_admins_suggestions_waiting_long( [self.admin_1_id], [], [], reviewable_suggestion_email_infos, ) ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.CURRICULUM_ADMIN_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body, self.admin_1_id, self.CURRICULUM_ADMIN_1_EMAIL, ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.ENABLE_ADMIN_NOTIFICATIONS_FOR_SUGGESTIONS_NEEDING_REVIEW, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_sent_to_admin_if_multi_suggestion_types_waiting_for_review( self, ) -> None: mocked_current_time = ( self.mocked_review_submission_datetime + datetime.timedelta(days=2, hours=1, minutes=5) ) suggestion_1 = ( self._create_translation_suggestion_in_lang_with_html_and_datetime( 'en', '<p>Translation 1</p>', self.mocked_review_submission_datetime, ) ) # Suggestion 2 has waited slighlty less time than suggestion 1 so that # the suggestions are not indistinguishable in terms of review wait # time. suggestion_2 = ( self._create_translation_suggestion_in_lang_with_html_and_datetime( 'fr', '<p>Translation 2</p>', self.mocked_review_submission_datetime + datetime.timedelta(minutes=5), ) ) # Similarly, suggestion 3 has waited less than both suggestion 1 and # suggestion 2 so that the suggestions are not indistinguishable in # terms of review wait time. suggestion_3 = ( self._create_question_suggestion_with_question_html_and_datetime( '<p>Question 1</p>', self.mocked_review_submission_datetime + datetime.timedelta(hours=1), ) ) reviewable_suggestion_email_infos = ( self._create_reviewable_suggestion_email_infos_from_suggestions( [suggestion_1, suggestion_2, suggestion_3] ) ) expected_email_html_body = ( 'Hi user1,' '<br><br>' 'There are suggestions on the ' '<a href="%s%s">Contributor Dashboard</a> that have been waiting ' 'for more than 0 days for review. Please take a look at the ' 'suggestions mentioned below and help them get reviewed by going ' 'to the <a href="%s%s">admin roles page</a> and either:' '<br><br><ul>' '<li>Add more reviewers to the suggestion types that have ' 'suggestions waiting too long for a review</li><br>' '<li>Find the existing reviewers and email reviewers directly ' 'about the suggestions waiting for a review</li><br>' '</ul><br>' 'Here are the suggestions that have been waiting too long for a ' 'review:' '<br><br>' '<ul>' '<li>The following English translation suggestion was submitted ' 'for review 2 days ago:' '<br>Translation 1</li><br>' '<li>The following français (French) translation suggestion was ' 'submitted for review 2 days ago:' '<br>Translation 2</li><br>' '</ul><br>' 'Thanks so much - we appreciate your help!<br>' 'Best Wishes!<br><br>' '- The Oppia Contributor Dashboard Team' % ( DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_ADMIN_URL, ) ) with self.log_new_error_ctx: with self.swap( suggestion_models, 'SUGGESTION_REVIEW_WAIT_TIME_THRESHOLD_IN_DAYS', 0, ): with self.swap( utils, 'get_current_utc_datetime', lambda: mocked_current_time, ): ( email_manager.send_mail_to_notify_admins_suggestions_waiting_long( [self.admin_1_id], [], [], reviewable_suggestion_email_infos, ) ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.CURRICULUM_ADMIN_1_EMAIL) messages.sort(key=lambda m: m.html) self.assertEqual(len(messages), 2) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. sent_email_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all() .filter(email_models.SentEmailModel.recipient_id == self.admin_1_id) .fetch() ) self.assertEqual(len(sent_email_models), 2) # Here, we are narrowing down the type of 'sent_email_models' from # Sequence to List. Because Sequence is a broader type and it does not # contain extra methods (e.g: .sort()), and below we are using .sort() # method by assuming that 'sent_email_models' is an instance of list. # So, to avoid error we used assertion here. assert isinstance(sent_email_models, list) email_sort_fn: Callable[[email_models.SentEmailModel], str] = ( lambda m: (m.html_body if isinstance(m.html_body, str) else '') ) sent_email_models.sort(key=email_sort_fn) sent_email_model = sent_email_models[0] self.assertEqual( sent_email_model.subject, email_manager.ADMIN_NOTIFICATION_FOR_SUGGESTIONS_NEEDING_REVIEW_EMAIL_DATA[ 'email_subject' ], ) self.assertEqual(sent_email_model.recipient_id, self.admin_1_id) self.assertEqual( sent_email_model.recipient_email, self.CURRICULUM_ADMIN_1_EMAIL ) self.assertEqual(sent_email_model.html_body, expected_email_html_body) self.assertEqual(sent_email_model.sender_id, feconf.SYSTEM_COMMITTER_ID) self.assertEqual( sent_email_model.sender_email, 'Site Admin <noreply@example.com>' ) self.assertEqual( sent_email_model.intent, feconf.EMAIL_INTENT_ADDRESS_CONTRIBUTOR_DASHBOARD_SUGGESTIONS, ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.ENABLE_ADMIN_NOTIFICATIONS_FOR_SUGGESTIONS_NEEDING_REVIEW, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_sent_to_multiple_admins(self) -> None: question_suggestion = ( self._create_question_suggestion_with_question_html_and_datetime( '<p>What is the meaning of life?</p>', self.mocked_review_submission_datetime, ) ) reviewable_suggestion_email_info = suggestion_services.create_reviewable_suggestion_email_info_from_suggestion( question_suggestion ) review_wait_time = 5 mocked_current_time = ( reviewable_suggestion_email_info.submission_datetime + datetime.timedelta(days=review_wait_time) ) expected_email_html_body_admin_1 = ( 'Hi user1,' '<br><br>' 'There are suggestions on the ' '<a href="%s%s">Contributor Dashboard</a> that have been waiting ' 'for more than 0 days for review. Please take a look at the ' 'suggestions mentioned below and help them get reviewed by going ' 'to the <a href="%s%s">admin roles page</a> and either:' '<br><br><ul>' '<li>Add more reviewers to the suggestion types that have ' 'suggestions waiting too long for a review</li><br>' '<li>Find the existing reviewers and email reviewers directly ' 'about the suggestions waiting for a review</li><br>' '</ul><br>' 'Here are the suggestions that have been waiting too long for a ' 'review:' '<br><br>' '<ul>' '<li>The following question suggestion was submitted for review 5 ' 'days ago:' '<br>What is the meaning of life?</li><br>' '</ul><br>' 'Thanks so much - we appreciate your help!<br>' 'Best Wishes!<br><br>' '- The Oppia Contributor Dashboard Team' % ( DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_ADMIN_URL, ) ) expected_email_html_body_admin_2 = ( 'Hi user2,' '<br><br>' 'There are suggestions on the ' '<a href="%s%s">Contributor Dashboard</a> that have been waiting ' 'for more than 0 days for review. Please take a look at the ' 'suggestions mentioned below and help them get reviewed by going ' 'to the <a href="%s%s">admin roles page</a> and either:' '<br><br><ul>' '<li>Add more reviewers to the suggestion types that have ' 'suggestions waiting too long for a review</li><br>' '<li>Find the existing reviewers and email reviewers directly ' 'about the suggestions waiting for a review</li><br>' '</ul><br>' 'Here are the suggestions that have been waiting too long for a ' 'review:' '<br><br>' '<ul>' '<li>The following question suggestion was submitted for review 5 ' 'days ago:' '<br>What is the meaning of life?</li><br>' '</ul><br>' 'Thanks so much - we appreciate your help!<br>' 'Best Wishes!<br><br>' '- The Oppia Contributor Dashboard Team' % ( DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_ADMIN_URL, ) ) with self.log_new_error_ctx: with self.swap( suggestion_models, 'SUGGESTION_REVIEW_WAIT_TIME_THRESHOLD_IN_DAYS', 0, ): with self.swap( utils, 'get_current_utc_datetime', lambda: mocked_current_time, ): ( email_manager.send_mail_to_notify_admins_suggestions_waiting_long( [self.admin_1_id, self.admin_2_id], [], [], [reviewable_suggestion_email_info], ) ) # Make sure correct emails are sent. messages = self._get_sent_email_messages(self.CURRICULUM_ADMIN_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body_admin_1) messages = self._get_sent_email_messages(self.CURRICULUM_ADMIN_2_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body_admin_2) # Make sure correct email models are stored. self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body_admin_1, self.admin_1_id, self.CURRICULUM_ADMIN_1_EMAIL, ) self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body_admin_2, self.admin_2_id, self.CURRICULUM_ADMIN_2_EMAIL, ) class NotifyReviewersNewSuggestionsTests(test_utils.EmailTestBase): """Tests the send_mail_to_notify_contributor_dashboard_reviewers method, which sends an email to reviewers with information regarding the suggestions that have waited the longest for review. """ target_id: str = 'exp1' skill_id: str = 'skill_123456' timezone = datetime.timezone.utc mocked_review_submission_datetime: datetime.datetime = datetime.datetime( 2023, 10, 23, 5, tzinfo=timezone ) mocked_review_submission_datetime = ( mocked_review_submission_datetime.replace(tzinfo=None) ) AUTHOR_USERNAME: Final = 'author' AUTHOR_EMAIL: Final = 'author@example.com' REVIEWER_1_USERNAME: Final = 'reviewer1' REVIEWER_1_EMAIL: Final = 'reviewer1@community.org' REVIEWER_2_USERNAME: Final = 'reviewer2' REVIEWER_2_EMAIL: Final = 'reviewer2@community.org' def setUp(self) -> None: super().setUp() self.signup(self.AUTHOR_EMAIL, self.AUTHOR_USERNAME) self.author_id = self.get_user_id_from_email(self.AUTHOR_EMAIL) self.signup(self.REVIEWER_1_EMAIL, self.REVIEWER_1_USERNAME) self.reviewer_1_id = self.get_user_id_from_email(self.REVIEWER_1_EMAIL) user_services.update_email_preferences( self.reviewer_1_id, True, False, False, False, feconf.DEFAULT_CONTRIBUTOR_DASHBOARD_EMAIL_PREFERENCE, ) self.signup(self.REVIEWER_2_EMAIL, self.REVIEWER_2_USERNAME) self.reviewer_2_id = self.get_user_id_from_email(self.REVIEWER_2_EMAIL) user_services.update_email_preferences( self.reviewer_2_id, True, False, False, False, feconf.DEFAULT_CONTRIBUTOR_DASHBOARD_EMAIL_PREFERENCE, ) self.log_new_error_counter = test_utils.CallCounter(logging.error) self.log_new_error_ctx = self.swap( logging, 'error', self.log_new_error_counter ) self.logged_info: List[str] = [] self.log_new_info_ctx = self.swap( logging, 'info', self._mock_logging_info ) self.save_new_valid_exploration(self.target_id, self.author_id) self.save_new_skill(self.skill_id, self.author_id) translation_suggestion = self._create_translation_suggestion_in_lang_with_html_and_datetime( # pylint: disable=line-too-long 'en', '<p>What is the meaning of life?</p>', self.mocked_review_submission_datetime, ) self.reviewable_suggestion_email_info = suggestion_services.create_reviewable_suggestion_email_info_from_suggestion( translation_suggestion ) def _create_reviewable_suggestion_email_infos_from_suggestions( self, suggestions: List[suggestion_registry.BaseSuggestion] ) -> List[suggestion_registry.ReviewableSuggestionEmailInfo]: """Creates a list of ReviewableSuggestionEmailInfo objects from the given suggestions. """ return [ ( suggestion_services.create_reviewable_suggestion_email_info_from_suggestion( suggestion ) ) for suggestion in suggestions ] def _assert_email_data_stored_in_sent_email_model_is_correct( self, expected_email_html_body: str, reviewer_id: Optional[str], reviewer_email: str, ) -> None: """Asserts that the created sent email model from the sent email contains the right information. """ sent_email_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all() .filter(email_models.SentEmailModel.recipient_id == reviewer_id) .fetch() ) self.assertEqual(len(sent_email_models), 1) sent_email_model = sent_email_models[0] self.assertEqual( sent_email_model.subject, email_manager.CONTRIBUTOR_DASHBOARD_REVIEWER_NOTIFICATION_EMAIL_DATA[ 'email_subject' ], ) self.assertEqual(sent_email_model.recipient_id, reviewer_id) self.assertEqual(sent_email_model.recipient_email, reviewer_email) self.assertEqual(sent_email_model.html_body, expected_email_html_body) self.assertEqual(sent_email_model.sender_id, feconf.SYSTEM_COMMITTER_ID) self.assertEqual( sent_email_model.sender_email, 'Site Admin <noreply@example.com>' ) self.assertEqual( sent_email_model.intent, feconf.EMAIL_INTENT_REVIEW_CONTRIBUTOR_DASHBOARD_SUGGESTIONS, ) def _mock_logging_info(self, msg: str, *args: str) -> None: """Mocks logging.info() by appending the log message to the logged info list. """ self.logged_info.append(msg % args) def _create_translation_suggestion_in_lang_with_html_and_datetime( self, language_code: str, translation_html: str, submission_datetime: datetime.datetime, ) -> suggestion_registry.BaseSuggestion: """Creates a translation suggestion in the given language_code with the given translation html and submission datetime. """ add_translation_change_dict = { 'cmd': exp_domain.CMD_ADD_WRITTEN_TRANSLATION, 'state_name': feconf.DEFAULT_INIT_STATE_NAME, 'content_id': 'content_0', 'language_code': language_code, 'content_html': feconf.DEFAULT_STATE_CONTENT_STR, 'translation_html': translation_html, 'data_format': 'html', } translation_suggestion = suggestion_services.create_suggestion( feconf.SUGGESTION_TYPE_TRANSLATE_CONTENT, feconf.ENTITY_TYPE_EXPLORATION, self.target_id, feconf.CURRENT_STATE_SCHEMA_VERSION, self.author_id, add_translation_change_dict, 'test description', ) translation_suggestion.last_updated = submission_datetime return translation_suggestion @test_utils.set_platform_parameters( [(param_list.ParamName.SERVER_CAN_SEND_EMAILS, False)] ) def test_email_not_sent_if_server_can_send_emails_is_false(self) -> None: with self.capture_logging(min_level=logging.ERROR) as logs: with self.log_new_error_ctx: reviewer_ids_by_language: DefaultDict[str, List[str]] = ( DefaultDict(list) ) suggestions_by_language: DefaultDict[ str, List[suggestion_registry.ReviewableSuggestionEmailInfo] ] = DefaultDict(list) reviewer_ids_by_language['en'] = [] suggestions_by_language['en'] = [] email_manager.send_reviewer_notifications( reviewer_ids_by_language, suggestions_by_language ) messages = self._get_all_sent_email_messages() self.assertEqual(len(messages), 0) self.assertEqual(self.log_new_error_counter.times_called, 1) self.assertEqual(logs[0], 'This app cannot send emails to users.') @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.CONTRIBUTOR_DASHBOARD_REVIEWER_EMAILS_IS_ENABLED, True, ), # pylint: disable=line-too-long ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_not_sent_if_no_reviewers_to_notify(self) -> None: with self.capture_logging(min_level=logging.ERROR) as logs: with self.log_new_error_ctx: reviewer_ids_by_language: DefaultDict[str, List[str]] = ( DefaultDict(list) ) suggestions_by_language: DefaultDict[ str, List[suggestion_registry.ReviewableSuggestionEmailInfo] ] = DefaultDict(list) reviewer_ids_by_language['en'] = [] suggestions_by_language['en'] = [] email_manager.send_reviewer_notifications( reviewer_ids_by_language, suggestions_by_language ) messages = self._get_all_sent_email_messages() self.assertEqual(len(messages), 0) self.assertEqual(self.log_new_error_counter.times_called, 1) self.assertEqual( logs[0], 'No reviewers found for language en to notify' ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.CONTRIBUTOR_DASHBOARD_REVIEWER_EMAILS_IS_ENABLED, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), # pylint: disable=line-too-long ] ) def test_email_sent_to_reviewer_with_translation_waiting_days_for_review( self, ) -> None: translation_suggestion = ( self._create_translation_suggestion_in_lang_with_html_and_datetime( 'fr', '<p>What is the meaning of life?</p>', self.mocked_review_submission_datetime, ) ) reviewable_suggestion_email_info = suggestion_services.create_reviewable_suggestion_email_info_from_suggestion( translation_suggestion ) review_wait_time = 2 mocked_current_time = ( reviewable_suggestion_email_info.submission_datetime + datetime.timedelta(days=review_wait_time) ) expected_email_html_body = ( 'Hi reviewer1' ',<br><br>There are new <a href="%s%s">opportunities</a>' ' to review translations that we think you might be interested' ' in on the Contributor Dashboard page. Here are some examples' ' of contributions that are waiting for review:' '<br><br>The following suggestions are available for review: ' '<br><br><ul><li>The following français (French) translation suggestion ' 'was submitted for review 2 days ago:<br>What is the' ' meaning of life?</li><br></ul><br>Please take some time ' 'to review any of the above contributions ' '(if they still need a review) or any other contributions ' 'on the dashboard. We appreciate your help!<br><br>Thanks again,' ' and happy reviewing!<br><br>The Oppia Contributor Dashboard Team' ) with self.log_new_error_ctx: with self.swap( utils, 'get_current_utc_datetime', lambda: mocked_current_time ): reviewer_ids_by_language: DefaultDict[str, List[str]] = ( DefaultDict(list) ) suggestions_by_language: DefaultDict[ str, List[suggestion_registry.ReviewableSuggestionEmailInfo] ] = DefaultDict( # pylint: disable=line-too-long list ) reviewer_ids_by_language['fr'] = [self.reviewer_1_id] suggestions_by_language['fr'] = [ reviewable_suggestion_email_info ] email_manager.send_reviewer_notifications( reviewer_ids_by_language, suggestions_by_language ) messages = self._get_sent_email_messages(self.REVIEWER_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual( messages[0].html, ( expected_email_html_body % (DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL) ), ) class NotifyAdminsContributorDashboardReviewersNeededTests( test_utils.EmailTestBase ): """Test emailing admins that Contributor Dashboard reviewers are needed in specific suggestion types. """ CURRICULUM_ADMIN_1_USERNAME: Final = 'user1' CURRICULUM_ADMIN_1_EMAIL: Final = 'user1@community.org' CURRICULUM_ADMIN_2_USERNAME: Final = 'user2' CURRICULUM_ADMIN_2_EMAIL: Final = 'user2@community.org' AUTHOR_EMAIL: Final = 'author@example.com' target_id: str = 'exp1' skill_id: str = 'skill_123456' def _create_translation_suggestion_with_language_code( self, language_code: str ) -> suggestion_registry.BaseSuggestion: """Creates a translation suggestion in the given language_code.""" add_translation_change_dict = { 'cmd': exp_domain.CMD_ADD_WRITTEN_TRANSLATION, 'state_name': feconf.DEFAULT_INIT_STATE_NAME, 'content_id': 'content_0', 'language_code': language_code, 'content_html': feconf.DEFAULT_STATE_CONTENT_STR, 'translation_html': '<p>This is the translated content.</p>', 'data_format': 'html', } return suggestion_services.create_suggestion( feconf.SUGGESTION_TYPE_TRANSLATE_CONTENT, feconf.ENTITY_TYPE_EXPLORATION, self.target_id, feconf.CURRENT_STATE_SCHEMA_VERSION, self.author_id, add_translation_change_dict, 'test description', ) def _create_question_suggestion(self) -> suggestion_registry.BaseSuggestion: """Creates a question suggestion.""" content_id_generator = translation_domain.ContentIdGenerator() add_question_change_dict: Dict[ str, Union[str, float, question_domain.QuestionDict] ] = { 'cmd': question_domain.CMD_CREATE_NEW_FULLY_SPECIFIED_QUESTION, 'question_dict': { 'id': 'test_id', 'version': 12, 'question_state_data': self._create_valid_question_data( 'default_state', content_id_generator ).to_dict(), 'language_code': constants.DEFAULT_LANGUAGE_CODE, 'question_state_data_schema_version': ( feconf.CURRENT_STATE_SCHEMA_VERSION ), 'linked_skill_ids': ['skill_1'], 'inapplicable_skill_misconception_ids': ['skillid12345-1'], 'next_content_id_index': ( content_id_generator.next_content_id_index ), }, 'skill_id': self.skill_id, 'skill_difficulty': 0.3, } return suggestion_services.create_suggestion( feconf.SUGGESTION_TYPE_ADD_QUESTION, feconf.ENTITY_TYPE_SKILL, self.skill_id, feconf.CURRENT_STATE_SCHEMA_VERSION, self.author_id, add_question_change_dict, 'test description', ) def _assert_email_data_stored_in_sent_email_model_is_correct( self, expected_email_html_body: str, admin_id: Optional[str], admin_email: str, ) -> None: """Asserts that the sent email model that was created from the email that was sent contains the right information. """ sent_email_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all() .filter(email_models.SentEmailModel.recipient_id == admin_id) .fetch() ) self.assertEqual(len(sent_email_models), 1) sent_email_model = sent_email_models[0] self.assertEqual( sent_email_model.subject, email_manager.ADMIN_NOTIFICATION_FOR_REVIEWER_SHORTAGE_EMAIL_DATA[ 'email_subject' ], ) self.assertEqual(sent_email_model.recipient_id, admin_id) self.assertEqual(sent_email_model.recipient_email, admin_email) self.assertEqual(sent_email_model.html_body, expected_email_html_body) self.assertEqual(sent_email_model.sender_id, feconf.SYSTEM_COMMITTER_ID) self.assertEqual( sent_email_model.sender_email, 'Site Admin <noreply@example.com>' ) self.assertEqual( sent_email_model.intent, feconf.EMAIL_INTENT_ADD_CONTRIBUTOR_DASHBOARD_REVIEWERS, ) def _mock_logging_info(self, msg: str, *args: str) -> None: """Mocks logging.info() by appending the log message to the logged info list. """ self.logged_info.append(msg % args) def setUp(self) -> None: super().setUp() self.signup(self.AUTHOR_EMAIL, 'author') self.author_id = self.get_user_id_from_email(self.AUTHOR_EMAIL) self.signup( self.CURRICULUM_ADMIN_1_EMAIL, self.CURRICULUM_ADMIN_1_USERNAME ) self.admin_1_id = self.get_user_id_from_email( self.CURRICULUM_ADMIN_1_EMAIL ) self.signup( self.CURRICULUM_ADMIN_2_EMAIL, self.CURRICULUM_ADMIN_2_USERNAME ) self.admin_2_id = self.get_user_id_from_email( self.CURRICULUM_ADMIN_2_EMAIL ) self.save_new_valid_exploration(self.target_id, self.author_id) self.save_new_skill(self.skill_id, self.author_id) self.log_new_error_counter = test_utils.CallCounter(logging.error) self.log_new_error_ctx = self.swap( logging, 'error', self.log_new_error_counter ) self.logged_info: List[str] = [] self.log_new_info_ctx = self.swap( logging, 'info', self._mock_logging_info ) self.suggestion_types_needing_reviewers: Dict[str, Set[str]] = { feconf.SUGGESTION_TYPE_ADD_QUESTION: set() } @test_utils.set_platform_parameters( [(param_list.ParamName.SERVER_CAN_SEND_EMAILS, False)] ) def test_email_not_sent_if_server_can_send_emails_is_false(self) -> None: with self.capture_logging(min_level=logging.ERROR) as logs: with self.log_new_error_ctx: email_manager.send_mail_to_notify_admins_that_reviewers_are_needed( # pylint: disable=line-too-long [self.admin_1_id], [], [], self.suggestion_types_needing_reviewers, ) messages = self._get_all_sent_email_messages() self.assertEqual(len(messages), 0) self.assertEqual(self.log_new_error_counter.times_called, 1) self.assertEqual(logs[0], 'This app cannot send emails to users.') @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.ENABLE_ADMIN_NOTIFICATIONS_FOR_REVIEWER_SHORTAGE, False, ), # pylint: disable=line-too-long ] ) def test_email_not_sent_if_notifying_admins_reviewers_needed_is_disabled( self, ) -> None: with self.capture_logging(min_level=logging.ERROR) as logs: with self.log_new_error_ctx: email_manager.send_mail_to_notify_admins_that_reviewers_are_needed( # pylint: disable=line-too-long [self.admin_1_id], [], [], self.suggestion_types_needing_reviewers, ) messages = self._get_all_sent_email_messages() self.assertEqual(len(messages), 0) self.assertEqual(self.log_new_error_counter.times_called, 1) self.assertEqual( logs[0], 'The "enable_admin_notifications_for_reviewer_shortage" ' 'property must be enabled on the admin config page in order to ' 'send admins the emails.', ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.ENABLE_ADMIN_NOTIFICATIONS_FOR_REVIEWER_SHORTAGE, True, ), # pylint: disable=line-too-long ] ) def test_email_not_sent_if_no_admins_to_notify(self) -> None: with self.capture_logging(min_level=logging.ERROR) as logs: with self.log_new_error_ctx: email_manager.send_mail_to_notify_admins_that_reviewers_are_needed( # pylint: disable=line-too-long [], [], [], self.suggestion_types_needing_reviewers ) messages = self._get_all_sent_email_messages() self.assertEqual(len(messages), 0) self.assertEqual(self.log_new_error_counter.times_called, 1) self.assertEqual(logs[0], 'There were no admins to notify.') @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.ENABLE_ADMIN_NOTIFICATIONS_FOR_REVIEWER_SHORTAGE, True, ), # pylint: disable=line-too-long ] ) def test_email_not_sent_if_no_suggestion_types_that_need_reviewers( self, ) -> None: with self.log_new_info_ctx: ( email_manager.send_mail_to_notify_admins_that_reviewers_are_needed( [self.admin_1_id], [], [], {} ) ) messages = self._get_all_sent_email_messages() self.assertEqual(len(messages), 0) self.assertEqual( self.logged_info[0], 'There were no suggestion types that needed more reviewers on the ' 'Contributor Dashboard.', ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.ENABLE_ADMIN_NOTIFICATIONS_FOR_REVIEWER_SHORTAGE, True, ), # pylint: disable=line-too-long (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_not_sent_if_admin_email_does_not_exist(self) -> None: with self.capture_logging(min_level=logging.ERROR) as logs: with self.log_new_error_ctx: email_manager.send_mail_to_notify_admins_that_reviewers_are_needed( # pylint: disable=line-too-long ['admin_id_without_email'], [], [], self.suggestion_types_needing_reviewers, ) messages = self._get_all_sent_email_messages() self.assertEqual(len(messages), 0) self.assertEqual(self.log_new_error_counter.times_called, 1) self.assertEqual( logs[0], 'There was no email for the given admin id: ' 'admin_id_without_email.', ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.ENABLE_ADMIN_NOTIFICATIONS_FOR_REVIEWER_SHORTAGE, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_sent_to_admin_if_question_suggestions_need_reviewers( self, ) -> None: self._create_question_suggestion() suggestion_types_needing_reviewers = ( suggestion_services.get_suggestion_types_that_need_reviewers() ) self.assertDictEqual( suggestion_types_needing_reviewers, {feconf.SUGGESTION_TYPE_ADD_QUESTION: set()}, ) expected_email_html_body = ( 'Hi user1,' '<br><br>' 'In the <a href="%s%s#/roles">admin roles page,</a> please add ' 'reviewers to the Contributor Dashboard community by entering ' 'their username(s) and allow reviewing for the suggestion types ' 'that need more reviewers bolded below.' '<br><br>' 'There have been <b>question suggestions</b> created on the ' '<a href="%s%s">Contributor Dashboard page</a> where there are not ' 'enough reviewers.' '<br><br>' 'Thanks so much - we appreciate your help!<br><br>' 'Best Wishes!<br>' '- The Oppia Contributor Dashboard Team' % ( DEV_OPPIA_SITE_URL, feconf.ADMIN_URL, DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, ) ) email_manager.send_mail_to_notify_admins_that_reviewers_are_needed( [self.admin_1_id], [], [], self.suggestion_types_needing_reviewers ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.CURRICULUM_ADMIN_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body, self.admin_1_id, self.CURRICULUM_ADMIN_1_EMAIL, ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.ENABLE_ADMIN_NOTIFICATIONS_FOR_REVIEWER_SHORTAGE, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_sent_to_admins_if_question_suggestions_need_reviewers( self, ) -> None: self._create_question_suggestion() suggestion_types_needing_reviewers = ( suggestion_services.get_suggestion_types_that_need_reviewers() ) self.assertDictEqual( suggestion_types_needing_reviewers, {feconf.SUGGESTION_TYPE_ADD_QUESTION: set()}, ) expected_email_html_body_for_admin_1 = ( 'Hi user1,' '<br><br>' 'In the <a href="%s%s#/roles">admin roles page,</a> please add ' 'reviewers to the Contributor Dashboard community by entering ' 'their username(s) and allow reviewing for the suggestion types ' 'that need more reviewers bolded below.' '<br><br>' 'There have been <b>question suggestions</b> created on the ' '<a href="%s%s">Contributor Dashboard page</a> where there are not ' 'enough reviewers.' '<br><br>' 'Thanks so much - we appreciate your help!<br><br>' 'Best Wishes!<br>' '- The Oppia Contributor Dashboard Team' % ( DEV_OPPIA_SITE_URL, feconf.ADMIN_URL, DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, ) ) expected_email_html_body_for_admin_2 = ( 'Hi user2,' '<br><br>' 'In the <a href="%s%s#/roles">admin roles page,</a> please add ' 'reviewers to the Contributor Dashboard community by entering ' 'their username(s) and allow reviewing for the suggestion types ' 'that need more reviewers bolded below.' '<br><br>' 'There have been <b>question suggestions</b> created on the ' '<a href="%s%s">Contributor Dashboard page</a> where there are not ' 'enough reviewers.' '<br><br>' 'Thanks so much - we appreciate your help!<br><br>' 'Best Wishes!<br>' '- The Oppia Contributor Dashboard Team' % ( DEV_OPPIA_SITE_URL, feconf.ADMIN_URL, DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, ) ) email_manager.send_mail_to_notify_admins_that_reviewers_are_needed( [self.admin_1_id, self.admin_2_id], [], [], suggestion_types_needing_reviewers, ) # Make sure correct emails are sent. messages = self._get_sent_email_messages(self.CURRICULUM_ADMIN_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body_for_admin_1) messages = self._get_sent_email_messages(self.CURRICULUM_ADMIN_2_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body_for_admin_2) # Make sure correct email models are stored. self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body_for_admin_1, self.admin_1_id, self.CURRICULUM_ADMIN_1_EMAIL, ) self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body_for_admin_2, self.admin_2_id, self.CURRICULUM_ADMIN_2_EMAIL, ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.ENABLE_ADMIN_NOTIFICATIONS_FOR_REVIEWER_SHORTAGE, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_admin_email_sent_if_translations_need_reviewers_for_one_lang( self, ) -> None: self._create_translation_suggestion_with_language_code('hi') suggestion_types_needing_reviewers = ( suggestion_services.get_suggestion_types_that_need_reviewers() ) self.assertDictEqual( suggestion_types_needing_reviewers, {feconf.SUGGESTION_TYPE_TRANSLATE_CONTENT: {'hi'}}, ) expected_email_html_body = ( 'Hi user1,' '<br><br>' 'In the <a href="%s%s#/roles">admin roles page,</a> please add ' 'reviewers to the Contributor Dashboard community by entering ' 'their username(s) and allow reviewing for the suggestion types ' 'that need more reviewers bolded below.' '<br><br>' 'There have been <b>हिन्दी (Hindi) translation suggestions</b> ' 'created on the <a href="%s%s">Contributor Dashboard page</a> where' ' there are not enough reviewers.<br><br>' 'Thanks so much - we appreciate your help!<br><br>' 'Best Wishes!<br>' '- The Oppia Contributor Dashboard Team' % ( DEV_OPPIA_SITE_URL, feconf.ADMIN_URL, DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, ) ) email_manager.send_mail_to_notify_admins_that_reviewers_are_needed( [self.admin_1_id], [], [], suggestion_types_needing_reviewers ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.CURRICULUM_ADMIN_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body, self.admin_1_id, self.CURRICULUM_ADMIN_1_EMAIL, ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.ENABLE_ADMIN_NOTIFICATIONS_FOR_REVIEWER_SHORTAGE, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_admin_emails_sent_if_translations_need_reviewers_for_one_lang( self, ) -> None: self._create_translation_suggestion_with_language_code('hi') suggestion_types_needing_reviewers = ( suggestion_services.get_suggestion_types_that_need_reviewers() ) self.assertDictEqual( suggestion_types_needing_reviewers, {feconf.SUGGESTION_TYPE_TRANSLATE_CONTENT: {'hi'}}, ) expected_email_html_body_for_admin_1 = ( 'Hi user1,' '<br><br>' 'In the <a href="%s%s#/roles">admin roles page,</a> please add ' 'reviewers to the Contributor Dashboard community by entering ' 'their username(s) and allow reviewing for the suggestion types ' 'that need more reviewers bolded below.' '<br><br>' 'There have been <b>हिन्दी (Hindi) translation suggestions</b> ' 'created on the <a href="%s%s">Contributor Dashboard page</a> where' ' there are not enough reviewers.<br><br>' 'Thanks so much - we appreciate your help!<br><br>' 'Best Wishes!<br>' '- The Oppia Contributor Dashboard Team' % ( DEV_OPPIA_SITE_URL, feconf.ADMIN_URL, DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, ) ) expected_email_html_body_for_admin_2 = ( 'Hi user2,' '<br><br>' 'In the <a href="%s%s#/roles">admin roles page,</a> please add ' 'reviewers to the Contributor Dashboard community by entering ' 'their username(s) and allow reviewing for the suggestion types ' 'that need more reviewers bolded below.' '<br><br>' 'There have been <b>हिन्दी (Hindi) translation suggestions</b> ' 'created on the <a href="%s%s">Contributor Dashboard page</a> where' ' there are not enough reviewers.<br><br>' 'Thanks so much - we appreciate your help!<br><br>' 'Best Wishes!<br>' '- The Oppia Contributor Dashboard Team' % ( DEV_OPPIA_SITE_URL, feconf.ADMIN_URL, DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, ) ) email_manager.send_mail_to_notify_admins_that_reviewers_are_needed( [self.admin_1_id, self.admin_2_id], [], [], suggestion_types_needing_reviewers, ) # Make sure correct emails are sent. messages = self._get_sent_email_messages(self.CURRICULUM_ADMIN_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body_for_admin_1) messages = self._get_sent_email_messages(self.CURRICULUM_ADMIN_2_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body_for_admin_2) # Make sure correct email models are stored. self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body_for_admin_1, self.admin_1_id, self.CURRICULUM_ADMIN_1_EMAIL, ) self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body_for_admin_2, self.admin_2_id, self.CURRICULUM_ADMIN_2_EMAIL, ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.ENABLE_ADMIN_NOTIFICATIONS_FOR_REVIEWER_SHORTAGE, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_admin_email_sent_if_translations_need_reviewers_for_multi_lang( self, ) -> None: self._create_translation_suggestion_with_language_code('fr') self._create_translation_suggestion_with_language_code('hi') suggestion_types_needing_reviewers = ( suggestion_services.get_suggestion_types_that_need_reviewers() ) self.assertDictEqual( suggestion_types_needing_reviewers, {feconf.SUGGESTION_TYPE_TRANSLATE_CONTENT: {'fr', 'hi'}}, ) expected_email_html_body = ( 'Hi user1,' '<br><br>' 'In the <a href="%s%s#/roles">admin roles page,</a> please add ' 'reviewers to the Contributor Dashboard community by entering ' 'their username(s) and allow reviewing for the suggestion types ' 'that need more reviewers bolded below.' '<br><br>' 'There have been <b>translation suggestions</b> created on the ' '<a href="%s%s">Contributor Dashboard page</a> in languages where ' 'there are not enough reviewers. The languages that need more ' 'reviewers are:' '<br><ul>' '<li><b>français (French)</b></li><br>' '<li><b>हिन्दी (Hindi)</b></li><br>' '</ul><br>' 'Thanks so much - we appreciate your help!<br><br>' 'Best Wishes!<br>' '- The Oppia Contributor Dashboard Team' % ( DEV_OPPIA_SITE_URL, feconf.ADMIN_URL, DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, ) ) email_manager.send_mail_to_notify_admins_that_reviewers_are_needed( [self.admin_1_id], [], [], suggestion_types_needing_reviewers ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.CURRICULUM_ADMIN_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body, self.admin_1_id, self.CURRICULUM_ADMIN_1_EMAIL, ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.ENABLE_ADMIN_NOTIFICATIONS_FOR_REVIEWER_SHORTAGE, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_admin_emails_sent_if_translations_need_reviewers_for_multi_lang( self, ) -> None: self._create_translation_suggestion_with_language_code('fr') self._create_translation_suggestion_with_language_code('hi') suggestion_types_needing_reviewers = ( suggestion_services.get_suggestion_types_that_need_reviewers() ) self.assertDictEqual( suggestion_types_needing_reviewers, {feconf.SUGGESTION_TYPE_TRANSLATE_CONTENT: {'fr', 'hi'}}, ) expected_email_html_body_for_admin_1 = ( 'Hi user1,' '<br><br>' 'In the <a href="%s%s#/roles">admin roles page,</a> please add ' 'reviewers to the Contributor Dashboard community by entering ' 'their username(s) and allow reviewing for the suggestion types ' 'that need more reviewers bolded below.' '<br><br>' 'There have been <b>translation suggestions</b> created on the ' '<a href="%s%s">Contributor Dashboard page</a> in languages where ' 'there are not enough reviewers. The languages that need more ' 'reviewers are:' '<br><ul>' '<li><b>français (French)</b></li><br>' '<li><b>हिन्दी (Hindi)</b></li><br>' '</ul><br>' 'Thanks so much - we appreciate your help!<br><br>' 'Best Wishes!<br>' '- The Oppia Contributor Dashboard Team' % ( DEV_OPPIA_SITE_URL, feconf.ADMIN_URL, DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, ) ) expected_email_html_body_for_admin_2 = ( 'Hi user2,' '<br><br>' 'In the <a href="%s%s#/roles">admin roles page,</a> please add ' 'reviewers to the Contributor Dashboard community by entering ' 'their username(s) and allow reviewing for the suggestion types ' 'that need more reviewers bolded below.' '<br><br>' 'There have been <b>translation suggestions</b> created on the ' '<a href="%s%s">Contributor Dashboard page</a> in languages where ' 'there are not enough reviewers. The languages that need more ' 'reviewers are:' '<br><ul>' '<li><b>français (French)</b></li><br>' '<li><b>हिन्दी (Hindi)</b></li><br>' '</ul><br>' 'Thanks so much - we appreciate your help!<br><br>' 'Best Wishes!<br>' '- The Oppia Contributor Dashboard Team' % ( DEV_OPPIA_SITE_URL, feconf.ADMIN_URL, DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, ) ) email_manager.send_mail_to_notify_admins_that_reviewers_are_needed( [self.admin_1_id, self.admin_2_id], [], [], suggestion_types_needing_reviewers, ) # Make sure correct emails are sent. messages = self._get_sent_email_messages(self.CURRICULUM_ADMIN_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body_for_admin_1) messages = self._get_sent_email_messages(self.CURRICULUM_ADMIN_2_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body_for_admin_2) # Make sure correct email models are stored. self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body_for_admin_1, self.admin_1_id, self.CURRICULUM_ADMIN_1_EMAIL, ) self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body_for_admin_2, self.admin_2_id, self.CURRICULUM_ADMIN_2_EMAIL, ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.ENABLE_ADMIN_NOTIFICATIONS_FOR_REVIEWER_SHORTAGE, True, ), # pylint: disable=line-too-long (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_sent_to_admins_if_mutli_suggestion_types_needing_reviewers( self, ) -> None: self._create_translation_suggestion_with_language_code('fr') self._create_translation_suggestion_with_language_code('hi') self._create_question_suggestion() suggestion_types_needing_reviewers = ( suggestion_services.get_suggestion_types_that_need_reviewers() ) self.assertDictEqual( suggestion_types_needing_reviewers, { feconf.SUGGESTION_TYPE_TRANSLATE_CONTENT: {'fr', 'hi'}, feconf.SUGGESTION_TYPE_ADD_QUESTION: set(), }, ) expected_email_html_body_for_admin_1 = ( 'Hi user1,' '<br><br>' 'In the <a href="%s%s#/roles">admin roles page,</a> please add ' 'reviewers to the Contributor Dashboard community by entering ' 'their username(s) and allow reviewing for the suggestion types ' 'that need more reviewers bolded below.' '<br><br>' 'There have been <b>translation suggestions</b> created on the ' '<a href="%s%s">Contributor Dashboard page</a> in languages where ' 'there are not enough reviewers. The languages that need more ' 'reviewers are:' '<br><ul>' '<li><b>français (French)</b></li><br>' '<li><b>हिन्दी (Hindi)</b></li><br>' '</ul><br>' 'Thanks so much - we appreciate your help!<br><br>' 'Best Wishes!<br>' '- The Oppia Contributor Dashboard Team' % ( DEV_OPPIA_SITE_URL, feconf.ADMIN_URL, DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, ) ) expected_email_html_body_for_admin_2 = ( 'Hi user2,' '<br><br>' 'In the <a href="%s%s#/roles">admin roles page,</a> please add ' 'reviewers to the Contributor Dashboard community by entering ' 'their username(s) and allow reviewing for the suggestion types ' 'that need more reviewers bolded below.' '<br><br>' 'There have been <b>translation suggestions</b> created on the ' '<a href="%s%s">Contributor Dashboard page</a> in languages where ' 'there are not enough reviewers. The languages that need more ' 'reviewers are:' '<br><ul>' '<li><b>français (French)</b></li><br>' '<li><b>हिन्दी (Hindi)</b></li><br>' '</ul><br>' 'Thanks so much - we appreciate your help!<br><br>' 'Best Wishes!<br>' '- The Oppia Contributor Dashboard Team' % ( DEV_OPPIA_SITE_URL, feconf.ADMIN_URL, DEV_OPPIA_SITE_URL, feconf.CONTRIBUTOR_DASHBOARD_URL, ) ) email_manager.send_mail_to_notify_admins_that_reviewers_are_needed( [self.admin_1_id], [self.admin_2_id], [], suggestion_types_needing_reviewers, ) # Make sure correct emails are sent. messages = self._get_sent_email_messages(self.CURRICULUM_ADMIN_1_EMAIL) messages.sort(key=lambda m: m.html) self.assertEqual(len(messages), 2) self.assertEqual(messages[1].html, expected_email_html_body_for_admin_1) messages = self._get_sent_email_messages(self.CURRICULUM_ADMIN_2_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body_for_admin_2) # Make sure correct email models are stored. sent_email_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all() .filter(email_models.SentEmailModel.recipient_id == self.admin_1_id) .fetch() ) # Here, we are narrowing down the type of 'sent_email_models' from # Sequence to List. Because Sequence is a broader type and it does not # contain extra methods (e.g: .sort()), and below we are using .sort() # method by assuming that 'sent_email_models' is an instance of list. # So, to avoid error we used assertion here. assert isinstance(sent_email_models, list) self.assertEqual(len(sent_email_models), 2) email_sort_fn: Callable[[email_models.SentEmailModel], str] = ( lambda m: (m.html_body if isinstance(m.html_body, str) else '') ) sent_email_models.sort(key=email_sort_fn) sent_email_model = sent_email_models[1] self.assertEqual( sent_email_model.subject, email_manager.ADMIN_NOTIFICATION_FOR_REVIEWER_SHORTAGE_EMAIL_DATA[ 'email_subject' ], ) self.assertEqual(sent_email_model.recipient_id, self.admin_1_id) self.assertEqual( sent_email_model.recipient_email, self.CURRICULUM_ADMIN_1_EMAIL ) self.assertEqual( sent_email_model.html_body, expected_email_html_body_for_admin_1 ) self.assertEqual(sent_email_model.sender_id, feconf.SYSTEM_COMMITTER_ID) self.assertEqual( sent_email_model.sender_email, 'Site Admin <noreply@example.com>' ) self.assertEqual( sent_email_model.intent, feconf.EMAIL_INTENT_ADD_CONTRIBUTOR_DASHBOARD_REVIEWERS, ) self._assert_email_data_stored_in_sent_email_model_is_correct( expected_email_html_body_for_admin_2, self.admin_2_id, self.CURRICULUM_ADMIN_2_EMAIL, ) class AccountDeletionEmailUnitTest(test_utils.EmailTestBase): """Unit test related to account deletion application emails.""" APPLICANT_USERNAME: Final = 'applicant' APPLICANT_EMAIL: Final = 'applicant@example.com' def setUp(self) -> None: super().setUp() self.signup(self.APPLICANT_EMAIL, self.APPLICANT_USERNAME) self.applicant_id = self.get_user_id_from_email(self.APPLICANT_EMAIL) @test_utils.set_platform_parameters( [(param_list.ParamName.SERVER_CAN_SEND_EMAILS, False)] ) def test_email_not_sent_if_server_can_send_emails_is_false(self) -> None: email_manager.send_account_deleted_email( self.applicant_id, self.APPLICANT_EMAIL ) messages = self._get_sent_email_messages(self.APPLICANT_EMAIL) self.assertEqual(len(messages), 0) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'admin@system.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.SYSTEM_EMAIL_NAME, '.'), ] ) def test_account_deletion_failed_email_is_sent_correctly(self) -> None: # Make sure there are no emails already sent. admin_email_address = 'admin@system.com' messages = self._get_sent_email_messages(admin_email_address) self.assertEqual(messages, []) # Send an account deletion failed email to admin. email_manager.send_account_deletion_failed_email( self.applicant_id, self.APPLICANT_EMAIL ) # Make sure emails are sent. messages = self._get_sent_email_messages(admin_email_address) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].to, ['admin@system.com']) self.assertEqual( messages[0].subject, 'WIPEOUT: Account deletion failed' ) self.assertIn( 'The Wipeout process failed for the user with ID \'%s\' and ' 'email \'%s\'.' % (self.applicant_id, self.APPLICANT_EMAIL), messages[0].html, ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ] ) def test_that_correct_account_deleted_email_is_sent(self) -> None: expected_email_subject = 'Account deleted' expected_email_html_body = ( 'Hi applicant@example.com,<br><br>' 'Your account was successfully deleted.<br><br>' '- The Oppia Team' ) email_manager.send_account_deleted_email( self.applicant_id, self.APPLICANT_EMAIL ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.APPLICANT_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) sent_email_model = all_models[0] self.assertEqual(sent_email_model.subject, expected_email_subject) self.assertEqual(sent_email_model.recipient_id, self.applicant_id) self.assertEqual(sent_email_model.recipient_email, self.APPLICANT_EMAIL) self.assertEqual(sent_email_model.sender_id, feconf.SYSTEM_COMMITTER_ID) self.assertEqual( sent_email_model.sender_email, 'Site Admin <noreply@example.com>' ) self.assertEqual( sent_email_model.intent, feconf.EMAIL_INTENT_ACCOUNT_DELETED ) class EmailPreferencesTests(test_utils.EmailTestBase): def test_can_users_receive_thread_email(self) -> None: gae_ids = ('someUser1', 'someUser2') exp_id = 'someExploration' usernames = ('username1', 'username2') emails = ('user1@example.com', 'user2@example.com') user_ids = [] for user_id, username, user_email in zip(gae_ids, usernames, emails): user_settings = user_services.create_new_user(user_id, user_email) user_ids.append(user_settings.user_id) user_services.set_username(user_settings.user_id, username) # Both users can receive all emails in default setting. self.assertListEqual( email_manager.can_users_receive_thread_email( user_ids, exp_id, True ), [True, True], ) self.assertTrue( email_manager.can_users_receive_thread_email( user_ids, exp_id, False ), msg=[True, True], ) # First user have muted feedback notifications for this exploration, # therefore he should receive only suggestion emails. user_services.set_email_preferences_for_exploration( user_ids[0], exp_id, mute_feedback_notifications=True ) self.assertListEqual( email_manager.can_users_receive_thread_email( user_ids, exp_id, True ), [True, True], ) self.assertListEqual( email_manager.can_users_receive_thread_email( user_ids, exp_id, False ), [False, True], ) # Second user have muted suggestion notifications for this exploration, # therefore he should receive only feedback emails. user_services.set_email_preferences_for_exploration( user_ids[1], exp_id, mute_suggestion_notifications=True ) self.assertListEqual( email_manager.can_users_receive_thread_email( user_ids, exp_id, True ), [True, False], ) self.assertListEqual( email_manager.can_users_receive_thread_email( user_ids, exp_id, False ), [False, True], ) # Both users have disabled all emails globally, therefore they # should not receive any emails. for user_id in user_ids: user_services.update_email_preferences( user_id, True, True, False, True, feconf.DEFAULT_CONTRIBUTOR_DASHBOARD_EMAIL_PREFERENCE, ) self.assertListEqual( email_manager.can_users_receive_thread_email( user_ids, exp_id, True ), [False, False], ) self.assertTrue( email_manager.can_users_receive_thread_email( user_ids, exp_id, False ), msg=[False, False], ) # Both users have unmuted feedback/suggestion emails for this # exploration, but all emails are still disabled globally, # therefore they should not receive any emails. user_services.set_email_preferences_for_exploration( user_ids[0], exp_id, mute_feedback_notifications=False ) user_services.set_email_preferences_for_exploration( user_ids[1], exp_id, mute_suggestion_notifications=False ) user_services.update_email_preferences( user_id, True, True, False, True, feconf.DEFAULT_CONTRIBUTOR_DASHBOARD_EMAIL_PREFERENCE, ) self.assertListEqual( email_manager.can_users_receive_thread_email( user_ids, exp_id, True ), [False, False], ) self.assertTrue( email_manager.can_users_receive_thread_email( user_ids, exp_id, False ), msg=[False, False], ) # Both user have enabled all emails globally, therefore they should # receive all emails. for user_id in user_ids: user_services.update_email_preferences( user_id, True, True, True, True, feconf.DEFAULT_CONTRIBUTOR_DASHBOARD_EMAIL_PREFERENCE, ) self.assertListEqual( email_manager.can_users_receive_thread_email( user_ids, exp_id, True ), [True, True], ) self.assertTrue( email_manager.can_users_receive_thread_email( user_ids, exp_id, False ), msg=[True, True], ) class ModeratorActionEmailsTests(test_utils.EmailTestBase): RECIPIENT_EMAIL: Final = 'a@example.com' RECIPIENT_USERNAME: Final = 'usera' def setUp(self) -> None: super().setUp() self.signup(self.MODERATOR_EMAIL, self.MODERATOR_USERNAME) self.moderator_id = self.get_user_id_from_email(self.MODERATOR_EMAIL) self.set_moderators([self.MODERATOR_USERNAME]) self.signup(self.RECIPIENT_EMAIL, self.RECIPIENT_USERNAME) self.recipient_id = self.get_user_id_from_email(self.RECIPIENT_EMAIL) def test_exception_raised_if_server_can_send_emails_is_false(self) -> None: with self.assertRaisesRegex( Exception, 'For moderator emails to be sent, please ensure that ' 'SERVER_CAN_SEND_EMAILS is set to True.', ): email_manager.require_moderator_email_prereqs_are_satisfied() @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), ( param_list.ParamName.UNPUBLISH_EXPLORATION_EMAIL_HTML_BODY, # pylint: disable=line-too-long ( 'I\'m writing to inform you that ' 'I have unpublished the above exploration.' ), ), ] ) def test_correct_email_draft_received_on_exploration_unpublish( self, ) -> None: expected_draft_text_body = ( 'I\'m writing to inform you that ' 'I have unpublished the above exploration.' ) d_text = email_manager.get_moderator_unpublish_exploration_email() self.assertEqual(d_text, expected_draft_text_body) @test_utils.set_platform_parameters( [(param_list.ParamName.SERVER_CAN_SEND_EMAILS, False)] ) def test_blank_draft_received_exploration_unpublish_exception_raised( self, ) -> None: expected_draft_text_body = '' d_text = email_manager.get_moderator_unpublish_exploration_email() self.assertEqual(d_text, expected_draft_text_body) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), ] ) def test_correct_moderator_action_email_sent(self) -> None: email_intent = 'unpublish_exploration' exploration_title = 'Title' email_html_body = 'Dummy email body.<br>' email_manager.send_moderator_action_email( self.moderator_id, self.recipient_id, email_intent, exploration_title, email_html_body, ) messages = self._get_sent_email_messages(self.RECIPIENT_EMAIL) self.assertEqual(len(messages), 1) class CDUserEmailTest(test_utils.EmailTestBase): """Test for assignment and removal of contribution reviewers.""" TRANSLATION_REVIEWER_EMAIL: Final = 'translationreviewer@example.com' QUESTION_SUBMITTER_EMAIL: Final = 'questionsubmitter@example.com' def setUp(self) -> None: super().setUp() self.signup(self.CURRICULUM_ADMIN_EMAIL, self.CURRICULUM_ADMIN_USERNAME) self.signup(self.TRANSLATION_REVIEWER_EMAIL, 'translator') self.signup(self.QUESTION_REVIEWER_EMAIL, 'question') self.signup(self.QUESTION_SUBMITTER_EMAIL, 'questionSuggestor') self.translation_reviewer_id = self.get_user_id_from_email( self.TRANSLATION_REVIEWER_EMAIL ) user_services.update_email_preferences( self.translation_reviewer_id, True, False, False, False, feconf.DEFAULT_CONTRIBUTOR_DASHBOARD_EMAIL_PREFERENCE, ) self.question_reviewer_id = self.get_user_id_from_email( self.QUESTION_REVIEWER_EMAIL ) user_services.update_email_preferences( self.question_reviewer_id, True, False, False, False, feconf.DEFAULT_CONTRIBUTOR_DASHBOARD_EMAIL_PREFERENCE, ) self.question_submitter_id = self.get_user_id_from_email( self.QUESTION_SUBMITTER_EMAIL ) user_services.update_email_preferences( self.question_submitter_id, True, False, False, False, feconf.DEFAULT_CONTRIBUTOR_DASHBOARD_EMAIL_PREFERENCE, ) @test_utils.set_platform_parameters( [(param_list.ParamName.SERVER_CAN_SEND_EMAILS, False)] ) def test_assign_translation_reviewer_email_for_can_send_emails_is_false( self, ) -> None: email_manager.send_email_to_new_cd_user( self.translation_reviewer_id, constants.CD_USER_RIGHTS_CATEGORY_REVIEW_TRANSLATION, language_code='hi', ) messages = self._get_sent_email_messages( self.TRANSLATION_REVIEWER_EMAIL ) self.assertEqual(len(messages), 0) @test_utils.set_platform_parameters( [(param_list.ParamName.SERVER_CAN_SEND_EMAILS, True)] ) def test_without_language_code_email_not_sent_to_new_translation_reviewer( self, ) -> None: with self.assertRaisesRegex( Exception, 'The language_code cannot be None' ): email_manager.send_email_to_new_cd_user( self.translation_reviewer_id, constants.CD_USER_RIGHTS_CATEGORY_REVIEW_TRANSLATION, ) @test_utils.set_platform_parameters( [(param_list.ParamName.SERVER_CAN_SEND_EMAILS, True)] ) def test_without_language_code_email_not_sent_to_removed_translation_reviewer( # pylint: disable=line-too-long self, ) -> None: with self.assertRaisesRegex( Exception, 'The language_code cannot be None' ): email_manager.send_email_to_removed_cd_user( self.translation_reviewer_id, constants.CD_USER_RIGHTS_CATEGORY_REVIEW_TRANSLATION, ) def test_assign_translation_reviewer_email_for_invalid_category( self, ) -> None: with self.assertRaisesRegex(Exception, 'Invalid category'): email_manager.send_email_to_new_cd_user( self.translation_reviewer_id, 'invalid_category' ) def test_schema_of_new_cd_user_email_data_constant(self) -> None: self.assertEqual( sorted(email_manager.NEW_CD_USER_EMAIL_DATA.keys()), [ constants.CD_USER_RIGHTS_CATEGORY_REVIEW_QUESTION, constants.CD_USER_RIGHTS_CATEGORY_SUBMIT_QUESTION, constants.CD_USER_RIGHTS_CATEGORY_REVIEW_TRANSLATION, ], ) for category_details in email_manager.NEW_CD_USER_EMAIL_DATA.values(): self.assertEqual(len(category_details), 5) self.assertTrue( 'description' in category_details or ('description_template' in category_details) ) self.assertTrue('category' in category_details) self.assertTrue( 'rights_message' in category_details or ('rights_message_template' in category_details) ) self.assertTrue( 'to_review' in category_details or ('to_submit' in category_details) ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ] ) def test_send_assigned_translation_reviewer_email(self) -> None: expected_email_subject = ( 'You have been invited to review Oppia translations' ) expected_email_html_body = ( 'Hi translator,<br><br>' 'This is to let you know that the Oppia team has added you as a ' 'reviewer for हिन्दी (hindi) language translations. This allows you' ' to review translation suggestions made by contributors in the ' 'हिन्दी (hindi) language.<br><br>' 'You can check the translation suggestions waiting for review in ' 'the <a href="https://www.oppia.org/contributor-dashboard">' 'Contributor Dashboard</a>.<br><br>' 'Thanks, and happy contributing!<br><br>' 'Best wishes,<br>' 'The Oppia Community' ) email_manager.send_email_to_new_cd_user( self.translation_reviewer_id, constants.CD_USER_RIGHTS_CATEGORY_REVIEW_TRANSLATION, language_code='hi', ) # Make sure correct email is sent. messages = self._get_sent_email_messages( self.TRANSLATION_REVIEWER_EMAIL ) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) sent_email_model = all_models[0] self.assertEqual(sent_email_model.subject, expected_email_subject) self.assertEqual( sent_email_model.recipient_id, self.translation_reviewer_id ) self.assertEqual( sent_email_model.recipient_email, self.TRANSLATION_REVIEWER_EMAIL ) self.assertEqual(sent_email_model.sender_id, feconf.SYSTEM_COMMITTER_ID) self.assertEqual( sent_email_model.sender_email, 'Site Admin <noreply@example.com>' ) self.assertEqual( sent_email_model.intent, feconf.EMAIL_INTENT_ONBOARD_CD_USER ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ] ) def test_send_assigned_question_reviewer_email(self) -> None: expected_email_subject = ( 'You have been invited to review Oppia questions' ) expected_email_html_body = ( 'Hi question,<br><br>' 'This is to let you know that the Oppia team has added you as a ' 'reviewer for questions. This allows you to review question ' 'suggestions made by contributors.<br><br>' 'You can check the question suggestions waiting for review in the ' '<a href="https://www.oppia.org/contributor-dashboard">' 'Contributor Dashboard</a>.<br><br>' 'Thanks, and happy contributing!<br><br>' 'Best wishes,<br>' 'The Oppia Community' ) email_manager.send_email_to_new_cd_user( self.question_reviewer_id, constants.CD_USER_RIGHTS_CATEGORY_REVIEW_QUESTION, language_code='hi', ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.QUESTION_REVIEWER_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) sent_email_model = all_models[0] self.assertEqual(sent_email_model.subject, expected_email_subject) self.assertEqual( sent_email_model.recipient_id, self.question_reviewer_id ) self.assertEqual( sent_email_model.recipient_email, self.QUESTION_REVIEWER_EMAIL ) self.assertEqual(sent_email_model.sender_id, feconf.SYSTEM_COMMITTER_ID) self.assertEqual( sent_email_model.sender_email, 'Site Admin <noreply@example.com>' ) self.assertEqual( sent_email_model.intent, feconf.EMAIL_INTENT_ONBOARD_CD_USER ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ] ) def test_send_assigned_question_submitter_email(self) -> None: expected_email_subject = ( 'You have been invited to submit Oppia questions' ) expected_email_html_body = ( 'Hi questionSuggestor,<br><br>' 'This is to let you know that the Oppia team has added you as a ' 'contributor to submit question suggestions ' 'for use in lessons.<br><br>' 'You can now start to submit questions in the ' '<a href="https://www.oppia.org/contributor-dashboard">' 'Contributor Dashboard</a>.<br><br>' 'Thanks, and happy contributing!<br><br>' 'Best wishes,<br>' 'The Oppia Community' ) email_manager.send_email_to_new_cd_user( self.question_submitter_id, constants.CD_USER_RIGHTS_CATEGORY_SUBMIT_QUESTION, language_code='hi', ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.QUESTION_SUBMITTER_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) sent_email_model = all_models[0] self.assertEqual(sent_email_model.subject, expected_email_subject) self.assertEqual( sent_email_model.recipient_id, self.question_submitter_id ) self.assertEqual( sent_email_model.recipient_email, self.QUESTION_SUBMITTER_EMAIL ) self.assertEqual(sent_email_model.sender_id, feconf.SYSTEM_COMMITTER_ID) self.assertEqual( sent_email_model.sender_email, 'Site Admin <noreply@example.com>' ) self.assertEqual( sent_email_model.intent, feconf.EMAIL_INTENT_ONBOARD_CD_USER ) @test_utils.set_platform_parameters( [(param_list.ParamName.SERVER_CAN_SEND_EMAILS, False)] ) def test_email_is_not_sent_server_can_send_emails_is_false(self) -> None: email_manager.send_email_to_removed_cd_user( self.translation_reviewer_id, constants.CD_USER_RIGHTS_CATEGORY_REVIEW_TRANSLATION, language_code='hi', ) messages = self._get_sent_email_messages( self.TRANSLATION_REVIEWER_EMAIL ) self.assertEqual(len(messages), 0) def test_remove_translation_reviewer_email_for_invalid_category( self, ) -> None: with self.assertRaisesRegex(Exception, 'Invalid category'): email_manager.send_email_to_removed_cd_user( self.translation_reviewer_id, 'invalid_category' ) def test_schema_of_removed_reviewer_email_data_constant(self) -> None: self.assertEqual( sorted(email_manager.REMOVED_CD_USER_EMAIL_DATA.keys()), [ constants.CD_USER_RIGHTS_CATEGORY_REVIEW_QUESTION, constants.CD_USER_RIGHTS_CATEGORY_SUBMIT_QUESTION, constants.CD_USER_RIGHTS_CATEGORY_REVIEW_TRANSLATION, ], ) for ( category_details ) in email_manager.REMOVED_CD_USER_EMAIL_DATA.values(): self.assertEqual(len(category_details), 3) self.assertTrue( 'role_description' in category_details or ('role_description_template' in category_details) ) self.assertTrue('category' in category_details) self.assertTrue( 'rights_message' in category_details or ('rights_message_template' in category_details) ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ] ) def test_send_removed_translation_reviewer_email(self) -> None: expected_email_subject = ( 'You have been unassigned as a translation reviewer' ) expected_email_html_body = ( 'Hi translator,<br><br>' 'The Oppia team has removed you from the translation reviewer role ' 'in the हिन्दी (hindi) language. You won\'t be able to review ' 'translation suggestions made by contributors in the हिन्दी (hindi)' ' language any more, but you can still contribute translations ' 'through the <a href="https://www.oppia.org/contributor-dashboard">' 'Contributor Dashboard</a>.<br><br>' 'Thanks, and happy contributing!<br><br>' 'Best wishes,<br>' 'The Oppia Community' ) email_manager.send_email_to_removed_cd_user( self.translation_reviewer_id, constants.CD_USER_RIGHTS_CATEGORY_REVIEW_TRANSLATION, language_code='hi', ) # Make sure correct email is sent. messages = self._get_sent_email_messages( self.TRANSLATION_REVIEWER_EMAIL ) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) sent_email_model = all_models[0] self.assertEqual(sent_email_model.subject, expected_email_subject) self.assertEqual( sent_email_model.recipient_id, self.translation_reviewer_id ) self.assertEqual( sent_email_model.recipient_email, self.TRANSLATION_REVIEWER_EMAIL ) self.assertEqual(sent_email_model.sender_id, feconf.SYSTEM_COMMITTER_ID) self.assertEqual( sent_email_model.sender_email, 'Site Admin <noreply@example.com>' ) self.assertEqual( sent_email_model.intent, feconf.EMAIL_INTENT_REMOVE_CD_USER ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ] ) def test_send_removed_question_reviewer_email(self) -> None: expected_email_subject = ( 'You have been unassigned as a question reviewer' ) expected_email_html_body = ( 'Hi question,<br><br>' 'The Oppia team has removed you from the question reviewer role. ' 'You won\'t be able to review question suggestions made by ' 'contributors any more, but you can still contribute questions ' 'through the <a href="https://www.oppia.org/contributor-dashboard">' 'Contributor Dashboard</a>.<br><br>' 'Thanks, and happy contributing!<br><br>' 'Best wishes,<br>' 'The Oppia Community' ) email_manager.send_email_to_removed_cd_user( self.question_reviewer_id, constants.CD_USER_RIGHTS_CATEGORY_REVIEW_QUESTION, language_code='hi', ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.QUESTION_REVIEWER_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) sent_email_model = all_models[0] self.assertEqual(sent_email_model.subject, expected_email_subject) self.assertEqual( sent_email_model.recipient_id, self.question_reviewer_id ) self.assertEqual( sent_email_model.recipient_email, self.QUESTION_REVIEWER_EMAIL ) self.assertEqual(sent_email_model.sender_id, feconf.SYSTEM_COMMITTER_ID) self.assertEqual( sent_email_model.sender_email, 'Site Admin <noreply@example.com>' ) self.assertEqual( sent_email_model.intent, feconf.EMAIL_INTENT_REMOVE_CD_USER ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Site Admin'), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'testadmin@example.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ] ) def test_send_removed_question_submitter_email(self) -> None: expected_email_subject = ( 'You have been unassigned as a question submitter' ) expected_email_html_body = ( 'Hi questionSuggestor,<br><br>' 'The Oppia team has removed you from the question submitter role. ' 'You won\'t be able to submit question suggestions ' 'any more, but you can still contribute questions ' 'through the <a href="https://www.oppia.org/contributor-dashboard">' 'Contributor Dashboard</a>.<br><br>' 'Thanks, and happy contributing!<br><br>' 'Best wishes,<br>' 'The Oppia Community' ) email_manager.send_email_to_removed_cd_user( self.question_submitter_id, constants.CD_USER_RIGHTS_CATEGORY_SUBMIT_QUESTION, language_code='hi', ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.QUESTION_SUBMITTER_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) sent_email_model = all_models[0] self.assertEqual(sent_email_model.subject, expected_email_subject) self.assertEqual( sent_email_model.recipient_id, self.question_submitter_id ) self.assertEqual( sent_email_model.recipient_email, self.QUESTION_SUBMITTER_EMAIL ) self.assertEqual(sent_email_model.sender_id, feconf.SYSTEM_COMMITTER_ID) self.assertEqual( sent_email_model.sender_email, 'Site Admin <noreply@example.com>' ) self.assertEqual( sent_email_model.intent, feconf.EMAIL_INTENT_REMOVE_CD_USER ) class NotMergeableChangesEmailUnitTest(test_utils.EmailTestBase): """Unit test related to not mergeable change list emails sent to admin.""" @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_SENDER_NAME, 'Name'), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'admin@system.com'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.SYSTEM_EMAIL_NAME, '.'), ] ) def test_not_mergeable_change_list_email_is_sent_correctly(self) -> None: # Make sure there are no emails already sent. admin_email_address = 'admin@system.com' messages = self._get_sent_email_messages(admin_email_address) self.assertEqual(messages, []) # Send an account deletion failed email to admin. email_manager.send_not_mergeable_change_list_to_admin_for_review( 'testExploration', 1, 2, [{'field1': 'value1'}] ) # Make sure emails are sent. messages = self._get_sent_email_messages(admin_email_address) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].to, ['admin@system.com']) self.assertEqual( messages[0].subject, 'Some changes were rejected due to a conflict' ) self.assertIn( 'Hi Admin,<br><br>' 'Some draft changes were rejected in exploration ' 'testExploration because the changes were conflicting and ' 'could not be saved. Please see the ' 'rejected change list below:<br>' 'Discarded change list: [{\'field1\': \'value1\'}] <br><br>' 'Frontend Version: 1<br>' 'Backend Version: 2<br><br>' 'Thanks!', messages[0].html, ) class MailchimpSecretTest(test_utils.GenericTestBase): """Tests for the verify_mailchimp_secret.""" def setUp(self) -> None: super().setUp() self.swap_webhook_secrets_return_none = self.swap_to_always_return( secrets_services, 'get_secret', None ) self.swap_webhook_secrets_return_secret = self.swap_with_checks( secrets_services, 'get_secret', lambda _: 'secret', expected_args=[ ('MAILCHIMP_WEBHOOK_SECRET',), ('MAILCHIMP_WEBHOOK_SECRET',), ], ) def test_cloud_secrets_return_none_logs_exception(self) -> None: with self.swap_webhook_secrets_return_none: with self.capture_logging(min_level=logging.WARNING) as logs: self.assertFalse( email_manager.verify_mailchimp_secret('secret') ) self.assertEqual( ['Mailchimp Webhook secret is not available.'], logs ) def test_cloud_secrets_return_secret_passes(self) -> None: with self.swap_webhook_secrets_return_secret: self.assertTrue(email_manager.verify_mailchimp_secret('secret')) self.assertFalse( email_manager.verify_mailchimp_secret('not-secret') ) class CurriculumAdminsChapterNotificationsReminderMailTests( test_utils.EmailTestBase ): CURRICULUM_ADMIN_1_USERNAME: Final = 'user1' CURRICULUM_ADMIN_1_EMAIL: Final = 'user1@community.org' CURRICULUM_ADMIN_2_USERNAME: Final = 'user2' CURRICULUM_ADMIN_2_EMAIL: Final = 'user2@community.org' def setUp(self) -> None: super().setUp() self.signup( self.CURRICULUM_ADMIN_1_EMAIL, self.CURRICULUM_ADMIN_1_USERNAME ) self.admin_1_id = self.get_user_id_from_email( self.CURRICULUM_ADMIN_1_EMAIL ) self.signup( self.CURRICULUM_ADMIN_2_EMAIL, self.CURRICULUM_ADMIN_2_USERNAME ) self.admin_2_id = self.get_user_id_from_email( self.CURRICULUM_ADMIN_2_EMAIL ) self.log_new_error_counter = test_utils.CallCounter(logging.error) @test_utils.set_platform_parameters( [(param_list.ParamName.SERVER_CAN_SEND_EMAILS, False)] ) def test_email_not_sent_if_server_can_send_emails_is_false(self) -> None: email_manager.send_reminder_mail_to_notify_curriculum_admins( [self.CURRICULUM_ADMIN_1_EMAIL, self.CURRICULUM_ADMIN_2_EMAIL], [] ) messages = self._get_sent_email_messages(self.CURRICULUM_ADMIN_1_EMAIL) self.assertEqual(len(messages), 0) messages = self._get_sent_email_messages(self.CURRICULUM_ADMIN_2_EMAIL) self.assertEqual(len(messages), 0) @test_utils.set_platform_parameters( [(param_list.ParamName.SERVER_CAN_SEND_EMAILS, True)] ) def test_email_not_sent_if_no_admins_to_notify(self) -> None: with self.capture_logging(min_level=logging.ERROR) as logs: email_manager.send_reminder_mail_to_notify_curriculum_admins([], []) messages = self._get_all_sent_email_messages() self.assertEqual(len(messages), 0) self.assertEqual( logs[0], 'There were no curriculum admins to notify.' ) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.SYSTEM_EMAIL_NAME, '.'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_not_sent_if_no_overdue_or_upcoming_chapters(self) -> None: story_publication_timeliness = story_domain.StoryPublicationTimeliness( 'story_1', 'Story', 'Topic', [], [] ) email_manager.send_reminder_mail_to_notify_curriculum_admins( [self.admin_1_id], [story_publication_timeliness] ) messages = self._get_all_sent_email_messages() self.assertEqual(len(messages), 0) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Oppia'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.SYSTEM_EMAIL_NAME, '.'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'admin@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_sent_if_chapters_are_overdue(self) -> None: story_publication_timeliness = story_domain.StoryPublicationTimeliness( 'story_1', 'Story', 'Topic', ['Chapter 1', 'Chapter 2'], [] ) expected_email_html_body = ( 'Dear Curriculum Admin, <br><br>' 'The following stories have unpublished chapters which are behind ' 'schedule. Please publish them or adjust the planned publication ' 'date.<br><br>' '<ol>' '<li>Story (Topic) - <a href="%s%s/story_1">Link</a><ul>' '<li>Chapter 1</li>' '<li>Chapter 2</li>' '</ul></li>' '</ol>' 'Regards,<br> Oppia Foundation' ) % (DEV_OPPIA_SITE_URL, feconf.STORY_EDITOR_URL_PREFIX) expected_email_subject = 'Chapter Publication Notifications' email_manager.send_reminder_mail_to_notify_curriculum_admins( [self.admin_1_id], [story_publication_timeliness] ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.CURRICULUM_ADMIN_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) sent_email_model = all_models[0] self.assertEqual(sent_email_model.subject, expected_email_subject) self.assertEqual(sent_email_model.sender_id, feconf.SYSTEM_COMMITTER_ID) self.assertEqual( sent_email_model.sender_email, '. <noreply@example.com>' ) self.assertEqual( sent_email_model.intent, feconf.EMAIL_INTENT_NOTIFY_CURRICULUM_ADMINS_CHAPTERS, ) self.assertEqual(sent_email_model.html_body, expected_email_html_body) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.EMAIL_FOOTER, EMAIL_FOOTER), (param_list.ParamName.EMAIL_SENDER_NAME, 'Oppia'), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'system@example.com'), (param_list.ParamName.SYSTEM_EMAIL_NAME, '.'), (param_list.ParamName.NOREPLY_EMAIL_ADDRESS, 'noreply@example.com'), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'admin@example.com'), ( param_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS, DEV_OPPIA_SITE_URL, ), ] ) def test_email_sent_if_chapters_are_upcoming(self) -> None: story_publication_timeliness = story_domain.StoryPublicationTimeliness( 'story_1', 'Story', 'Topic', [], ['Chapter 1', 'Chapter 2'] ) expected_email_html_body = ( 'Dear Curriculum Admin, <br><br>' 'The following stories have unpublished chapters which are due for' ' publication in the next %s' ' days.' ' Please ensure' ' they are published on or before the planned date or adjust the ' 'planned publication date.' '<br><br>' '<ol>' '<li>Story (Topic) - <a href="%s%s/story_1">Link</a><ul>' '<li>Chapter 1</li>' '<li>Chapter 2</li>' '</ul></li>' '</ol>' 'Regards,<br> Oppia Foundation' ) % ( str(constants.CHAPTER_PUBLICATION_NOTICE_PERIOD_IN_DAYS), DEV_OPPIA_SITE_URL, feconf.STORY_EDITOR_URL_PREFIX, ) expected_email_subject = 'Chapter Publication Notifications' email_manager.send_reminder_mail_to_notify_curriculum_admins( [self.admin_1_id], [story_publication_timeliness] ) # Make sure correct email is sent. messages = self._get_sent_email_messages(self.CURRICULUM_ADMIN_1_EMAIL) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_html_body) # Make sure correct email model is stored. all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) sent_email_model = all_models[0] self.assertEqual(sent_email_model.subject, expected_email_subject) self.assertEqual(sent_email_model.sender_id, feconf.SYSTEM_COMMITTER_ID) self.assertEqual( sent_email_model.sender_email, '. <noreply@example.com>' ) self.assertEqual( sent_email_model.intent, feconf.EMAIL_INTENT_NOTIFY_CURRICULUM_ADMINS_CHAPTERS, ) self.assertEqual(sent_email_model.html_body, expected_email_html_body) class VoiceoverRegenerationNotificationEmailUnitTests(test_utils.EmailTestBase): """Tests for the voiceover regeneration notification emails.""" @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'dummy@system.com'), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'admin@system.com'), (param_list.ParamName.EMAIL_SENDER_NAME, 'Name'), ] ) def test_sends_email_to_voiceover_admins_on_regeneration_failure( self, ) -> None: date = '10-10-2025' time = '10:00 AM' exploration_id = 'exp_id_123' exploration_title = 'Test Exploration' number_of_requested_voiceovers = 10 number_of_successful_voiceovers = 8 number_of_failed_voiceovers = 2 language_descriptions = ['English', 'Hindi'] author_username = 'test_author' email_manager.send_emails_to_voiceover_admins( date, time, exploration_id, exploration_title, number_of_requested_voiceovers, number_of_successful_voiceovers, number_of_failed_voiceovers, language_descriptions, author_username, ) optional_message = ( '<br>You have also been cc’d on a separate email, sent to the ' 'voiceover tech lead, to address the failed voiceover synthesis. ' 'Please follow up on that email as needed.<br><br>' ) exploration_link = 'https://www.oppia.org/create/%s' % exploration_id expected_email_body = ( email_manager.VOICEOVER_ADMINS_REGENERATION_NOTIFICATION_EMAIL[ 'email_body_template' ] % ( author_username, exploration_link, exploration_title, ', '.join(language_descriptions), date, time, number_of_requested_voiceovers, number_of_successful_voiceovers, number_of_failed_voiceovers, optional_message, ) ) # Make sure correct email is sent. messages = self._get_sent_email_messages( email_manager.VOICEOVER_ADMIN_GOOGLE_GROUP ) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_body) # Make sure correct email model is stored. all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) sent_email_model = all_models[0] self.assertEqual( sent_email_model.subject, 'Report on Automatic Voiceovers Generated for Test Exploration', ) self.assertEqual( sent_email_model.recipient_email, email_manager.VOICEOVER_ADMIN_GOOGLE_GROUP, ) self.assertEqual(sent_email_model.sender_id, 'admin') self.assertEqual(sent_email_model.html_body, expected_email_body) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'dummy@system.com'), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'admin@system.com'), (param_list.ParamName.EMAIL_SENDER_NAME, 'Name'), ] ) def test_sends_correct_email_to_voiceover_admins_on_successful_regeneration( self, ) -> None: date = '10-10-2025' time = '10:00 AM' exploration_id = 'exp_id_123' exploration_title = 'Test Exploration' number_of_requested_voiceovers = 10 number_of_successful_voiceovers = 10 number_of_failed_voiceovers = 0 language_descriptions = ['English', 'Hindi'] author_username = 'test_author' email_manager.send_emails_to_voiceover_admins( date, time, exploration_id, exploration_title, number_of_requested_voiceovers, number_of_successful_voiceovers, number_of_failed_voiceovers, language_descriptions, author_username, ) optional_message = '<br>' exploration_link = 'https://www.oppia.org/create/%s' % exploration_id expected_email_body = ( email_manager.VOICEOVER_ADMINS_REGENERATION_NOTIFICATION_EMAIL[ 'email_body_template' ] % ( author_username, exploration_link, exploration_title, ', '.join(language_descriptions), date, time, number_of_requested_voiceovers, number_of_successful_voiceovers, number_of_failed_voiceovers, optional_message, ) ) # Make sure correct email is sent. messages = self._get_sent_email_messages( email_manager.VOICEOVER_ADMIN_GOOGLE_GROUP ) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_body) # Make sure correct email model is stored. all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) sent_email_model = all_models[0] self.assertEqual( sent_email_model.subject, 'Report on Automatic Voiceovers Generated for Test Exploration', ) self.assertEqual( sent_email_model.recipient_email, email_manager.VOICEOVER_ADMIN_GOOGLE_GROUP, ) self.assertEqual(sent_email_model.sender_id, 'admin') self.assertEqual(sent_email_model.html_body, expected_email_body) @test_utils.set_platform_parameters( [ (param_list.ParamName.SERVER_CAN_SEND_EMAILS, True), (param_list.ParamName.SYSTEM_EMAIL_ADDRESS, 'dummy@system.com'), (param_list.ParamName.ADMIN_EMAIL_ADDRESS, 'admin@system.com'), (param_list.ParamName.EMAIL_SENDER_NAME, 'Name'), ] ) def test_sends_email_to_tech_leads_on_regeneration_failure(self) -> None: date = '10-10-2025' time = '10:00 AM' exploration_id = 'exp_id_123' exploration_title = 'Test Exploration' exploration_link = 'https://www.oppia.org/create/%s' % exploration_id document_link_1 = ( 'https://docs.google.com/document/d/1Wwd0Eg2jA3rnsiER6cf-' 'ixOm3oh0T_c0lXnaRwbSWH4/edit?tab=t.0#heading=h.uc9ozkinrt05' ) document_link_2 = ( 'https://docs.google.com/document/d/1Wwd0Eg2jA3rnsiER6cf-' 'ixOm3oh0T_c0lXnaRwbSWH4/edit?tab=t.0#heading=h.y85o1y4ceo9y' ) language_descriptions = ['English', 'Hindi'] voiceover_regeneration_error_messages: List[ Dict[str, str | List[Tuple[str, str]]] ] = [ { 'exploration_id': 'exploration_id_1', 'language_accent': 'English (India)', 'error_messages': [ ('content0', 'Error message 1 for English (India)'), ('content1', 'Error message 2 for English (India)'), ], }, { 'exploration_id': 'exploration_id_2', 'language_accent': 'Hindi (India)', 'error_messages': [ ('content0', 'Error message 1 for Hindi (India)'), ('content1', 'Error message 2 for Hindi (India)'), ], }, ] expected_email_body = ( email_manager.VOICEOVER_TECH_LEADS_REGENERATION_NOTIFICATION_EMAIL[ 'email_body_template' ] % ( exploration_link, exploration_title, ', '.join(language_descriptions), date, time, document_link_1, document_link_2, document_link_1, document_link_2, ) ) email_manager.send_emails_to_voiceover_tech_leads( exploration_id, exploration_title, date, time, language_descriptions, voiceover_regeneration_error_messages, ) # Make sure correct email is sent. messages = self._get_sent_email_messages( email_manager.VOICEOVER_TECH_LEADS_GOOGLE_GROUP ) self.assertEqual(len(messages), 1) self.assertEqual(messages[0].html, expected_email_body) # Make sure correct email model is stored. all_models: Sequence[email_models.SentEmailModel] = ( email_models.SentEmailModel.get_all().fetch() ) sent_email_model = all_models[0] self.assertEqual( sent_email_model.recipient_email, email_manager.VOICEOVER_TECH_LEADS_GOOGLE_GROUP, ) self.assertEqual(sent_email_model.sender_id, 'admin') self.assertEqual(sent_email_model.html_body, expected_email_body) self.assertEqual( sent_email_model.subject, '[Attention needed] Automatic Voiceover Generation Failed', ) class EmailRetryQueueTests(test_utils.EmailTestBase): """Tests the retry logic when email sending fails.""" USER_A_EMAIL = 'a@example.com' def setUp(self) -> None: super().setUp() self.signup(self.USER_A_EMAIL, 'userA') self.user_a_id = self.get_user_id_from_email(self.USER_A_EMAIL) def test_failed_send_mail_enqueues_retry_task(self) -> None: def mock_send_mail(*_args: str, **_kwargs: str) -> None: raise Exception('Simulated email failure') enqueued_tasks = [] def mock_enqueue_task( url: str, payload: dict[str, str], _delay: int ) -> None: enqueued_tasks.append((url, payload)) send_mail_swap = self.swap(email_services, 'send_mail', mock_send_mail) enqueue_task_swap = self.swap( taskqueue_services, 'enqueue_task', mock_enqueue_task ) with send_mail_swap, enqueue_task_swap: email_manager._send_email( # pylint: disable=protected-access self.user_a_id, feconf.SYSTEM_COMMITTER_ID, feconf.EMAIL_INTENT_SIGNUP, 'Subject', 'Body', 'sender@example.com', ) self.assertEqual(len(enqueued_tasks), 1) self.assertEqual( enqueued_tasks[0][0], feconf.TASK_URL_RETRY_FAILED_EMAIL ) self.assertEqual(enqueued_tasks[0][1]['subject'], 'Subject')