Merge pull request #34340 from openedx/mroytman/COSMO-210-idv-approva…

…l-management-command

Add new management command to approve submitted ID verification attempts.

lms/djangoapps/verify_student/management/commands/approve_id_verifications.py

@@ -0,0 +1,152 @@
+"""
+Django admin commands related to verify_student
+"""
+
+
+import logging
+import os
+import time
+from pprint import pformat
+
+from django.contrib.auth.models import User  # lint-amnesty, pylint: disable=imported-auth-user
+from django.core.management.base import BaseCommand, CommandError
+
+from lms.djangoapps.verify_student.models import SoftwareSecurePhotoVerification
+from lms.djangoapps.verify_student.utils import earliest_allowed_verification_date
+
+
+log = logging.getLogger(__name__)
+
+
+class Command(BaseCommand):
+    """
+    This command manually approves ID verification attempts for a provided set of learners whose ID verification
+    attempt is in the submitted state.
+
+    This command differs from the similar manual_verifications command in that it approves the
+    SoftwareSecurePhotoVerification instance instead of creating a ManualVerification instance. This is advantageous
+    because it ensures that the approval is registered with the Name Affirmation application to approve corresponding
+    verified names. Creating a ManualVerification instance does not effect any change in the Name Affirmation
+    application.
+
+    Example usage:
+        $ ./manage.py lms idv_verifications <absolute path of file with user IDs (one per line)>
+    """
+    help = 'Manually approves ID verifications for users with an ID verification attempt in the submitted state.'
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            'user_ids_file',
+            action='store',
+            help='Path of the file to read user IDs from.',
+            type=str,
+        )
+
+        parser.add_argument(
+            '--batch-size',
+            default=10000,
+            help='Maximum records to write in one query.',
+            type=int,
+        )
+
+        parser.add_argument(
+            '--sleep_time',
+            action='store',
+            dest='sleep_time',
+            type=int,
+            default=10,
+            help='Sleep time in seconds between update of batches'
+        )
+
+    def handle(self, *args, **options):
+        user_ids_file = options['user_ids_file']
+        batch_size = options['batch_size']
+        sleep_time = options['sleep_time']
+
+        if user_ids_file:
+            if not os.path.exists(user_ids_file):
+                raise CommandError('Pass the correct absolute path to user ID file as the first positional argument.')
+
+        total_users, failed_user_ids, total_invalid_users = self._approve_verifications_from_file(
+            user_ids_file,
+            batch_size,
+            sleep_time,
+        )
+
+        if failed_user_ids:
+            log.error('Completed ID verification approvals. {} of {} failed.'.format(
+                len(failed_user_ids),
+                total_users,
+            ))
+            log.error(f'Failed user IDs:{pformat(sorted(failed_user_ids))}')
+        else:
+            log.info(f'Successfully approved ID verification attempts for {total_users} user IDs.')
+
+    def _approve_verifications_from_file(self, user_ids_file, batch_size, sleep_time):
+        """
+        Manually approve ID verification attempts for the user provided in the user IDs file.
+
+        Arguments:
+            user_ids_file (str): path of the file containing user ids.
+            batch_size (int): limits the number of verifications written to db at once
+            sleep_time (int): sleep time in seconds between update of batches
+
+        Returns:
+            (total_users, failed_user_ids): a tuple containing count of users processed and a list containing
+             user IDs whose verifications could not be processed.
+        """
+        failed_user_ids = []
+        user_ids = []
+
+        with open(user_ids_file) as file_handler:
+            user_ids_strs = [line.rstrip() for line in file_handler]
+            log.info(f'Received request to manually approve ID verification attempts for {len(user_ids_strs)} users.')
+
+        total_invalid_users = 0
+        for user_id_str in user_ids_strs:
+            try:
+                user_id = int(user_id_str)
+                user_ids.append(user_id)
+            except ValueError:
+                total_invalid_users += 1
+                log.info(f'Skipping user ID {user_id_str}, invalid user ID.')
+
+        total_users = len(user_ids)
+        log.info(f'Attempting to manually approve ID verification attempts for {total_users} users.')
+
+        for n in range(0, total_users, batch_size):
+            failed_user_ids.extend(self._approve_id_verifications(user_ids[n:n + batch_size]))
+
+            # If we have one or more batches left to process, sleep for sleep_time.
+            if n + batch_size < total_users:
+                time.sleep(sleep_time)
+
+        return total_users, failed_user_ids, total_invalid_users
+
+    def _approve_id_verifications(self, user_ids):
+        """
+        This command manually approves ID verification attempts for a provided set of learners whose ID verification
+        attempt is in the submitted state.
+
+        Arguments:
+            user_ids (list): user IDs of the users whose ID verification attempt should be manually approved
+
+        Returns:
+            failed_user_ids: list of user IDs for which a ID verification attempt approval was not performed
+        """
+        existing_id_verifications = SoftwareSecurePhotoVerification.objects.filter(
+            user_id__in=user_ids,
+            status='submitted',
+            created_at__gte=earliest_allowed_verification_date(),
+        )
+
+        found_user_ids = existing_id_verifications.values_list('user_id', flat=True)
+        failed_user_ids = set(user_ids) - set(found_user_ids)
+
+        for user_id in failed_user_ids:
+            log.info(f'Skipping user ID {user_id}, either no user or no IDV verification attempt found.')
+
+        for verification in existing_id_verifications:
+            verification.approve(service='idv_verifications command')
+
+        return list(failed_user_ids)

lms/djangoapps/verify_student/management/commands/tests/test_approve_id_verifications.py

@@ -0,0 +1,128 @@
+"""  # lint-amnesty, pylint: disable=cyclic-import
+Tests for django admin commands in the verify_student module
+
+"""
+
+import logging
+import os
+import tempfile
+
+import pytest
+from django.core.management import CommandError, call_command
+from django.test import TestCase
+from testfixtures import LogCapture
+
+from common.djangoapps.student.tests.factories import UserFactory, UserProfileFactory
+from lms.djangoapps.verify_student.models import SoftwareSecurePhotoVerification
+
+LOGGER_NAME = 'lms.djangoapps.verify_student.management.commands.approve_id_verifications'
+
+
+class TestApproveIDVerificationsCommand(TestCase):
+    """
+    Tests for django admin commands in the verify_student module
+    """
+    tmp_file_path = os.path.join(tempfile.gettempdir(), 'tmp-user-ids.txt')
+
+    def setUp(self):
+        super().setUp()
+        self.user1_profile = UserProfileFactory.create(user=UserFactory.create())
+        self.user2_profile = UserProfileFactory.create(user=UserFactory.create())
+        self.user3_profile = UserProfileFactory.create(user=UserFactory.create())
+        self.invalid_user_id = '12345'
+
+        self.create_user_ids_file(
+            self.tmp_file_path,
+            [
+                str(self.user1_profile.id),
+                str(self.user2_profile.id),
+                str(self.user3_profile.id),
+                str(self.invalid_user_id),
+                'invalid_user_id',
+            ]
+        )
+
+    @staticmethod
+    def create_user_ids_file(file_path, user_ids):
+        """
+        Write the email_ids list to the temp file.
+        """
+        with open(file_path, 'w') as temp_file:
+            temp_file.write(str("\n".join(user_ids)))
+
+    def test_approve_id_verifications(self):
+        """
+        Tests that the approve_id_verifications management command executes successfully.
+        """
+        # Create SoftwareSecurePhotoVerification instances for the users.
+        for user in [self.user1_profile, self.user2_profile, self.user3_profile]:
+            SoftwareSecurePhotoVerification.objects.create(
+                user=user.user,
+                name=user.name,
+                status='submitted',
+            )
+
+        assert SoftwareSecurePhotoVerification.objects.filter(status='approved').count() == 0
+
+        call_command('approve_id_verifications', self.tmp_file_path)
+
+        assert SoftwareSecurePhotoVerification.objects.filter(status='approved').count() == 3
+
+    def test_user_does_not_exist_log(self):
+        """
+        Tests that the approve_id_verifications management command logs an error when an invalid user ID is
+        provided as input.
+        """
+        expected_log = (
+            (LOGGER_NAME,
+             'INFO',
+             'Received request to manually approve ID verification attempts for 5 users.'
+             ),
+            (LOGGER_NAME,
+             'INFO',
+             'Skipping user ID invalid_user_id, invalid user ID.'
+             ),
+            (LOGGER_NAME,
+             'INFO',
+             'Attempting to manually approve ID verification attempts for 4 users.'
+             ),
+            (LOGGER_NAME,
+             'INFO',
+             'Skipping user ID 3, either no user or no IDV verification attempt found.'
+             ),
+            (LOGGER_NAME,
+             'INFO',
+             'Skipping user ID 12345, either no user or no IDV verification attempt found.'
+             ),
+            (LOGGER_NAME,
+             'ERROR',
+             'Completed ID verification approvals. 2 of 4 failed.',
+             ),
+            (LOGGER_NAME,
+             'ERROR',
+             f"Failed user IDs:[{self.user3_profile.user.id}, {self.invalid_user_id}]"
+             )
+        )
+
+        # Create SoftwareSecurePhotoVerification instances for the users.
+        for user in [self.user1_profile, self.user2_profile]:
+            SoftwareSecurePhotoVerification.objects.create(
+                user=user.user,
+                name=user.name,
+                status='submitted',
+            )
+
+        with LogCapture(LOGGER_NAME, level=logging.INFO) as logger:
+            call_command('approve_id_verifications', self.tmp_file_path)
+
+            logger.check_present(
+                *expected_log,
+                order_matters=False,
+            )
+
+    def test_invalid_file_path(self):
+        """
+        Verify command raises the CommandError for invalid file path.
+        """
+        with pytest.raises(CommandError):
+            call_command('approve_id_verifications', 'invalid/user_id/file/path')