Release 1.144.5

.pre-commit-config.yaml

@@ -47,7 +47,7 @@ repos:
           - "config/keycloak/*"
         additional_dependencies: ["gibberish-detector"]
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: "v0.15.7"
+    rev: "v0.15.8"
     hooks:
       - id: ruff-format
       - id: ruff
@@ -60,8 +60,20 @@ repos:
       - id: shellcheck
         args: ["--severity=warning"]
   - repo: https://github.com/rhysd/actionlint
-    rev: v1.7.11
+    rev: v1.7.12
     hooks:
       - id: actionlint
         name: actionlint
         description: Runs actionlint to lint GitHub Actions workflow files
+  - repo: local
+    hooks:
+      - id: drf-serializer-orm-check
+        name: DRF Serializer ORM Check
+        description: "Detects Django ORM queries inside DRF serializer methods (N+1 risk)"
+        entry: drf-lint
+        args: [--baseline, drf_lint_baseline.json]
+        language: python
+        files: 'serializers\.py$'
+        additional_dependencies:
+          - mitol-drf-lint
+          - "setuptools<82"

RELEASE.rst

@@ -1,6 +1,16 @@
 Release Notes
 =============
 
+Version 1.144.5
+---------------
+
+- Add management command to check/repair B2B program enrollments (#3449)
+- chore: add drf-lint pre-commit hook with baseline (#3453)
+- [pre-commit.ci] pre-commit autoupdate (#3438)
+- Add test to the program enrollment dialog (#3445)
+- fix: fall back to audit enrollment for non-upgradable runs in verified program enrollment API (#3451)
+- Refactor program cert generation to support new cert requirements (#3439)
+
 Version 1.144.4 (Released April 02, 2026)
 ---------------
 

b2b/management/commands/repair_b2b_program_enrollments.py

@@ -0,0 +1,81 @@
+"""Check and repair B2B program enrollments."""
+
+import logging
+
+from django.core.management import BaseCommand
+from django.db.models import Q
+
+from b2b.models import ContractProgramItem
+from courses.constants import ENROLL_CHANGE_STATUS_UNENROLLED
+from courses.models import Program, ProgramEnrollment
+
+log = logging.getLogger(__name__)
+
+
+class Command(BaseCommand):
+    """Check and repair B2B program enrollments."""
+
+    help = """Check and repair B2B program enrollments.
+
+Unenroll users from B2B programs that belong to contracts that the user doesn't have an attachment for."""
+
+    def add_arguments(self, parser):
+        """Add command line arguments."""
+
+        parser.add_argument(
+            "--user", type=str, help="Check specified username or email."
+        )
+        parser.add_argument(
+            "--commit",
+            action="store_true",
+            default=False,
+            help="Commit changes. Otherwise, this will produce output but won't make any changes.",
+        )
+
+    def handle(self, *args, **kwargs):  # noqa: ARG002
+        """Perform check and repair operation."""
+
+        specific_user = kwargs.pop("user", None)
+        commit = kwargs.pop("commit", False)
+
+        if not commit:
+            self.stdout.write(
+                self.style.WARNING("Commit flag not set - no changes will be made.")
+            )
+
+        b2b_programs = Program.objects.filter(
+            b2b_only=True,
+            contract_memberships__isnull=False,
+        ).distinct()
+
+        for program in b2b_programs:
+            program_contract_ids = ContractProgramItem.objects.filter(
+                program=program
+            ).values_list("contract_id", flat=True)
+
+            extraneous_enrollments = (
+                ProgramEnrollment.objects.filter(program=program)
+                .exclude(user__b2b_contracts__in=program_contract_ids)
+                .select_related("user", "program")
+                .distinct()
+            )
+
+            if specific_user:
+                extraneous_enrollments = extraneous_enrollments.filter(
+                    Q(user__email=specific_user) | Q(user__username=specific_user)
+                )
+
+            count = extraneous_enrollments.count()
+            if count > 0:
+                self.stdout.write(
+                    f"Program {program.readable_id}: unenrolling {count} user(s):"
+                )
+
+                for enrollment in extraneous_enrollments:
+                    self.stdout.write(f"\t{enrollment.user.email}")
+                    if commit:
+                        enrollment.deactivate_and_save(
+                            ENROLL_CHANGE_STATUS_UNENROLLED, no_user=True
+                        )
+
+                self.stdout.write("\n")

courses/api.py

@@ -1042,11 +1042,16 @@ def _has_earned_program_cert(user, program):
     """
     program_course_ids = [course[0].id for course in program.courses]
 
-    passed_courses = Course.objects.filter(
+    cert_courses = Course.objects.filter(
         id__in=program_course_ids,
         courseruns__courseruncertificates__user=user,
         courseruns__courseruncertificates__is_revoked=False,
     )
+    grade_courses = Course.objects.filter(
+        id__in=program_course_ids,
+        courseruns__grades__user=user,
+        courseruns__grades__passed=True,
+    ).exclude(courseruns__enrollment_modes__mode_slug=EDX_ENROLLMENT_VERIFIED_MODE)
     root = ProgramRequirement.get_root_nodes().get(program=program)
 
     def _has_earned(node):
@@ -1059,8 +1064,8 @@ def _has_earned(node):
                 node.operator_value
             )
         elif node.is_course:
-            # has passed the reference course
-            return node.course in passed_courses
+            # has passed the referenced course
+            return node.course in [*cert_courses, *grade_courses]
         elif node.is_program:
             # has earned certificate for the required sub-program
             return ProgramCertificate.objects.filter(
@@ -1073,9 +1078,8 @@ def _has_earned(node):
 
 def generate_program_certificate(user, program, force_create=False):  # noqa: FBT002
     """
-    Create a program certificate if the user has a course certificate
-    for each course in the program. Also, It will create the
-    program enrollment if it does not exist for the user.
+    Create a program certificate if the user has a verified enrollment in the
+    program and certificates or passing grades in the program's required courses.
 
     Args:
         user (User): a Django user.
@@ -1090,13 +1094,26 @@ def generate_program_certificate(user, program, force_create=False):  # noqa: FB
     """
     from hubspot_sync.task_helpers import sync_hubspot_user  # noqa: PLC0415
 
+    if (
+        not force_create
+        and not program.enrollments.filter(
+            user=user, enrollment_mode=EDX_ENROLLMENT_VERIFIED_MODE
+        ).exists()
+    ):
+        log.warning(
+            "Skipping program enrollment generation for %s in %s: no verified enrollment",
+            user,
+            program,
+        )
+        return (
+            None,
+            False,
+        )
+
     existing_cert_queryset = ProgramCertificate.all_objects.filter(
         user=user, program=program
     )
     if existing_cert_queryset.exists():
-        ProgramEnrollment.objects.get_or_create(
-            program=program, user=user, defaults={"active": True, "change_status": None}
-        )
         return existing_cert_queryset.first(), False
 
     if not force_create and not _has_earned_program_cert(user, program):
@@ -1113,17 +1130,6 @@ def generate_program_certificate(user, program, force_create=False):  # noqa: FB
         if not program_cert.verifiable_credential_id:
             create_verifiable_credential(program_cert)
 
-        _, created = ProgramEnrollment.objects.get_or_create(
-            program=program, user=user, defaults={"active": True, "change_status": None}
-        )
-
-        if created:
-            log.info(
-                "Program enrollment for [%s] in program [%s] is created.",
-                user.edx_username,
-                program.title,
-            )
-
     return program_cert, True