Merge pull request #18622 from edx/ahsan/LEARNER-5852-Student-un-enroll-refund-for-verified-learners

Added Management command for bulk unenrollment of users

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

@@ -0,0 +1,64 @@
+import logging
+
+import unicodecsv
+from django.core.exceptions import ObjectDoesNotExist
+from django.core.management.base import BaseCommand, CommandError
+from django.db.models import Q
+from opaque_keys import InvalidKeyError
+from opaque_keys.edx.keys import CourseKey
+
+from student.models import CourseEnrollment, User
+
+
+logger = logging.getLogger(__name__)  # pylint: disable=invalid-name
+
+
+class Command(BaseCommand):
+
+    help = """"
+    Un-enroll bulk users from the courses.
+    It expect that the data will be provided in a csv file format with
+    first row being the header and columns will be as follows:
+    user_id, username, email, course_id, is_verified, verification_date
+    """
+
+    def add_arguments(self, parser):
+        parser.add_argument('-p', '--csv_path',
+                            metavar='csv_path',
+                            dest='csv_path',
+                            required=True,
+                            help='Path to CSV file.')
+
+    def handle(self, *args, **options):
+        csv_path = options['csv_path']
+        with open(csv_path) as csvfile:
+            reader = unicodecsv.DictReader(csvfile)
+            for row in reader:
+                username = row['username']
+                email = row['email']
+                course_key = row['course_id']
+                try:
+                    user = User.objects.get(Q(username=username) | Q(email=email))
+                except ObjectDoesNotExist:
+                    user = None
+                    msg = 'User with username {} or email {} does not exist'.format(username, email)
+                    logger.warning(msg)
+
+                try:
+                    course_id = CourseKey.from_string(course_key)
+                except InvalidKeyError:
+                    course_id = None
+                    msg = 'Invalid course id {course_id}, skipping un-enrollement for {username}, {email}'.format(**row)
+                    logger.warning(msg)
+
+                if user and course_id:
+                    enrollment = CourseEnrollment.get_enrollment(user, course_id)
+                    if not enrollment:
+                        msg = 'Enrollment for the user {} in course {} does not exist!'.format(username, course_key)
+                        logger.info(msg)
+                    else:
+                        try:
+                            CourseEnrollment.unenroll(user, course_id, skip_refund=True)
+                        except Exception as err:
+                            msg = 'Error un-enrolling User {} from course {}: '.format(username, course_key, err)
+                            logger.error(msg, exc_info=True)

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

@@ -0,0 +1,104 @@
+from tempfile import NamedTemporaryFile
+
+from django.core.management import call_command
+from xmodule.modulestore.tests.django_utils import SharedModuleStoreTestCase
+from xmodule.modulestore.tests.factories import CourseFactory
+from testfixtures import LogCapture
+
+from course_modes.tests.factories import CourseModeFactory
+from student.tests.factories import UserFactory
+from student.models import CourseEnrollment, User
+
+
+LOGGER_NAME = 'student.management.commands.bulk_unenroll'
+
+
+class BulkUnenrollTests(SharedModuleStoreTestCase):
+    """Test Bulk un-enroll command works fine for all test cases."""
+    def setUp(self):
+        super(BulkUnenrollTests, self).setUp()
+        self.course = CourseFactory.create()
+        self.audit_mode = CourseModeFactory.create(
+            course_id=self.course.id,
+            mode_slug='audit',
+            mode_display_name='Audit',
+        )
+
+        self.user_info = [
+            ('amy', 'amy@pond.com', 'password'),
+            ('rory', 'rory@theroman.com', 'password'),
+            ('river', 'river@song.com', 'password')
+        ]
+        self.enrollments = []
+        self.users = []
+
+        for username, email, password in self.user_info:
+            user = UserFactory.create(username=username, email=email, password=password)
+            self.users.append(user)
+            self.enrollments.append(CourseEnrollment.enroll(user, self.course.id, mode='audit'))
+
+    def _write_test_csv(self, csv, lines=None):
+        """Write a test csv file with the lines procided"""
+        csv.write("user_id,username,email,course_id\n")
+        csv.writelines(lines)
+        csv.seek(0)
+        return csv
+
+    def test_user_not_exist(self):
+        """Verify that warning user not exist is logged for non existing user."""
+        with NamedTemporaryFile() as csv:
+            csv = self._write_test_csv(csv, lines="111,test,test@example.com,course-v1:edX+DemoX+Demo_Course\n")
+
+            with LogCapture(LOGGER_NAME) as log:
+                call_command("bulk_unenroll", "--csv_path={}".format(csv.name))
+                log.check(
+                    (
+                        LOGGER_NAME,
+                        'WARNING',
+                        'User with username {} or email {} does not exist'.format('test', 'test@example.com')
+                    )
+                )
+
+    def test_invalid_course_key(self):
+        """Verify in case of invalid course key warning is logged."""
+        with NamedTemporaryFile() as csv:
+            csv = self._write_test_csv(csv, lines="111,amy,amy@pond.com,test_course\n")
+
+            with LogCapture(LOGGER_NAME) as log:
+                call_command("bulk_unenroll", "--csv_path={}".format(csv.name))
+                log.check(
+                    (
+                        LOGGER_NAME,
+                        'WARNING',
+                        'Invalid course id {}, skipping un-enrollement for {}, {}'.format(
+                            'test_course', 'amy', 'amy@pond.com')
+                    )
+                )
+
+    def test_user_not_enrolled(self):
+        """Verify in case of user not enrolled in course warning is logged."""
+        with NamedTemporaryFile() as csv:
+            csv = self._write_test_csv(csv, lines="111,amy,amy@pond.com,course-v1:edX+DemoX+Demo_Course\n")
+
+            with LogCapture(LOGGER_NAME) as log:
+                call_command("bulk_unenroll", "--csv_path={}".format(csv.name))
+                log.check(
+                    (
+                        LOGGER_NAME,
+                        'INFO',
+                        'Enrollment for the user {} in course {} does not exist!'.format(
+                            'amy', 'course-v1:edX+DemoX+Demo_Course')
+                    )
+                )
+
+    def test_bulk_un_enroll(self):
+        """Verify users are unenrolled using the command."""
+        lines = (str(enrollment.user.id) + "," + enrollment.user.username + "," +
+                 enrollment.user.email + "," + str(enrollment.course.id) + "\n"
+                 for enrollment in self.enrollments)
+        with NamedTemporaryFile() as csv:
+            csv = self._write_test_csv(csv, lines=lines)\
+
+            call_command("bulk_unenroll", "--csv_path={}".format(csv.name))
+            for enrollment in CourseEnrollment.objects.all():
+                self.assertEqual(enrollment.is_active, False)

lms/djangoapps/shoppingcart/models.py

@@ -1882,7 +1882,7 @@ class CertificateItem(OrderItem):
         """
 
         # Only refund verified cert unenrollments that are within bounds of the expiration date
-        if (not course_enrollment.refundable()) or skip_refund:
+        if skip_refund or (not course_enrollment.refundable()):
             return
 
         target_certs = CertificateItem.objects.filter(course_id=course_enrollment.course_id, user_id=course_enrollment.user, status='purchased', mode='verified')