fix: issues

Author: Anas12091101

courses/management/commands/test_unenroll_enrollment.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":

courses/management/commands/unenroll_learners.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",

courses/management/commands/unenroll_learners_test.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}"