feat: Add initial certificate generation checks for updated (V2) course certificates (#27090)

MICROBA-923

lms/djangoapps/certificates/generation.py

@@ -46,6 +46,14 @@ def generate_allowlist_certificate(user, course_key):
     return cert
 
 
+def generate_course_certificate(user, course_key):
+    """
+    Generate a regular certificate for this user, in this course run. This method should be called from a task.
+    """
+    # TODO: Implementation will be added in MICROBA-1039
+    log.warning(f'Ignoring course certificate generation for {user.id}: {course_key}')
+
+
 def _generate_certificate(user, course_id):
     """
     Generate a certificate for this user, in this course run.

lms/djangoapps/certificates/generation_handler.py

@@ -21,6 +21,7 @@ from lms.djangoapps.certificates.models import (
 from lms.djangoapps.certificates.queue import XQueueCertInterface
 from lms.djangoapps.certificates.tasks import CERTIFICATE_DELAY_SECONDS, generate_certificate
 from lms.djangoapps.certificates.utils import emit_certificate_event, has_html_certificates_enabled
+from lms.djangoapps.grades.api import CourseGradeFactory
 from lms.djangoapps.instructor.access import list_with_level
 from lms.djangoapps.verify_student.services import IDVerificationService
 from openedx.core.djangoapps.certificates.api import auto_certificate_generation_enabled
@@ -34,7 +35,7 @@ WAFFLE_FLAG_NAMESPACE = LegacyWaffleFlagNamespace(name='certificates_revamp')
 # .. toggle_name: certificates_revamp.use_allowlist
 # .. toggle_implementation: CourseWaffleFlag
 # .. toggle_default: False
-# .. toggle_description: Waffle flag to enable the course certificates allowlist (aka V2 of the certificate whitelist)
+# .. toggle_description: Waffle flag to enable the course certificates allowlist (aka v2 of the certificate whitelist)
 #   on a per-course run basis.
 # .. toggle_use_cases: temporary
 # .. toggle_creation_date: 2021-01-27
@@ -151,6 +152,11 @@ def _can_generate_allowlist_certificate(user, course_key):
                  f'for {user.id}.')
         return False
 
+    if not _is_on_certificate_allowlist(user, course_key):
+        log.info(f'{user.id} : {course_key} is not on the certificate allowlist. Allowlist certificate cannot be '
+                 f'generated.')
+        return False
+
     if not auto_certificate_generation_enabled():
         # Automatic certificate generation is globally disabled
         log.info(f'Automatic certificate generation is globally disabled. Allowlist certificate cannot be generated'
@@ -177,11 +183,6 @@ def _can_generate_allowlist_certificate(user, course_key):
         log.info(f'{user.id} does not have a verified id. Allowlist certificate cannot be generated for {course_key}.')
         return False
 
-    if not _is_on_certificate_allowlist(user, course_key):
-        log.info(f'{user.id} : {course_key} is not on the certificate allowlist. Allowlist certificate cannot be '
-                 f'generated.')
-        return False
-
     log.info(f'{user.id} : {course_key} is on the certificate allowlist')
     return _can_generate_allowlist_certificate_for_status(user, course_key)
 
@@ -196,9 +197,49 @@ def _can_generate_v2_certificate(user, course_key):
         log.info(f'{course_key} is not using v2 course certificates. Certificate cannot be generated.')
         return False
 
-    # TODO: Further implementation will be added in MICROBA-923
-    log.warning(f'Ignoring check on V2 course certificates for {user.id}: {course_key}')
-    return False
+    if not auto_certificate_generation_enabled():
+        # Automatic certificate generation is globally disabled
+        log.info(f'Automatic certificate generation is globally disabled. Certificate cannot be generated for '
+                 f'{user.id} : {course_key}.')
+        return False
+
+    if CertificateInvalidation.has_certificate_invalidation(user, course_key):
+        # The invalidation list prevents certificate generation
+        log.info(f'{user.id} : {course_key} is on the certificate invalidation list. Certificate cannot be generated.')
+        return False
+
+    enrollment_mode, __ = CourseEnrollment.enrollment_mode_for_user(user, course_key)
+    if enrollment_mode is None:
+        log.info(f'{user.id} : {course_key} does not have an enrollment. Certificate cannot be generated.')
+        return False
+
+    if not modes_api.is_eligible_for_certificate(enrollment_mode):
+        log.info(f'{user.id} : {course_key} has an enrollment mode of {enrollment_mode}, which is not eligible for an '
+                 f'allowlist certificate. Certificate cannot be generated.')
+        return False
+
+    if not IDVerificationService.user_is_verified(user):
+        log.info(f'{user.id} does not have a verified id. Certificate cannot be generated for {course_key}.')
+        return False
+
+    if _is_ccx_course(course_key):
+        log.info(f'{course_key} is a CCX course. Certificate cannot be generated for {user.id}.')
+        return False
+
+    course = modulestore().get_course(course_key, depth=0)
+    if _is_beta_tester(user, course):
+        log.info(f'{user.id} is a beta tester in {course_key}. Certificate cannot be generated.')
+        return False
+
+    if not _has_passing_grade(user, course):
+        log.info(f'{user.id} does not have a passing grade in {course_key}. Certificate cannot be generated.')
+        return False
+
+    if not _can_generate_certificate_for_status(user, course_key):
+        return False
+
+    log.info(f'V2 certificate can be generated for {user.id} : {course_key}')
+    return True
 
 
 def is_using_certificate_allowlist_and_is_on_allowlist(user, course_key):
@@ -212,7 +253,7 @@ def is_using_certificate_allowlist_and_is_on_allowlist(user, course_key):
 
 def is_using_certificate_allowlist(course_key):
     """
-    Check if the course run is using the allowlist, aka V2 of certificate whitelisting
+    Check if the course run is using the allowlist, aka v2 of certificate whitelisting
     """
     return CERTIFICATES_USE_ALLOWLIST.is_enabled(course_key)
 
@@ -250,6 +291,47 @@ def _can_generate_allowlist_certificate_for_status(user, course_key):
     return True
 
 
+def _can_generate_certificate_for_status(user, course_key):
+    """
+    Check if the user's certificate status can handle regular (non-allowlist) certificate generation
+    """
+    cert = GeneratedCertificate.certificate_for_student(user, course_key)
+    if cert is None:
+        return True
+
+    if cert.status == CertificateStatuses.downloadable:
+        log.info(f'Certificate with status {cert.status} already exists for {user.id} : {course_key}, and is not '
+                 f'eligible for generation. Certificate cannot be generated as it is already in a final state.')
+        return False
+
+    log.info(f'Certificate with status {cert.status} already exists for {user.id} : {course_key}, and is eligible for '
+             f'generation')
+    return True
+
+
+def _is_beta_tester(user, course):
+    """
+    Check if the user is a beta tester in this course run
+    """
+    beta_testers_queryset = list_with_level(course, 'beta')
+    return beta_testers_queryset.filter(username=user.username).exists()
+
+
+def _is_ccx_course(course_key):
+    """
+    Check if the course is a CCX (custom edX course)
+    """
+    return hasattr(course_key, 'ccx')
+
+
+def _has_passing_grade(user, course):
+    """
+    Check if the user has a passing grade in this course run
+    """
+    course_grade = CourseGradeFactory().read(user, course)
+    return course_grade.passed
+
+
 def generate_user_certificates(student, course_key, course=None, insecure=False, generation_mode='batch',
                                forced_grade=None):
     """

lms/djangoapps/certificates/tasks.py

@@ -10,7 +10,11 @@ from django.contrib.auth import get_user_model
 from edx_django_utils.monitoring import set_code_owner_attribute
 from opaque_keys.edx.keys import CourseKey
 
-from lms.djangoapps.certificates.generation import generate_allowlist_certificate, generate_user_certificates
+from lms.djangoapps.certificates.generation import (
+    generate_allowlist_certificate,
+    generate_course_certificate,
+    generate_user_certificates
+)
 from lms.djangoapps.verify_student.services import IDVerificationService
 
 log = getLogger(__name__)
@@ -49,8 +53,7 @@ def generate_certificate(self, **kwargs):
         return
 
     if v2_certificate:
-        # TODO: will be implemented in MICROBA-923
-        log.warning(f'Ignoring v2 certificate task request for {student.id}: {course_key}')
+        generate_course_certificate(user=student, course_key=course_key)
         return
 
     if expected_verification_status:

lms/djangoapps/certificates/tests/test_generation_handler.py

@@ -17,6 +17,7 @@ from lms.djangoapps.certificates.generation_handler import (
     is_using_certificate_allowlist,
     _is_using_v2_course_certificates,
     _can_generate_allowlist_certificate,
+    _can_generate_certificate_for_status,
     _can_generate_v2_certificate,
     can_generate_certificate_task,
     generate_allowlist_certificate_task,
@@ -36,7 +37,10 @@ from xmodule.modulestore.tests.factories import CourseFactory
 
 log = logging.getLogger(__name__)
 
+BETA_TESTER_METHOD = 'lms.djangoapps.certificates.generation_handler._is_beta_tester'
+CCX_COURSE_METHOD = 'lms.djangoapps.certificates.generation_handler._is_ccx_course'
 ID_VERIFIED_METHOD = 'lms.djangoapps.verify_student.services.IDVerificationService.user_is_verified'
+PASSING_GRADE_METHOD = 'lms.djangoapps.certificates.generation_handler._has_passing_grade'
 AUTO_GENERATION_NAMESPACE = waffle.WAFFLE_NAMESPACE
 AUTO_GENERATION_NAME = waffle.AUTO_CERTIFICATE_GENERATION
 AUTO_GENERATION_SWITCH_NAME = f'{AUTO_GENERATION_NAMESPACE}.{AUTO_GENERATION_NAME}'
@@ -259,6 +263,9 @@ class AllowlistTests(ModuleStoreTestCase):
 @override_switch(AUTO_GENERATION_SWITCH_NAME, active=True)
 @override_waffle_flag(CERTIFICATES_USE_UPDATED, active=True)
 @mock.patch(ID_VERIFIED_METHOD, mock.Mock(return_value=True))
+@mock.patch(CCX_COURSE_METHOD, mock.Mock(return_value=False))
+@mock.patch(PASSING_GRADE_METHOD, mock.Mock(return_value=True))
+@ddt.ddt
 class CertificateTests(ModuleStoreTestCase):
     """
     Tests for handling course certificates
@@ -281,13 +288,11 @@ class CertificateTests(ModuleStoreTestCase):
     def test_handle_valid(self):
         """
         Test handling of a valid user/course run combo.
-
-        Note: these assertions are placeholders for now. They will be updated as the implementation is added.
         """
-        assert not _can_generate_v2_certificate(self.user, self.course_run_key)
+        assert _can_generate_v2_certificate(self.user, self.course_run_key)
         assert can_generate_certificate_task(self.user, self.course_run_key)
-        assert not generate_certificate_task(self.user, self.course_run_key)
-        assert not generate_regular_certificate_task(self.user, self.course_run_key)
+        assert generate_certificate_task(self.user, self.course_run_key)
+        assert generate_regular_certificate_task(self.user, self.course_run_key)
 
     @override_waffle_flag(CERTIFICATES_USE_UPDATED, active=False)
     def test_handle_invalid(self):
@@ -311,3 +316,128 @@ class CertificateTests(ModuleStoreTestCase):
         Test the updated flag without the override
         """
         assert not _is_using_v2_course_certificates(self.course_run_key)
+
+    @ddt.data(
+        (CertificateStatuses.deleted, True),
+        (CertificateStatuses.deleting, True),
+        (CertificateStatuses.downloadable, False),
+        (CertificateStatuses.error, True),
+        (CertificateStatuses.generating, True),
+        (CertificateStatuses.notpassing, True),
+        (CertificateStatuses.restricted, True),
+        (CertificateStatuses.unavailable, True),
+        (CertificateStatuses.audit_passing, True),
+        (CertificateStatuses.audit_notpassing, True),
+        (CertificateStatuses.honor_passing, True),
+        (CertificateStatuses.unverified, True),
+        (CertificateStatuses.invalidated, True),
+        (CertificateStatuses.requesting, True))
+    @ddt.unpack
+    def test_generation_status(self, status, expected_response):
+        """
+        Test handling of certificate statuses
+        """
+        u = UserFactory()
+        cr = CourseFactory()
+        key = cr.id  # pylint: disable=no-member
+        GeneratedCertificateFactory(
+            user=u,
+            course_id=key,
+            mode=GeneratedCertificate.MODES.verified,
+            status=status,
+        )
+
+        assert _can_generate_certificate_for_status(u, key) == expected_response
+
+    def test_generation_status_for_none(self):
+        """
+        Test handling of certificate statuses for a non-existent cert
+        """
+        assert _can_generate_certificate_for_status(None, None)
+
+    def test_can_generate_auto_disabled(self):
+        """
+        Test handling when automatic generation is disabled
+        """
+        with override_waffle_switch(AUTO_GENERATION_SWITCH, active=False):
+            assert not _can_generate_v2_certificate(self.user, self.course_run_key)
+
+    def test_can_generate_not_verified(self):
+        """
+        Test handling when the user's id is not verified
+        """
+        with mock.patch(ID_VERIFIED_METHOD, return_value=False):
+            assert not _can_generate_v2_certificate(self.user, self.course_run_key)
+
+    def test_can_generate_ccx(self):
+        """
+        Test handling when the course is a CCX (custom edX) course
+        """
+        with mock.patch(CCX_COURSE_METHOD, return_value=True):
+            assert not _can_generate_v2_certificate(self.user, self.course_run_key)
+
+    def test_can_generate_beta_tester(self):
+        """
+        Test handling when the user is a beta tester
+        """
+        with mock.patch(BETA_TESTER_METHOD, return_value=True):
+            assert not _can_generate_v2_certificate(self.user, self.course_run_key)
+
+    def test_can_generate_failing_grade(self):
+        """
+        Test handling when the user does not have a passing grade
+        """
+        with mock.patch(PASSING_GRADE_METHOD, return_value=False):
+            assert not _can_generate_v2_certificate(self.user, self.course_run_key)
+
+    def test_can_generate_not_enrolled(self):
+        """
+        Test handling when user is not enrolled
+        """
+        u = UserFactory()
+        cr = CourseFactory()
+        key = cr.id  # pylint: disable=no-member
+        assert not _can_generate_v2_certificate(u, key)
+
+    def test_can_generate_audit(self):
+        """
+        Test handling when user is enrolled in audit mode
+        """
+        u = UserFactory()
+        cr = CourseFactory()
+        key = cr.id  # pylint: disable=no-member
+        CourseEnrollmentFactory(
+            user=u,
+            course_id=key,
+            is_active=True,
+            mode="audit",
+        )
+
+        assert not _can_generate_v2_certificate(u, key)
+
+    def test_can_generate_invalidated(self):
+        """
+        Test handling when user is on the invalidate list
+        """
+        u = UserFactory()
+        cr = CourseFactory()
+        key = cr.id  # pylint: disable=no-member
+        CourseEnrollmentFactory(
+            user=u,
+            course_id=key,
+            is_active=True,
+            mode="verified",
+        )
+        cert = GeneratedCertificateFactory(
+            user=u,
+            course_id=key,
+            mode=GeneratedCertificate.MODES.verified,
+            status=CertificateStatuses.downloadable
+        )
+        CertificateInvalidationFactory.create(
+            generated_certificate=cert,
+            invalidated_by=self.user,
+            active=True
+        )
+
+        assert not _can_generate_v2_certificate(u, key)