diff --git a/lms/djangoapps/verify_student/management/commands/approve_id_verifications.py b/lms/djangoapps/verify_student/management/commands/approve_id_verifications.py new file mode 100644 index 000000000000..edada5fd7a2a --- /dev/null +++ b/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 + """ + 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) diff --git a/lms/djangoapps/verify_student/management/commands/tests/test_approve_id_verifications.py b/lms/djangoapps/verify_student/management/commands/tests/test_approve_id_verifications.py new file mode 100644 index 000000000000..5e616d04fbe0 --- /dev/null +++ b/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')