From e11eccbbcfba1da38f04ded92c3a9be3b007cc54 Mon Sep 17 00:00:00 2001 From: Muhammad Anas Date: Fri, 22 May 2026 17:38:42 +0500 Subject: [PATCH 1/4] feat: add unenroll_learners management command for bulk unenrollment --- .../commands/unenroll_enrollment.py | 92 ----- .../management/commands/unenroll_learners.py | 218 +++++++++++ .../commands/unenroll_learners_test.py | 345 ++++++++++++++++++ courses/management/utils.py | 128 +++++++ 4 files changed, 691 insertions(+), 92 deletions(-) delete mode 100644 courses/management/commands/unenroll_enrollment.py create mode 100644 courses/management/commands/unenroll_learners.py create mode 100644 courses/management/commands/unenroll_learners_test.py 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..8551c178e0 --- /dev/null +++ b/courses/management/commands/unenroll_learners.py @@ -0,0 +1,218 @@ +""" +Management command to unenroll learners from course runs in both edX and MITx Online. + +**Usage:** + +1. Unenroll a single user from a course run: +./manage.py unenroll_learners --users=user@example.com --run=course-v1:MITx+6.00.1x+2024 + +2. Unenroll specific users (inline) from a course run: +./manage.py unenroll_learners --users=user1@example.com,user2@example.com --run=course-v1:MITx+6.00.1x+2024 + +3. Unenroll users listed in a CSV file (columns: user, courseware_id): +./manage.py unenroll_learners --csv=unenrollments.csv + +4. Dry run (preview only, no changes): +./manage.py unenroll_learners --csv=unenrollments.csv --dry-run + +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 -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): + group = parser.add_mutually_exclusive_group(required=True) + group.add_argument( + "--csv", + type=str, + help="Path to a CSV file with columns: user, courseware_id", + ) + group.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 (required with --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( + "--dry-run", + action="store_true", + dest="dry_run", + help="Preview which enrollments would be unenrolled without making changes", + ) + + 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( # noqa: TRY301 + "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 _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" + ) + ) + + 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") + dry_run = options.get("dry_run") + + if users_str and not courseware_id: + raise CommandError("--run is required when using --users") # noqa: EM101 + + if csv_path: + entries = self._parse_csv(csv_path) + else: + entries = self._parse_inline_users(users_str, courseware_id) + + if not entries: + raise CommandError("No valid entries found to process") # noqa: EM101 + + 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 + ) + + # 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..ed74b35b24 --- /dev/null +++ b/courses/management/commands/unenroll_learners_test.py @@ -0,0 +1,345 @@ +"""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 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", + stdout=out, + ) + + mock_bulk_unenroll.assert_called_once_with( + [("a@b.com", "run-1"), ("c@d.com", "run-1")], + keep_failed_enrollments=False, + ) + 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, + stdout=StringIO(), + ) + + mock_bulk_unenroll.assert_called_once_with( + [("a@b.com", "run-1")], + keep_failed_enrollments=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", + stdout=out, + stderr=err, + ) + + output = out.getvalue() + assert "1 succeeded" in output + assert "1 failed" in output + assert "1 skipped" in output + + +@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, stdout=out) + + mock_bulk_unenroll.assert_called_once_with( + [("a@b.com", "run-1"), ("c@d.com", "run-2")], + keep_failed_enrollments=False, + ) + 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, 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, + ) + + +@pytest.mark.django_db() +class TestBulkUnenrollDryRun: + """Tests for --dry-run flag""" + + def test_dry_run_no_changes(self, mock_bulk_unenroll): + """Dry run should not call bulk_unenroll_learners""" + enrollment = CourseRunEnrollmentFactory.create(active=True) + out = StringIO() + call_command( + "unenroll_learners", + users=enrollment.user.email, + run=enrollment.run.courseware_id, + dry_run=True, + stdout=out, + ) + + mock_bulk_unenroll.assert_not_called() + output = out.getvalue() + assert "DRY RUN" in output + assert "Would unenroll" 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, + dry_run=True, + 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, + dry_run=True, + stdout=out, + stderr=err, + ) + + mock_bulk_unenroll.assert_not_called() + assert "User not found" in err.getvalue() + + +@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""" + from courses.management.utils import bulk_unenroll_learners + + 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""" + from courses.management.utils import bulk_unenroll_learners + + 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""" + from courses.management.utils import bulk_unenroll_learners + + 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""" + from courses.management.utils import bulk_unenroll_learners + + 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""" + from courses.management.utils import bulk_unenroll_learners + + 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""" + from courses.management.utils import bulk_unenroll_learners + + 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 diff --git a/courses/management/utils.py b/courses/management/utils.py index 698cc3a3d4..6da7ec574d 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,126 @@ 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): + """ + 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. + + 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, + ) + 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): + """ + 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. + + 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, + ) + 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""" From 163556ae93afafc024337bfb87bdf897c9983499 Mon Sep 17 00:00:00 2001 From: Muhammad Anas Date: Fri, 22 May 2026 18:36:28 +0500 Subject: [PATCH 2/4] fix: issues --- .../commands/test_unenroll_enrollment.py | 149 ------------------ .../management/commands/unenroll_learners.py | 18 +-- .../commands/unenroll_learners_test.py | 46 +++--- courses/management/utils.py | 8 +- 4 files changed, 28 insertions(+), 193 deletions(-) delete mode 100644 courses/management/commands/test_unenroll_enrollment.py 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_learners.py b/courses/management/commands/unenroll_learners.py index 8551c178e0..78556fb831 100644 --- a/courses/management/commands/unenroll_learners.py +++ b/courses/management/commands/unenroll_learners.py @@ -84,7 +84,7 @@ def _parse_csv(self, csv_path): if not reader.fieldnames or not {"user", "courseware_id"}.issubset( set(reader.fieldnames) ): - raise CommandError( # noqa: TRY301 + raise CommandError( "CSV file must have 'user' and 'courseware_id' columns" # noqa: EM101 ) for row_num, row in enumerate(reader, start=2): @@ -113,11 +113,7 @@ def _parse_inline_users(self, users_str, courseware_id): 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() - ] + return [(u.strip(), courseware_id) for u in users_str.split(",") if u.strip()] def _dry_run(self, entries): """Preview which enrollments would be unenrolled without making changes.""" @@ -136,9 +132,7 @@ def _dry_run(self, entries): continue if cw_id not in run_cache: - run_cache[cw_id] = CourseRun.objects.filter( - courseware_id=cw_id - ).first() + 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( @@ -196,12 +190,10 @@ def handle(self, *args, **options): # noqa: ARG002 self._dry_run(entries) return - summary = bulk_unenroll_learners( - entries, keep_failed_enrollments=keep_failed - ) + summary = bulk_unenroll_learners(entries, keep_failed_enrollments=keep_failed) # Print details - for user_id, cw_id, status, message in summary["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": diff --git a/courses/management/commands/unenroll_learners_test.py b/courses/management/commands/unenroll_learners_test.py index ed74b35b24..02dd77442e 100644 --- a/courses/management/commands/unenroll_learners_test.py +++ b/courses/management/commands/unenroll_learners_test.py @@ -9,10 +9,11 @@ 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() +@pytest.fixture def mock_bulk_unenroll(mocker): """Mock bulk_unenroll_learners to avoid edX API calls""" return mocker.patch( @@ -87,7 +88,12 @@ def test_inline_users_mixed_results(self, mock_bulk_unenroll): "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"), + ( + "c@d.com", + "run-1", + "skipped", + "No active enrollment for c@d.com in run-1", + ), ], } @@ -135,10 +141,12 @@ def test_csv_success(self, mock_bulk_unenroll): ], } - csv_path = self._write_csv([ - {"user": "a@b.com", "courseware_id": "run-1"}, - {"user": "c@d.com", "courseware_id": "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, stdout=out) @@ -180,10 +188,12 @@ def test_csv_skips_empty_rows(self, mock_bulk_unenroll): ], } - csv_path = self._write_csv([ - {"user": "", "courseware_id": "course-v1:x+y+z"}, - {"user": "a@b.com", "courseware_id": "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, stdout=out) @@ -257,8 +267,6 @@ class TestBulkUnenrollLearnersUtil: def test_successful_unenrollment(self, mocker): """Should unenroll active enrollments successfully""" - from courses.management.utils import bulk_unenroll_learners - enrollment = CourseRunEnrollmentFactory.create(active=True) mocker.patch( "courses.management.utils.deactivate_run_enrollment", @@ -275,8 +283,6 @@ def test_successful_unenrollment(self, mocker): def test_user_not_found(self): """Should skip when user doesn't exist""" - from courses.management.utils import bulk_unenroll_learners - run = CourseRunFactory.create() result = bulk_unenroll_learners( [("nonexistent@example.com", run.courseware_id)] @@ -287,20 +293,14 @@ def test_user_not_found(self): def test_course_run_not_found(self): """Should skip when course run doesn't exist""" - from courses.management.utils import bulk_unenroll_learners - user = UserFactory.create() - result = bulk_unenroll_learners( - [(user.email, "course-v1:fake+fake+fake")] - ) + 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""" - from courses.management.utils import bulk_unenroll_learners - enrollment = CourseRunEnrollmentFactory.create(active=False) result = bulk_unenroll_learners( [(enrollment.user.email, enrollment.run.courseware_id)] @@ -311,8 +311,6 @@ def test_no_active_enrollment(self): def test_deactivation_failure(self, mocker): """Should count as failed when deactivate_run_enrollment returns None""" - from courses.management.utils import bulk_unenroll_learners - enrollment = CourseRunEnrollmentFactory.create(active=True) mocker.patch( "courses.management.utils.deactivate_run_enrollment", @@ -328,8 +326,6 @@ def test_deactivation_failure(self, mocker): def test_keep_failed_enrollments_passed(self, mocker): """Should pass keep_failed_enrollments to deactivate_run_enrollment""" - from courses.management.utils import bulk_unenroll_learners - enrollment = CourseRunEnrollmentFactory.create(active=True) mock_deactivate = mocker.patch( "courses.management.utils.deactivate_run_enrollment", diff --git a/courses/management/utils.py b/courses/management/utils.py index 6da7ec574d..07da86cc0b 100644 --- a/courses/management/utils.py +++ b/courses/management/utils.py @@ -81,9 +81,7 @@ def unenroll_learner_from_run(user, course_run, *, keep_failed_enrollments=False 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() + enrollment = CourseRunEnrollment.objects.filter(user=user, run=course_run).first() if enrollment is None or not enrollment.active: return ( None, @@ -142,9 +140,7 @@ def bulk_unenroll_learners(entries, *, keep_failed_enrollments=False): # Resolve course run (with caching) if cw_id not in run_cache: - run_cache[cw_id] = CourseRun.objects.filter( - courseware_id=cw_id - ).first() + 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}" From 1c63130445d65ae74a3f26b7859da178585566b5 Mon Sep 17 00:00:00 2001 From: Muhammad Anas Date: Fri, 22 May 2026 19:56:31 +0500 Subject: [PATCH 3/4] refactor: made the command dry-run bydefault and added no-email flag --- courses/api.py | 6 +- .../management/commands/unenroll_learners.py | 107 ++++++++++--- .../commands/unenroll_learners_test.py | 148 +++++++++++++++++- courses/management/utils.py | 12 +- 4 files changed, 238 insertions(+), 35 deletions(-) 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/unenroll_learners.py b/courses/management/commands/unenroll_learners.py index 78556fb831..090d3e186a 100644 --- a/courses/management/commands/unenroll_learners.py +++ b/courses/management/commands/unenroll_learners.py @@ -1,22 +1,24 @@ """ 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 a single user from a course run: -./manage.py unenroll_learners --users=user@example.com --run=course-v1:MITx+6.00.1x+2024 +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 (inline) from a course run: -./manage.py unenroll_learners --users=user1@example.com,user2@example.com --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 +./manage.py unenroll_learners --csv=unenrollments.csv --commit -4. Dry run (preview only, no changes): -./manage.py unenroll_learners --csv=unenrollments.csv --dry-run +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 -k +./manage.py unenroll_learners --users=user1@example.com --run=course-v1:MITx+6.00.1x+2024 --commit -k """ import csv @@ -37,13 +39,12 @@ class Command(BaseCommand): help = "Unenroll learners from course runs in both edX and MITx Online" def add_arguments(self, parser): - group = parser.add_mutually_exclusive_group(required=True) - group.add_argument( + parser.add_argument( "--csv", type=str, help="Path to a CSV file with columns: user, courseware_id", ) - group.add_argument( + parser.add_argument( "--users", type=str, help="Comma-separated list of user emails or usernames", @@ -51,7 +52,9 @@ def add_arguments(self, parser): parser.add_argument( "--run", type=str, - help="The 'courseware_id' value for a CourseRun (required with --users)", + 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", @@ -61,10 +64,17 @@ def add_arguments(self, parser): help="Keep local enrollment records even if edX unenrollment fails", ) parser.add_argument( - "--dry-run", + "--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="dry_run", - help="Preview which enrollments would be unenrolled without making changes", + dest="no_email", + help="Suppress unenrollment notification emails to learners", ) def _parse_csv(self, csv_path): @@ -115,6 +125,26 @@ def _parse_inline_users(self, users_str, courseware_id): """ 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 @@ -161,26 +191,51 @@ def _dry_run(self, entries): 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 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") - dry_run = options.get("dry_run") + 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 + "to unenroll all active learners from a course run." + ) + if csv_path: entries = self._parse_csv(csv_path) - else: + 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 "") @@ -190,7 +245,11 @@ def handle(self, *args, **options): # noqa: ARG002 self._dry_run(entries) return - summary = bulk_unenroll_learners(entries, keep_failed_enrollments=keep_failed) + 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"]: diff --git a/courses/management/commands/unenroll_learners_test.py b/courses/management/commands/unenroll_learners_test.py index 02dd77442e..8085b82730 100644 --- a/courses/management/commands/unenroll_learners_test.py +++ b/courses/management/commands/unenroll_learners_test.py @@ -48,12 +48,14 @@ def test_inline_users_success(self, mock_bulk_unenroll): "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 @@ -71,12 +73,14 @@ def test_inline_users_keep_failed_enrollments(self, mock_bulk_unenroll): 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): @@ -103,6 +107,7 @@ def test_inline_users_mixed_results(self, mock_bulk_unenroll): "unenroll_learners", users="a@b.com,b@c.com,c@d.com", run="run-1", + commit=True, stdout=out, stderr=err, ) @@ -112,6 +117,23 @@ def test_inline_users_mixed_results(self, mock_bulk_unenroll): 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: @@ -149,11 +171,12 @@ def test_csv_success(self, mock_bulk_unenroll): ) out = StringIO() - call_command("unenroll_learners", csv=csv_path, stdout=out) + 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() @@ -196,28 +219,28 @@ def test_csv_skips_empty_rows(self, mock_bulk_unenroll): ) out = StringIO() - call_command("unenroll_learners", csv=csv_path, stdout=out) + 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 --dry-run flag""" + """Tests for default dry-run behavior (no --commit flag)""" - def test_dry_run_no_changes(self, mock_bulk_unenroll): - """Dry run should not call bulk_unenroll_learners""" + 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, - dry_run=True, stdout=out, ) @@ -225,6 +248,7 @@ def test_dry_run_no_changes(self, mock_bulk_unenroll): 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""" @@ -235,7 +259,6 @@ def test_dry_run_skips_inactive(self, mock_bulk_unenroll): "unenroll_learners", users=enrollment.user.email, run=enrollment.run.courseware_id, - dry_run=True, stdout=out, stderr=err, ) @@ -252,7 +275,6 @@ def test_dry_run_user_not_found(self, mock_bulk_unenroll): "unenroll_learners", users="nonexistent@example.com", run=run.courseware_id, - dry_run=True, stdout=out, stderr=err, ) @@ -261,6 +283,100 @@ def test_dry_run_user_not_found(self, mock_bulk_unenroll): 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""" @@ -339,3 +455,19 @@ def test_keep_failed_enrollments_passed(self, mocker): 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 07da86cc0b..69159ead81 100644 --- a/courses/management/utils.py +++ b/courses/management/utils.py @@ -66,7 +66,9 @@ 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): +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. @@ -75,6 +77,7 @@ def unenroll_learner_from_run(user, course_run, *, keep_failed_enrollments=False 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) @@ -92,6 +95,7 @@ def unenroll_learner_from_run(user, course_run, *, keep_failed_enrollments=False enrollment, change_status=ENROLL_CHANGE_STATUS_UNENROLLED, keep_failed_enrollments=keep_failed_enrollments, + send_notification=send_notification, ) if result: return ( @@ -104,7 +108,9 @@ def unenroll_learner_from_run(user, course_run, *, keep_failed_enrollments=False ) -def bulk_unenroll_learners(entries, *, keep_failed_enrollments=False): +def bulk_unenroll_learners( + entries, *, keep_failed_enrollments=False, send_notification=True +): """ Unenroll multiple learners from course runs in both edX and MITx Online. @@ -116,6 +122,7 @@ def bulk_unenroll_learners(entries, *, keep_failed_enrollments=False): 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) @@ -154,6 +161,7 @@ def bulk_unenroll_learners(entries, *, keep_failed_enrollments=False): user, course_run, keep_failed_enrollments=keep_failed_enrollments, + send_notification=send_notification, ) if result: log.info(message) From d2b1c4e3b23d077fedd0a635ddc5577d6ed397e1 Mon Sep 17 00:00:00 2001 From: Muhammad Anas Date: Fri, 22 May 2026 20:11:40 +0500 Subject: [PATCH 4/4] fix: issue --- courses/management/commands/unenroll_learners.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/courses/management/commands/unenroll_learners.py b/courses/management/commands/unenroll_learners.py index 090d3e186a..4cb98cc85a 100644 --- a/courses/management/commands/unenroll_learners.py +++ b/courses/management/commands/unenroll_learners.py @@ -208,8 +208,7 @@ def _resolve_entries(self, csv_path, users_str, courseware_id): if not csv_path and not users_str and not courseware_id: raise CommandError( - "Provide --csv, --users with --run, or --run alone " # noqa: EM101 - "to unenroll all active learners from a course run." + "Provide --csv, --users with --run, or --run alone." # noqa: EM101 ) if csv_path: