Merge pull request #23395 from edx/aehsan/prod-1361/command_for_account_recovery

Command added to recover learner accounts

common/djangoapps/student/management/commands/recover_account.py

+"""
+Management command to recover learners accounts
+"""
+
+import logging
+from os import path
+import unicodecsv
+
+from django.db.models import Q
+from django.contrib.sites.models import Site
+from django.core.management.base import BaseCommand, CommandError
+from django.contrib.auth import get_user_model
+from django.conf import settings
+from django.contrib.auth.tokens import default_token_generator
+from django.urls import reverse
+from django.utils.http import int_to_base36
+from edx_ace import ace
+from edx_ace.recipient import Recipient
+
+from openedx.core.djangoapps.ace_common.template_context import get_base_template_context
+from openedx.core.djangoapps.lang_pref import LANGUAGE_KEY
+from openedx.core.djangoapps.site_configuration import helpers as configuration_helpers
+from openedx.core.djangoapps.user_api.preferences.api import get_user_preference
+from openedx.core.djangoapps.user_authn.message_types import PasswordReset
+from openedx.core.lib.celery.task_utils import emulate_http_request
+
+logger = logging.getLogger(__name__)  # pylint: disable=invalid-name
+
+
+class Command(BaseCommand):
+    """
+        Management command to recover account for the learners got their accounts taken
+        over due to bad passwords. Learner's email address will be updated and password
+        reset email would be sent to these learner's.
+    """
+
+    help = """
+        Change the email address of each user specified in the csv file and
+        send password reset email.
+
+        csv file is expected to have one row per user with the format:
+        username, email, new_email
+
+        Example:
+            $ ... recover_account csv_file_path
+        """
+
+    def add_arguments(self, parser):
+        """ Add argument to the command parser. """
+        parser.add_argument(
+            '--csv_file_path',
+            required=True,
+            help='Csv file path'
+        )
+
+    def handle(self, *args, **options):
+        """ Main handler for the command."""
+        file_path = options['csv_file_path']
+
+        if not path.isfile(file_path):
+            raise CommandError('File not found.')
+
+        with open(file_path, 'rb') as csv_file:
+            csv_reader = list(unicodecsv.DictReader(csv_file))
+
+        successful_updates = []
+        failed_updates = []
+        site = Site.objects.get_current()
+
+        for row in csv_reader:
+            username = row['username']
+            email = row['email']
+            new_email = row['new_email']
+
+            try:
+                user = get_user_model().objects.get(Q(username__iexact=username) | Q(email__iexact=email))
+                user.email = new_email
+                user.save()
+                self.send_password_reset_email(user, email, site)
+                successful_updates.append(new_email)
+            except Exception as exc:  # pylint: disable=broad-except
+                logger.exception('Unable to send email to {email} and exception was {exp}'.
+                                 format(email=email, exp=exc)
+                                 )
+
+                failed_updates.append(email)
+
+        logger.info('Successfully updated {successful} accounts. Failed to update {failed} '
+                    'accounts'.format(successful=successful_updates, failed=failed_updates)
+                    )
+
+    def send_password_reset_email(self, user, email, site):
+        """
+        Send email to learner with reset password link
+        :param user:
+        :param email:
+        :param site:
+        """
+        message_context = get_base_template_context(site)
+        message_context.update({
+            'email': email,
+            'platform_name': configuration_helpers.get_value('PLATFORM_NAME', settings.PLATFORM_NAME),
+            'reset_link': '{protocol}://{site}{link}?track=pwreset'.format(
+                protocol='http',
+                site=configuration_helpers.get_value('SITE_NAME', settings.SITE_NAME),
+                link=reverse('password_reset_confirm', kwargs={
+                    'uidb36': int_to_base36(user.id),
+                    'token': default_token_generator.make_token(user),
+                }),
+            )
+        })
+
+        with emulate_http_request(site, user):
+            msg = PasswordReset().personalize(
+                recipient=Recipient(user.username, email),
+                language=get_user_preference(user, LANGUAGE_KEY),
+                user_context=message_context,
+            )
+            ace.send(msg)

common/djangoapps/student/management/tests/test_recover_account.py

+"""
+Test cases for recover account management command
+"""
+import re
+from tempfile import NamedTemporaryFile
+import six
+
+from django.core import mail
+from django.core.management import call_command, CommandError
+from django.test import TestCase, RequestFactory
+
+from testfixtures import LogCapture
+from student.tests.factories import UserFactory
+
+
+LOGGER_NAME = 'student.management.commands.recover_account'
+
+
+class RecoverAccountTests(TestCase):
+    """
+    Test account recovery and exception handling
+    """
+
+    request_factory = RequestFactory()
+
+    def setUp(self):
+        super(RecoverAccountTests, self).setUp()
+        self.user = UserFactory.create(username='amy', email='amy@edx.com', password='password')
+
+    def _write_test_csv(self, csv, lines):
+        """Write a test csv file with the lines provided"""
+        csv.write(b"username,email,new_email\n")
+        for line in lines:
+            csv.write(six.b(line))
+        csv.seek(0)
+        return csv
+
+    def test_account_recovery(self):
+        """
+        Test account is recovered. Send email to learner and then reset password. After
+        reset password login to make sure account is recovered
+        :return:
+        """
+
+        with NamedTemporaryFile() as csv:
+            csv = self._write_test_csv(csv, lines=['amy,amy@edx.com,amy@newemail.com\n'])
+            call_command("recover_account", "--csv_file_path={}".format(csv.name))
+
+            self.assertEqual(len(mail.outbox), 1)
+
+        reset_link = re.findall("(http.+pwreset)", mail.outbox[0].body)[0]
+        request_params = {'new_password1': 'password1', 'new_password2': 'password1'}
+        self.client.get(reset_link)
+        resp = self.client.post(reset_link, data=request_params)
+
+        # Verify the response status code is: 302 with password reset because 302 means success
+        self.assertEqual(resp.status_code, 302)
+
+        self.assertTrue(self.client.login(username=self.user.username, password='password1'))
+
+        # try to login with previous password
+        self.assertFalse(self.client.login(username=self.user.username, password='password'))
+
+    def test_file_not_found_error(self):
+        """
+        Test command error raised when csv path is invalid
+        :return:
+        """
+        with self.assertRaises(CommandError):
+            call_command("recover_account", "--csv_file_path={}".format('test'))
+
+    def test_exception_raised(self):
+        """
+        Test user matching query does not exist exception raised
+        :return:
+        """
+        with NamedTemporaryFile() as csv:
+            csv = self._write_test_csv(csv, lines=['amm,amy@myedx.com,amy@newemail.com\n'])
+
+            expected_message = 'Unable to send email to amy@myedx.com and ' \
+                               'exception was User matching query does not exist.'
+
+            with LogCapture(LOGGER_NAME) as log:
+                call_command("recover_account", "--csv_file_path={}".format(csv.name))
+
+                log.check_present(
+                    (LOGGER_NAME, 'ERROR', expected_message)
+                )
+
+    def test_successfull_users_logged(self):
+        """
+        Test accumulative logs for all successfull and failed learners.
+        """
+        with NamedTemporaryFile() as csv:
+            csv = self._write_test_csv(csv, lines=['amy,amy@edx.com,amy@newemail.com\n'])
+
+            expected_message = "Successfully updated ['amy@newemail.com'] accounts. Failed to update [] accounts"
+
+            with LogCapture(LOGGER_NAME) as log:
+                call_command("recover_account", "--csv_file_path={}".format(csv.name))
+
+                log.check_present(
+                    (LOGGER_NAME, 'INFO', expected_message)
+                )