diff --git a/courses/api.py b/courses/api.py index 7007c08bbb..4a6c058828 100644 --- a/courses/api.py +++ b/courses/api.py @@ -369,6 +369,7 @@ def deactivate_run_enrollment( run_enrollment, change_status, keep_failed_enrollments=None, + send_notification=True, # noqa: FBT002 ): """ Helper method to deactivate a CourseRunEnrollment @@ -379,6 +380,8 @@ def deactivate_run_enrollment( keep_failed_enrollments: (boolean): If True, keeps the local enrollment record in the database even if the enrollment fails in edX. If None, defaults to the value of the IGNORE_EDX_FAILURES feature flag. + send_notification (bool): If True, sends unenrollment email to the learner. + Defaults to True. Returns: CourseRunEnrollment: The deactivated enrollment @@ -404,7 +407,8 @@ def deactivate_run_enrollment( edx_unenrolled = False else: edx_unenrolled = True - mail_api.send_course_run_unenrollment_email(run_enrollment) + if send_notification: + mail_api.send_course_run_unenrollment_email(run_enrollment) if edx_unenrolled: run_enrollment.edx_enrolled = False run_enrollment.edx_emails_subscription = False diff --git a/courses/management/commands/test_unenroll_enrollment.py b/courses/management/commands/test_unenroll_enrollment.py deleted file mode 100644 index 6cf58b53fb..0000000000 --- a/courses/management/commands/test_unenroll_enrollment.py +++ /dev/null @@ -1,149 +0,0 @@ -"""Tests for Unenroll Enrollment management command""" - -from types import SimpleNamespace - -import pytest -import reversion -from django.core.management.base import CommandError -from reversion.models import Version - -from courses.constants import ENROLL_CHANGE_STATUS_UNENROLLED -from courses.factories import ( - CourseRunEnrollmentFactory, - CourseRunFactory, -) -from courses.management.commands import unenroll_enrollment -from ecommerce.factories import LineFactory, OrderFactory, ProductFactory -from ecommerce.models import OrderStatus -from users.factories import UserFactory - -pytestmark = [pytest.mark.django_db] - - -@pytest.fixture -def patches(mocker): # pylint: disable=missing-docstring - edx_unenroll = mocker.patch("courses.api.unenroll_edx_course_run") - log_exception = mocker.patch("courses.api.log.exception") - sync_hubspot_line_by_line_id = mocker.patch( - "hubspot_sync.task_helpers.sync_hubspot_line_by_line_id" - ) - return SimpleNamespace( - edx_unenroll=edx_unenroll, - log_exception=log_exception, - sync_hubspot_line_by_line_id=sync_hubspot_line_by_line_id, - ) - - -def test_unenroll_enrollment_no_argument(): - """Test that command throws error when no input is provided""" - - with pytest.raises(CommandError) as command_error: - unenroll_enrollment.Command().handle() - assert str(command_error.value) == "Could not find a user with =" - - -def test_unenroll_enrollment_invalid_run(): - """ - Test that unenroll_enrollment management command throws proper error when - no valid course run is supplied - """ - - test_user = UserFactory.create() - with pytest.raises(CommandError) as command_error: - unenroll_enrollment.Command().handle(user=test_user.edx_username) - assert ( - str(command_error.value) - == f"Could not find course run with courseware_id={None}" - ) - - with pytest.raises(CommandError) as command_error: - unenroll_enrollment.Command().handle(user=test_user.edx_username, run="test") - assert ( - str(command_error.value) == "Could not find course run with courseware_id=test" - ) - - -def test_unenroll_enrollment_invalid_user(): - """Test that the command throws proper error when user is invalid arguments""" - run = CourseRunFactory.create() - - with pytest.raises(CommandError) as command_error: - unenroll_enrollment.Command().handle( - user="test", - run=run.courseware_id, - ) - assert ( - str(command_error.value) - == "Could not find a user with =test" - ) - - -def test_unenroll_enrollment(patches): - """ - Test that user unenrolled from the course properly - """ - enrollment = CourseRunEnrollmentFactory.create(edx_enrolled=True) - with reversion.create_revision(): - product = ProductFactory.create(purchasable_object=enrollment.run) - version = Version.objects.get_for_object(product).first() - order = OrderFactory.create(state=OrderStatus.PENDING, purchaser=enrollment.user) - LineFactory.create( - order=order, purchased_object=enrollment.run, product_version=version - ) - assert enrollment.change_status is None - assert enrollment.active is True - assert enrollment.edx_enrolled is True - unenroll_enrollment.Command().handle( - run=enrollment.run.courseware_id, - user=enrollment.user.edx_username, - ) - patches.edx_unenroll.assert_called_once_with(enrollment) - patches.sync_hubspot_line_by_line_id.assert_called_once() - enrollment.refresh_from_db() - assert enrollment.change_status == ENROLL_CHANGE_STATUS_UNENROLLED - assert enrollment.active is False - assert enrollment.edx_enrolled is False - - -def test_unenroll_enrollment_without_edx(mocker): - """ - Test that user unenrolled from the course properly without edx - """ - enrollment = CourseRunEnrollmentFactory.create(edx_enrolled=True) - with reversion.create_revision(): - product = ProductFactory.create(purchasable_object=enrollment.run) - version = Version.objects.get_for_object(product).first() - order = OrderFactory.create(state=OrderStatus.PENDING, purchaser=enrollment.user) - LineFactory.create( - order=order, purchased_object=enrollment.run, product_version=version - ) - sync_hubspot_line_by_line_id = mocker.patch( - "hubspot_sync.task_helpers.sync_hubspot_line_by_line_id" - ) - assert enrollment.change_status is None - assert enrollment.active is True - assert enrollment.edx_enrolled is True - # User will not be unenrolled - # Unenrolling without mocker and keep_failed_enrollments argument - unenroll_enrollment.Command().handle( - run=enrollment.run.courseware_id, - user=enrollment.user.edx_username, - ) - enrollment.refresh_from_db() - # Enrollment will remain as it is - assert enrollment.change_status is None - assert enrollment.active is True - assert enrollment.edx_enrolled is True - # User will not unenrolled - # Unenrolling with keep_failed_enrollments argument - unenroll_enrollment.Command().handle( - run=enrollment.run.courseware_id, - user=enrollment.user.edx_username, - keep_failed_enrollments=True, - ) - enrollment.refresh_from_db() - assert enrollment.change_status == ENROLL_CHANGE_STATUS_UNENROLLED - assert enrollment.active is False - # Enrollment will remain edx_enrolled - assert enrollment.edx_enrolled is True - sync_hubspot_line_by_line_id.assert_called_once() diff --git a/courses/management/commands/unenroll_enrollment.py b/courses/management/commands/unenroll_enrollment.py deleted file mode 100644 index dfc7ba80ad..0000000000 --- a/courses/management/commands/unenroll_enrollment.py +++ /dev/null @@ -1,92 +0,0 @@ -""" -Management command to unenroll enrollment for a course run for the given User - -Check the usages of this command below: - -**Unenroll enrollment** - -1. Unenroll enrollment for user -./manage.py unenroll_enrollment -—user= -—run= - -**Keep failed enrollments** - -2. Keep failed enrollments -./manage.py unenroll_enrollment -—user= -—run= -k or --keep-failed-enrollments -""" - -from django.contrib.auth import get_user_model -from django.core.management.base import CommandError - -from courses.api import deactivate_run_enrollment -from courses.constants import ENROLL_CHANGE_STATUS_UNENROLLED -from courses.management.utils import EnrollmentChangeCommand, enrollment_summary -from courses.models import CourseRun -from users.api import fetch_user - -User = get_user_model() - - -class Command(EnrollmentChangeCommand): - """Sets a user's enrollment to 'unenrolled' and deactivates it""" - - help = "Sets a user's enrollment to 'unenrolled' and deactivates it" - - def add_arguments(self, parser): - parser.add_argument( - "--user", - type=str, - help="The id, email, or username of the enrolled User", - required=True, - ) - parser.add_argument( - "--run", - type=str, - help="The 'courseware_id' value for an enrolled CourseRun", - required=True, - ) - parser.add_argument( - "-k", - "--keep-failed-enrollments", - action="store_true", - dest="keep_failed_enrollments", - help="If provided, enrollment records will be kept even if edX enrollment fails", - ) - - super().add_arguments(parser) - - def handle(self, *args, **options): # noqa: ARG002 - """Handle command execution""" - username = options.get("user", "") - try: - user = fetch_user(username) - except User.DoesNotExist: - raise CommandError( # noqa: B904 - f"Could not find a user with ={username}" # noqa: EM102 - ) - courseware_id = options.get("run") - course_run = CourseRun.objects.filter(courseware_id=courseware_id).first() - if course_run is None: - raise CommandError( - f"Could not find course run with courseware_id={courseware_id}" # noqa: EM102 - ) - - keep_failed_enrollments = options.get("keep_failed_enrollments") - enrollment, _ = self.fetch_enrollment(user, options) - run_enrollment = deactivate_run_enrollment( - enrollment, - change_status=ENROLL_CHANGE_STATUS_UNENROLLED, - keep_failed_enrollments=keep_failed_enrollments, - ) - - if run_enrollment: - success_msg = f"Unenrolled enrollment for user: {enrollment.user.edx_username} ({enrollment.user.email})\nEnrollment affected: {enrollment_summary(run_enrollment)}" - - self.stdout.write(self.style.SUCCESS(success_msg)) - else: - self.stdout.write( - self.style.ERROR( - "Failed to unenroll the enrollment - 'for' user: {} ({}) from course ({})\n".format( - user.edx_username, user.email, options["run"] - ) - ) - ) diff --git a/courses/management/commands/unenroll_learners.py b/courses/management/commands/unenroll_learners.py new file mode 100644 index 0000000000..4cb98cc85a --- /dev/null +++ b/courses/management/commands/unenroll_learners.py @@ -0,0 +1,268 @@ +""" +Management command to unenroll learners from course runs in both edX and MITx Online. + +By default, the command runs in dry-run mode (preview only). Use --commit to apply changes. + +**Usage:** + +1. Unenroll all active learners from a course run (preview): +./manage.py unenroll_learners --run=course-v1:MITx+6.00.1x+2024 + +2. Unenroll specific users from a course run: +./manage.py unenroll_learners --users=user1@example.com,user2@example.com --run=course-v1:MITx+6.00.1x+2024 --commit + +3. Unenroll users listed in a CSV file (columns: user, courseware_id): +./manage.py unenroll_learners --csv=unenrollments.csv --commit + +4. Suppress unenrollment emails: +./manage.py unenroll_learners --run=course-v1:MITx+6.00.1x+2024 --commit --no-email + +5. Keep local enrollment records even if edX unenrollment fails: +./manage.py unenroll_learners --users=user1@example.com --run=course-v1:MITx+6.00.1x+2024 --commit -k +""" + +import csv + +from django.contrib.auth import get_user_model +from django.core.management.base import BaseCommand, CommandError + +from courses.management.utils import bulk_unenroll_learners +from courses.models import CourseRun, CourseRunEnrollment +from users.api import fetch_user + +User = get_user_model() + + +class Command(BaseCommand): + """Unenroll learners from course runs in both edX and MITx Online""" + + help = "Unenroll learners from course runs in both edX and MITx Online" + + def add_arguments(self, parser): + parser.add_argument( + "--csv", + type=str, + help="Path to a CSV file with columns: user, courseware_id", + ) + parser.add_argument( + "--users", + type=str, + help="Comma-separated list of user emails or usernames", + ) + parser.add_argument( + "--run", + type=str, + help="The 'courseware_id' value for a CourseRun. " + "When used alone, unenrolls ALL active learners from this run. " + "When used with --users, scopes unenrollment to those users.", + ) + parser.add_argument( + "-k", + "--keep-failed-enrollments", + action="store_true", + dest="keep_failed_enrollments", + help="Keep local enrollment records even if edX unenrollment fails", + ) + parser.add_argument( + "--commit", + action="store_true", + dest="commit", + help="Actually perform unenrollments. Without this flag, " + "the command runs in dry-run mode (preview only).", + ) + parser.add_argument( + "--no-email", + action="store_true", + dest="no_email", + help="Suppress unenrollment notification emails to learners", + ) + + def _parse_csv(self, csv_path): + """ + Parse CSV file and return list of (user_identifier, courseware_id) tuples. + + Args: + csv_path (str): Path to the CSV file + + Returns: + list[tuple[str, str]]: List of (user_identifier, courseware_id) tuples + """ + entries = [] + try: + with open(csv_path, newline="") as f: # noqa: PTH123 + reader = csv.DictReader(f) + if not reader.fieldnames or not {"user", "courseware_id"}.issubset( + set(reader.fieldnames) + ): + raise CommandError( + "CSV file must have 'user' and 'courseware_id' columns" # noqa: EM101 + ) + for row_num, row in enumerate(reader, start=2): + user_val = row["user"].strip() + courseware_id = row["courseware_id"].strip() + if not user_val or not courseware_id: + self.stderr.write( + self.style.WARNING( + f"Row {row_num}: Skipping empty user or courseware_id" + ) + ) + continue + entries.append((user_val, courseware_id)) + except FileNotFoundError: + raise CommandError(f"CSV file not found: {csv_path}") # noqa: B904, EM102 + return entries + + def _parse_inline_users(self, users_str, courseware_id): + """ + Parse inline users string and return list of (user_identifier, courseware_id) tuples. + + Args: + users_str (str): Comma-separated user emails/usernames + courseware_id (str): The courseware_id for the course run + + Returns: + list[tuple[str, str]]: List of (user_identifier, courseware_id) tuples + """ + return [(u.strip(), courseware_id) for u in users_str.split(",") if u.strip()] + + def _entries_for_run(self, courseware_id): + """ + Build entries list for all active enrollments in a course run. + + Args: + courseware_id (str): The courseware_id for the course run + + Returns: + list[tuple[str, str]]: List of (user_email, courseware_id) tuples + """ + course_run = CourseRun.objects.filter(courseware_id=courseware_id).first() + if course_run is None: + raise CommandError( + f"Could not find course run with courseware_id={courseware_id}" # noqa: EM102 + ) + enrollments = CourseRunEnrollment.objects.filter( + run=course_run, active=True + ).select_related("user") + return [(e.user.email, courseware_id) for e in enrollments] + + def _dry_run(self, entries): + """Preview which enrollments would be unenrolled without making changes.""" + succeeded = 0 + skipped = 0 + run_cache = {} + + for user_identifier, cw_id in entries: + try: + user = fetch_user(user_identifier) + except User.DoesNotExist: + self.stderr.write( + self.style.WARNING(f"SKIP: User not found: {user_identifier}") + ) + skipped += 1 + continue + + if cw_id not in run_cache: + run_cache[cw_id] = CourseRun.objects.filter(courseware_id=cw_id).first() + course_run = run_cache[cw_id] + if course_run is None: + self.stderr.write( + self.style.WARNING(f"SKIP: Course run not found: {cw_id}") + ) + skipped += 1 + continue + + enrollment = CourseRunEnrollment.objects.filter( + user=user, run=course_run + ).first() + if enrollment is None or not enrollment.active: + self.stderr.write( + self.style.WARNING( + f"SKIP: No active enrollment for {user.email} in {cw_id}" + ) + ) + skipped += 1 + else: + self.stdout.write(f" Would unenroll: {user.email} from {cw_id}") + succeeded += 1 + + self.stdout.write("") + self.stdout.write( + self.style.SUCCESS( + f"Summary: {succeeded} would be unenrolled, {skipped} skipped" + ) + ) + if succeeded > 0: + self.stdout.write( + self.style.WARNING("Re-run with --commit to apply changes.") + ) + + def _resolve_entries(self, csv_path, users_str, courseware_id): + """Validate arguments and return the list of (user, courseware_id) entries.""" + if csv_path and (users_str or courseware_id): + raise CommandError( + "--csv cannot be combined with --users or --run" # noqa: EM101 + ) + + if users_str and not courseware_id: + raise CommandError("--run is required when using --users") # noqa: EM101 + + if not csv_path and not users_str and not courseware_id: + raise CommandError( + "Provide --csv, --users with --run, or --run alone." # noqa: EM101 + ) + + if csv_path: + entries = self._parse_csv(csv_path) + elif users_str: + entries = self._parse_inline_users(users_str, courseware_id) + else: + entries = self._entries_for_run(courseware_id) + + if not entries: + raise CommandError("No valid entries found to process") # noqa: EM101 + + return entries + + def handle(self, *args, **options): # noqa: ARG002 + """Handle command execution""" + csv_path = options.get("csv") + users_str = options.get("users") + courseware_id = options.get("run") + keep_failed = options.get("keep_failed_enrollments") + commit = options.get("commit") + no_email = options.get("no_email") + + entries = self._resolve_entries(csv_path, users_str, courseware_id) + + dry_run = not commit + self.stdout.write( + f"Processing {len(entries)} unenrollment(s)..." + + (" (DRY RUN)" if dry_run else "") + ) + + if dry_run: + self._dry_run(entries) + return + + summary = bulk_unenroll_learners( + entries, + keep_failed_enrollments=keep_failed, + send_notification=not no_email, + ) + + # Print details + for _user_id, _cw_id, status, message in summary["details"]: + if status == "succeeded": + self.stdout.write(self.style.SUCCESS(f" {message}")) + elif status == "skipped": + self.stderr.write(self.style.WARNING(f" SKIP: {message}")) + else: + self.stderr.write(self.style.ERROR(f" FAILED: {message}")) + + self.stdout.write("") + self.stdout.write( + self.style.SUCCESS( + f"Summary: {summary['succeeded']} succeeded, " + f"{summary['failed']} failed, {summary['skipped']} skipped" + ) + ) diff --git a/courses/management/commands/unenroll_learners_test.py b/courses/management/commands/unenroll_learners_test.py new file mode 100644 index 0000000000..8085b82730 --- /dev/null +++ b/courses/management/commands/unenroll_learners_test.py @@ -0,0 +1,473 @@ +"""Tests for unenroll_learners management command""" + +import csv +import tempfile +from io import StringIO + +import pytest +from django.core.management import call_command +from django.core.management.base import CommandError + +from courses.factories import CourseRunEnrollmentFactory, CourseRunFactory +from courses.management.utils import bulk_unenroll_learners +from users.factories import UserFactory + + +@pytest.fixture +def mock_bulk_unenroll(mocker): + """Mock bulk_unenroll_learners to avoid edX API calls""" + return mocker.patch( + "courses.management.commands.unenroll_learners.bulk_unenroll_learners", + return_value={ + "succeeded": 0, + "failed": 0, + "skipped": 0, + "details": [], + }, + ) + + +@pytest.mark.django_db() +class TestBulkUnenrollInlineUsers: + """Tests for --users flag""" + + def test_inline_users_success(self, mock_bulk_unenroll): + """Unenrolling inline users should call bulk_unenroll_learners with correct entries""" + mock_bulk_unenroll.return_value = { + "succeeded": 2, + "failed": 0, + "skipped": 0, + "details": [ + ("a@b.com", "run-1", "succeeded", "Unenrolled: a@b.com from run-1"), + ("c@d.com", "run-1", "succeeded", "Unenrolled: c@d.com from run-1"), + ], + } + + out = StringIO() + call_command( + "unenroll_learners", + users="a@b.com,c@d.com", + run="run-1", + commit=True, + stdout=out, + ) + + mock_bulk_unenroll.assert_called_once_with( + [("a@b.com", "run-1"), ("c@d.com", "run-1")], + keep_failed_enrollments=False, + send_notification=True, + ) + output = out.getvalue() + assert "2 succeeded" in output + assert "0 failed" in output + + def test_inline_users_missing_run(self): + """--users without --run should raise CommandError""" + with pytest.raises(CommandError, match="--run is required"): + call_command("unenroll_learners", users="user@example.com") + + def test_inline_users_keep_failed_enrollments(self, mock_bulk_unenroll): + """--keep-failed-enrollments flag should be passed through""" + call_command( + "unenroll_learners", + users="a@b.com", + run="run-1", + keep_failed_enrollments=True, + commit=True, + stdout=StringIO(), + ) + + mock_bulk_unenroll.assert_called_once_with( + [("a@b.com", "run-1")], + keep_failed_enrollments=True, + send_notification=True, + ) + + def test_inline_users_mixed_results(self, mock_bulk_unenroll): + """Command should display correct summary from bulk_unenroll_learners result""" + mock_bulk_unenroll.return_value = { + "succeeded": 1, + "failed": 1, + "skipped": 1, + "details": [ + ("a@b.com", "run-1", "succeeded", "Unenrolled: a@b.com from run-1"), + ("b@c.com", "run-1", "failed", "Failed to unenroll b@c.com from run-1"), + ( + "c@d.com", + "run-1", + "skipped", + "No active enrollment for c@d.com in run-1", + ), + ], + } + + out = StringIO() + err = StringIO() + call_command( + "unenroll_learners", + users="a@b.com,b@c.com,c@d.com", + run="run-1", + commit=True, + stdout=out, + stderr=err, + ) + + output = out.getvalue() + assert "1 succeeded" in output + assert "1 failed" in output + assert "1 skipped" in output + + def test_no_email_flag(self, mock_bulk_unenroll): + """--no-email should pass send_notification=False""" + call_command( + "unenroll_learners", + users="a@b.com", + run="run-1", + commit=True, + no_email=True, + stdout=StringIO(), + ) + + mock_bulk_unenroll.assert_called_once_with( + [("a@b.com", "run-1")], + keep_failed_enrollments=False, + send_notification=False, + ) + + +@pytest.mark.django_db() +class TestBulkUnenrollCSV: + """Tests for --csv flag""" + + def _write_csv(self, rows): + """Write a CSV file and return its path""" + f = tempfile.NamedTemporaryFile( # noqa: SIM115 + mode="w", suffix=".csv", delete=False, newline="" + ) + writer = csv.DictWriter(f, fieldnames=["user", "courseware_id"]) + writer.writeheader() + for row in rows: + writer.writerow(row) + f.close() + return f.name + + def test_csv_success(self, mock_bulk_unenroll): + """CSV-based unenrollment should pass parsed entries to bulk_unenroll_learners""" + mock_bulk_unenroll.return_value = { + "succeeded": 2, + "failed": 0, + "skipped": 0, + "details": [ + ("a@b.com", "run-1", "succeeded", "Unenrolled: a@b.com from run-1"), + ("c@d.com", "run-2", "succeeded", "Unenrolled: c@d.com from run-2"), + ], + } + + csv_path = self._write_csv( + [ + {"user": "a@b.com", "courseware_id": "run-1"}, + {"user": "c@d.com", "courseware_id": "run-2"}, + ] + ) + + out = StringIO() + call_command("unenroll_learners", csv=csv_path, commit=True, stdout=out) + + mock_bulk_unenroll.assert_called_once_with( + [("a@b.com", "run-1"), ("c@d.com", "run-2")], + keep_failed_enrollments=False, + send_notification=True, + ) + assert "2 succeeded" in out.getvalue() + + def test_csv_missing_columns(self): + """CSV without required columns should raise CommandError""" + f = tempfile.NamedTemporaryFile( # noqa: SIM115 + mode="w", suffix=".csv", delete=False, newline="" + ) + writer = csv.DictWriter(f, fieldnames=["email", "course"]) + writer.writeheader() + writer.writerow({"email": "a@b.com", "course": "x"}) + f.close() + + with pytest.raises(CommandError, match="must have 'user' and 'courseware_id'"): + call_command("unenroll_learners", csv=f.name, stdout=StringIO()) + + def test_csv_file_not_found(self): + """Non-existent CSV path should raise CommandError""" + with pytest.raises(CommandError, match="CSV file not found"): + call_command( + "unenroll_learners", csv="/nonexistent/file.csv", stdout=StringIO() + ) + + def test_csv_skips_empty_rows(self, mock_bulk_unenroll): + """Rows with empty user or courseware_id should not be passed to the util""" + mock_bulk_unenroll.return_value = { + "succeeded": 1, + "failed": 0, + "skipped": 0, + "details": [ + ("a@b.com", "run-1", "succeeded", "Unenrolled: a@b.com from run-1"), + ], + } + + csv_path = self._write_csv( + [ + {"user": "", "courseware_id": "course-v1:x+y+z"}, + {"user": "a@b.com", "courseware_id": "run-1"}, + ] + ) + + out = StringIO() + call_command("unenroll_learners", csv=csv_path, commit=True, stdout=out) + + # Only the valid row should be passed + mock_bulk_unenroll.assert_called_once_with( + [("a@b.com", "run-1")], + keep_failed_enrollments=False, + send_notification=True, + ) + + +@pytest.mark.django_db() +class TestBulkUnenrollDryRun: + """Tests for default dry-run behavior (no --commit flag)""" + + def test_default_is_dry_run(self, mock_bulk_unenroll): + """Without --commit, command should run in dry-run mode""" + enrollment = CourseRunEnrollmentFactory.create(active=True) + out = StringIO() + call_command( + "unenroll_learners", + users=enrollment.user.email, + run=enrollment.run.courseware_id, + stdout=out, + ) + + mock_bulk_unenroll.assert_not_called() + output = out.getvalue() + assert "DRY RUN" in output + assert "Would unenroll" in output + assert "Re-run with --commit" in output + + def test_dry_run_skips_inactive(self, mock_bulk_unenroll): + """Dry run should report skipped for inactive enrollments""" + enrollment = CourseRunEnrollmentFactory.create(active=False) + out = StringIO() + err = StringIO() + call_command( + "unenroll_learners", + users=enrollment.user.email, + run=enrollment.run.courseware_id, + stdout=out, + stderr=err, + ) + + mock_bulk_unenroll.assert_not_called() + assert "No active enrollment" in err.getvalue() + + def test_dry_run_user_not_found(self, mock_bulk_unenroll): + """Dry run should report skipped for non-existent users""" + run = CourseRunFactory.create() + out = StringIO() + err = StringIO() + call_command( + "unenroll_learners", + users="nonexistent@example.com", + run=run.courseware_id, + stdout=out, + stderr=err, + ) + + mock_bulk_unenroll.assert_not_called() + assert "User not found" in err.getvalue() + + +@pytest.mark.django_db() +class TestBulkUnenrollRunAlone: + """Tests for --run used alone (unenroll all active learners from a run)""" + + def test_run_alone_unenrolls_all_active(self, mock_bulk_unenroll): + """--run alone should build entries for all active enrollments in the run""" + run = CourseRunFactory.create() + e1 = CourseRunEnrollmentFactory.create(run=run, active=True) + e2 = CourseRunEnrollmentFactory.create(run=run, active=True) + CourseRunEnrollmentFactory.create(run=run, active=False) # inactive, excluded + + out = StringIO() + call_command( + "unenroll_learners", + run=run.courseware_id, + commit=True, + stdout=out, + ) + + call_args = mock_bulk_unenroll.call_args + entries = call_args[0][0] + emails = {e[0] for e in entries} + assert emails == {e1.user.email, e2.user.email} + assert all(e[1] == run.courseware_id for e in entries) + + def test_run_alone_nonexistent_run(self): + """--run alone with a nonexistent course run should raise CommandError""" + with pytest.raises(CommandError, match="Could not find course run"): + call_command( + "unenroll_learners", + run="course-v1:fake+fake+fake", + commit=True, + stdout=StringIO(), + ) + + def test_run_alone_no_active_enrollments(self): + """--run alone with no active enrollments should raise CommandError""" + run = CourseRunFactory.create() + CourseRunEnrollmentFactory.create(run=run, active=False) + + with pytest.raises(CommandError, match="No valid entries found"): + call_command( + "unenroll_learners", + run=run.courseware_id, + commit=True, + stdout=StringIO(), + ) + + def test_run_alone_dry_run_by_default(self, mock_bulk_unenroll): + """--run alone without --commit should be dry-run""" + run = CourseRunFactory.create() + CourseRunEnrollmentFactory.create(run=run, active=True) + + out = StringIO() + call_command( + "unenroll_learners", + run=run.courseware_id, + stdout=out, + ) + + mock_bulk_unenroll.assert_not_called() + assert "DRY RUN" in out.getvalue() + + +@pytest.mark.django_db() +class TestBulkUnenrollNoArgs: + """Tests for missing arguments""" + + def test_no_args_raises_error(self): + """Command with no arguments should raise CommandError""" + with pytest.raises(CommandError, match="Provide --csv"): + call_command("unenroll_learners", stdout=StringIO()) + + def test_csv_with_users_raises_error(self): + """--csv combined with --users should raise CommandError""" + with pytest.raises(CommandError, match="--csv cannot be combined"): + call_command( + "unenroll_learners", + csv="file.csv", + users="a@b.com", + stdout=StringIO(), + ) + + def test_csv_with_run_raises_error(self): + """--csv combined with --run should raise CommandError""" + with pytest.raises(CommandError, match="--csv cannot be combined"): + call_command( + "unenroll_learners", + csv="file.csv", + run="run-1", + stdout=StringIO(), + ) + + +@pytest.mark.django_db() +class TestBulkUnenrollLearnersUtil: + """Tests for the bulk_unenroll_learners utility function""" + + def test_successful_unenrollment(self, mocker): + """Should unenroll active enrollments successfully""" + enrollment = CourseRunEnrollmentFactory.create(active=True) + mocker.patch( + "courses.management.utils.deactivate_run_enrollment", + return_value=enrollment, + ) + + result = bulk_unenroll_learners( + [(enrollment.user.email, enrollment.run.courseware_id)] + ) + + assert result["succeeded"] == 1 + assert result["failed"] == 0 + assert result["skipped"] == 0 + + def test_user_not_found(self): + """Should skip when user doesn't exist""" + run = CourseRunFactory.create() + result = bulk_unenroll_learners( + [("nonexistent@example.com", run.courseware_id)] + ) + + assert result["skipped"] == 1 + assert result["succeeded"] == 0 + + def test_course_run_not_found(self): + """Should skip when course run doesn't exist""" + user = UserFactory.create() + result = bulk_unenroll_learners([(user.email, "course-v1:fake+fake+fake")]) + + assert result["skipped"] == 1 + assert result["succeeded"] == 0 + + def test_no_active_enrollment(self): + """Should skip when enrollment is inactive""" + enrollment = CourseRunEnrollmentFactory.create(active=False) + result = bulk_unenroll_learners( + [(enrollment.user.email, enrollment.run.courseware_id)] + ) + + assert result["skipped"] == 1 + assert result["succeeded"] == 0 + + def test_deactivation_failure(self, mocker): + """Should count as failed when deactivate_run_enrollment returns None""" + enrollment = CourseRunEnrollmentFactory.create(active=True) + mocker.patch( + "courses.management.utils.deactivate_run_enrollment", + return_value=None, + ) + + result = bulk_unenroll_learners( + [(enrollment.user.email, enrollment.run.courseware_id)] + ) + + assert result["failed"] == 1 + assert result["succeeded"] == 0 + + def test_keep_failed_enrollments_passed(self, mocker): + """Should pass keep_failed_enrollments to deactivate_run_enrollment""" + enrollment = CourseRunEnrollmentFactory.create(active=True) + mock_deactivate = mocker.patch( + "courses.management.utils.deactivate_run_enrollment", + return_value=enrollment, + ) + + bulk_unenroll_learners( + [(enrollment.user.email, enrollment.run.courseware_id)], + keep_failed_enrollments=True, + ) + + mock_deactivate.assert_called_once() + assert mock_deactivate.call_args[1]["keep_failed_enrollments"] is True + + def test_send_notification_passed(self, mocker): + """Should pass send_notification to deactivate_run_enrollment""" + enrollment = CourseRunEnrollmentFactory.create(active=True) + mock_deactivate = mocker.patch( + "courses.management.utils.deactivate_run_enrollment", + return_value=enrollment, + ) + + bulk_unenroll_learners( + [(enrollment.user.email, enrollment.run.courseware_id)], + send_notification=False, + ) + + mock_deactivate.assert_called_once() + assert mock_deactivate.call_args[1]["send_notification"] is False diff --git a/courses/management/utils.py b/courses/management/utils.py index 698cc3a3d4..69159ead81 100644 --- a/courses/management/utils.py +++ b/courses/management/utils.py @@ -1,11 +1,15 @@ """Utility functions/classes for course management commands""" import json +import logging +from django.contrib.auth import get_user_model from django.core.management.base import BaseCommand, CommandError from mitol.common.utils.collections import has_equal_properties from courses import mail_api +from courses.api import deactivate_run_enrollment +from courses.constants import ENROLL_CHANGE_STATUS_UNENROLLED from courses.models import CourseRun, CourseRunEnrollment, Program, ProgramEnrollment from main import settings from openedx.api import enroll_in_edx_course_runs @@ -14,6 +18,10 @@ NoEdxApiAuthError, UnknownEdxApiEnrollException, ) +from users.api import fetch_user + +User = get_user_model() +log = logging.getLogger(__name__) def enrollment_summary(enrollment): @@ -58,6 +66,130 @@ def create_or_update_enrollment(model_cls, defaults=None, **kwargs): return enrollment, created +def unenroll_learner_from_run( + user, course_run, *, keep_failed_enrollments=False, send_notification=True +): + """ + Unenroll a single learner from a course run in both edX and MITx Online. + + Args: + user (User): The user to unenroll + course_run (CourseRun): The course run to unenroll from + keep_failed_enrollments (bool): If True, keeps the local enrollment record + even if the edX unenrollment fails. + send_notification (bool): If True, sends unenrollment email to the learner. + + Returns: + tuple[CourseRunEnrollment | None, str]: (enrollment_result, message) + enrollment_result is the deactivated enrollment on success, None on failure. + message is a human-readable status string. + """ + enrollment = CourseRunEnrollment.objects.filter(user=user, run=course_run).first() + if enrollment is None or not enrollment.active: + return ( + None, + f"No active enrollment for {user.email} in {course_run.courseware_id}", + ) + + result = deactivate_run_enrollment( + enrollment, + change_status=ENROLL_CHANGE_STATUS_UNENROLLED, + keep_failed_enrollments=keep_failed_enrollments, + send_notification=send_notification, + ) + if result: + return ( + result, + f"Unenrolled: {user.email} from {course_run.courseware_id}", + ) + return ( + None, + f"Failed to unenroll {user.email} from {course_run.courseware_id}", + ) + + +def bulk_unenroll_learners( + entries, *, keep_failed_enrollments=False, send_notification=True +): + """ + Unenroll multiple learners from course runs in both edX and MITx Online. + + Resolves users and course runs from string identifiers, logs progress, + and returns a summary of results. + + Args: + entries (list[tuple[str, str]]): List of (user_identifier, courseware_id) tuples. + user_identifier can be email, username, or user id. + keep_failed_enrollments (bool): If True, keeps local enrollment records + even if the edX unenrollment fails. + send_notification (bool): If True, sends unenrollment email to each learner. + + Returns: + dict: Summary with keys 'succeeded', 'failed', 'skipped' (int counts) + and 'details' (list of (user_identifier, courseware_id, status, message) tuples). + """ + run_cache = {} + succeeded = 0 + failed = 0 + skipped = 0 + details = [] + + for user_identifier, cw_id in entries: + # Resolve user + try: + user = fetch_user(user_identifier) + except User.DoesNotExist: + msg = f"User not found: {user_identifier}" + log.warning(msg) + skipped += 1 + details.append((user_identifier, cw_id, "skipped", msg)) + continue + + # Resolve course run (with caching) + if cw_id not in run_cache: + run_cache[cw_id] = CourseRun.objects.filter(courseware_id=cw_id).first() + course_run = run_cache[cw_id] + if course_run is None: + msg = f"Course run not found: {cw_id}" + log.warning(msg) + skipped += 1 + details.append((user_identifier, cw_id, "skipped", msg)) + continue + + # Perform unenrollment + result, message = unenroll_learner_from_run( + user, + course_run, + keep_failed_enrollments=keep_failed_enrollments, + send_notification=send_notification, + ) + if result: + log.info(message) + succeeded += 1 + details.append((user_identifier, cw_id, "succeeded", message)) + elif "No active enrollment" in message: + log.warning(message) + skipped += 1 + details.append((user_identifier, cw_id, "skipped", message)) + else: + log.error(message) + failed += 1 + details.append((user_identifier, cw_id, "failed", message)) + + log.info( + "Bulk unenroll complete: %d succeeded, %d failed, %d skipped", + succeeded, + failed, + skipped, + ) + return { + "succeeded": succeeded, + "failed": failed, + "skipped": skipped, + "details": details, + } + + class EnrollmentChangeCommand(BaseCommand): """Base class for management commands that change enrollment status"""