diff --git a/common/djangoapps/student/api.py b/common/djangoapps/student/api.py index c5bd7b861871..da31d84ea9a1 100644 --- a/common/djangoapps/student/api.py +++ b/common/djangoapps/student/api.py @@ -5,10 +5,13 @@ from typing import TYPE_CHECKING +import csv +import io import logging from django.contrib.auth import get_user_model from django.conf import settings +from opaque_keys import InvalidKeyError from opaque_keys.edx.keys import CourseKey from common.djangoapps.student.models import CourseEnrollment @@ -159,3 +162,136 @@ def get_course_enrollments( course_enrollments = course_enrollments.filter(course_id__in=course_ids) return course_enrollments + + +# Header cells (case-insensitive, whitespace-trimmed) that mark an optional first row as a header. +BULK_UNENROLL_CSV_HEADERS = frozenset({"course_id", "course id"}) + + +class BulkUnenrollCsvTooManyRows(Exception): + """ + Raised as soon as a bulk-unenroll CSV passes the caller's row limit, so an + oversized file is never parsed in full. Carries the limit so it can be reported. + """ + + def __init__(self, max_rows): + self.max_rows = max_rows + super().__init__(f"CSV exceeds the maximum of {max_rows} rows.") + + +class BulkUnenrollCsvUnreadable(Exception): + """ + Raised when the upload cannot be read as CSV text at all. + + Distinct from the per-row errors this parser collects: a file that is not UTF-8 + (a ``.xls`` renamed to ``.csv``) or is malformed CSV has no rows to report + against. Callers turn this into a 400 rather than a 500. + """ + + +def _iter_csv_rows(text): + """ + Yield CSV rows, reporting a malformed file as ``BulkUnenrollCsvUnreadable``. + + ``csv.reader`` raises ``csv.Error`` lazily, from the middle of iteration, so the + guard has to wrap the stepping rather than the reader's construction. + """ + reader = csv.reader(io.StringIO(text)) + while True: + try: + yield next(reader) + except StopIteration: + return + except csv.Error as exc: + raise BulkUnenrollCsvUnreadable(f"File is not readable as CSV: {exc}") from exc + + +def parse_bulk_unenroll_csv(file_obj, max_rows=None): + """ + Parse a bulk-unenroll CSV into course keys. + + UTF-8 (BOM optional), exactly **one** non-empty cell per row: a course id. An + optional ``course_id`` / ``course id`` header and blank rows are skipped, and + duplicates are collapsed to the first occurrence without being reported. A row + with more than one cell is an error — deliberately narrower than the + ``bulk_unenroll`` management command's ``username,course_id``, whose username + column the whole-course worker would silently ignore while unenrolling everyone. + + Arguments: + file_obj: a file-like object opened in binary or text mode. + max_rows: optional cap on *data* rows (every non-blank, non-header row, + including duplicates and invalid ones). ``None`` means no limit. + + Raises: + BulkUnenrollCsvTooManyRows: if ``max_rows`` is exceeded. + BulkUnenrollCsvUnreadable: if the bytes are not UTF-8 or not parseable CSV. + + Returns: + (course_keys, errors) — course keys de-duplicated in input order, and + ``{"row": int, "value": str, "error": str}`` dicts whose 1-based row number + counts every physical row, so it matches what a spreadsheet shows. + """ + raw = file_obj.read() + if isinstance(raw, bytes): + try: + text = raw.decode("utf-8-sig") + except UnicodeDecodeError as exc: + raise BulkUnenrollCsvUnreadable( + "File is not valid UTF-8 text. Re-save it as a UTF-8 CSV and try again." + ) from exc + else: + # Strip a leading BOM if the caller handed us already-decoded text. + # Spelled as an escape: the literal character is invisible in a diff. + text = raw.removeprefix("\ufeff") + + course_keys = [] + errors = [] + seen = set() + data_rows = 0 + + for row_number, row in enumerate(_iter_csv_rows(text), start=1): + cells = [cell.strip() for cell in row] + non_empty = [cell for cell in cells if cell] + + if not non_empty: + # Blank row (possibly all-whitespace or trailing newline) — skip. + continue + + # Skip a single-cell header row if it names the column. + if row_number == 1 and len(non_empty) == 1 and non_empty[0].lower() in BULK_UNENROLL_CSV_HEADERS: + continue + + # Count every data row, not just the ones that survive to `course_keys`: + # duplicates and invalid rows still cost time and response size. + data_rows += 1 + if max_rows is not None and data_rows > max_rows: + raise BulkUnenrollCsvTooManyRows(max_rows) + + raw_value = ",".join(cells).strip(",") + + if len(non_empty) > 1: + errors.append({ + "row": row_number, + "value": raw_value, + "error": "Expected a single course_id column", + }) + continue + + value = non_empty[0] + try: + course_key = CourseKey.from_string(value) + except InvalidKeyError: + errors.append({ + "row": row_number, + "value": value, + "error": "Invalid course id", + }) + continue + + if course_key in seen: + # Duplicate — collapse silently so a course is never double-queued. + continue + seen.add(course_key) + course_keys.append(course_key) + + return course_keys, errors diff --git a/common/djangoapps/student/tests/test_api.py b/common/djangoapps/student/tests/test_api.py index 4757790e727f..e00a999097a8 100644 --- a/common/djangoapps/student/tests/test_api.py +++ b/common/djangoapps/student/tests/test_api.py @@ -2,13 +2,23 @@ Test Student api.py """ +import csv +import io + +import ddt + +from django.test import SimpleTestCase +from opaque_keys.edx.keys import CourseKey from xmodule.modulestore.tests.django_utils import SharedModuleStoreTestCase from xmodule.modulestore.tests.factories import CourseFactory from common.djangoapps.student.api import ( + BulkUnenrollCsvTooManyRows, + BulkUnenrollCsvUnreadable, is_user_enrolled_in_course, is_user_staff_or_instructor_in_course, get_course_enrollments, + parse_bulk_unenroll_csv, ) from common.djangoapps.student.models import CourseEnrollment from common.djangoapps.student.tests.factories import ( @@ -100,3 +110,121 @@ def test_get_filtered_course_enrollments(self): result = get_course_enrollments(self.user, True, course_ids=[course_2.id]) self.assertEqual(list(expected), list(result)) + + +@ddt.ddt +class TestParseBulkUnenrollCsv(SimpleTestCase): + """ + Tests for parse_bulk_unenroll_csv (pure parser, no DB access). + """ + + DEMO_2024 = CourseKey.from_string("course-v1:edX+DemoX+2024") + DEMO_2025 = CourseKey.from_string("course-v1:edX+DemoX+2025") + + @staticmethod + def _file(content): + """Build a binary file-like object (mimicking a Django UploadedFile).""" + if isinstance(content, str): + content = content.encode("utf-8") + return io.BytesIO(content) + + @ddt.data( + # (content, expected keys) — one course id per row, header optional. + ("course-v1:edX+DemoX+2024\ncourse-v1:edX+DemoX+2025\n", 2), + ("course_id\ncourse-v1:edX+DemoX+2024\n", 1), # header skipped + ("Course ID\ncourse-v1:edX+DemoX+2024\n", 1), # header variant + ("\ufeffcourse_id\ncourse-v1:edX+DemoX+2024\n", 1), # Excel's BOM + ("course-v1:edX+DemoX+2024\ncourse-v1:edX+DemoX+2024\n", 1), # duplicate collapsed + ("course-v1:edX+DemoX+2024\n\n \ncourse-v1:edX+DemoX+2025\n", 2), # blank rows + ("course-v1:edX+DemoX+2024,\n", 1), # stray trailing comma + ("", 0), # empty file + ) + @ddt.unpack + def test_valid_files_parse_without_errors(self, content, expected_count): + keys, errors = parse_bulk_unenroll_csv(self._file(content)) + assert not errors + assert len(keys) == expected_count + assert keys == list(dict.fromkeys(keys)) # de-duplicated, input order kept + + def test_already_decoded_text_with_a_bom(self): + """Callers may hand us text rather than bytes; the BOM still has to go.""" + keys, errors = parse_bulk_unenroll_csv(io.StringIO("\ufeffcourse-v1:edX+DemoX+2024\n")) + assert not errors + assert keys == [self.DEMO_2024] + + def test_legacy_username_course_id_rejected(self): + """ + The management command accepts ``username,course_id``; this parser must not. + The whole-course worker would ignore the username and unenroll everyone. + """ + keys, errors = parse_bulk_unenroll_csv(self._file( + "course-v1:edX+DemoX+2024\nalice,course-v1:edX+DemoX+2025\n" + )) + assert keys == [self.DEMO_2024] + assert errors == [{ + "row": 2, + "value": "alice,course-v1:edX+DemoX+2025", + "error": "Expected a single course_id column", + }] + + def test_invalid_course_id_reports_the_spreadsheet_row_number(self): + """Row numbers count every physical row, header included, so they line up.""" + keys, errors = parse_bulk_unenroll_csv(self._file( + "course_id\ncourse-v1:edX+DemoX+2024\nbad-key\n" + )) + assert keys == [self.DEMO_2024] + assert errors == [{"row": 3, "value": "bad-key", "error": "Invalid course id"}] + + def test_a_file_of_nothing_but_bad_ids_yields_errors_and_no_keys(self): + keys, errors = parse_bulk_unenroll_csv(self._file("foo\nbar\n")) + assert not keys + assert [e["row"] for e in errors] == [1, 2] + assert all(e["error"] == "Invalid course id" for e in errors) + + # --- the row limit guards the *file*, not the de-duplicated course list --- + + @ddt.data( + "course-v1:edX+DemoX+2024\ncourse-v1:edX+DemoX+2025\ncourse-v1:edX+DemoX+2026\n", + "course-v1:edX+DemoX+2024\n" * 3, # duplicates still cost time and size + "bad-1\nbad-2\nbad-3\n", # so do invalid rows + ) + def test_max_rows_counts_every_data_row(self, content): + with self.assertRaises(BulkUnenrollCsvTooManyRows): + parse_bulk_unenroll_csv(self._file(content), max_rows=2) + + def test_max_rows_ignores_header_and_blank_rows(self): + keys, errors = parse_bulk_unenroll_csv( + self._file("course_id\ncourse-v1:edX+DemoX+2024\n\ncourse-v1:edX+DemoX+2025\n\n"), + max_rows=2, + ) + assert not errors + assert len(keys) == 2 + + def test_max_rows_none_is_unlimited(self): + keys, errors = parse_bulk_unenroll_csv(self._file("bad-1\nbad-2\nbad-3\n")) + assert not keys + assert len(errors) == 3 + + # Unreadable files have no row to attach an error to, so they are raised + # rather than collected; the upload view turns them into a 400, not a 500. + + @ddt.data( + b"course-v1:edX+D\xe9moX+2024\n", # Latin-1 export: never valid UTF-8 + b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1", # a real .xls saved as .csv + ) + def test_non_utf8_bytes_are_rejected_as_unreadable(self, content): + with self.assertRaises(BulkUnenrollCsvUnreadable): + parse_bulk_unenroll_csv(self._file(content)) + + def test_malformed_csv_is_rejected_as_unreadable(self): + """ + An unterminated quoted field swallows the rest of the file and trips csv's + field-size limit. csv.reader raises that mid-iteration, after the first row + was already handed back, so the guard has to wrap the stepping rather than + the reader's construction. + """ + runaway_quote = '"' + "a" * (csv.field_size_limit() + 10) + with self.assertRaises(BulkUnenrollCsvUnreadable): + parse_bulk_unenroll_csv(self._file( + "course-v1:edX+DemoX+2024\n" + runaway_quote + )) diff --git a/lms/djangoapps/support/migrations/0007_bulkunenrollbatch_bulkunenrollcoursestate_and_more.py b/lms/djangoapps/support/migrations/0007_bulkunenrollbatch_bulkunenrollcoursestate_and_more.py new file mode 100644 index 000000000000..5027dd8e8443 --- /dev/null +++ b/lms/djangoapps/support/migrations/0007_bulkunenrollbatch_bulkunenrollcoursestate_and_more.py @@ -0,0 +1,79 @@ +# Generated by Django 4.2.24 on 2026-08-10 06:04 + +from django.conf import settings +from django.db import migrations, models +import django.db.models.deletion +import django.utils.timezone +import model_utils.fields +import opaque_keys.edx.django.models +import uuid + + +class Migration(migrations.Migration): + + dependencies = [ + migrations.swappable_dependency(settings.AUTH_USER_MODEL), + ('support', '0006_alter_historicalusersocialauth_extra_data_and_more'), + ] + + operations = [ + migrations.CreateModel( + name='BulkUnenrollBatch', + fields=[ + ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('created', model_utils.fields.AutoCreatedField(default=django.utils.timezone.now, editable=False, verbose_name='created')), + ('modified', model_utils.fields.AutoLastModifiedField(default=django.utils.timezone.now, editable=False, verbose_name='modified')), + ('uuid', models.UUIDField(db_index=True, default=uuid.uuid4, editable=False, unique=True)), + ('csv_filename', models.CharField(blank=True, default='', max_length=255)), + ('reason', models.CharField(blank=True, default='', max_length=255)), + ('state', models.CharField(choices=[('validated', 'Validated'), ('pending', 'Pending'), ('cancelled', 'Cancelled'), ('running', 'Running'), ('succeeded', 'Succeeded'), ('partial', 'Partial'), ('failed', 'Failed')], default='validated', max_length=20)), + ('total_courses', models.PositiveIntegerField(default=0)), + ('requester', models.ForeignKey(on_delete=django.db.models.deletion.DO_NOTHING, to=settings.AUTH_USER_MODEL)), + ], + options={ + 'abstract': False, + }, + ), + migrations.CreateModel( + name='BulkUnenrollCourseState', + fields=[ + ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('created', model_utils.fields.AutoCreatedField(default=django.utils.timezone.now, editable=False, verbose_name='created')), + ('modified', model_utils.fields.AutoLastModifiedField(default=django.utils.timezone.now, editable=False, verbose_name='modified')), + ('course_id', opaque_keys.edx.django.models.CourseKeyField(db_index=True, max_length=255)), + ('active_count', models.PositiveIntegerField(default=0)), + ('state', models.CharField(choices=[('pending', 'Pending'), ('running', 'Running'), ('succeeded', 'Succeeded'), ('failed', 'Failed'), ('skipped', 'Skipped'), ('cancelled', 'Cancelled')], default='pending', max_length=20)), + ('error', models.CharField(blank=True, default='', max_length=255)), + ('total_enrollments', models.PositiveIntegerField(default=0)), + ('unenrolled', models.PositiveIntegerField(default=0)), + ('already_inactive', models.PositiveIntegerField(default=0)), + ('failed_count', models.PositiveIntegerField(default=0)), + ('chunks_total', models.PositiveIntegerField(default=0)), + ('chunks_finished', models.PositiveIntegerField(default=0)), + ('attempt', models.PositiveIntegerField(default=1)), + ('started', models.DateTimeField(blank=True, null=True)), + ('finished', models.DateTimeField(blank=True, null=True)), + ('batch', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='courses', to='support.bulkunenrollbatch')), + ], + options={ + 'unique_together': {('batch', 'course_id')}, + }, + ), + migrations.CreateModel( + name='BulkUnenrollChunk', + fields=[ + ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('created', model_utils.fields.AutoCreatedField(default=django.utils.timezone.now, editable=False, verbose_name='created')), + ('modified', model_utils.fields.AutoLastModifiedField(default=django.utils.timezone.now, editable=False, verbose_name='modified')), + ('attempt', models.PositiveIntegerField(default=1)), + ('chunk_index', models.PositiveIntegerField()), + ('continuation', models.PositiveIntegerField(default=0)), + ('state', models.CharField(choices=[('pending', 'Pending'), ('finished', 'Finished')], default='pending', max_length=20)), + ('finished', models.DateTimeField(blank=True, null=True)), + ('course_state', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='chunks', to='support.bulkunenrollcoursestate')), + ], + options={ + 'unique_together': {('course_state', 'attempt', 'chunk_index', 'continuation')}, + }, + ), + ] diff --git a/lms/djangoapps/support/models.py b/lms/djangoapps/support/models.py index d8f6e2b9a1d1..99a2dbac718e 100644 --- a/lms/djangoapps/support/models.py +++ b/lms/djangoapps/support/models.py @@ -1,9 +1,17 @@ """ Models used to implement support related models in such as SSO History model """ +import uuid + from django.contrib.auth import get_user_model from django.db.models import ForeignKey, DO_NOTHING, CASCADE, TextChoices -from django.db.models.fields import BooleanField, CharField, DateTimeField +from django.db.models.fields import ( + BooleanField, + CharField, + DateTimeField, + PositiveIntegerField, + UUIDField, +) from model_utils.models import TimeStampedModel from opaque_keys.edx.django.models import CourseKeyField @@ -82,3 +90,133 @@ def status_message(self): if self.status == self.CourseResetStatus.IN_PROGRESS: return f"In progress - Started on {self.modified} by {self.reset_by.username}" return self.status + + +class BulkUnenrollBatch(TimeStampedModel): + """ + A single bulk-unenroll upload: the batch layer over per-course work. + + Holds the operator-supplied metadata and the aggregate state the UI polls; + per-course rows hang off it via ``BulkUnenrollCourseState``. ``uuid`` is the + public identifier used in URLs, so the auto-increment pk is never exposed. + + .. no_pii: + """ + class State(TextChoices): + """Lifecycle states for a bulk-unenroll batch.""" + VALIDATED = "validated" # dry-run parsed & counted; awaiting confirm + PENDING = "pending" # confirmed; queued for the workers + CANCELLED = "cancelled" + RUNNING = "running" + SUCCEEDED = "succeeded" + PARTIAL = "partial" + FAILED = "failed" + + uuid = UUIDField(default=uuid.uuid4, unique=True, editable=False, db_index=True) + requester = ForeignKey(User, on_delete=DO_NOTHING) + csv_filename = CharField(max_length=255, default="", blank=True) + reason = CharField(max_length=255, default="", blank=True) + state = CharField(max_length=20, choices=State.choices, default=State.VALIDATED) + total_courses = PositiveIntegerField(default=0) + + def __str__(self): + return f"BulkUnenrollBatch {self.uuid} ({self.state}, {self.total_courses} courses)" + + +class BulkUnenrollCourseState(TimeStampedModel): + """ + Per-course row within a ``BulkUnenrollBatch`` — one row per *distinct* valid + course id in the upload (duplicate lines collapse), and the unit of progress, + retry, and resumability. + + ``active_count`` is the dry-run preview count set at upload; the remaining + counters and chunk-tracking fields are populated by the Celery layer. + + .. no_pii: + """ + class State(TextChoices): + """Lifecycle states for a single course within a batch.""" + PENDING = "pending" + RUNNING = "running" + SUCCEEDED = "succeeded" + FAILED = "failed" + SKIPPED = "skipped" + # Work was stopped (unlike SKIPPED: never eligible); some learners may + # already be unenrolled. Terminal — a late chunk must not flip it back. + CANCELLED = "cancelled" + + #: States a course can never leave — the single definition shared by the + #: engine's finalizers and the status endpoint's ``courses_finished`` count. + TERMINAL_STATES = frozenset({ + State.SUCCEEDED, State.FAILED, State.SKIPPED, State.CANCELLED, + }) + + batch = ForeignKey(BulkUnenrollBatch, on_delete=CASCADE, related_name="courses") + course_id = CourseKeyField(max_length=255, db_index=True) + active_count = PositiveIntegerField(default=0) + state = CharField(max_length=20, choices=State.choices, default=State.PENDING) + error = CharField(max_length=255, default="", blank=True) + + # --- Mutation progress (populated by the Celery workers) --- + # Active enrollments measured at fan-out (vs active_count, the dry-run preview). + total_enrollments = PositiveIntegerField(default=0) + unenrolled = PositiveIntegerField(default=0) + # Learners already inactive when a chunk reached them — not an error. + already_inactive = PositiveIntegerField(default=0) + failed_count = PositiveIntegerField(default=0) + # Finalization primitive: the course is done when chunks_finished == + # chunks_total (and > 0). Updated with F() so concurrent chunks don't clobber. + chunks_total = PositiveIntegerField(default=0) + chunks_finished = PositiveIntegerField(default=0) + # Generation counter, bumped by every retry: a straggler from a superseded + # attempt is discarded instead of claiming an identity the new attempt needs. + attempt = PositiveIntegerField(default=1) + started = DateTimeField(null=True, blank=True) + finished = DateTimeField(null=True, blank=True) + + class Meta: + unique_together = (("batch", "course_id"),) + + def __str__(self): + return f"{self.course_id} ({self.state}) in batch {self.batch.uuid}" + + +class BulkUnenrollChunk(TimeStampedModel): + """ + One chunk of learners within a ``BulkUnenrollCourseState`` — the completion ledger. + + Celery delivers at-least-once, so counting "chunks finished" per invocation is + unsafe: a redelivery could push ``chunks_finished`` to ``chunks_total`` while a + different chunk had never run, finalizing the course with learners still + enrolled. This table gives every chunk a durable identity and one atomic + completion claim — a conditional ``pending -> finished`` flip, where only the + winner records its counters. Re-doing the *work* stays harmless (every level + filters on ``is_active=True``); it is the *accounting* that must happen once. + + .. no_pii: + """ + class State(TextChoices): + """Lifecycle states for a single chunk.""" + PENDING = "pending" + FINISHED = "finished" + + course_state = ForeignKey(BulkUnenrollCourseState, on_delete=CASCADE, related_name="chunks") + #: Which generation of the course's fan-out this chunk belongs to. Rows from + #: earlier attempts are kept as history and never collide with the current one. + attempt = PositiveIntegerField(default=1) + #: 0-based position within that attempt's fan-out. + chunk_index = PositiveIntegerField() + #: Which hand-off of the chunk this row is (0 = as fanned out, +1 per timeout + #: tail). Numbered within its chunk so it never collides with a queued sibling. + continuation = PositiveIntegerField(default=0) + state = CharField(max_length=20, choices=State.choices, default=State.PENDING) + finished = DateTimeField(null=True, blank=True) + + class Meta: + unique_together = (("course_state", "attempt", "chunk_index", "continuation"),) + + def __str__(self): + return ( + f"chunk {self.chunk_index}.{self.continuation} ({self.state}) " + f"of {self.course_state_id} attempt {self.attempt}" + ) diff --git a/lms/djangoapps/support/rest_api/serializers.py b/lms/djangoapps/support/rest_api/serializers.py index 3ce5ac3989da..105104ac8bf0 100644 --- a/lms/djangoapps/support/rest_api/serializers.py +++ b/lms/djangoapps/support/rest_api/serializers.py @@ -8,6 +8,7 @@ from django.conf import settings from rest_framework import serializers +from lms.djangoapps.support.models import BulkUnenrollBatch, BulkUnenrollCourseState from openedx.core.djangoapps.content.course_overviews.models import CourseOverview @@ -56,3 +57,39 @@ def to_representation(self, instance): "run": course_key.run, "number": course_key.course, } + + +class BulkUnenrollCourseStateSerializer(serializers.ModelSerializer): + """ + Per-course row within a bulk-unenroll batch. + + Carries both the dry-run preview (``active_count``, set at upload) and the + worker-populated mutation counters, so the polling UI can show real progress + for a run that may last hours rather than just a state label. + """ + + course_id = serializers.CharField() + + class Meta: + model = BulkUnenrollCourseState + fields = ( + "course_id", "state", "active_count", "error", + "unenrolled", "already_inactive", "failed_count", + "chunks_total", "chunks_finished", + ) + + +class BulkUnenrollBatchSerializer(serializers.ModelSerializer): + """Aggregate view of a bulk-unenroll batch (the public ``batch_id`` is the uuid).""" + + batch_id = serializers.UUIDField(source="uuid", read_only=True) + # A uuid is not recognizable: who/which-file is how an operator spots their + # own run in the list (or a colleague's, for an in-flight batch). + requester = serializers.CharField(source="requester.username", read_only=True) + + class Meta: + model = BulkUnenrollBatch + fields = ( + "batch_id", "state", "reason", "total_courses", + "requester", "csv_filename", "created", "modified", + ) diff --git a/lms/djangoapps/support/rest_api/v1/tests/test_views.py b/lms/djangoapps/support/rest_api/v1/tests/test_views.py index 58de5c965ecc..79f26a30e2ca 100644 --- a/lms/djangoapps/support/rest_api/v1/tests/test_views.py +++ b/lms/djangoapps/support/rest_api/v1/tests/test_views.py @@ -2,13 +2,19 @@ Tests for support views. """ +import csv +import uuid as uuid_lib +from unittest.mock import patch + import ddt +from django.core.files.uploadedfile import SimpleUploadedFile +from django.test import override_settings from django.urls import reverse from rest_framework.test import APIClient from common.djangoapps.student.models import CourseEnrollment from common.djangoapps.student.models.user import CourseAccessRole -from common.djangoapps.student.roles import CourseInstructorRole, CourseStaffRole +from common.djangoapps.student.roles import CourseInstructorRole, CourseStaffRole, SupportStaffRole from common.djangoapps.student.tests.factories import ( TEST_PASSWORD, AdminFactory, @@ -17,6 +23,7 @@ SuperuserFactory, UserFactory ) +from lms.djangoapps.support.models import BulkUnenrollBatch, BulkUnenrollChunk, BulkUnenrollCourseState from lms.djangoapps.support.tests.test_views import SupportViewTestCase from openedx.core.djangoapps.content.course_overviews.tests.factories import CourseOverviewFactory @@ -382,3 +389,904 @@ def test_put_api_course_instructor_cannot_manage_other_courses(self): self.assertEqual(resp.data["email"], self.user.email) self.assertEqual(resp.data["results"][0]["status"], "failed") self.assertIn("do not have instructor access", resp.data["results"][0]["error"]) + + +class BulkUnenrollEndpointTestMixin: + """ + Shared fixtures and the global-staff gate every bulk-unenroll endpoint must have. + + Each endpoint declares its ``url_name`` and implements ``_request``. The gate is + asserted per endpoint because each view names its own ``permission_classes``, so + one that forgets ``IsAdminUser`` — or settles for the support-role gate — fails here. + """ + + #: URL name under ``lms.djangoapps.support.rest_api:v1``. + url_name = None + #: Patch the engine hand-off so the endpoint can be tested without running the + #: workers (tasks run eagerly under test). Sets ``self.mock_dispatch``. + patch_dispatch = False + + def setUp(self): + super().setUp() + self.admin = AdminFactory() # global staff (is_staff=True) + self.plain_user = UserFactory() + self.client = APIClient() + if self.patch_dispatch: + patcher = patch("lms.djangoapps.support.rest_api.v1.views._dispatch_bulk_unenroll") + self.mock_dispatch = patcher.start() + self.addCleanup(patcher.stop) + + def _login(self, user): + self.client.login(username=user.username, password=TEST_PASSWORD) + + def _url(self, batch_id=None): + name = f"lms.djangoapps.support.rest_api:v1:{self.url_name}" + if batch_id is None: + return reverse(name) + return reverse(name, kwargs={"batch_id": str(batch_id)}) + + def _request(self, batch_id=None): + """One call to the endpoint under test; ``None`` means the class's own fixture.""" + raise NotImplementedError + + def test_plain_user_forbidden(self): + self._login(self.plain_user) + self.assertEqual(self._request().status_code, 403) + + def test_support_staff_role_forbidden(self): + """A SupportStaffRole holder is not global staff — this gate is the narrower one.""" + support_user = UserFactory() + SupportStaffRole().add_users(support_user) + self._login(support_user) + self.assertEqual(self._request().status_code, 403) + + def test_anonymous_forbidden(self): + self.assertIn(self._request().status_code, (401, 403)) + + +class BatchScopedEndpointTestMixin(BulkUnenrollEndpointTestMixin): + """A bulk-unenroll endpoint addressed by ````.""" + + def test_unknown_batch_returns_404(self): + self._login(self.admin) + self.assertEqual(self._request(uuid_lib.uuid4()).status_code, 404) + + +@ddt.ddt +class BulkUnenrollAPIViewTest(BulkUnenrollEndpointTestMixin, SupportViewTestCase): + """ + Tests for BulkUnenrollAPIView (POST /api/support/v1/bulk_unenroll/). + """ + + url_name = "bulk_unenroll" + + def setUp(self): + super().setUp() + self.url = self._url() + + # Two real courses with active enrollments. + self.course_a = CourseOverviewFactory.create(org="edX", run="A", display_name="A") + self.course_b = CourseOverviewFactory.create(org="edX", run="B", display_name="B") + self._enroll(self.course_a.id, 3) + self._enroll(self.course_b.id, 2) + # An inactive enrollment must NOT be counted. + CourseEnrollment.objects.create( + user=UserFactory(), course_id=self.course_a.id, is_active=False, + ) + + @staticmethod + def _enroll(course_id, count): + for _ in range(count): + CourseEnrollment.objects.create(user=UserFactory(), course_id=course_id, is_active=True) + + @staticmethod + def _csv(text): + return SimpleUploadedFile("courses.csv", text.encode("utf-8"), content_type="text/csv") + + def _request(self, batch_id=None): + return self.client.post( + self.url, {"file": self._csv(f"{self.course_a.id}\n")}, format="multipart", + ) + + def test_upload_creates_validated_batch(self): + self._login(self.admin) + content = f"course_id\n{self.course_a.id}\n{self.course_b.id}\n" + resp = self.client.post(self.url, {"file": self._csv(content)}, format="multipart") + + self.assertEqual(resp.status_code, 200) + data = resp.json() + self.assertEqual(data["state"], "validated") + self.assertEqual(data["total_courses"], 2) + self.assertEqual(data["totals"]["active"], 5) # 3 + 2, inactive excluded + self.assertEqual(data["errors"], []) + counts = {c["course_id"]: c["active_count"] for c in data["courses"]} + self.assertEqual(counts[str(self.course_a.id)], 3) + self.assertEqual(counts[str(self.course_b.id)], 2) + + batch = BulkUnenrollBatch.objects.get(uuid=data["batch_id"]) + self.assertEqual(batch.requester, self.admin) + self.assertEqual(batch.courses.count(), 2) + + def test_upload_mutates_nothing(self): + self._login(self.admin) + active_before = CourseEnrollment.objects.filter(is_active=True).count() + self.client.post(self.url, {"file": self._csv(f"{self.course_a.id}\n")}, format="multipart") + self.assertEqual(CourseEnrollment.objects.filter(is_active=True).count(), active_before) + + def test_errors_reported_without_aborting(self): + self._login(self.admin) + content = f"{self.course_a.id}\nnot-a-key\nalice,{self.course_b.id}\n" + resp = self.client.post(self.url, {"file": self._csv(content)}, format="multipart") + + self.assertEqual(resp.status_code, 200) + data = resp.json() + self.assertEqual(data["total_courses"], 1) # only course_a is valid + rows = {e["row"]: e["error"] for e in data["errors"]} + self.assertEqual(rows[2], "Invalid course id") + self.assertEqual(rows[3], "Expected a single course_id column") + + def test_all_invalid_returns_400_no_batch(self): + self._login(self.admin) + resp = self.client.post(self.url, {"file": self._csv("foo\nbar\n")}, format="multipart") + self.assertEqual(resp.status_code, 400) + self.assertEqual(BulkUnenrollBatch.objects.count(), 0) + + def test_missing_file_returns_400(self): + self._login(self.admin) + resp = self.client.post(self.url, {}, format="multipart") + self.assertEqual(resp.status_code, 400) + + @ddt.data("latin1", "runaway_quote") + def test_unreadable_upload_returns_400_not_500(self, kind): + """ + Bad input, not a server fault: the operator is told to re-save the file. + """ + self._login(self.admin) + if kind == "latin1": + upload = SimpleUploadedFile( + "courses.csv", b"course-v1:edX+D\xe9moX+2024\n", content_type="text/csv", + ) + else: + upload = self._csv('"' + "a" * (csv.field_size_limit() + 10)) + + resp = self.client.post(self.url, {"file": upload}, format="multipart") + self.assertEqual(resp.status_code, 400) + self.assertEqual(BulkUnenrollBatch.objects.count(), 0) + + @override_settings(BULK_UNENROLL_MAX_FILE_BYTES=10) + def test_file_over_size_limit_returns_413(self): + self._login(self.admin) + content = f"course_id\n{self.course_a.id}\n" # > 10 bytes + resp = self.client.post(self.url, {"file": self._csv(content)}, format="multipart") + self.assertEqual(resp.status_code, 413) + self.assertEqual(BulkUnenrollBatch.objects.count(), 0) + + @ddt.data("valid", "duplicate", "invalid") + @override_settings(BULK_UNENROLL_MAX_ROWS=2) + def test_row_limit_counts_every_data_row(self, kind): + """The limit guards the uploaded file, not the de-duplicated course list.""" + content = { + "valid": f"{self.course_a.id}\n{self.course_b.id}\ncourse-v1:edX+Third+2099\n", + "duplicate": f"{self.course_a.id}\n" * 3, # collapses to 1 key, 3 rows + "invalid": f"{self.course_a.id}\nnot-a-key\nalso-not-a-key\n", + }[kind] + self._login(self.admin) + resp = self.client.post(self.url, {"file": self._csv(content)}, format="multipart") + self.assertEqual(resp.status_code, 400) + self.assertEqual(BulkUnenrollBatch.objects.count(), 0) + + def test_course_not_found_flagged_not_dropped(self): + self._login(self.admin) + missing = "course-v1:edX+Ghost+2099" + content = f"{self.course_a.id}\n{missing}\n" + resp = self.client.post(self.url, {"file": self._csv(content)}, format="multipart") + + self.assertEqual(resp.status_code, 200) + data = resp.json() + self.assertEqual(data["total_courses"], 2) # not dropped + by_id = {c["course_id"]: c for c in data["courses"]} + self.assertEqual(by_id[missing]["error"], "Course not found") + self.assertEqual(by_id[missing]["active_count"], 0) + self.assertEqual(by_id[str(self.course_a.id)]["error"], "") + + def test_query_count_independent_of_enrollment_count(self): + from django.db import connection + from django.test.utils import CaptureQueriesContext + + self._login(self.admin) + few = CourseOverviewFactory.create(org="edX", run="Few", display_name="Few") + many = CourseOverviewFactory.create(org="edX", run="Many", display_name="Many") + self._enroll(few.id, 2) + self._enroll(many.id, 40) + + def upload(course_id): + with CaptureQueriesContext(connection) as ctx: + resp = self.client.post( + self.url, {"file": self._csv(f"{course_id}\n")}, format="multipart" + ) + self.assertEqual(resp.status_code, 200) + return len(ctx.captured_queries) + + # Warm process-global caches so first-request overhead doesn't skew this. + upload(self.course_a.id) + + # Same query count at 2 and 40 enrollments: no per-enrollment fan-out. + self.assertEqual(upload(few.id), upload(many.id)) + + +class BulkUnenrollListAPIViewTest(BulkUnenrollEndpointTestMixin, SupportViewTestCase): + """ + Tests for BulkUnenrollAPIView's list route (GET /api/support/v1/bulk_unenroll/). + """ + + url_name = "bulk_unenroll" + + def setUp(self): + super().setUp() + self.url = self._url() + + def _request(self, batch_id=None): + return self.client.get(self.url) + + def _batch(self, state, requester=None, reason="cleanup", filename="courses.csv"): + return BulkUnenrollBatch.objects.create( + requester=requester or self.admin, + state=state, + reason=reason, + csv_filename=filename, + total_courses=2, + ) + + def test_lists_batches_newest_first(self): + older = self._batch(BulkUnenrollBatch.State.SUCCEEDED) + newer = self._batch(BulkUnenrollBatch.State.RUNNING) + self._login(self.admin) + + resp = self.client.get(self.url) + + self.assertEqual(resp.status_code, 200) + ids = [row["batch_id"] for row in resp.json()["results"]] + self.assertEqual(ids, [str(newer.uuid), str(older.uuid)]) + + def test_row_carries_the_fields_needed_to_identify_a_run(self): + """A uuid is unrecognizable; the operator identifies a batch by who/why/when.""" + self._batch(BulkUnenrollBatch.State.RUNNING, reason="LP-860 cleanup", filename="q3.csv") + self._login(self.admin) + + row = self.client.get(self.url).json()["results"][0] + + self.assertEqual(row["reason"], "LP-860 cleanup") + self.assertEqual(row["csv_filename"], "q3.csv") + self.assertEqual(row["requester"], self.admin.username) + self.assertEqual(row["state"], "running") + self.assertEqual(row["total_courses"], 2) + self.assertIn("created", row) + self.assertIn("modified", row) + + def test_state_filter_accepts_several_states(self): + pending = self._batch(BulkUnenrollBatch.State.PENDING) + running = self._batch(BulkUnenrollBatch.State.RUNNING) + self._batch(BulkUnenrollBatch.State.SUCCEEDED) + self._batch(BulkUnenrollBatch.State.VALIDATED) + self._login(self.admin) + + resp = self.client.get(self.url, {"state": "pending,running"}) + + self.assertEqual(resp.status_code, 200) + self.assertEqual( + {row["batch_id"] for row in resp.json()["results"]}, + {str(pending.uuid), str(running.uuid)}, + ) + + def test_unknown_state_returns_400(self): + self._login(self.admin) + resp = self.client.get(self.url, {"state": "running,bogus"}) + self.assertEqual(resp.status_code, 400) + + def test_lists_batches_from_every_requester(self): + """Recovering a lost batch id is the point; a colleague's run is the same problem.""" + other_admin = AdminFactory() + mine = self._batch(BulkUnenrollBatch.State.RUNNING) + theirs = self._batch(BulkUnenrollBatch.State.RUNNING, requester=other_admin) + self._login(self.admin) + + resp = self.client.get(self.url) + + self.assertEqual( + {row["batch_id"] for row in resp.json()["results"]}, + {str(mine.uuid), str(theirs.uuid)}, + ) + + def test_query_count_is_flat_in_batch_count(self): + """select_related('requester') — the username must not cost a query per row.""" + from django.db import connection + from django.test.utils import CaptureQueriesContext + + self._login(self.admin) + self._batch(BulkUnenrollBatch.State.RUNNING) + self.client.get(self.url) # warm caches + + with CaptureQueriesContext(connection) as one_batch: + self.client.get(self.url) + + for _ in range(4): + self._batch(BulkUnenrollBatch.State.RUNNING, requester=AdminFactory()) + + with CaptureQueriesContext(connection) as five_batches: + self.client.get(self.url) + + self.assertEqual(len(one_batch.captured_queries), len(five_batches.captured_queries)) + + +@ddt.ddt +class BulkUnenrollConfirmAPIViewTest(BatchScopedEndpointTestMixin, SupportViewTestCase): + """ + Tests for BulkUnenrollConfirmAPIView + (POST /api/support/v1/bulk_unenroll//confirm/). + """ + + url_name = "bulk_unenroll_confirm" + patch_dispatch = True + + def setUp(self): + super().setUp() + self.course = CourseOverviewFactory.create(org="edX", run="A", display_name="A") + + def _request(self, batch_id=None): + return self._confirm(self._make_batch().uuid if batch_id is None else batch_id) + + def _make_batch(self, state=BulkUnenrollBatch.State.VALIDATED): + """Create a persisted batch (as the upload endpoint would) in the given state.""" + batch = BulkUnenrollBatch.objects.create( + requester=self.admin, + csv_filename="courses.csv", + total_courses=1, + state=state, + ) + BulkUnenrollCourseState.objects.create(batch=batch, course_id=self.course.id, active_count=3) + return batch + + def _confirm(self, batch_id, reason="cleanup"): + data = {} if reason is None else {"reason": reason} + return self.client.post(self._url(batch_id), data, format="json") + + # --- happy path --- + + def test_confirm_flips_validated_to_pending(self): + self._login(self.admin) + batch = self._make_batch() + resp = self._confirm(batch.uuid, reason="Partner offboarding LP-860") + + self.assertEqual(resp.status_code, 202) + data = resp.json() + self.assertEqual(data["batch_id"], str(batch.uuid)) + self.assertEqual(data["state"], "pending") + + batch.refresh_from_db() + self.assertEqual(batch.state, "pending") + self.assertEqual(batch.reason, "Partner offboarding LP-860") + + def test_confirm_hands_off_to_the_engine_without_mutating_in_request(self): + """Confirm dispatches the async engine; the request thread unenrolls nobody.""" + self._login(self.admin) + CourseEnrollment.objects.create(user=UserFactory(), course_id=self.course.id, is_active=True) + active_before = CourseEnrollment.objects.filter(is_active=True).count() + self._confirm(self._make_batch().uuid) + self.mock_dispatch.assert_called_once() + self.assertEqual(CourseEnrollment.objects.filter(is_active=True).count(), active_before) + + @ddt.data(None, " ") + def test_missing_or_blank_reason_returns_400(self, reason): + self._login(self.admin) + batch = self._make_batch() + resp = self._confirm(batch.uuid, reason=reason) + self.assertEqual(resp.status_code, 400) + batch.refresh_from_db() + self.assertEqual(batch.state, "validated") # unchanged + + @ddt.data( + BulkUnenrollBatch.State.PENDING, + BulkUnenrollBatch.State.RUNNING, + BulkUnenrollBatch.State.CANCELLED, + BulkUnenrollBatch.State.SUCCEEDED, + ) + def test_non_validated_batch_returns_409(self, state): + """Confirm is valid exactly once: every other state is a conflict, not a re-run.""" + self._login(self.admin) + batch = self._make_batch(state=state) + resp = self._confirm(batch.uuid) + self.assertEqual(resp.status_code, 409) + batch.refresh_from_db() + self.assertEqual(batch.state, state) # not re-dispatched + self.mock_dispatch.assert_not_called() + + def test_failed_dispatch_restores_the_batch_and_reports_503(self): + """ + The flip commits before the publish, so a batch left 'pending' after a failed + publish is stranded: no worker holds it and confirm would refuse to re-run. + """ + self._login(self.admin) + batch = self._make_batch() + self.mock_dispatch.side_effect = OSError("broker unreachable") + + resp = self._confirm(batch.uuid, reason="Partner offboarding") + + self.assertEqual(resp.status_code, 503) + batch.refresh_from_db() + self.assertEqual(batch.state, "validated") # confirmable again + self.assertEqual(batch.reason, "") + + def test_concurrent_confirm_does_not_double_dispatch(self): + """ + The state check has to happen on a locked, freshly-read row, or two + dispatchers run over the same batch. + """ + self._login(self.admin) + batch = self._make_batch() + stale = BulkUnenrollBatch.objects.get(pk=batch.pk) # says 'validated' + # The competing confirm commits first. + BulkUnenrollBatch.objects.filter(pk=batch.pk).update(state=BulkUnenrollBatch.State.PENDING) + + with patch.object(BulkUnenrollBatch.objects, "get", return_value=stale): + resp = self._confirm(batch.uuid) + + self.assertEqual(resp.status_code, 409) + self.mock_dispatch.assert_not_called() + + +class BulkUnenrollStatusAPIViewTest(BatchScopedEndpointTestMixin, SupportViewTestCase): + """ + Tests for BulkUnenrollStatusAPIView + (GET /api/support/v1/bulk_unenroll//). + """ + + url_name = "bulk_unenroll_status" + + def setUp(self): + super().setUp() + self.batch = BulkUnenrollBatch.objects.create( + requester=self.admin, + csv_filename="courses.csv", + reason="cleanup", + total_courses=3, + state=BulkUnenrollBatch.State.PENDING, + ) + # Three courses with different states and active counts. + self.rows = [ + BulkUnenrollCourseState.objects.create( + batch=self.batch, + course_id=CourseOverviewFactory.create(org="edX", run=f"R{i}", display_name=f"C{i}").id, + active_count=count, + state=state, + ) + for i, (count, state) in enumerate([ + (10, BulkUnenrollCourseState.State.PENDING), + (20, BulkUnenrollCourseState.State.FAILED), + (30, BulkUnenrollCourseState.State.PENDING), + ]) + ] + + def _request(self, batch_id=None): + return self.client.get(self._url(batch_id or self.batch.uuid)) + + # --- happy path --- + + def test_status_returns_summary_and_paginated_courses(self): + self._login(self.admin) + resp = self.client.get(self._url(self.batch.uuid)) + + self.assertEqual(resp.status_code, 200) + data = resp.json() + self.assertEqual(data["batch_id"], str(self.batch.uuid)) + self.assertEqual(data["state"], "pending") + self.assertEqual(data["reason"], "cleanup") + self.assertEqual(data["total_courses"], 3) + # Summary is the sum of all course active counts (10 + 20 + 30). + self.assertEqual(data["totals"]["active"], 60) + self.assertEqual(data["courses"]["count"], 3) + self.assertEqual(len(data["courses"]["results"]), 3) + self.assertIn("next", data["courses"]) + self.assertIn("previous", data["courses"]) + + def test_state_filter_narrows_courses_but_not_summary(self): + """Filtering the listed rows must not narrow the batch-level progress numbers.""" + self._login(self.admin) + self.rows[0].unenrolled = 5 + self.rows[0].save() + self.rows[1].unenrolled = 10 + self.rows[1].save() + + resp = self.client.get(self._url(self.batch.uuid), {"state": "failed"}) + + self.assertEqual(resp.status_code, 200) + data = resp.json() + # Only the one failed course is listed... + self.assertEqual(data["courses"]["count"], 1) + self.assertEqual(data["courses"]["results"][0]["state"], "failed") + # ...but the summary still reflects the whole batch. + self.assertEqual(data["totals"]["active"], 60) + self.assertEqual(data["totals"]["unenrolled"], 15) + + def test_state_filter_accepts_a_comma_separated_list(self): + """ + The UI offers grouped choices ("in progress" == pending,running), so a comma + list must select the union rather than silently matching nothing. + """ + self._login(self.admin) + resp = self.client.get(self._url(self.batch.uuid), {"state": "pending,failed"}) + + self.assertEqual(resp.status_code, 200) + data = resp.json() + self.assertEqual(data["courses"]["count"], 3) + self.assertEqual( + sorted(row["state"] for row in data["courses"]["results"]), + ["failed", "pending", "pending"], + ) + self.assertEqual(data["totals"]["active"], 60) + + def test_state_filter_rejects_an_unknown_state(self): + """A typo must say so rather than silently listing nothing.""" + self._login(self.admin) + resp = self.client.get(self._url(self.batch.uuid), {"state": "pending,bogus"}) + + self.assertEqual(resp.status_code, 400) + self.assertIn("bogus", str(resp.json())) + + def test_course_rows_expose_progress_counters(self): + """Each course row carries the worker-populated counters, not just the preview.""" + self._login(self.admin) + row = self.rows[1] + row.unenrolled = 7 + row.already_inactive = 2 + row.failed_count = 1 + row.chunks_total = 4 + row.chunks_finished = 3 + row.save() + + resp = self.client.get(self._url(self.batch.uuid)) + self.assertEqual(resp.status_code, 200) + listed = {r["course_id"]: r for r in resp.json()["courses"]["results"]} + got = listed[str(row.course_id)] + + self.assertEqual(got["unenrolled"], 7) + self.assertEqual(got["already_inactive"], 2) + self.assertEqual(got["failed_count"], 1) + self.assertEqual(got["chunks_total"], 4) + self.assertEqual(got["chunks_finished"], 3) + # The dry-run preview count is still there alongside them. + self.assertEqual(got["active_count"], 20) + + def test_totals_aggregate_progress_across_whole_batch(self): + """Batch-level totals sum the counters so the UI needn't page through 1000 rows.""" + self._login(self.admin) + for row, (unenrolled, inactive, failed) in zip(self.rows, [(5, 1, 0), (10, 0, 2), (0, 0, 0)]): + row.unenrolled = unenrolled + row.already_inactive = inactive + row.failed_count = failed + row.save() + + data = self.client.get(self._url(self.batch.uuid)).json() + + self.assertEqual(data["totals"]["active"], 60) # unchanged, still the preview sum + self.assertEqual(data["totals"]["unenrolled"], 15) + self.assertEqual(data["totals"]["already_inactive"], 1) + self.assertEqual(data["totals"]["failed"], 2) + # Courses finished = those in a terminal state (1 failed, 0 succeeded). + self.assertEqual(data["totals"]["courses_finished"], 1) + + def test_totals_progress_defaults_to_zero_before_any_work(self): + """A freshly uploaded batch reports zeros, not nulls — the UI renders them directly.""" + self._login(self.admin) + data = self.client.get(self._url(self.batch.uuid)).json() + + self.assertEqual(data["totals"]["unenrolled"], 0) + self.assertEqual(data["totals"]["already_inactive"], 0) + self.assertEqual(data["totals"]["failed"], 0) + + def test_status_query_count_is_flat_in_course_count(self): + """Polling cost must not grow with batch size — the summary stays one aggregate.""" + from django.db import connection + from django.test.utils import CaptureQueriesContext + + self._login(self.admin) + + def poll(): + with CaptureQueriesContext(connection) as ctx: + resp = self.client.get(self._url(self.batch.uuid)) + self.assertEqual(resp.status_code, 200) + return len(ctx) + + # Warm session/waffle/site caches: the first request's middleware lookups + # would otherwise swamp the signal. + poll() + small = poll() + for i in range(20): + BulkUnenrollCourseState.objects.create( + batch=self.batch, + course_id=CourseOverviewFactory.create(org="edX", run=f"Q{i}", display_name=f"Q{i}").id, + active_count=1, + ) + self.assertEqual(poll(), small) + + +@ddt.ddt +class BulkUnenrollCancelAPIViewTest(BatchScopedEndpointTestMixin, SupportViewTestCase): + """POST /api/support/v1/bulk_unenroll//cancel/.""" + + url_name = "bulk_unenroll_cancel" + + def _make_batch(self, state): + return BulkUnenrollBatch.objects.create( + requester=self.admin, total_courses=0, state=state, + ) + + def _cancel(self, batch_id): + return self.client.post(self._url(batch_id), {}, format="json") + + def _request(self, batch_id=None): + if batch_id is None: + batch_id = self._make_batch(BulkUnenrollBatch.State.RUNNING).uuid + return self._cancel(batch_id) + + @ddt.data( + BulkUnenrollBatch.State.VALIDATED, + BulkUnenrollBatch.State.PENDING, + BulkUnenrollBatch.State.RUNNING, + ) + def test_cancel_a_batch_still_in_flight(self, state): + self._login(self.admin) + batch = self._make_batch(state) + resp = self._cancel(batch.uuid) + self.assertIn(resp.status_code, (200, 202)) + batch.refresh_from_db() + self.assertEqual(batch.state, "cancelled") + + def test_cancel_sweeps_unfinished_courses_and_keeps_finished_ones(self): + """ + Unfinished rows left ``running`` would make a terminal batch read as still + working; rows that already earned a terminal state keep it. + """ + self._login(self.admin) + batch = self._make_batch(BulkUnenrollBatch.State.RUNNING) + rows = {} + for i, state in enumerate([ + BulkUnenrollCourseState.State.SUCCEEDED, + BulkUnenrollCourseState.State.FAILED, + BulkUnenrollCourseState.State.RUNNING, + BulkUnenrollCourseState.State.PENDING, + ]): + rows[state] = BulkUnenrollCourseState.objects.create( + batch=batch, + course_id=CourseOverviewFactory.create(org="edX", run=f"C{i}", display_name=f"C{i}").id, + state=state, + ) + + resp = self._cancel(batch.uuid) + self.assertIn(resp.status_code, (200, 202)) + + for state, row in rows.items(): + row.refresh_from_db() + + # Work that never finished is now reported as cancelled... + self.assertEqual(rows[BulkUnenrollCourseState.State.RUNNING].state, "cancelled") + self.assertEqual(rows[BulkUnenrollCourseState.State.PENDING].state, "cancelled") + self.assertIsNotNone(rows[BulkUnenrollCourseState.State.RUNNING].finished) + # ...and work that had already finished keeps the state it earned. + self.assertEqual(rows[BulkUnenrollCourseState.State.SUCCEEDED].state, "succeeded") + self.assertEqual(rows[BulkUnenrollCourseState.State.FAILED].state, "failed") + + def test_cancelled_courses_count_as_finished(self): + """The progress view's "N of M courses finished" must reach M after cancel.""" + self._login(self.admin) + batch = self._make_batch(BulkUnenrollBatch.State.RUNNING) + BulkUnenrollCourseState.objects.create( + batch=batch, + course_id=CourseOverviewFactory.create(org="edX", run="D0", display_name="D0").id, + state=BulkUnenrollCourseState.State.RUNNING, + ) + self._cancel(batch.uuid) + + status_url = reverse( + "lms.djangoapps.support.rest_api:v1:bulk_unenroll_status", + kwargs={"batch_id": str(batch.uuid)}, + ) + totals = self.client.get(status_url).json()["totals"] + self.assertEqual(totals["courses_finished"], 1) + + @ddt.data( + BulkUnenrollBatch.State.SUCCEEDED, + BulkUnenrollBatch.State.FAILED, + BulkUnenrollBatch.State.PARTIAL, + BulkUnenrollBatch.State.CANCELLED, + ) + def test_cancel_settled_batch_returns_409(self, state): + """Cancel stops work in flight; a batch that already settled has none left.""" + self._login(self.admin) + batch = self._make_batch(state) + resp = self._cancel(batch.uuid) + self.assertEqual(resp.status_code, 409) + batch.refresh_from_db() + self.assertEqual(batch.state, state) + + def test_cancel_does_not_overwrite_a_terminal_state_set_after_the_read(self): + """ + A stale write must not resurrect a batch the engine just finalized. + """ + self._login(self.admin) + batch = self._make_batch(BulkUnenrollBatch.State.RUNNING) + stale = BulkUnenrollBatch.objects.get(pk=batch.pk) # says 'running' + # The engine's finalizer commits first. + BulkUnenrollBatch.objects.filter(pk=batch.pk).update(state=BulkUnenrollBatch.State.SUCCEEDED) + + with patch.object(BulkUnenrollBatch.objects, "get", return_value=stale): + resp = self._cancel(batch.uuid) + + self.assertEqual(resp.status_code, 409) + batch.refresh_from_db() + self.assertEqual(batch.state, "succeeded") + + +@ddt.ddt +class BulkUnenrollRetryAPIViewTest(BatchScopedEndpointTestMixin, SupportViewTestCase): + """POST /api/support/v1/bulk_unenroll//retry/.""" + + url_name = "bulk_unenroll_retry" + patch_dispatch = True + + def setUp(self): + super().setUp() + self.course_ok = CourseOverviewFactory.create(org="edX", run="OK", display_name="OK") + self.course_bad = CourseOverviewFactory.create(org="edX", run="BAD", display_name="BAD") + + def _make_batch(self, state=BulkUnenrollBatch.State.PARTIAL, with_failure=True): + """Build a batch with one succeeded course and, optionally, one failed.""" + batch = BulkUnenrollBatch.objects.create( + requester=self.admin, total_courses=2, state=state, + ) + BulkUnenrollCourseState.objects.create( + batch=batch, course_id=self.course_ok.id, + state=BulkUnenrollCourseState.State.SUCCEEDED, unenrolled=5, + chunks_total=1, chunks_finished=1, + ) + if with_failure: + BulkUnenrollCourseState.objects.create( + batch=batch, course_id=self.course_bad.id, + state=BulkUnenrollCourseState.State.FAILED, failed_count=3, + chunks_total=1, chunks_finished=1, error="boom", + ) + return batch + + def _retry(self, batch_id): + return self.client.post(self._url(batch_id), {}, format="json") + + def _request(self, batch_id=None): + return self._retry(self._make_batch().uuid if batch_id is None else batch_id) + + def test_retry_resets_failed_courses_and_redispatches(self): + self._login(self.admin) + batch = self._make_batch() + resp = self._retry(batch.uuid) + + self.assertEqual(resp.status_code, 202) + self.mock_dispatch.assert_called_once() + batch.refresh_from_db() + self.assertEqual(batch.state, "pending") + failed = batch.courses.get(course_id=self.course_bad.id) + self.assertEqual(failed.state, "pending") # reset + self.assertEqual(failed.failed_count, 0) + self.assertEqual(failed.chunks_finished, 0) + self.assertEqual(failed.error, "") + ok = batch.courses.get(course_id=self.course_ok.id) + self.assertEqual(ok.state, "succeeded") # untouched + self.assertEqual(ok.unenrolled, 5) + + def test_retry_with_no_failed_courses_returns_409(self): + """ + A batch can be in a retryable *state* yet have no failed rows to re-run; + dispatching would then queue nothing at all. + """ + self._login(self.admin) + batch = self._make_batch(state=BulkUnenrollBatch.State.PARTIAL, with_failure=False) + resp = self._retry(batch.uuid) + self.assertEqual(resp.status_code, 409) + self.mock_dispatch.assert_not_called() + + @ddt.data( + BulkUnenrollBatch.State.VALIDATED, + BulkUnenrollBatch.State.PENDING, + BulkUnenrollBatch.State.RUNNING, + BulkUnenrollBatch.State.CANCELLED, + BulkUnenrollBatch.State.SUCCEEDED, + ) + def test_retry_unsettled_or_clean_batch_returns_409(self, state): + """Retry needs a settled batch with failures; anything else is a conflict.""" + self._login(self.admin) + resp = self._retry(self._make_batch(state=state).uuid) + self.assertEqual(resp.status_code, 409) + self.mock_dispatch.assert_not_called() + + def test_failed_dispatch_restores_the_batch_and_reports_503(self): + """A retry that cannot reach the broker must not leave the batch pending.""" + self._login(self.admin) + batch = self._make_batch() + self.mock_dispatch.side_effect = OSError("broker unreachable") + + resp = self._retry(batch.uuid) + + self.assertEqual(resp.status_code, 503) + batch.refresh_from_db() + self.assertEqual(batch.state, "partial") # retryable again + failed = batch.courses.get(course_id=self.course_bad.id) + self.assertEqual(failed.state, "failed") # and the reset was rolled back + # Including the diagnostics: a 'failed' row reporting no error and no counts + # would strip the operator of the very information they retry against. + self.assertEqual(failed.error, "boom") + self.assertEqual(failed.failed_count, 3) + self.assertEqual(failed.chunks_total, 1) + self.assertEqual(failed.chunks_finished, 1) + self.assertEqual(failed.attempt, 2) # but the spent generation stays spent + + def test_failed_dispatch_does_not_claw_back_work_already_claimed(self): + """ + A publish error does not prove non-delivery: if a dispatcher already claimed + the courses, the rollback must not mark running work 'failed'. + """ + self._login(self.admin) + batch = self._make_batch() + + def raise_after_the_dispatcher_claimed_it(dispatched_batch): + BulkUnenrollCourseState.objects.filter( + batch=dispatched_batch, state=BulkUnenrollCourseState.State.PENDING, + ).update(state=BulkUnenrollCourseState.State.RUNNING) + BulkUnenrollBatch.objects.filter(pk=dispatched_batch.pk).update( + state=BulkUnenrollBatch.State.RUNNING, + ) + raise OSError("connection reset after publish") + + self.mock_dispatch.side_effect = raise_after_the_dispatcher_claimed_it + resp = self._retry(batch.uuid) + + self.assertEqual(resp.status_code, 503) + batch.refresh_from_db() + self.assertEqual(batch.state, "running") # the live run is left alone + failed = batch.courses.get(course_id=self.course_bad.id) + self.assertEqual(failed.state, "running") # not clawed back to 'failed' + + def test_retry_starts_a_new_attempt_instead_of_reusing_chunk_identities(self): + """ + The re-run fans out from index 0 again, so reusing (course, index) would let + a straggler from the previous attempt squat the new chunk 0. + """ + self._login(self.admin) + batch = self._make_batch() + failed_state = batch.courses.get(course_id=self.course_bad.id) + ok_state = batch.courses.get(course_id=self.course_ok.id) + old_chunk = BulkUnenrollChunk.objects.create( + course_state=failed_state, chunk_index=0, attempt=1, + state=BulkUnenrollChunk.State.FINISHED, + ) + + resp = self._retry(batch.uuid) + + self.assertEqual(resp.status_code, 202) + failed_state.refresh_from_db() + ok_state.refresh_from_db() + self.assertEqual(failed_state.attempt, 2) # re-run under a fresh generation + self.assertEqual(ok_state.attempt, 1) # untouched; it is not re-run + # The old row survives as history: the new attempt's chunk 0 is a new identity. + self.assertTrue(BulkUnenrollChunk.objects.filter(pk=old_chunk.pk).exists()) + BulkUnenrollChunk.objects.create(course_state=failed_state, chunk_index=0, attempt=2) + + def test_concurrent_retry_does_not_double_dispatch(self): + """Two retries on the same batch must not both re-queue the failed courses.""" + self._login(self.admin) + batch = self._make_batch() + stale = BulkUnenrollBatch.objects.get(pk=batch.pk) # says 'partial' + # The competing retry commits first and puts the batch back in flight. + BulkUnenrollBatch.objects.filter(pk=batch.pk).update(state=BulkUnenrollBatch.State.PENDING) + + with patch.object(BulkUnenrollBatch.objects, "get", return_value=stale): + resp = self._retry(batch.uuid) + + self.assertEqual(resp.status_code, 409) + self.mock_dispatch.assert_not_called() diff --git a/lms/djangoapps/support/rest_api/v1/urls.py b/lms/djangoapps/support/rest_api/v1/urls.py index 8b6343e18be4..68d97cb74a65 100644 --- a/lms/djangoapps/support/rest_api/v1/urls.py +++ b/lms/djangoapps/support/rest_api/v1/urls.py @@ -1,17 +1,52 @@ """ -URL definitions for the course_modes v1 API. +URL definitions for the support v1 API. """ from django.urls import re_path -from .views import CourseTeamManageAPIView +from .views import ( + BulkUnenrollAPIView, + BulkUnenrollCancelAPIView, + BulkUnenrollConfirmAPIView, + BulkUnenrollRetryAPIView, + BulkUnenrollStatusAPIView, + CourseTeamManageAPIView, +) app_name = "v1" +# A ``batch_id`` is the batch's uuid: 32 hex digits, optionally hyphen-grouped. +BATCH_ID = r"(?P[0-9a-fA-F-]+)" + urlpatterns = [ re_path( r"manage_course_team/?$", CourseTeamManageAPIView.as_view(), name="manage_course_team", ), + re_path( + rf"bulk_unenroll/{BATCH_ID}/confirm/?$", + BulkUnenrollConfirmAPIView.as_view(), + name="bulk_unenroll_confirm", + ), + re_path( + rf"bulk_unenroll/{BATCH_ID}/cancel/?$", + BulkUnenrollCancelAPIView.as_view(), + name="bulk_unenroll_cancel", + ), + re_path( + rf"bulk_unenroll/{BATCH_ID}/retry/?$", + BulkUnenrollRetryAPIView.as_view(), + name="bulk_unenroll_retry", + ), + re_path( + rf"bulk_unenroll/{BATCH_ID}/?$", + BulkUnenrollStatusAPIView.as_view(), + name="bulk_unenroll_status", + ), + re_path( + r"bulk_unenroll/?$", + BulkUnenrollAPIView.as_view(), + name="bulk_unenroll", + ), ] diff --git a/lms/djangoapps/support/rest_api/v1/views.py b/lms/djangoapps/support/rest_api/v1/views.py index 73bbdf04da62..c2df80aa48c5 100644 --- a/lms/djangoapps/support/rest_api/v1/views.py +++ b/lms/djangoapps/support/rest_api/v1/views.py @@ -2,24 +2,122 @@ API Views for course team management in support app. """ +import logging + +from django.conf import settings from django.contrib.auth import get_user_model from django.core.exceptions import ObjectDoesNotExist -from django.db.models import Q +from django.core.exceptions import ValidationError as DjangoValidationError +from django.db import transaction +from django.db.models import Count, F, Q, Sum +from django.utils.timezone import now +from edx_rest_framework_extensions.paginators import DefaultPagination from opaque_keys.edx.keys import CourseKey from rest_framework import status from rest_framework.exceptions import NotFound, ValidationError from rest_framework.generics import GenericAPIView -from rest_framework.permissions import IsAuthenticated +from rest_framework.parsers import FormParser, MultiPartParser +from rest_framework.permissions import IsAdminUser, IsAuthenticated from rest_framework.response import Response +from common.djangoapps.student.api import ( + BulkUnenrollCsvTooManyRows, + BulkUnenrollCsvUnreadable, + parse_bulk_unenroll_csv, +) from common.djangoapps.student.models import CourseEnrollment from common.djangoapps.student.models.user import CourseAccessRole +from lms.djangoapps.support.models import BulkUnenrollBatch, BulkUnenrollCourseState from openedx.core.djangoapps.content.course_overviews.models import CourseOverview -from ..serializers import CourseTeamManageSerializer +from ..serializers import ( + BulkUnenrollBatchSerializer, + BulkUnenrollCourseStateSerializer, + CourseTeamManageSerializer, +) User = get_user_model() +log = logging.getLogger(__name__) + + +def _dispatch_bulk_unenroll(batch): + """ + Queue the level-1 dispatcher (``bulk_unenroll_batch``) for a confirmed batch. + + The request thread never touches enrollments — all fetching and mutation happen + on the workers. Kept as a module-level seam so ``confirm``/``retry`` have a + stable call site and tests can assert the hand-off without running the engine. + """ + from lms.djangoapps.support.tasks import bulk_unenroll_batch # lazy: avoid import cycle + bulk_unenroll_batch.apply_async( + args=[str(batch.uuid)], + routing_key=settings.BULK_UNENROLL_ROUTING_KEY, + ) + + +def _dispatch_or_rollback(batch, action, rollback): + """ + Dispatch a committed batch; on failure run ``rollback`` and return a 503. + + Shared by confirm and retry, which both commit a state transition and only then + publish — a batch left ``pending`` with no dispatcher is stranded, since nothing + owns it and neither endpoint would accept it again. ``rollback`` differs per + endpoint (they undo different rows) but must always be *conditional*: a publish + error does not prove the message was undelivered, so a dispatcher that did + receive it may already be working, and only untouched rows may be reverted. + + Returns the 503 ``Response`` on failure, or ``None`` when the dispatch landed. + """ + try: + _dispatch_bulk_unenroll(batch) + except Exception: # pylint: disable=broad-except + log.exception("bulk_unenroll: failed to queue %s for batch %s", action, batch.uuid) + rollback() + return Response( + {"detail": f"Could not queue the {action}. Please try again."}, + status=status.HTTP_503_SERVICE_UNAVAILABLE, + ) + return None + + +def _batch_response(batch, http_status): + """The `{batch_id, state}` body confirm/cancel/retry all answer with.""" + return Response({"batch_id": str(batch.uuid), "state": batch.state}, status=http_status) + + +def _conflict(detail): + """A 409 for a lifecycle action the batch's current state does not allow.""" + return Response({"detail": detail}, status=status.HTTP_409_CONFLICT) + + +#: Per-course progress a retry clears, and the value it clears each one to. Named +#: once so the reset and the rollback that undoes it cannot drift apart. +RESET_FIELDS = { + "unenrolled": 0, "already_inactive": 0, "failed_count": 0, + "total_enrollments": 0, "chunks_total": 0, "chunks_finished": 0, + "started": None, "finished": None, "error": "", +} + + +def _parse_state_filter(request, state_enum): + """ + Read a comma-separated ``?state=`` query param, or ``None`` if absent. + + Shared by the batch list and the per-batch course list so the two cannot + disagree about what ``?state=`` means. Unknown values raise rather than being + dropped: a filter that silently returns nothing is worse than one that rejects. + """ + raw = request.query_params.get("state") + if not raw: + return None + + states = [value.strip() for value in raw.split(",") if value.strip()] + unknown = [value for value in states if value not in set(state_enum.values)] + if unknown: + raise ValidationError({"state": f"Unknown state(s): {', '.join(unknown)}."}) + return states + class CourseTeamManageAPIView(GenericAPIView): """ @@ -569,3 +667,341 @@ def _make_result(self, data, outcome, error=None): if error: result["error"] = error return result + + +class BulkUnenrollAPIView(GenericAPIView): + """ + Use case: + - POST a single-column CSV of course ids to stage a bulk-unenroll batch. + - Parses and validates the file, counts active enrollments per course, and + persists a *validated* batch — a dry-run preview. Nothing is unenrolled + here: the batch is confirmed separately and run by ``support/tasks.py``. + - GET lists batches, newest first, optionally filtered by ``?state=``. + - Global staff only (not the support-role gate): deactivating whole courses + has a far wider blast radius than the per-learner unenroll. + """ + + permission_classes = (IsAuthenticated, IsAdminUser) + parser_classes = (MultiPartParser, FormParser) + serializer_class = BulkUnenrollBatchSerializer + pagination_class = DefaultPagination + + def get(self, request): + """ + List batches, newest first, optionally filtered by state. + + Otherwise a batch is only reachable by its uuid, surfaced once in the URL + after confirming — an operator who closed that tab has no way back to a run + with hours left. Not scoped to ``request.user``: every caller is global + staff, and losing a colleague's batch is the same operational problem. + """ + batches = BulkUnenrollBatch.objects.select_related("requester").order_by("-created") + + # Comma-separated so one request can ask for all in-flight states at once. + states = _parse_state_filter(request, BulkUnenrollBatch.State) + if states: + batches = batches.filter(state__in=states) + + page = self.paginate_queryset(batches) + return self.get_paginated_response(self.get_serializer(page, many=True).data) + + def post(self, request): + """Parse + validate the CSV, count active enrollments, persist a validated batch.""" + file_obj = request.FILES.get("file") + if not file_obj: + raise ValidationError({"file": "A CSV file is required."}) + + max_file_bytes = settings.BULK_UNENROLL_MAX_FILE_BYTES + max_rows = settings.BULK_UNENROLL_MAX_ROWS + + if file_obj.size > max_file_bytes: + return Response( + {"detail": f"File exceeds the maximum size of {max_file_bytes} bytes."}, + status=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, + ) + + # Enforced *while parsing*, over physical data rows: an oversized file — + # duplicates and invalid ids included — is never parsed in full. + try: + course_keys, errors = parse_bulk_unenroll_csv(file_obj, max_rows=max_rows) + except BulkUnenrollCsvTooManyRows as exc: + return Response( + {"detail": f"File exceeds the maximum of {exc.max_rows} rows."}, + status=status.HTTP_400_BAD_REQUEST, + ) + except BulkUnenrollCsvUnreadable as exc: + # A file we cannot decode or parse is bad input, not a server fault: + # answer it like any other rejected upload instead of raising a 500. + return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) + + if not course_keys: + return Response( + {"detail": "No valid course ids found.", "errors": errors}, + status=status.HTTP_400_BAD_REQUEST, + ) + + # One query: which of the course ids actually have a CourseOverview. + existing = set( + CourseOverview.objects.filter(id__in=course_keys).values_list("id", flat=True) + ) + # One grouped query: active-enrollment count per course (0 if absent). + counts = dict( + CourseEnrollment.objects + .filter(course_id__in=course_keys, is_active=True) + .order_by() # no sort on a grouped query (CourseEnrollment has Meta.ordering) + .values_list("course_id") + .annotate(active=Count("id")) + ) + + with transaction.atomic(): + batch = BulkUnenrollBatch.objects.create( + requester=request.user, + csv_filename=getattr(file_obj, "name", "")[:255], + total_courses=len(course_keys), + state=BulkUnenrollBatch.State.VALIDATED, + ) + course_states = [ + BulkUnenrollCourseState( + batch=batch, + course_id=course_key, + active_count=counts.get(course_key, 0), + # Flag (do not drop) course ids with no CourseOverview — likely + # a typo, but enrollments can exist without one, so it's a signal. + error="" if course_key in existing else "Course not found", + ) + for course_key in course_keys + ] + BulkUnenrollCourseState.objects.bulk_create(course_states) + + response = BulkUnenrollBatchSerializer(batch).data + response["totals"] = {"active": sum(counts.values())} + response["courses"] = BulkUnenrollCourseStateSerializer(course_states, many=True).data + response["errors"] = errors + return Response(response, status=status.HTTP_200_OK) + + +class BulkUnenrollConfirmAPIView(GenericAPIView): + """ + Use case: + - Confirm a validated batch after reviewing its dry-run preview: flips + ``validated -> pending`` and dispatches the async work. Nothing is + unenrolled here — that happens in ``support/tasks.py``. + - ``reason`` is collected here rather than at upload, so it can be attached + to every audit row and the preview-first UI need not demand it up front. + - ``409`` if the batch is not ``validated``: confirm is only valid once. + - Global staff only, matching upload. + """ + + permission_classes = (IsAuthenticated, IsAdminUser) + pagination_class = None + + def post(self, request, batch_id): + """Validate the reason + state, flip to ``pending``, dispatch the worker.""" + batch = _get_batch_or_404(batch_id) + + reason = (request.data.get("reason") or "").strip() + if not reason: + raise ValidationError({"reason": "A non-blank reason is required to confirm."}) + + # Claim the transition on a locked, freshly-read row, or two concurrent + # confirms could both pass the guard and dispatch two engines. + with transaction.atomic(): + batch = BulkUnenrollBatch.objects.select_for_update().get(pk=batch.pk) + if batch.state != BulkUnenrollBatch.State.VALIDATED: + return _conflict(f"Batch is '{batch.state}', not 'validated'; it cannot be confirmed.") + batch.reason = reason[:255] + batch.state = BulkUnenrollBatch.State.PENDING + batch.save(update_fields=["reason", "state", "modified"]) + + # Dispatched after the transaction commits, so the worker can never read a + # row this request still holds locked. + failure = _dispatch_or_rollback(batch, "batch", lambda: ( + BulkUnenrollBatch.objects.filter( + pk=batch.pk, state=BulkUnenrollBatch.State.PENDING, + ).update(state=BulkUnenrollBatch.State.VALIDATED, reason="", modified=now()) + )) + if failure: + return failure + + return _batch_response(batch, status.HTTP_202_ACCEPTED) + + +class BulkUnenrollStatusAPIView(GenericAPIView): + """ + Use case: + - Polled by the UI to show a batch's progress: the batch-level summary plus + a page of its per-course rows, optionally filtered by ``?state=``. + - A batch can hold 1000 courses, so the course list is paginated and the + summary is one aggregate query rather than a sum over 1000 rows in Python. + - Global staff only, matching upload/confirm. + """ + + permission_classes = (IsAuthenticated, IsAdminUser) + serializer_class = BulkUnenrollCourseStateSerializer + pagination_class = DefaultPagination + + def get(self, request, batch_id): + """Return the batch summary plus a paginated (optionally filtered) course list.""" + batch = _get_batch_or_404(batch_id) + + courses = batch.courses.all().order_by("course_id") + # Optional comma-separated filter (same as the batch list), e.g. + # ?state=failed or the UI's "in progress" == ?state=pending,running. + states = _parse_state_filter(request, BulkUnenrollCourseState.State) + if states: + courses = courses.filter(state__in=states) + + # One aggregate query, over the *unfiltered* batch: the state filter + # narrows the listed rows only, so progress stays batch-wide. + totals = batch.courses.aggregate( + active=Sum("active_count"), + unenrolled=Sum("unenrolled"), + already_inactive=Sum("already_inactive"), + failed=Sum("failed_count"), + # "Finished" is exactly the engine's terminal set, so the progress the + # UI shows and the state the engine acts on cannot drift apart. + courses_finished=Count( + "pk", filter=Q(state__in=BulkUnenrollCourseState.TERMINAL_STATES), + ), + ) + + page = self.paginate_queryset(courses) + courses_data = self.get_serializer(page, many=True).data + + response = BulkUnenrollBatchSerializer(batch).data + # Coalesce to 0: Sum() returns None on an empty batch, and the UI renders + # these numbers directly rather than null-checking each one. + response["totals"] = { + "active": totals["active"] or 0, + "unenrolled": totals["unenrolled"] or 0, + "already_inactive": totals["already_inactive"] or 0, + "failed": totals["failed"] or 0, + "courses_finished": totals["courses_finished"] or 0, + } + response["courses"] = self.get_paginated_response(courses_data).data + return Response(response, status=status.HTTP_200_OK) + + +def _get_batch_or_404(batch_id): + """Look up a batch by its public uuid, raising DRF NotFound on miss/bad id.""" + try: + return BulkUnenrollBatch.objects.get(uuid=batch_id) + except (BulkUnenrollBatch.DoesNotExist, DjangoValidationError, ValueError, TypeError) as exc: + raise NotFound("Batch not found.") from exc + + +class BulkUnenrollCancelAPIView(GenericAPIView): + """ + Use case: + - Cancel a batch that is still ``validated``, ``pending``, or ``running``; + ``409`` once it has finished. + - The engine notices: the dispatcher stops queuing and each running chunk + stops within ``BULK_UNENROLL_CANCEL_CHECK_EVERY`` learners. Learners + already processed are **not** re-enrolled — cancel stops future work only. + - Unfinished course rows are swept to ``cancelled`` in the same transaction, + so a terminal batch never reads as still working. Courses that already + finished keep the state they earned. + - Global staff only. + """ + + permission_classes = (IsAuthenticated, IsAdminUser) + pagination_class = None + + #: States from which a cancel is still meaningful. + CANCELLABLE = ( + BulkUnenrollBatch.State.VALIDATED, + BulkUnenrollBatch.State.PENDING, + BulkUnenrollBatch.State.RUNNING, + ) + + def post(self, request, batch_id): + """Mark the batch cancelled if it is still in flight.""" + batch = _get_batch_or_404(batch_id) + # Locked re-read: the engine may have finalized this batch since the row was + # read, and a stale write would resurrect a terminal batch as 'cancelled'. + with transaction.atomic(): + batch = BulkUnenrollBatch.objects.select_for_update().get(pk=batch.pk) + if batch.state not in self.CANCELLABLE: + return _conflict(f"Batch is '{batch.state}'; it can no longer be cancelled.") + batch.state = BulkUnenrollBatch.State.CANCELLED + batch.save(update_fields=["state", "modified"]) + # Conditional on purpose: a course the engine just finalized keeps + # its earned state; only unfinished work is reported cancelled. + batch.courses.filter( + state__in=( + BulkUnenrollCourseState.State.PENDING, + BulkUnenrollCourseState.State.RUNNING, + ), + ).update( + state=BulkUnenrollCourseState.State.CANCELLED, finished=now(), modified=now(), + ) + return _batch_response(batch, status.HTTP_200_OK) + + +class BulkUnenrollRetryAPIView(GenericAPIView): + """ + Use case: + - Retry the ``failed`` courses of a settled batch: resets those rows to + ``pending`` (clearing counters, chunk tracking, and error) and + re-dispatches. The dispatcher only queues ``pending`` courses, so the + successes are left untouched. + - Re-running is safe: the engine skips learners already inactive. + - ``409`` if the batch is still in flight or has no failed courses. + - Global staff only. + """ + + permission_classes = (IsAuthenticated, IsAdminUser) + pagination_class = None + + #: A retry only makes sense once the batch has settled with failures. + RETRYABLE = ( + BulkUnenrollBatch.State.FAILED, + BulkUnenrollBatch.State.PARTIAL, + ) + + def post(self, request, batch_id): + """Reset failed courses to pending and re-dispatch.""" + batch = _get_batch_or_404(batch_id) + + # One locked transaction for the whole claim: two retries on the same batch + # would otherwise both pass the guard and dispatch two engines over it. + with transaction.atomic(): + batch = BulkUnenrollBatch.objects.select_for_update().get(pk=batch.pk) + previous_state = batch.state + if batch.state not in self.RETRYABLE: + return _conflict(f"Batch is '{batch.state}'; nothing to retry.") + + failed = batch.courses.filter(state=BulkUnenrollCourseState.State.FAILED) + if not failed.exists(): + return _conflict("Batch has no failed courses to retry.") + + # Bump the generation (not delete the ledger) so an old-attempt straggler + # can't claim chunk 0; the snapshot lets the rollback restore diagnostics. + before = {row.pop("pk"): row for row in failed.values("pk", *RESET_FIELDS)} + failed.update( + state=BulkUnenrollCourseState.State.PENDING, + attempt=F("attempt") + 1, + modified=now(), + **{field: RESET_FIELDS[field] for field in RESET_FIELDS}, + ) + batch.state = BulkUnenrollBatch.State.PENDING + batch.save(update_fields=["state", "modified"]) + + # Restores courses (row by row — each carries its own counters back) and + # the batch. The bumped attempt stays bumped: those identities are spent. + def _undo_reset(): + for pk, fields in before.items(): + BulkUnenrollCourseState.objects.filter( + pk=pk, state=BulkUnenrollCourseState.State.PENDING, + ).update( + state=BulkUnenrollCourseState.State.FAILED, modified=now(), **fields, + ) + BulkUnenrollBatch.objects.filter( + pk=batch.pk, state=BulkUnenrollBatch.State.PENDING, + ).update(state=previous_state, modified=now()) + + failure = _dispatch_or_rollback(batch, "retry", _undo_reset) + if failure: + return failure + + return _batch_response(batch, status.HTTP_202_ACCEPTED) diff --git a/lms/djangoapps/support/tasks.py b/lms/djangoapps/support/tasks.py index e274f80e0e4d..d0d4d54deebd 100644 --- a/lms/djangoapps/support/tasks.py +++ b/lms/djangoapps/support/tasks.py @@ -3,11 +3,24 @@ from datetime import datetime import logging from celery import shared_task +from celery.exceptions import SoftTimeLimitExceeded from completion.models import BlockCompletion from edx_django_utils.monitoring import set_code_owner_attribute -from common.djangoapps.student.models.course_enrollment import CourseEnrollment +from django.conf import settings +from django.contrib.auth import get_user_model +from django.db import transaction +from django.db.models import F +from django.utils import timezone +from opaque_keys.edx.keys import CourseKey + +from common.djangoapps.student.models.course_enrollment import ( + ENROLLED_TO_UNENROLLED, + CourseEnrollment, + ManualEnrollmentAudit, +) from common.djangoapps.student.models.user import get_user_by_username_or_email +from lms.djangoapps.support.models import BulkUnenrollBatch, BulkUnenrollChunk, BulkUnenrollCourseState from lms.djangoapps.courseware.courses import get_course from lms.djangoapps.courseware.models import StudentModule from lms.djangoapps.instructor.enrollment import reset_student_attempts @@ -25,6 +38,489 @@ log = logging.getLogger(__name__) +User = get_user_model() + + +@shared_task( + rate_limit=settings.BULK_UNENROLL_CHUNK_RATE_LIMIT, + soft_time_limit=settings.BULK_UNENROLL_SOFT_TIME_LIMIT, +) +@set_code_owner_attribute +def bulk_unenroll_chunk(batch_uuid, course_id, user_ids, chunk_index, attempt, continuation=0): + """ + Deactivate a chunk of learners in one course — level 3, the only place + enrollments are mutated. + + Unenrolls via ``CourseEnrollment.unenroll(skip_refund=True)`` under + ``select_for_update`` on the *active* enrollment, writes a + ``ManualEnrollmentAudit`` per learner, and tallies the outcomes. Re-running is + safe: an already-inactive learner counts as ``already_inactive``, and one bad + learner never aborts the chunk. ``(attempt, chunk_index, continuation)`` is the + chunk's identity; the *accounting* is claimed exactly once by + ``_record_chunk_completion``. Work stops as soon as ``_chunk_stop_reason`` + reports a revocation (checked up front and every ``check_every`` learners). + """ + batch = BulkUnenrollBatch.objects.get(uuid=batch_uuid) + course_key = CourseKey.from_string(course_id) + state = BulkUnenrollCourseState.objects.get(batch=batch, course_id=course_key) + + # Hard stop before touching anything: cancelled, superseded by a retry, or a + # course that has already settled (see _chunk_stop_reason). + stop_reason = _chunk_stop_reason(state.pk, attempt) + if stop_reason: + log.info( + "bulk_unenroll_chunk: discarding chunk %s of course %s (batch %s) because " + "%s", + chunk_index, course_key, batch_uuid, stop_reason, + ) + return + + # A non-positive interval falls back to checking every learner: never divide + # by zero, and never disable the checks that bound a revoked chunk's overrun. + check_every = max(1, settings.BULK_UNENROLL_CANCEL_CHECK_EVERY or 1) + processed_ids = [] + unenrolled = already_inactive = failed = 0 + remaining = None + try: + for user in User.objects.filter(id__in=user_ids).iterator(): + try: + with transaction.atomic(): + enrollment = CourseEnrollment.objects.select_for_update().get( + user=user, course_id=course_key, is_active=True, + ) + CourseEnrollment.unenroll(user, course_key, skip_refund=True) + ManualEnrollmentAudit.create_manual_enrollment_audit( + batch.requester, user.email, ENROLLED_TO_UNENROLLED, + reason=batch.reason, enrollment=enrollment, + ) + unenrolled += 1 + except CourseEnrollment.DoesNotExist: + # Already inactive — someone else got there first, or this is a retry. + already_inactive += 1 + except SoftTimeLimitExceeded: + # The chunk is out of time; this is not a bad learner. Re-raised so + # the handler below can hand the untouched tail to a fresh chunk. + raise + except Exception: # pylint: disable=broad-except + # Never let one learner kill the chunk; record and move on. + failed += 1 + log.exception( + "bulk_unenroll_chunk: failed to unenroll user %s in course %s (batch %s)", + user.id, course_key, batch_uuid, + ) + + processed_ids.append(user.id) + # Honour a mid-chunk revocation (cancel / retry / course failed) + # without a DB check per learner. + if len(processed_ids) % check_every == 0 and _chunk_stop_reason(state.pk, attempt): + break + except SoftTimeLimitExceeded: + done = set(processed_ids) + remaining = [user_id for user_id in user_ids if user_id not in done] + log.warning( + "bulk_unenroll_chunk: soft time limit hit in chunk %s of course %s (batch %s); " + "re-queueing %s learner(s)", + chunk_index, course_key, batch_uuid, len(remaining), + ) + + _record_chunk_completion( + state.pk, attempt, chunk_index, continuation, unenrolled, already_inactive, failed, + tail=(batch_uuid, course_id, remaining) if remaining else None, + ) + + +def _record_chunk_completion( + state_pk, attempt, chunk_index, continuation, unenrolled, already_inactive, failed, tail=None, +): + """ + Claim this chunk's completion, then roll its tally into the course counters. + + A conditional ``pending -> finished`` update on the chunk's durable row means + exactly one delivery is ever counted; a duplicate is dropped here rather than + inflating the counters or letting ``chunks_finished`` reach ``chunks_total`` + with real chunks outstanding. Claim and tally commit together — a claim without + its counters would wedge the course ``running`` forever. + + ``tail`` is ``(batch_uuid, course_id, user_ids)`` when a soft-time-limit left a + remainder. It is queued as ``continuation + 1`` of *this* chunk (never a fresh + fan-out index, which would collide with a chunk still in flight) and only by the + delivery that won the claim. A chunk that handed on a tail does not count as + finished — its chain reports once, via its last continuation — so ``chunks_total`` + stays the fan-out's number and is never written here. The course row is locked + and its ``attempt`` re-checked, since a retry can land after the caller's check. + """ + finalized_batch_pk = None + with transaction.atomic(): + state = BulkUnenrollCourseState.objects.select_for_update().get(pk=state_pk) + if state.attempt != attempt: + log.info( + "bulk_unenroll_chunk: course-state %s moved to attempt %s while chunk %s " + "of attempt %s was running; discarding its tally", + state_pk, state.attempt, chunk_index, attempt, + ) + return + + chunk, _ = BulkUnenrollChunk.objects.get_or_create( + course_state_id=state_pk, attempt=attempt, + chunk_index=chunk_index, continuation=continuation, + ) + claimed = BulkUnenrollChunk.objects.filter( + pk=chunk.pk, state=BulkUnenrollChunk.State.PENDING, + ).update( + state=BulkUnenrollChunk.State.FINISHED, + finished=timezone.now(), + modified=timezone.now(), + ) + if not claimed: + log.info( + "bulk_unenroll_chunk: chunk %s of course-state %s already reported in; " + "discarding duplicate tally", + chunk_index, state_pk, + ) + return + + BulkUnenrollCourseState.objects.filter(pk=state_pk).update( + unenrolled=F("unenrolled") + unenrolled, + already_inactive=F("already_inactive") + already_inactive, + failed_count=F("failed_count") + failed, + # A chunk that handed on a tail is not finished: its chain counts once, + # via its last continuation. Only the fan-out ever writes chunks_total. + chunks_finished=F("chunks_finished") + (0 if tail else 1), + modified=timezone.now(), + ) + finalized_batch_pk = _finalize_course_if_complete(state_pk) + + # Publish/settle only after the accounting commits: the tail must not be + # runnable before its row exists, and course+batch locks together can deadlock. + if tail: + batch_uuid, course_id, user_ids = tail + try: + # The tail keeps this chunk's index and takes its next continuation — + # that triple is the identity the worker claims against. + _queue_engine_task(bulk_unenroll_chunk, [ + batch_uuid, course_id, list(user_ids), chunk_index, attempt, continuation + 1, + ]) + except Exception as exc: # pylint: disable=broad-except + # The tail is unreachable (a redelivery discards itself as a duplicate), + # so the chain can never finish — fail the course so retry can re-run it. + log.exception( + "bulk_unenroll_chunk: could not queue the continuation of chunk %s " + "(course-state %s, attempt %s); failing the course", + chunk_index, state_pk, attempt, + ) + _fail_course(state_pk, f"Could not queue remaining learners: {exc}") + raise + if finalized_batch_pk: + _finalize_batch_if_complete(finalized_batch_pk) + + +def _fail_course(state_pk, error): + """ + Mark a course failed outright, without waiting for its chunks to report in. + + Used when the *fan-out itself* breaks: the course then expects work that will + never run, so chunk accounting can never complete it and it would wedge its + batch ``running`` forever — the one state the retry endpoint refuses. ``failed`` + is terminal, lets the batch settle, and is what retry picks up (bumping + ``attempt``, so anything still in flight is discarded). The batch is settled + here too, after the course row's lock is released. + """ + with transaction.atomic(): + state = BulkUnenrollCourseState.objects.select_for_update().get(pk=state_pk) + if state.state in BulkUnenrollCourseState.TERMINAL_STATES: + return + state.state = BulkUnenrollCourseState.State.FAILED + # The column is 255 chars; a broker traceback can be far longer. + state.error = error[:255] + state.finished = timezone.now() + state.save(update_fields=["state", "error", "finished", "modified"]) + batch_pk = state.batch_id + _finalize_batch_if_complete(batch_pk) + + +def _fail_unqueued_courses(course_pks, error): + """ + Fail the specific courses a dispatcher never managed to queue. + + ``_fail_course``'s counterpart one level up: nothing else will ever queue them, + so ``pending`` is a state they can never leave and their batch never settles. + + Only pass courses the caller knows it did not reach — "still ``pending``" is not + the test, since a published course stays pending until a worker claims it. The + update is nonetheless conditional on ``pending``, because a publish error does + not prove the message was undelivered: a course a worker already claimed is left + to run normally. + + Returns the number of courses failed. + """ + if not course_pks: + return 0 + return BulkUnenrollCourseState.objects.filter( + pk__in=course_pks, state=BulkUnenrollCourseState.State.PENDING, + ).update( + state=BulkUnenrollCourseState.State.FAILED, + # The column is 255 chars; a broker traceback can be far longer. + error=error[:255], + finished=timezone.now(), + modified=timezone.now(), + ) + + +@shared_task +@set_code_owner_attribute +def bulk_unenroll_batch(batch_uuid): + """ + Start a whole bulk-unenroll batch — level 1, the dispatcher. + + Marks the batch ``running`` and queues one ``bulk_unenroll_course`` per + still-pending course. Fan-out is intentionally unbounded: concurrency is + governed by the (ideally dedicated) queue's worker count, the same shape + bulk-email uses. + + **Resumable, not exclusive** — it queues whatever is still ``pending``, and + re-queuing a course already under way is harmless because + ``bulk_unenroll_course`` claims its own ``pending -> running`` flip. But nothing + guarantees a redelivery (acked on delivery, no retry policy), so a fan-out that + raises fails the courses it could not queue (``_fail_unqueued_courses``), + leaving a settled batch the operator can retry. + """ + batch = BulkUnenrollBatch.objects.get(uuid=batch_uuid) + # `modified` is set explicitly at every queryset .update() in this feature: + # TimeStampedModel maintains it on save(), not on UPDATE. + started = BulkUnenrollBatch.objects.filter( + pk=batch.pk, + state__in=(BulkUnenrollBatch.State.PENDING, BulkUnenrollBatch.State.RUNNING), + ).update(state=BulkUnenrollBatch.State.RUNNING, modified=timezone.now()) + if not started: + # Cancelled or already finished — there is nothing left to dispatch. + log.info( + "bulk_unenroll_batch: batch %s is not dispatchable (state=%s); nothing queued", + batch_uuid, batch.state, + ) + return + + # Materialized up front: queued courses are still 'pending' too, so position + # in this list — not row state — separates "queued" from "never reached". + pending = list( + batch.courses + .filter(state=BulkUnenrollCourseState.State.PENDING) + .values_list("pk", "course_id") + ) + queued = 0 + try: + for _, course_id in pending: + # Cheap EXISTS check: stop queuing as soon as a cancel lands. + if BulkUnenrollBatch.objects.filter( + pk=batch.pk, state=BulkUnenrollBatch.State.CANCELLED, + ).exists(): + break + _queue_engine_task(bulk_unenroll_course, [batch_uuid, str(course_id)]) + queued += 1 + except Exception as exc: # pylint: disable=broad-except + # Nothing else will ever queue the unreached courses, so fail them. The + # slice starts at the ambiguous raiser; _fail_unqueued_courses is safe there. + unreached = [pk for pk, _ in pending[queued:]] + log.exception( + "bulk_unenroll_batch: fan-out of batch %s failed after queueing %s of %s " + "course(s); failing the %s it never reached", + batch_uuid, queued, len(pending), len(unreached), + ) + _fail_unqueued_courses(unreached, f"Could not queue unenroll work: {exc}") + _finalize_batch_if_complete(batch.pk) + raise + + _finalize_batch_if_complete(batch.pk) + + +def _chunk_stop_reason(state_pk, attempt): + """ + Why this chunk must not (or must no longer) touch enrollments — or ``None``. + + One query for the three ways a chunk's mandate can be revoked: the batch was + cancelled; a retry bumped ``attempt``, superseding this chunk's generation; or + the course already settled — most importantly as ``failed``, which a broken + fan-out sets while chunks it did publish are still queued. That last case is + what stops a destructive operation from outliving the status the operator sees. + + Called up front *and* every ``BULK_UNENROLL_CANCEL_CHECK_EVERY`` learners, so a + mid-chunk revocation is honoured within a bounded number of learners. + """ + row = ( + BulkUnenrollCourseState.objects + .filter(pk=state_pk) + .values_list("state", "attempt", "batch__state") + .first() + ) + if row is None: + return "its course-state row no longer exists" + course_state, current_attempt, batch_state = row + if batch_state == BulkUnenrollBatch.State.CANCELLED: + return "the batch was cancelled" + if current_attempt != attempt: + return f"attempt {attempt} was superseded by attempt {current_attempt}" + if course_state in BulkUnenrollCourseState.TERMINAL_STATES: + return f"the course already settled as '{course_state}'" + return None + + +def _queue_engine_task(task, args): + """Publish one engine task on the (optionally dedicated) bulk-unenroll queue.""" + task.apply_async(args=args, routing_key=settings.BULK_UNENROLL_ROUTING_KEY) + + +def _finalize_batch_if_complete(batch_pk): + """ + Mark the batch done once every one of its courses has reached a terminal state. + + Claimed transactionally (``select_for_update`` + a running-only guard) so that + when courses finish concurrently, exactly one of them derives the batch's final + state: ``succeeded`` (no course failed), ``failed`` (none succeeded), or + ``partial`` (a mix). Skipped courses count as neither. + """ + with transaction.atomic(): + batch = BulkUnenrollBatch.objects.select_for_update().get(pk=batch_pk) + if batch.state != BulkUnenrollBatch.State.RUNNING: + return + course_states = list(batch.courses.values_list("state", flat=True)) + if any(cs not in BulkUnenrollCourseState.TERMINAL_STATES for cs in course_states): + return + failed = sum(1 for cs in course_states if cs == BulkUnenrollCourseState.State.FAILED) + succeeded = sum(1 for cs in course_states if cs == BulkUnenrollCourseState.State.SUCCEEDED) + if failed == 0: + batch.state = BulkUnenrollBatch.State.SUCCEEDED + elif succeeded == 0: + batch.state = BulkUnenrollBatch.State.FAILED + else: + batch.state = BulkUnenrollBatch.State.PARTIAL + batch.save(update_fields=["state", "modified"]) + + +@shared_task +@set_code_owner_attribute +def bulk_unenroll_course(batch_uuid, course_id): + """ + Fan a single course out into chunk tasks — level 2, where the enrollment fetch + happens (on the worker). + + Streams the course's *active* enrollment user ids with + ``values_list(...).iterator()`` so a 50k-enrollment course is never materialized, + queues a ``bulk_unenroll_chunk`` per ``BULK_UNENROLL_CHUNK_SIZE`` group, and + records ``total_enrollments`` / ``chunks_total`` for the chunk-level finalizer. + A course with nothing active finalizes ``succeeded`` immediately. + + The fan-out is **claimed** via the ``pending -> running`` flip, so a redelivery + cannot queue a second chunk set. That same flip means a redelivery would skip a + *broken* fan-out, so one that raises fails the course on its way out + (``_fail_course``) instead of leaving it ``running`` with no ``chunks_total``. + """ + batch = BulkUnenrollBatch.objects.get(uuid=batch_uuid) + course_key = CourseKey.from_string(course_id) + state = BulkUnenrollCourseState.objects.get(batch=batch, course_id=course_key) + + claimed = BulkUnenrollCourseState.objects.filter( + pk=state.pk, state=BulkUnenrollCourseState.State.PENDING, + ).update( + state=BulkUnenrollCourseState.State.RUNNING, + started=timezone.now(), + modified=timezone.now(), + ) + if not claimed: + log.info( + "bulk_unenroll_course: course %s in batch %s is not pending (state=%s); " + "skipping duplicate fan-out", + course_id, batch_uuid, state.state, + ) + return + + # Read back the generation this fan-out belongs to: every chunk it queues + # carries it, so a straggler from an earlier attempt can be told apart. + attempt = BulkUnenrollCourseState.objects.values_list("attempt", flat=True).get(pk=state.pk) + + chunk_size = settings.BULK_UNENROLL_CHUNK_SIZE + active_user_ids = ( + CourseEnrollment.objects + .filter(course_id=course_key, is_active=True) + .values_list("user_id", flat=True) + ) + + total = 0 + chunks = 0 + buffer = [] + try: + for user_id in active_user_ids.iterator(chunk_size=2000): + buffer.append(user_id) + total += 1 + if len(buffer) >= chunk_size: + _queue_engine_task(bulk_unenroll_chunk, [batch_uuid, course_id, buffer, chunks, attempt, 0]) + chunks += 1 + buffer = [] + if buffer: + _queue_engine_task(bulk_unenroll_chunk, [batch_uuid, course_id, buffer, chunks, attempt, 0]) + chunks += 1 + + BulkUnenrollCourseState.objects.filter(pk=state.pk).update( + total_enrollments=total, + chunks_total=chunks, + modified=timezone.now(), + ) + except Exception as exc: # pylint: disable=broad-except + # A fan-out that dies partway never records chunks_total, and the claim + # means a redelivery skips it — fail the course so retry can re-run it. + log.exception( + "bulk_unenroll_course: fan-out of course %s in batch %s failed after " + "queueing %s chunk(s); failing the course", + course_id, batch_uuid, chunks, + ) + _fail_course(state.pk, f"Could not queue unenroll work: {exc}") + raise + + if chunks == 0: + # Nothing to unenroll — done, but only while this task's claim still holds: + # a cancel landing mid-fan-out has already settled the course terminally. + settled = BulkUnenrollCourseState.objects.filter( + pk=state.pk, state=BulkUnenrollCourseState.State.RUNNING, + ).update( + state=BulkUnenrollCourseState.State.SUCCEEDED, + finished=timezone.now(), + modified=timezone.now(), + ) + if settled: + _finalize_batch_if_complete(batch.pk) + else: + # Covers the case where every chunk has already finished (e.g. eager mode) + # by the time chunks_total is recorded. + finalized_batch_pk = _finalize_course_if_complete(state.pk) + if finalized_batch_pk: + _finalize_batch_if_complete(finalized_batch_pk) + + +def _finalize_course_if_complete(state_pk): + """ + Mark a course done once every one of its chunks has reported in. + + The finalizer is claimed transactionally (``select_for_update`` + a + not-already-terminal guard) so that when concurrent chunks finish together, + exactly one of them flips the course to its terminal state — the reliable + "is the course truly finished?" primitive. + + Returns the batch pk when *this* call finalized the course, so the caller can + settle the batch after the course row's lock is released; ``None`` otherwise. + """ + with transaction.atomic(): + state = BulkUnenrollCourseState.objects.select_for_update().get(pk=state_pk) + if not state.chunks_total or state.chunks_finished < state.chunks_total: + return None + if state.state in BulkUnenrollCourseState.TERMINAL_STATES: + return None + state.state = ( + BulkUnenrollCourseState.State.SUCCEEDED + if state.failed_count == 0 + else BulkUnenrollCourseState.State.FAILED + ) + state.finished = timezone.now() + state.save(update_fields=["state", "finished", "modified"]) + return state.batch_id + def update_audit_status(audit_instance, status): audit_instance.status = status diff --git a/lms/djangoapps/support/tests/test_bulk_unenroll_tasks.py b/lms/djangoapps/support/tests/test_bulk_unenroll_tasks.py new file mode 100644 index 000000000000..26959059ae47 --- /dev/null +++ b/lms/djangoapps/support/tests/test_bulk_unenroll_tasks.py @@ -0,0 +1,853 @@ +""" +Tests for the bulk-unenroll Celery engine (lms/djangoapps/support/tasks.py). +""" +from unittest.mock import patch + +from celery.exceptions import SoftTimeLimitExceeded +from django.db import connection +from django.test import TestCase, override_settings +from django.test.utils import CaptureQueriesContext + +from common.djangoapps.student.models.course_enrollment import ( + ENROLLED_TO_UNENROLLED, + CourseEnrollment, + ManualEnrollmentAudit, +) +from common.djangoapps.student.tests.factories import UserFactory +from lms.djangoapps.support.models import BulkUnenrollBatch, BulkUnenrollChunk, BulkUnenrollCourseState +from lms.djangoapps.support.tasks import ( + _chunk_stop_reason, + _finalize_batch_if_complete, + bulk_unenroll_batch, + bulk_unenroll_chunk, + bulk_unenroll_course, +) +from openedx.core.djangoapps.content.course_overviews.tests.factories import CourseOverviewFactory + + +class BulkUnenrollTaskTestCase(TestCase): + """ + One batch holding one course, plus the helpers every engine test needs. + Subclasses declare the situation under test via the class attributes below. + """ + + batch_state = BulkUnenrollBatch.State.RUNNING + course_state = BulkUnenrollCourseState.State.PENDING + chunks_total = 0 + + def setUp(self): + super().setUp() + self.requester = UserFactory() + self.course = CourseOverviewFactory.create(org="edX", run="A", display_name="A") + self.batch = BulkUnenrollBatch.objects.create( + requester=self.requester, reason="offboarding", total_courses=1, + state=self.batch_state, + ) + self.state = BulkUnenrollCourseState.objects.create( + batch=self.batch, course_id=self.course.id, + state=self.course_state, chunks_total=self.chunks_total, + ) + + def _enroll(self, count, is_active=True): + """Create `count` learners enrolled in `self.course`; return them.""" + users = [UserFactory() for _ in range(count)] + for user in users: + CourseEnrollment.objects.create( + user=user, course_id=self.course.id, is_active=is_active, + ) + return users + + def _active(self): + return CourseEnrollment.objects.filter(course_id=self.course.id, is_active=True).count() + + def _refresh(self): + self.state.refresh_from_db() + self.batch.refresh_from_db() + return self.state + + def _run_chunk(self, users, chunk_index=0, attempt=1): + bulk_unenroll_chunk( + str(self.batch.uuid), str(self.course.id), [u.id for u in users], + chunk_index, attempt, + ) + + def _timeout_on(self, victim): + """Patch unenroll so it raises Celery's soft-timeout when it reaches `victim`.""" + real_unenroll = CourseEnrollment.unenroll + + def timing_out(user, course_id, skip_refund=False): + if user.id == victim.id: + raise SoftTimeLimitExceeded() + return real_unenroll(user, course_id, skip_refund=skip_refund) + + return patch.object(CourseEnrollment, "unenroll", side_effect=timing_out) + + +class BulkUnenrollChunkTaskTest(BulkUnenrollTaskTestCase): + """The level-3 chunk task: actually deactivates a batch of learners.""" + + course_state = BulkUnenrollCourseState.State.RUNNING + chunks_total = 1 + + def test_unenrolls_active_learners_and_writes_audit(self): + users = self._enroll(3) + self._run_chunk(users) + + self.assertEqual(self._active(), 0) + audits = ManualEnrollmentAudit.objects.filter(enrollment__course_id=self.course.id) + self.assertEqual(audits.count(), 3) + for audit in audits: + self.assertEqual(audit.state_transition, ENROLLED_TO_UNENROLLED) + self.assertEqual(audit.reason, "offboarding") + self.assertEqual(audit.enrolled_by, self.requester) + self.assertEqual(self._refresh().unenrolled, 3) + + def test_skip_refund_true_is_passed_to_unenroll(self): + """Refunds are never issued for a bulk run.""" + users = self._enroll(2) + real_unenroll = CourseEnrollment.unenroll + seen = [] + + def spy(user, course_id, skip_refund=False): + seen.append(skip_refund) + return real_unenroll(user, course_id, skip_refund=skip_refund) + + with patch.object(CourseEnrollment, "unenroll", side_effect=spy): + self._run_chunk(users) + + self.assertEqual(seen, [True, True]) + self.assertEqual(self._active(), 0) + + def test_one_failing_learner_does_not_abort_the_chunk(self): + users = self._enroll(3) + bad = users[1] + real_unenroll = CourseEnrollment.unenroll + + def flaky(user, course_id, skip_refund=False): + if user.id == bad.id: + raise ValueError("boom") + return real_unenroll(user, course_id, skip_refund=skip_refund) + + with patch.object(CourseEnrollment, "unenroll", side_effect=flaky): + self._run_chunk(users) + + state = self._refresh() + self.assertEqual(state.unenrolled, 2) + self.assertEqual(state.failed_count, 1) + self.assertEqual(state.state, "failed") + self.assertTrue( + CourseEnrollment.objects.get(user=bad, course_id=self.course.id).is_active + ) + + def test_already_inactive_learners_counted_not_unenrolled(self): + """The is_active filter is what makes every level of the engine re-runnable.""" + users = self._enroll(2, is_active=False) + self._run_chunk(users) + + state = self._refresh() + self.assertEqual(state.unenrolled, 0) + self.assertEqual(state.already_inactive, 2) + self.assertEqual(ManualEnrollmentAudit.objects.count(), 0) + + def test_duplicate_delivery_of_one_chunk_reports_in_exactly_once(self): + """ + Otherwise a redelivery pushes chunks_finished to chunks_total while a + different chunk has never run, succeeding a course with learners enrolled. + """ + self.state.chunks_total = 2 # chunk 1 will never run in this test + self.state.save() + users = self._enroll(2) + self._run_chunk(users, chunk_index=0) + self._run_chunk(users, chunk_index=0) # duplicate delivery of the SAME chunk + + state = self._refresh() + self.assertEqual(state.chunks_finished, 1) + self.assertEqual(state.unenrolled, 2) + self.assertEqual(state.already_inactive, 0) # the redelivery's tally is discarded + self.assertEqual(state.state, "running") # not finalized: chunk 1 is outstanding + self.assertIsNone(state.finished) + + def test_a_failed_counter_update_leaves_the_chunk_unclaimed(self): + """ + A claim without its tally would leave the chunk finished but uncounted, and + every redelivery would then discard itself as a duplicate — course wedged. + """ + users = self._enroll(2) + boom = Exception("counter update died") + real_filter = BulkUnenrollCourseState.objects.filter + + def fail_on_counter_update(*args, **kwargs): + queryset = real_filter(*args, **kwargs) + original_update = queryset.update + + def exploding_update(**fields): + if "chunks_finished" in fields: + raise boom + return original_update(**fields) + + queryset.update = exploding_update + return queryset + + with patch.object(BulkUnenrollCourseState.objects, "filter", side_effect=fail_on_counter_update): + with self.assertRaises(Exception): + self._run_chunk(users, chunk_index=0) + + # The ledger must not remember a chunk whose tally was never recorded. + self.assertFalse( + BulkUnenrollChunk.objects.filter( + course_state=self.state, chunk_index=0, + state=BulkUnenrollChunk.State.FINISHED, + ).exists() + ) + # ...so a redelivery can still count it. + self._run_chunk(users, chunk_index=0) + state = self._refresh() + self.assertEqual(state.chunks_finished, 1) + self.assertEqual(state.already_inactive, 2) # they were removed by the failed run + + def test_chunk_from_a_superseded_attempt_is_discarded(self): + """ + Its learner set is stale, and claiming chunk 0 of the new attempt would get + the real chunk dropped as a duplicate. + """ + users = self._enroll(2) + BulkUnenrollCourseState.objects.filter(pk=self.state.pk).update(attempt=2) + + self._run_chunk(users, chunk_index=0, attempt=1) # superseded generation + + self.assertEqual(self._active(), 2) + self.assertEqual(self._refresh().chunks_finished, 0) + self.assertFalse(BulkUnenrollChunk.objects.filter(course_state=self.state).exists()) + + def test_attempt_bumped_while_the_chunk_runs_discards_its_tally(self): + """ + The attempt is checked before the work and acted on after it, so a retry can + land in between — its counters belong to a different fan-out. + """ + users = self._enroll(2) + real_unenroll = CourseEnrollment.unenroll + + def retry_lands_mid_chunk(user, course_id, skip_refund=False): + BulkUnenrollCourseState.objects.filter(pk=self.state.pk).update(attempt=2) + return real_unenroll(user, course_id, skip_refund=skip_refund) + + with patch.object(CourseEnrollment, "unenroll", side_effect=retry_lands_mid_chunk): + self._run_chunk(users, chunk_index=0, attempt=1) + + state = self._refresh() + self.assertEqual(state.chunks_finished, 0) # nothing counted for attempt 2 + self.assertEqual(state.unenrolled, 0) + self.assertEqual(state.state, "running") + + def test_soft_time_limit_hands_on_the_tail_without_failing_the_course(self): + """ + A timeout is not a bad learner, and the chain it starts counts as finished + only when its last continuation reports — chunks_total stays the fan-out's. + """ + users = self._enroll(3) + with self._timeout_on(users[1]): + with patch("lms.djangoapps.support.tasks.bulk_unenroll_chunk.apply_async") as mock_apply: + self._run_chunk(users, chunk_index=0) + + state = self._refresh() + self.assertEqual(state.failed_count, 0) # a timeout is not a learner failure + self.assertEqual(state.chunks_total, 1) # only the fan-out writes the denominator + self.assertEqual(state.chunks_finished, 0) # the chain has not finished yet + self.assertEqual(state.state, "running") + self.assertIsNone(state.finished) + + args = mock_apply.call_args.kwargs["args"] + still_active = set( + CourseEnrollment.objects + .filter(course_id=self.course.id, is_active=True) + .values_list("user_id", flat=True) + ) + self.assertEqual(set(args[2]), still_active) # exactly the learners not yet done + # Numbered within its own chunk: a fresh fan-out index would collide with a + # sibling that is queued but has not reported yet. + self.assertEqual(args[3], 0) # same chunk_index... + self.assertEqual(args[5], 1) # ...next continuation + + def test_timeout_reporting_before_the_fan_out_records_totals_does_not_finalize(self): + """ + A chunk can time out while the fan-out is still streaming a huge course, so + its report can land while chunks_total is still 0. Counting the tail into + the denominator there would finalize the course at 1/1 — reported succeeded + with real chunks still being queued, all then discarded as settled. + """ + BulkUnenrollCourseState.objects.filter(pk=self.state.pk).update(chunks_total=0) + users = self._enroll(3) + with self._timeout_on(users[1]): + with patch("lms.djangoapps.support.tasks.bulk_unenroll_chunk.apply_async"): + self._run_chunk(users, chunk_index=0) + + state = self._refresh() + self.assertEqual(state.state, "running") # not finalized out from under the fan-out + self.assertEqual(state.chunks_total, 0) + self.assertEqual(state.chunks_finished, 0) + + def test_course_finalizes_only_when_the_last_chunk_reports(self): + self.state.chunks_total = 2 + self.state.save() + first, second = self._enroll(1), self._enroll(1) + + self._run_chunk(first, chunk_index=0) + state = self._refresh() + self.assertEqual(state.chunks_finished, 1) + self.assertEqual(state.state, "running") # one chunk still outstanding + self.assertIsNone(state.finished) + + self._run_chunk(second, chunk_index=1) + state = self._refresh() + self.assertEqual(state.chunks_finished, 2) + self.assertEqual(state.unenrolled, 2) # F() accumulation, no clobbering + self.assertEqual(state.state, "succeeded") + self.assertIsNotNone(state.finished) + + +class BulkUnenrollCourseTaskTest(BulkUnenrollTaskTestCase): + """The level-2 per-course task: fetches enrollments and fans out chunks.""" + + def _run(self): + bulk_unenroll_course(str(self.batch.uuid), str(self.course.id)) + + @override_settings(BULK_UNENROLL_CHUNK_SIZE=2) + def test_splits_active_enrollments_into_chunks(self): + self._enroll(5) + with patch("lms.djangoapps.support.tasks.bulk_unenroll_chunk.apply_async") as mock_apply: + self._run() + + self.assertEqual(mock_apply.call_count, 3) # 2 + 2 + 1 + sizes = sorted(len(c.kwargs["args"][2]) for c in mock_apply.call_args_list) + self.assertEqual(sizes, [1, 2, 2]) + state = self._refresh() + self.assertEqual(state.total_enrollments, 5) + self.assertEqual(state.chunks_total, 3) + self.assertEqual(state.state, "running") + self.assertIsNotNone(state.started) + + @override_settings(BULK_UNENROLL_CHUNK_SIZE=2) + def test_only_active_enrollments_are_queued(self): + self._enroll(3, is_active=True) + self._enroll(2, is_active=False) + with patch("lms.djangoapps.support.tasks.bulk_unenroll_chunk.apply_async") as mock_apply: + self._run() + + queued = sum(len(c.kwargs["args"][2]) for c in mock_apply.call_args_list) + self.assertEqual(queued, 3) + self.assertEqual(self._refresh().total_enrollments, 3) + + def test_empty_course_finalizes_succeeded_without_queuing(self): + with patch("lms.djangoapps.support.tasks.bulk_unenroll_chunk.apply_async") as mock_apply: + self._run() + + mock_apply.assert_not_called() + state = self._refresh() + self.assertEqual(state.chunks_total, 0) + self.assertEqual(state.state, "succeeded") + self.assertIsNotNone(state.finished) + + def test_an_empty_course_cancelled_mid_fan_out_is_not_finalized_succeeded(self): + """A cancel landing after the claim has already settled the course.""" + real_values_list = BulkUnenrollCourseState.objects.values_list + + def cancel_lands_after_the_claim(*args, **kwargs): + BulkUnenrollCourseState.objects.filter(pk=self.state.pk).update( + state=BulkUnenrollCourseState.State.CANCELLED, + ) + return real_values_list(*args, **kwargs) + + with patch.object( + BulkUnenrollCourseState.objects, "values_list", + side_effect=cancel_lands_after_the_claim, + ): + self._run() + + state = self._refresh() + self.assertEqual(state.state, "cancelled") # not resurrected as succeeded + self.assertIsNone(state.finished) + + @override_settings(BULK_UNENROLL_CHUNK_SIZE=2) + def test_duplicate_course_delivery_fans_out_once(self): + """ + A second chunk set would report in against the same chunks_total and + finalize the course before the first set had finished. + """ + self._enroll(5) + with patch("lms.djangoapps.support.tasks.bulk_unenroll_chunk.apply_async") as mock_apply: + self._run() + self._run() # duplicate delivery + + self.assertEqual(mock_apply.call_count, 3) # 2 + 2 + 1, queued once + self.assertEqual(self._refresh().chunks_total, 3) + + @override_settings(BULK_UNENROLL_CHUNK_SIZE=2) + def test_end_to_end_unenrolls_all_and_finalizes(self): + self._enroll(5) + queued = [] + with patch( + "lms.djangoapps.support.tasks.bulk_unenroll_chunk.apply_async", + side_effect=lambda *a, **k: queued.append(k["args"]), + ): + self._run() + + # Chunks execute after the course task has finished queuing them. + for args in queued: + bulk_unenroll_chunk(*args) + + self.assertEqual(self._active(), 0) + state = self._refresh() + self.assertEqual(state.unenrolled, 5) + self.assertEqual(state.chunks_finished, 3) + self.assertEqual(state.state, "succeeded") + + def test_query_count_independent_of_enrollment_count(self): + """The worker streams user ids; it must not query per enrollment.""" + def run_for(count): + course = CourseOverviewFactory.create(org="edX", run=f"R{count}", display_name=f"C{count}") + batch = BulkUnenrollBatch.objects.create( + requester=self.requester, total_courses=1, state=BulkUnenrollBatch.State.RUNNING, + ) + BulkUnenrollCourseState.objects.create(batch=batch, course_id=course.id) + for _ in range(count): + CourseEnrollment.objects.create(user=UserFactory(), course_id=course.id, is_active=True) + with patch("lms.djangoapps.support.tasks.bulk_unenroll_chunk.apply_async"): + with CaptureQueriesContext(connection) as ctx: + bulk_unenroll_course(str(batch.uuid), str(course.id)) + return len(ctx.captured_queries) + + run_for(2) # warm caches + self.assertEqual(run_for(3), run_for(30)) + + +class BulkUnenrollBatchDispatcherTest(TestCase): + """The level-1 dispatcher + the batch-level finalizer.""" + + def setUp(self): + super().setUp() + self.requester = UserFactory() + + def _batch_with_courses(self, count, course_state=BulkUnenrollCourseState.State.PENDING): + """Build a pending batch with `count` courses; return (batch, courses).""" + batch = BulkUnenrollBatch.objects.create( + requester=self.requester, total_courses=count, state=BulkUnenrollBatch.State.PENDING, + ) + courses = [] + for i in range(count): + course = CourseOverviewFactory.create(org="edX", run=f"D{i}", display_name=f"D{i}") + BulkUnenrollCourseState.objects.create(batch=batch, course_id=course.id, state=course_state) + courses.append(course) + return batch, courses + + def _finalize_states(self, course_states): + """Build a running batch whose courses already sit in `course_states`.""" + batch = BulkUnenrollBatch.objects.create( + requester=self.requester, total_courses=len(course_states), + state=BulkUnenrollBatch.State.RUNNING, + ) + for i, cstate in enumerate(course_states): + course = CourseOverviewFactory.create(org="edX", run=f"F{i}", display_name=f"F{i}") + BulkUnenrollCourseState.objects.create(batch=batch, course_id=course.id, state=cstate) + _finalize_batch_if_complete(batch.pk) + batch.refresh_from_db() + return batch.state + + def test_marks_running_and_queues_one_task_per_course(self): + batch, courses = self._batch_with_courses(3) + with patch("lms.djangoapps.support.tasks.bulk_unenroll_course.apply_async") as mock_apply: + bulk_unenroll_batch(str(batch.uuid)) + + self.assertEqual(mock_apply.call_count, 3) + queued = sorted(c.kwargs["args"][1] for c in mock_apply.call_args_list) + self.assertEqual(queued, sorted(str(co.id) for co in courses)) + batch.refresh_from_db() + self.assertEqual(batch.state, "running") + + def test_duplicate_dispatcher_delivery_fans_each_course_out_once(self): + """ + Re-queuing a course task is cheap and is how a half-finished fan-out + resumes; fanning one out into chunks twice is not. + """ + batch, _ = self._batch_with_courses(3) + for course_state in batch.courses.all(): + CourseEnrollment.objects.create( + user=UserFactory(), course_id=course_state.course_id, is_active=True, + ) + + course_args = [] + with patch( + "lms.djangoapps.support.tasks.bulk_unenroll_course.apply_async", + side_effect=lambda *a, **k: course_args.append(k["args"]), + ): + bulk_unenroll_batch(str(batch.uuid)) + bulk_unenroll_batch(str(batch.uuid)) # duplicate delivery + + chunk_args = [] + with patch( + "lms.djangoapps.support.tasks.bulk_unenroll_chunk.apply_async", + side_effect=lambda *a, **k: chunk_args.append(k["args"]), + ): + for args in course_args: + bulk_unenroll_course(*args) + + self.assertEqual(len(chunk_args), 3) # one per course, however many deliveries + + def test_dispatcher_redelivery_resumes_a_half_finished_fan_out(self): + """ + A worker that dies mid-queue leaves the batch 'running' with courses + 'pending'; an exclusive claim would strand them forever. + """ + batch, _ = self._batch_with_courses(3) + # Simulate a fan-out that got one course out before dying. + BulkUnenrollBatch.objects.filter(pk=batch.pk).update(state=BulkUnenrollBatch.State.RUNNING) + first = batch.courses.order_by("pk").first() + BulkUnenrollCourseState.objects.filter(pk=first.pk).update( + state=BulkUnenrollCourseState.State.RUNNING, + ) + + with patch("lms.djangoapps.support.tasks.bulk_unenroll_course.apply_async") as mock_apply: + bulk_unenroll_batch(str(batch.uuid)) + + queued = sorted(c.kwargs["args"][1] for c in mock_apply.call_args_list) + still_pending = sorted( + str(cs.course_id) + for cs in batch.courses.filter(state=BulkUnenrollCourseState.State.PENDING) + ) + self.assertEqual(queued, still_pending) # exactly the unfinished work + self.assertEqual(len(queued), 2) + + def test_batch_state_is_derived_from_its_courses(self): + """Succeeded when none failed, failed when none succeeded, partial in between.""" + for course_states, expected in [ + (["succeeded", "succeeded"], "succeeded"), + (["failed", "failed"], "failed"), + (["succeeded", "failed"], "partial"), + (["succeeded", "pending"], "running"), # not settled while work remains + ]: + with self.subTest(course_states=course_states): + self.assertEqual(self._finalize_states(course_states), expected) + + def test_dispatch_end_to_end_unenrolls_all_and_finalizes(self): + batch, courses = self._batch_with_courses(2) + for course in courses: + for _ in range(2): + CourseEnrollment.objects.create(user=UserFactory(), course_id=course.id, is_active=True) + + course_args = [] + with patch( + "lms.djangoapps.support.tasks.bulk_unenroll_course.apply_async", + side_effect=lambda *a, **k: course_args.append(k["args"]), + ): + bulk_unenroll_batch(str(batch.uuid)) + + for cargs in course_args: + chunk_args = [] + with patch( + "lms.djangoapps.support.tasks.bulk_unenroll_chunk.apply_async", + side_effect=lambda *a, _sink=chunk_args, **k: _sink.append(k["args"]), + ): + bulk_unenroll_course(*cargs) + for chargs in chunk_args: + bulk_unenroll_chunk(*chargs) + + batch.refresh_from_db() + self.assertEqual(batch.state, "succeeded") + self.assertEqual( + CourseEnrollment.objects.filter( + course_id__in=[c.id for c in courses], is_active=True, + ).count(), + 0, + ) + + +class BulkUnenrollCancellationTest(BulkUnenrollTaskTestCase): + """Cancellation is honoured by the dispatcher and the chunk worker.""" + + batch_state = BulkUnenrollBatch.State.CANCELLED + chunks_total = 1 + + def test_chunk_removes_nobody_when_batch_already_cancelled(self): + self._run_chunk(self._enroll(3)) + self.assertEqual(self._active(), 3) + self.assertEqual(self._refresh().unenrolled, 0) + + @override_settings(BULK_UNENROLL_CANCEL_CHECK_EVERY=2) + def test_chunk_stops_partway_when_cancelled_midway(self): + users = self._enroll(4) + # Start check clear (proceed), then cancelled on the check after learner 2. + with patch( + "lms.djangoapps.support.tasks._chunk_stop_reason", + side_effect=[None, "the batch was cancelled"], + ): + self._run_chunk(users) + self.assertEqual(4 - self._active(), 2) # stopped after the 2nd + + @override_settings(BULK_UNENROLL_CANCEL_CHECK_EVERY=0) + def test_a_zero_check_interval_still_honours_a_cancel_mid_chunk(self): + """ + 0 must degrade to checking every learner, never to checking none: the check + bounds how many more learners a revoked chunk can still unenroll (and a + divide-by-zero would escape uncaught, leaving the chunk never recorded). + """ + users = self._enroll(4) + with patch( + "lms.djangoapps.support.tasks._chunk_stop_reason", + side_effect=[None, "the batch was cancelled"], + ): + self._run_chunk(users) + self.assertEqual(4 - self._active(), 1) # stopped after the 1st + + def test_dispatcher_queues_nothing_and_stays_cancelled(self): + with patch("lms.djangoapps.support.tasks.bulk_unenroll_course.apply_async") as mock_apply: + bulk_unenroll_batch(str(self.batch.uuid)) + mock_apply.assert_not_called() + self.batch.refresh_from_db() + self.assertEqual(self.batch.state, "cancelled") + + +class BulkUnenrollBrokerFailureTest(BulkUnenrollTaskTestCase): + """ + What happens when the broker, not the work, is what breaks. + + Both levels publish *after* claiming their work, so a publish that never lands + leaves the course waiting on chunks that will never run — and the claim is what + stops a redelivery repairing it. The engine has to settle the course itself, or + the batch never reaches the settled state the retry endpoint requires. + """ + + def _second_course(self): + """Add a second course to self.batch and return its state row.""" + course = CourseOverviewFactory.create(org="edX", run="B", display_name="B") + self.batch.total_courses = 2 + self.batch.save() + return BulkUnenrollCourseState.objects.create(batch=self.batch, course_id=course.id) + + # --- level 2: the per-course fan-out --- + + @override_settings(BULK_UNENROLL_CHUNK_SIZE=2) + def test_fan_out_that_dies_midway_fails_the_course_and_settles_the_batch(self): + """ + The course was claimed 'running' before the first publish, so a redelivery + skips it and nothing can ever record chunks_total. + """ + self._enroll(5) # 3 chunks at size 2 + with patch( + "lms.djangoapps.support.tasks.bulk_unenroll_chunk.apply_async", + side_effect=[None, OSError("broker gone"), None], + ): + with self.assertRaises(OSError): + bulk_unenroll_course(str(self.batch.uuid), str(self.course.id)) + + state = self._refresh() + self.assertEqual(state.state, "failed") + self.assertIn("Could not queue", state.error) + self.assertIsNotNone(state.finished) + # chunks_total was never recorded, so no chunk count can finalize this course. + self.assertEqual(state.chunks_total, 0) + self.assertEqual(self.batch.state, "failed") # settled, so retry can reach it + + @override_settings(BULK_UNENROLL_CHUNK_SIZE=2) + def test_chunks_already_queued_stop_once_the_broken_fan_out_fails_the_course(self): + """ + The chunks the fan-out *did* publish are still on the queue. They must not + deactivate enrollments after the API has reported the batch finished. + """ + users = self._enroll(4) + with patch( + "lms.djangoapps.support.tasks.bulk_unenroll_chunk.apply_async", + side_effect=[None, OSError("broker gone")], + ): + with self.assertRaises(OSError): + bulk_unenroll_course(str(self.batch.uuid), str(self.course.id)) + + self.assertEqual(self._refresh().state, "failed") + self.assertEqual(self.batch.state, "failed") + + # Now deliver the chunk that really was published, as a worker would. + self._run_chunk(users[:2], chunk_index=0) + self.assertEqual(self._active(), 4) # nobody removed post-settlement + self.assertEqual(self._refresh().unenrolled, 0) + + def test_a_chunk_stops_mid_pass_when_a_retry_supersedes_its_attempt(self): + """ + Left running, the old chunk keeps unenrolling into a generation whose + counters no longer accept its tally — work done, never reported. + """ + BulkUnenrollCourseState.objects.filter(pk=self.state.pk).update( + state=BulkUnenrollCourseState.State.RUNNING, chunks_total=1, + ) + users = self._enroll(4) + state_pk = self.state.pk + real_stop_reason = _chunk_stop_reason + calls = [] + + def bump_attempt_after_first_check(pk, attempt): + calls.append(pk) + reason = real_stop_reason(pk, attempt) + if len(calls) == 1: + # The up-front check has passed; a retry now moves the course to a + # new generation, and the next mid-pass check has to notice. + BulkUnenrollCourseState.objects.filter(pk=state_pk).update(attempt=2) + return reason + + with override_settings(BULK_UNENROLL_CANCEL_CHECK_EVERY=2): + with patch( + "lms.djangoapps.support.tasks._chunk_stop_reason", + side_effect=bump_attempt_after_first_check, + ): + self._run_chunk(users, chunk_index=0) + + self.assertEqual(4 - self._active(), 2) # stopped at the first check + # Its tally is discarded anyway, so every extra learner is unreported work. + self.assertEqual(self._refresh().unenrolled, 0) + + # --- level 3: the timeout continuation --- + + def test_continuation_that_cannot_be_queued_fails_the_course(self): + """ + This chunk's ledger row is 'finished', so no redelivery will re-queue the + tail — a chain whose last link never runs can never report finished. + """ + BulkUnenrollCourseState.objects.filter(pk=self.state.pk).update( + state=BulkUnenrollCourseState.State.RUNNING, chunks_total=1, + ) + users = self._enroll(3) + + with self._timeout_on(users[1]): + with patch( + "lms.djangoapps.support.tasks.bulk_unenroll_chunk.apply_async", + side_effect=OSError("broker gone"), + ): + with self.assertRaises(OSError): + self._run_chunk(users, chunk_index=0) + + state = self._refresh() + self.assertEqual(state.state, "failed") + self.assertIn("Could not queue", state.error) + self.assertEqual(self.batch.state, "failed") + # The chain never reported finished; only the failure releases the course. + self.assertEqual(state.chunks_total, 1) + self.assertEqual(state.chunks_finished, 0) + + # --- level 1: the dispatcher --- + + def test_dispatcher_that_cannot_queue_fails_the_courses_it_never_reached(self): + """ + The dispatcher is acked on delivery with no retry, so a course it never + queued would sit 'pending' forever and its batch would never settle. + """ + second = self._second_course() + with patch( + "lms.djangoapps.support.tasks.bulk_unenroll_course.apply_async", + side_effect=OSError("broker gone"), + ): + with self.assertRaises(OSError): + bulk_unenroll_batch(str(self.batch.uuid)) + + state = self._refresh() + second.refresh_from_db() + self.assertEqual(state.state, "failed") + self.assertEqual(second.state, "failed") + self.assertIn("Could not queue", state.error) + self.assertEqual(self.batch.state, "failed") # settled, so retry can reach it + + def test_dispatcher_failure_leaves_the_courses_it_did_queue_alone(self): + """ + A published course stays 'pending' until a worker claims it, so failing + every pending row would discard work already on the queue. + """ + self._second_course() + with patch( + "lms.djangoapps.support.tasks.bulk_unenroll_course.apply_async", + side_effect=[None, OSError("broker gone")], + ): + with self.assertRaises(OSError): + bulk_unenroll_batch(str(self.batch.uuid)) + + states = sorted(self.batch.courses.values_list("state", flat=True)) + self.assertEqual(states, ["failed", "pending"]) # the queued one is untouched + self.batch.refresh_from_db() + self.assertEqual(self.batch.state, "running") # its worker has yet to run + + def test_a_course_claimed_between_the_failure_and_the_cleanup_is_not_clobbered(self): + """The cleanup is conditional on 'pending', so a worker mid-claim wins.""" + second = self._second_course() + second.state = BulkUnenrollCourseState.State.RUNNING # a worker got there first + second.save() + + with patch( + "lms.djangoapps.support.tasks.bulk_unenroll_course.apply_async", + side_effect=OSError("broker gone"), + ): + with self.assertRaises(OSError): + bulk_unenroll_batch(str(self.batch.uuid)) + + second.refresh_from_db() + self.assertEqual(second.state, "running") + self.assertEqual(self._refresh().state, "failed") + + +class BulkUnenrollModifiedTimestampTest(BulkUnenrollTaskTestCase): + """ + `modified` must track the lifecycle, not just the calls that happen to save(). + + TimeStampedModel maintains it in `save()`, so every queryset `.update()` leaves + it behind — and it is what an operator reads to answer "is this run moving?". + """ + + batch_state = BulkUnenrollBatch.State.PENDING + + def test_dispatch_and_claim_advance_the_modified_timestamps(self): + batch_before = BulkUnenrollBatch.objects.values_list("modified", flat=True).get(pk=self.batch.pk) + state_before = BulkUnenrollCourseState.objects.values_list("modified", flat=True).get(pk=self.state.pk) + + with patch("lms.djangoapps.support.tasks.bulk_unenroll_course.apply_async"): + bulk_unenroll_batch(str(self.batch.uuid)) + self.batch.refresh_from_db() + self.assertEqual(self.batch.state, "running") + self.assertGreater(self.batch.modified, batch_before) + + # The course claim is a queryset .update() too, so it needs the same care. + self._enroll(1) + with patch("lms.djangoapps.support.tasks.bulk_unenroll_chunk.apply_async"): + bulk_unenroll_course(str(self.batch.uuid), str(self.course.id)) + self.state.refresh_from_db() + self.assertGreater(self.state.modified, state_before) + + +class BulkUnenrollContinuationIdentityTest(BulkUnenrollTaskTestCase): + """ + A timed-out chunk's continuation must not take an identity a queued chunk owns. + + A ledger row only appears when a chunk *reports*, so while chunks are in flight + the ledger is not a record of what has been handed out: deriving "the next free + index" from it would hand the tail an index a real chunk is already carrying. + The second to report is then dropped as a duplicate and the course never finishes. + """ + + course_state = BulkUnenrollCourseState.State.RUNNING + chunks_total = 2 # a fan-out that queued two chunks: 0 and 1 are spoken for + + def test_timed_out_chunk_and_its_sibling_both_get_counted(self): + """ + Chunk 0 times out while chunk 1 is still queued: every learner in both must + end up unenrolled *and* counted, and the course must finalize. + """ + chunk_zero = self._enroll(3) + chunk_one = self._enroll(2) + + with self._timeout_on(chunk_zero[1]): + with patch("lms.djangoapps.support.tasks.bulk_unenroll_chunk.apply_async") as mock_apply: + self._run_chunk(chunk_zero, chunk_index=0) + self.assertEqual(mock_apply.call_count, 1) + continuation_args = mock_apply.call_args.kwargs["args"] + + self._run_chunk(chunk_one, chunk_index=1) + bulk_unenroll_chunk(*continuation_args) + + self.assertEqual(self._active(), 0) + state = self._refresh() + self.assertEqual(state.unenrolled, 5) + self.assertEqual(state.chunks_total, 2) # the fan-out's number, tails and all + self.assertEqual(state.chunks_finished, 2) # chunk 0's chain counted once, at its end + self.assertEqual(state.state, "succeeded") diff --git a/lms/djangoapps/support/tests/test_models.py b/lms/djangoapps/support/tests/test_models.py new file mode 100644 index 000000000000..8e15409d6370 --- /dev/null +++ b/lms/djangoapps/support/tests/test_models.py @@ -0,0 +1,65 @@ +""" +Tests for support app models. +""" +from django.db import IntegrityError +from django.test import TestCase +from opaque_keys.edx.keys import CourseKey + +from common.djangoapps.student.tests.factories import UserFactory +from lms.djangoapps.support.models import BulkUnenrollBatch, BulkUnenrollChunk, BulkUnenrollCourseState + +COURSE_ID = CourseKey.from_string("course-v1:edX+DemoX+2024") + + +class BulkUnenrollModelsTest(TestCase): + """ + The uniqueness constraints the engine's claims are built on. + + Everything else about these models (field defaults, reverse accessors) is + Django's own behaviour and is exercised by the task and API tests instead. + """ + + def setUp(self): + super().setUp() + self.requester = UserFactory.create() + + def _batch(self): + return BulkUnenrollBatch.objects.create(requester=self.requester) + + def _course_state(self, batch=None, course=COURSE_ID): + return BulkUnenrollCourseState.objects.create( + batch=batch or self._batch(), course_id=course, + ) + + def test_a_course_appears_at_most_once_per_batch(self): + """One row per course per batch — the unit the whole engine addresses.""" + batch = self._batch() + self._course_state(batch=batch) + with self.assertRaises(IntegrityError): + self._course_state(batch=batch) + + def test_the_same_course_may_appear_in_other_batches(self): + """The constraint is per batch: re-running a course later must stay possible.""" + self._course_state(batch=self._batch()) + self._course_state(batch=self._batch()) + assert BulkUnenrollCourseState.objects.filter(course_id=COURSE_ID).count() == 2 + + def test_chunk_identity_is_unique_within_a_course(self): + """The uniqueness that makes the completion claim a claim.""" + course_state = self._course_state() + BulkUnenrollChunk.objects.create(course_state=course_state, chunk_index=0) + with self.assertRaises(IntegrityError): + BulkUnenrollChunk.objects.create(course_state=course_state, chunk_index=0) + + def test_a_retry_or_continuation_gets_a_fresh_identity(self): + """ + attempt and continuation are part of the key, so a re-run's chunk 0 and a + timed-out chunk's tail never collide with the row already in the ledger. + """ + course_state = self._course_state() + for attempt, continuation in [(1, 0), (1, 1), (2, 0)]: + BulkUnenrollChunk.objects.create( + course_state=course_state, chunk_index=0, + attempt=attempt, continuation=continuation, + ) + assert course_state.chunks.count() == 3 diff --git a/lms/envs/common.py b/lms/envs/common.py index a43ec942c361..b7f158d1fc0c 100644 --- a/lms/envs/common.py +++ b/lms/envs/common.py @@ -2105,6 +2105,37 @@ # parallel, and what the SES rate is. BULK_EMAIL_RETRY_DELAY_BETWEEN_SENDS = 0.02 +############################# Bulk Unenroll ################################### + +# Upload guardrails (lms/djangoapps/support/rest_api). The row limit counts every +# data row, duplicates included: it guards the file, not the de-duped course list. +BULK_UNENROLL_MAX_FILE_BYTES = 5 * 1024 * 1024 # 5 MB +BULK_UNENROLL_MAX_ROWS = 2000 + +# --- Async engine (lms/djangoapps/support/tasks.py) --- +# Worst case is ~10M unenrollments from one upload; these knobs throttle a real +# run without a code deploy. + +# Learners deactivated per chunk task. Smaller = finer-grained progress/retry; +# larger = fewer task round-trips. +BULK_UNENROLL_CHUNK_SIZE = 500 + +# Celery ``rate_limit`` for the chunk task, per worker process. ``None`` leaves +# throughput to the queue's worker concurrency; "10/s" adds a per-worker cap. +BULK_UNENROLL_CHUNK_RATE_LIMIT = None + +# Per-chunk soft time limit (seconds). On overrun the chunk records what it got +# through and queues the untouched tail as a fresh chunk. +BULK_UNENROLL_SOFT_TIME_LIMIT = 300 + +# A running chunk re-checks for revocation every N learners, bounding the +# overrun after a cancel to ~N without a DB check per learner. +BULK_UNENROLL_CANCEL_CHECK_EVERY = 50 + +# Queue the engine's tasks run on. In production, point this at a DEDICATED queue +# whose worker concurrency bounds the run, so it never starves interactive tasks. +BULK_UNENROLL_ROUTING_KEY = Derived(lambda settings: settings.DEFAULT_PRIORITY_QUEUE) + ############################# Email Opt In #################################### # Minimum age for organization-wide email opt in