feat: add new management command to approve submitted ID verification attempts
This pull requests adds a new management command approve_id_verifications to manually approve submitted ID verification attempts (i.e. instances of the SoftwareSecurePhotoVerifications model).
This commit is contained in:
michaelroytman
@@ -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)
|
||||
@@ -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')
|
||||
Reference in New Issue
Block a user