diff --git a/lms/djangoapps/certificates/api.py b/lms/djangoapps/certificates/api.py
index 865717bf708f1dbe21be262dde9ff6423ec64153..b2756610354722c12c21f941414367ae8524f8ab 100644
--- a/lms/djangoapps/certificates/api.py
+++ b/lms/djangoapps/certificates/api.py
@@ -11,6 +11,7 @@ certificates models or any other certificates modules.
 import logging
 
 import six
+from django.contrib.auth import get_user_model
 from django.db.models import Q
 from eventtracking import tracker
 from opaque_keys.edx.django.models import CourseKeyField
@@ -18,6 +19,7 @@ from organizations.api import get_course_organization_id
 
 from lms.djangoapps.branding import api as branding_api
 from lms.djangoapps.certificates.generation_handler import (
+    is_using_certificate_allowlist as _is_using_certificate_allowlist,
     is_using_certificate_allowlist_and_is_on_allowlist as _is_using_certificate_allowlist_and_is_on_allowlist,
     generate_allowlist_certificate_task as _generate_allowlist_certificate_task,
     generate_user_certificates as _generate_user_certificates,
@@ -30,6 +32,7 @@ from lms.djangoapps.certificates.models import (
     CertificateStatuses,
     CertificateTemplate,
     CertificateTemplateAsset,
+    CertificateWhitelist,
     ExampleCertificateSet,
     GeneratedCertificate,
     certificate_status_for_student
@@ -41,6 +44,7 @@ from openedx.core.djangoapps.certificates.api import certificates_viewable_for_c
 from openedx.core.djangoapps.content.course_overviews.models import CourseOverview
 
 log = logging.getLogger("edx.certificate")
+User = get_user_model()
 MODES = GeneratedCertificate.MODES
 
 
@@ -553,3 +557,13 @@ def is_using_certificate_allowlist_and_is_on_allowlist(user, course_key):
     2) if the user is on the allowlist for this course run
     """
     return _is_using_certificate_allowlist_and_is_on_allowlist(user, course_key)
+
+
+def get_allowlisted_users(course_key):
+    """
+    Return the users who are on the allowlist for this course run
+    """
+    if not _is_using_certificate_allowlist(course_key):
+        return User.objects.none()
+
+    return User.objects.filter(certificatewhitelist__course_id=course_key, certificatewhitelist__whitelist=True)
diff --git a/lms/djangoapps/certificates/generation_handler.py b/lms/djangoapps/certificates/generation_handler.py
index 8354db367fd0843dc8f8e3d507203f1d5a1c9df5..1bb7f8d62912673d242a55d35ade1f1d26a6357b 100644
--- a/lms/djangoapps/certificates/generation_handler.py
+++ b/lms/djangoapps/certificates/generation_handler.py
@@ -72,7 +72,7 @@ def can_generate_allowlist_certificate(user, course_key):
     Check if an allowlist certificate can be generated (created if it doesn't already exist, or updated if it does
     exist) for this user, in this course run.
     """
-    if not _is_using_certificate_allowlist(course_key):
+    if not is_using_certificate_allowlist(course_key):
         # This course run is not using the allowlist feature
         log.info(
             '{course} is not using the certificate allowlist. Certificate cannot be generated.'.format(
@@ -130,10 +130,10 @@ def is_using_certificate_allowlist_and_is_on_allowlist(user, course_key):
     1) the course run is using the allowlist, and
     2) if the user is on the allowlist for this course run
     """
-    return _is_using_certificate_allowlist(course_key) and _is_on_certificate_allowlist(user, course_key)
+    return is_using_certificate_allowlist(course_key) and _is_on_certificate_allowlist(user, course_key)
 
 
-def _is_using_certificate_allowlist(course_key):
+def is_using_certificate_allowlist(course_key):
     """
     Check if the course run is using the allowlist, aka V2 of certificate whitelisting
     """
@@ -193,6 +193,12 @@ def generate_user_certificates(student, course_key, course=None, insecure=False,
         forced_grade - a string indicating to replace grade parameter. if present grading
                        will be skipped.
     """
+    if is_using_certificate_allowlist_and_is_on_allowlist(student, course_key):
+        # Note that this will launch an asynchronous task, and so cannot return the certificate status. This is a
+        # change from the older certificate code that tries to immediately create a cert.
+        log.info(f'{course_key} is using allowlist certificates, and the user {student.id} is on its allowlist. '
+                 f'Attempt will be made to regenerate an allowlist certificate.')
+        return generate_allowlist_certificate_task(student, course_key)
 
     if not course:
         course = modulestore().get_course(course_key, depth=0)
diff --git a/lms/djangoapps/certificates/tests/test_api.py b/lms/djangoapps/certificates/tests/test_api.py
index df375cfb553fa572e62a37c260ddee62d20def5a..c0c02c6542192f27a215fd31caa634ea987df3f9 100644
--- a/lms/djangoapps/certificates/tests/test_api.py
+++ b/lms/djangoapps/certificates/tests/test_api.py
@@ -14,6 +14,7 @@ from django.test import RequestFactory, TestCase
 from django.test.utils import override_settings
 from django.urls import reverse
 from django.utils import timezone
+from edx_toggles.toggles.testutils import override_waffle_flag
 from freezegun import freeze_time
 from mock import patch
 from opaque_keys.edx.keys import CourseKey
@@ -24,7 +25,10 @@ from xmodule.modulestore.tests.factories import CourseFactory
 from common.djangoapps.course_modes.models import CourseMode
 from common.djangoapps.course_modes.tests.factories import CourseModeFactory
 from common.djangoapps.student.models import CourseEnrollment
-from common.djangoapps.student.tests.factories import UserFactory
+from common.djangoapps.student.tests.factories import (
+    CourseEnrollmentFactory,
+    UserFactory
+)
 from common.djangoapps.util.testing import EventTestMixin
 from lms.djangoapps.certificates.api import (
     cert_generation_enabled,
@@ -32,6 +36,7 @@ from lms.djangoapps.certificates.api import (
     example_certificates_status,
     generate_example_certificates,
     generate_user_certificates,
+    get_allowlisted_users,
     get_certificate_for_user,
     get_certificates_for_user,
     get_certificates_for_user_by_course_keys,
@@ -41,6 +46,7 @@ from lms.djangoapps.certificates.api import (
     is_certificate_invalid,
     set_cert_generation_enabled
 )
+from lms.djangoapps.certificates.generation_handler import CERTIFICATES_USE_ALLOWLIST
 from lms.djangoapps.certificates.models import (
     CertificateGenerationConfiguration,
     CertificateStatuses,
@@ -49,7 +55,11 @@ from lms.djangoapps.certificates.models import (
     certificate_status_for_student
 )
 from lms.djangoapps.certificates.queue import XQueueAddToQueueError, XQueueCertInterface
-from lms.djangoapps.certificates.tests.factories import CertificateInvalidationFactory, GeneratedCertificateFactory
+from lms.djangoapps.certificates.tests.factories import (
+    CertificateWhitelistFactory,
+    GeneratedCertificateFactory,
+    CertificateInvalidationFactory
+)
 from lms.djangoapps.courseware.tests.factories import GlobalStaffFactory
 from lms.djangoapps.grades.tests.utils import mock_passing_grade
 from openedx.core.djangoapps.site_configuration.tests.test_util import with_site_configuration
@@ -782,3 +792,87 @@ class CertificatesBrandingTest(ModuleStoreTestCase):
         assert self.configuration['urls']['ABOUT'] in data['company_about_url']
         assert self.configuration['urls']['PRIVACY'] in data['company_privacy_url']
         assert self.configuration['urls']['TOS_AND_HONOR'] in data['company_tos_url']
+
+
+@override_waffle_flag(CERTIFICATES_USE_ALLOWLIST, active=True)
+class AllowlistTests(ModuleStoreTestCase):
+    """
+    Tests for handling allowlist certificates
+    """
+
+    def setUp(self):
+        super().setUp()
+
+        # Create users, a course run, and enrollments
+        self.user = UserFactory()
+        self.user2 = UserFactory()
+        self.user3 = UserFactory()
+        self.user4 = UserFactory()
+
+        self.course_run = CourseFactory()
+        self.course_run_key = self.course_run.id  # pylint: disable=no-member
+        self.second_course_run = CourseFactory()
+        self.second_course_run_key = self.second_course_run.id  # pylint: disable=no-member
+        self.third_course_run = CourseFactory()
+        self.third_course_run_key = self.third_course_run.id  # pylint: disable=no-member
+
+        CourseEnrollmentFactory(
+            user=self.user,
+            course_id=self.course_run_key,
+            is_active=True,
+            mode="verified",
+        )
+        CourseEnrollmentFactory(
+            user=self.user2,
+            course_id=self.course_run_key,
+            is_active=True,
+            mode="verified",
+        )
+        CourseEnrollmentFactory(
+            user=self.user3,
+            course_id=self.course_run_key,
+            is_active=True,
+            mode="verified",
+        )
+        CourseEnrollmentFactory(
+            user=self.user4,
+            course_id=self.second_course_run_key,
+            is_active=True,
+            mode="verified",
+        )
+
+        # Add user to the allowlist
+        CertificateWhitelistFactory.create(course_id=self.course_run_key, user=self.user)
+        # Add user to the allowlist, but set whitelist to false
+        CertificateWhitelistFactory.create(course_id=self.course_run_key, user=self.user2, whitelist=False)
+        # Add user to the allowlist in the other course
+        CertificateWhitelistFactory.create(course_id=self.second_course_run_key, user=self.user4)
+
+    def test_get_users_allowlist(self):
+        """
+        Test that allowlisted users are returned correctly
+        """
+        users = get_allowlisted_users(self.course_run_key)
+        assert 1 == users.count()
+        assert users[0].id == self.user.id
+
+        users = get_allowlisted_users(self.second_course_run_key)
+        assert 1 == users.count()
+        assert users[0].id == self.user4.id
+
+        users = get_allowlisted_users(self.third_course_run_key)
+        assert 0 == users.count()
+
+    @override_waffle_flag(CERTIFICATES_USE_ALLOWLIST, active=False)
+    def test_get_users_allowlist_false(self):
+        """
+        Test
+        """
+        users = get_allowlisted_users(self.course_run_key)
+        assert 0 == users.count()
+
+        users = get_allowlisted_users(self.second_course_run_key)
+        assert 0 == users.count()
+
+        users = get_allowlisted_users(self.third_course_run_key)
+        assert 0 == users.count()
diff --git a/lms/djangoapps/certificates/tests/test_generation_handler.py b/lms/djangoapps/certificates/tests/test_generation_handler.py
index 47b8df44b93412931259b97ee6fc5b6d4cb5db70..a5b8cf06b11e0cf174700260f00e876dd89af3fd 100644
--- a/lms/djangoapps/certificates/tests/test_generation_handler.py
+++ b/lms/djangoapps/certificates/tests/test_generation_handler.py
@@ -16,7 +16,7 @@ from common.djangoapps.student.tests.factories import CourseEnrollmentFactory, U
 from lms.djangoapps.certificates.generation_handler import CERTIFICATES_USE_ALLOWLIST
 from lms.djangoapps.certificates.generation_handler import (
     _can_generate_allowlist_certificate_for_status,
-    _is_using_certificate_allowlist,
+    is_using_certificate_allowlist,
     can_generate_allowlist_certificate,
     generate_allowlist_certificate_task,
     is_using_certificate_allowlist_and_is_on_allowlist
@@ -68,14 +68,14 @@ class AllowlistTests(ModuleStoreTestCase):
         """
         Test the allowlist flag
         """
-        assert _is_using_certificate_allowlist(self.course_run_key)
+        assert is_using_certificate_allowlist(self.course_run_key)
 
     @override_waffle_flag(CERTIFICATES_USE_ALLOWLIST, active=False)
     def test_is_using_allowlist_false(self):
         """
         Test the allowlist flag without the override
         """
-        assert not _is_using_certificate_allowlist(self.course_run_key)
+        assert not is_using_certificate_allowlist(self.course_run_key)
 
     def test_is_using_allowlist_and_is_on_list(self):
         """
diff --git a/lms/djangoapps/certificates/tests/test_queue.py b/lms/djangoapps/certificates/tests/test_queue.py
index fa8346b0af6c7c64d3de6c624d5ba00418c54f77..c96dda2c9f2c89e7b9dd0ce96991bff376e6f9e1 100644
--- a/lms/djangoapps/certificates/tests/test_queue.py
+++ b/lms/djangoapps/certificates/tests/test_queue.py
@@ -10,21 +10,19 @@ import ddt
 import freezegun
 import pytz
 import six
+from django.conf import settings
+from django.test import TestCase
+from django.test.utils import override_settings
+from mock import Mock, patch
+from opaque_keys.edx.locator import CourseLocator
+from testfixtures import LogCapture
+
 # It is really unfortunate that we are using the XQueue client
 # code from the capa library.  In the future, we should move this
 # into a shared library.  We import it here so we can mock it
 # and verify that items are being correctly added to the queue
 # in our `XQueueCertInterface` implementation.
 from capa.xqueue_interface import XQueueInterface
-from django.conf import settings  # lint-amnesty, pylint: disable=wrong-import-order
-from django.test import TestCase  # lint-amnesty, pylint: disable=wrong-import-order
-from django.test.utils import override_settings  # lint-amnesty, pylint: disable=wrong-import-order
-from mock import Mock, patch  # lint-amnesty, pylint: disable=wrong-import-order
-from opaque_keys.edx.locator import CourseLocator  # lint-amnesty, pylint: disable=wrong-import-order
-from testfixtures import LogCapture  # lint-amnesty, pylint: disable=wrong-import-order
-from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
-from xmodule.modulestore.tests.factories import CourseFactory
-
 from common.djangoapps.course_modes.models import CourseMode
 from common.djangoapps.student.tests.factories import CourseEnrollmentFactory, UserFactory
 from lms.djangoapps.certificates.models import (
@@ -37,6 +35,8 @@ from lms.djangoapps.certificates.queue import LOGGER, XQueueCertInterface
 from lms.djangoapps.certificates.tests.factories import CertificateWhitelistFactory, GeneratedCertificateFactory
 from lms.djangoapps.grades.tests.utils import mock_passing_grade
 from lms.djangoapps.verify_student.tests.factories import SoftwareSecurePhotoVerificationFactory
+from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
+from xmodule.modulestore.tests.factories import CourseFactory
 
 
 @ddt.ddt
diff --git a/lms/djangoapps/certificates/views/xqueue.py b/lms/djangoapps/certificates/views/xqueue.py
index 33b67247f2b80656fd5d8fbba845a3f9c3f36fdf..f9c76938cd259399125d309efd69b47fb79466a4 100644
--- a/lms/djangoapps/certificates/views/xqueue.py
+++ b/lms/djangoapps/certificates/views/xqueue.py
@@ -16,7 +16,11 @@ from xmodule.modulestore.django import modulestore
 
 from common.djangoapps.util.json_request import JsonResponse, JsonResponseBadRequest
 from common.djangoapps.util.request_rate_limiter import BadRequestRateLimiter
-from lms.djangoapps.certificates.api import generate_user_certificates
+from lms.djangoapps.certificates.api import (
+    generate_allowlist_certificate_task,
+    generate_user_certificates,
+    is_using_certificate_allowlist_and_is_on_allowlist
+)
 from lms.djangoapps.certificates.models import (
     CertificateStatuses,
     ExampleCertificate,
@@ -47,7 +51,11 @@ def request_certificate(request):
             course = modulestore().get_course(course_key, depth=2)
 
             status = certificate_status_for_student(student, course_key)['status']
-            if status in [CertificateStatuses.unavailable, CertificateStatuses.notpassing, CertificateStatuses.error]:
+            if is_using_certificate_allowlist_and_is_on_allowlist(student, course_key):
+                log.info(f'{course_key} is using allowlist certificates, and the user {student.id} is on its '
+                         f'allowlist. Attempt will be made to generate an allowlist certificate.')
+                generate_allowlist_certificate_task(student, course_key)
+            elif status in [CertificateStatuses.unavailable, CertificateStatuses.notpassing, CertificateStatuses.error]:
                 log_msg = u'Grading and certification requested for user %s in course %s via /request_certificate call'
                 log.info(log_msg, username, course_key)
                 status = generate_user_certificates(student, course_key, course=course)
@@ -94,6 +102,18 @@ def update_certificate(request):
                 'content': 'unable to lookup key'
             }), content_type='application/json')
 
+        user = cert.user
+        if is_using_certificate_allowlist_and_is_on_allowlist(user, course_key):
+            log.warning(f'{course_key} is using allowlist certificates, and the user {user.id} is on its allowlist. '
+                        f'Request to update the certificate will be ignored.')
+            return HttpResponse(  # pylint: disable=http-response-with-content-type-json, http-response-with-json-dumps
+                json.dumps({
+                    'return_code': 1,
+                    'content': 'allowlist certificate'
+                }),
+                content_type='application/json'
+            )
+
         if 'error' in xqueue_body:
             cert.status = status.error
             if 'error_reason' in xqueue_body:
diff --git a/lms/djangoapps/courseware/views/views.py b/lms/djangoapps/courseware/views/views.py
index a73d9930e6f4e4d15f20384a0c4cd5420f5100b1..100876d4093fe576270b8d9951dda610a0188e9b 100644
--- a/lms/djangoapps/courseware/views/views.py
+++ b/lms/djangoapps/courseware/views/views.py
@@ -967,7 +967,7 @@ def course_about(request, course_id):
             'active_reg_button': active_reg_button,
             'is_shib_course': is_shib_course,
             # We do not want to display the internal courseware header, which is used when the course is found in the
-            # context. This value is therefor explicitly set to render the appropriate header.
+            # context. This value is therefore explicitly set to render the appropriate header.
             'disable_courseware_header': True,
             'pre_requisite_courses': pre_requisite_courses,
             'course_image_urls': overview.image_urls,
@@ -1569,6 +1569,12 @@ def generate_user_cert(request, course_id):
     if not course:
         return HttpResponseBadRequest(_("Course is not valid"))
 
+    if certs_api.is_using_certificate_allowlist_and_is_on_allowlist(student, course_key):
+        log.info(f'{course_key} is using allowlist certificates, and the user {student.id} is on its allowlist. '
+                 f'Attempt will be made to generate an allowlist certificate.')
+        certs_api.generate_allowlist_certificate_task(student, course_key)
+        return HttpResponse()
+
     if not is_course_passed(student, course):
         log.info(u"User %s has not passed the course: %s", student.username, course_id)
         return HttpResponseBadRequest(_("Your certificate will be available when you pass the course."))
diff --git a/lms/djangoapps/instructor_task/tasks_helper/certs.py b/lms/djangoapps/instructor_task/tasks_helper/certs.py
index 930f29598a95ebee69b19a2108142366a3601116..6f8b67077807f9ea3ccdfc2979cdf413b6ae9a09 100644
--- a/lms/djangoapps/instructor_task/tasks_helper/certs.py
+++ b/lms/djangoapps/instructor_task/tasks_helper/certs.py
@@ -3,13 +3,20 @@ Instructor tasks related to certificates.
 """
 
 
+import logging
+
 from time import time
 
 from django.contrib.auth import get_user_model
 from django.db.models import Q
 
 from common.djangoapps.student.models import CourseEnrollment
-from lms.djangoapps.certificates.api import generate_user_certificates
+from lms.djangoapps.certificates.api import (
+    generate_allowlist_certificate_task,
+    generate_user_certificates,
+    get_allowlisted_users,
+    is_using_certificate_allowlist_and_is_on_allowlist
+)
 from lms.djangoapps.certificates.models import CertificateStatuses, GeneratedCertificate
 from xmodule.modulestore.django import modulestore
 
@@ -17,6 +24,8 @@ from .runner import TaskProgress
 
 User = get_user_model()
 
+log = logging.getLogger(__name__)
+
 
 def generate_students_certificates(
         _xmodule_instance_args, _entry_id, course_id, task_input, action_name):
@@ -63,6 +72,8 @@ def generate_students_certificates(
             course_id, students_to_generate_certs_for, statuses_to_regenerate
         )
 
+    log.info(f'About to attempt certificate generation for {len(students_require_certs)} users in course {course_id}. '
+             f'The student_set is {student_set} and statuses_to_regenerate is {statuses_to_regenerate}')
     if statuses_to_regenerate:
         # Mark existing generated certificates as 'unavailable' before regenerating
         # We need to call this method after "students_require_certificate" otherwise "students_require_certificate"
@@ -78,17 +89,17 @@ def generate_students_certificates(
     # Generate certificate for each student
     for student in students_require_certs:
         task_progress.attempted += 1
-        status = generate_user_certificates(
-            student,
-            course_id,
-            course=course
-        )
-
-        if CertificateStatuses.is_passing_status(status):
-            task_progress.succeeded += 1
+        if is_using_certificate_allowlist_and_is_on_allowlist(student, course_id):
+            log.info(f'{course_id} is using allowlist certificates, and the user {student.id} is on its allowlist. '
+                     f'Attempt will be made to generate an allowlist certificate.')
+            generate_allowlist_certificate_task(student, course_id)
         else:
-            task_progress.failed += 1
-
+            log.info(f'Attempt will be made to generate a certificate for user {student.id} in {course_id}.')
+            generate_user_certificates(
+                student,
+                course_id,
+                course=course
+            )
     return task_progress.update_task_state(extra_meta=current_step)
 
 
@@ -127,7 +138,7 @@ def students_require_certificate(course_id, enrolled_students, statuses_to_regen
 def _invalidate_generated_certificates(course_id, enrolled_students, certificate_statuses):
     """
     Invalidate generated certificates for all enrolled students in the given course having status in
-    'certificate_statuses'.
+    'certificate_statuses', if the student is not on the course's allowlist.
 
     Generated Certificates are invalidated by marking its status 'unavailable' and updating verify_uuid, download_uuid,
     download_url and grade with empty string.
@@ -142,12 +153,20 @@ def _invalidate_generated_certificates(course_id, enrolled_students, certificate
         status__in=certificate_statuses,
     )
 
+    allowlisted_users = get_allowlisted_users(course_id)
+
     # Mark generated certificates as 'unavailable' and update download_url, download_uui, verify_uuid and
-    # grade with empty string for each row
-    certificates.update(
-        status=CertificateStatuses.unavailable,
-        verify_uuid='',
-        download_uuid='',
-        download_url='',
-        grade='',
-    )
+    # grade with empty string for each cert that is not allowlisted. We loop over the certs and save each individually
+    # in order to save a history of the change.
+    for c in certificates:
+        if c.user in allowlisted_users:
+            log.info(f'Certificate for user {c.user.id} will not be invalidated because they are on the allowlist for '
+                     f'course {course_id}')
+        else:
+            log.info(f'About to invalidate certificate for user {c.user.id} in course {course_id}')
+            c.status = CertificateStatuses.unavailable
+            c.verify_uuid = ''
+            c.download_uuid = ''
+            c.download_url = ''
+            c.grade = ''
+            c.save()
diff --git a/lms/djangoapps/instructor_task/tests/test_tasks_helper.py b/lms/djangoapps/instructor_task/tests/test_tasks_helper.py
index dcec0f3fb17a1d282779ab0ed6c69c54fb46064e..cab878ca2cd17dfa6f919badc40da06ec9f9fe44 100644
--- a/lms/djangoapps/instructor_task/tests/test_tasks_helper.py
+++ b/lms/djangoapps/instructor_task/tests/test_tasks_helper.py
@@ -21,6 +21,7 @@ import unicodecsv
 from django.conf import settings
 from django.test.utils import override_settings
 from edx_django_utils.cache import RequestCache
+from edx_toggles.toggles.testutils import override_waffle_flag
 from freezegun import freeze_time
 from pytz import UTC
 
@@ -29,6 +30,7 @@ from capa.tests.response_xml_factory import MultipleChoiceResponseXMLFactory
 from common.djangoapps.course_modes.models import CourseMode
 from common.djangoapps.student.models import CourseEnrollment, CourseEnrollmentAllowed
 from common.djangoapps.student.tests.factories import CourseEnrollmentFactory, UserFactory
+from lms.djangoapps.certificates.generation_handler import CERTIFICATES_USE_ALLOWLIST
 from lms.djangoapps.certificates.models import CertificateStatuses, GeneratedCertificate
 from lms.djangoapps.certificates.tests.factories import CertificateWhitelistFactory, GeneratedCertificateFactory
 from lms.djangoapps.courseware.models import StudentModule
@@ -37,7 +39,10 @@ from lms.djangoapps.grades.models import PersistentCourseGrade, PersistentSubsec
 from lms.djangoapps.grades.subsection_grade import CreateSubsectionGrade
 from lms.djangoapps.grades.transformer import GradesTransformer
 from lms.djangoapps.instructor_analytics.basic import UNAVAILABLE, list_problem_responses
-from lms.djangoapps.instructor_task.tasks_helper.certs import generate_students_certificates
+from lms.djangoapps.instructor_task.tasks_helper.certs import (
+    generate_students_certificates,
+    _invalidate_generated_certificates
+)
 from lms.djangoapps.instructor_task.tasks_helper.enrollments import upload_may_enroll_csv, upload_students_csv
 from lms.djangoapps.instructor_task.tasks_helper.grades import (
     ENROLLED_IN_COURSE,
@@ -2025,11 +2030,11 @@ class TestCertificateGeneration(InstructorTaskModuleTestCase):
             'action_name': 'certificates generated',
             'total': 10,
             'attempted': 8,
-            'succeeded': 5,
-            'failed': 3,
+            'succeeded': 0,
+            'failed': 0,
             'skipped': 2
         }
-        with self.assertNumQueries(141):
+        with self.assertNumQueries(157):
             self.assertCertificatesGenerated(task_input, expected_results)
 
         expected_results = {
@@ -2076,7 +2081,7 @@ class TestCertificateGeneration(InstructorTaskModuleTestCase):
             'action_name': 'certificates generated',
             'total': 3,
             'attempted': 3,
-            'succeeded': 3,
+            'succeeded': 0,
             'failed': 0,
             'skipped': 0
         }
@@ -2129,7 +2134,7 @@ class TestCertificateGeneration(InstructorTaskModuleTestCase):
             'action_name': 'certificates generated',
             'total': expected_certs,
             'attempted': expected_certs,
-            'succeeded': expected_certs,
+            'succeeded': 0,
             'failed': 0,
             'skipped': 0
         }
@@ -2161,7 +2166,7 @@ class TestCertificateGeneration(InstructorTaskModuleTestCase):
             'action_name': 'certificates generated',
             'total': 1,
             'attempted': 1,
-            'succeeded': 1,
+            'succeeded': 0,
             'failed': 0,
             'skipped': 0,
         }
@@ -2182,7 +2187,7 @@ class TestCertificateGeneration(InstructorTaskModuleTestCase):
             'total': 1,
             'attempted': 1,
             'succeeded': 0,
-            'failed': 1,
+            'failed': 0,
             'skipped': 0,
         }
         self.assertCertificatesGenerated(task_input, expected_results)
@@ -2234,7 +2239,7 @@ class TestCertificateGeneration(InstructorTaskModuleTestCase):
             'action_name': 'certificates generated',
             'total': 10,
             'attempted': 5,
-            'succeeded': 5,
+            'succeeded': 0,
             'failed': 0,
             'skipped': 5
         }
@@ -2310,8 +2315,8 @@ class TestCertificateGeneration(InstructorTaskModuleTestCase):
             'action_name': 'certificates generated',
             'total': 10,
             'attempted': 5,
-            'succeeded': 2,
-            'failed': 3,
+            'succeeded': 0,
+            'failed': 0,
             'skipped': 5
         }
 
@@ -2407,7 +2412,7 @@ class TestCertificateGeneration(InstructorTaskModuleTestCase):
             'action_name': 'certificates generated',
             'total': 10,
             'attempted': 8,
-            'succeeded': 8,
+            'succeeded': 0,
             'failed': 0,
             'skipped': 2
         }
@@ -2498,13 +2503,45 @@ class TestCertificateGeneration(InstructorTaskModuleTestCase):
             'action_name': 'certificates generated',
             'total': 7,
             'attempted': 7,
-            'succeeded': 7,
+            'succeeded': 0,
             'failed': 0,
             'skipped': 0,
         }
 
         self.assertCertificatesGenerated(task_input, expected_results)
 
+    @override_waffle_flag(CERTIFICATES_USE_ALLOWLIST, active=True)
+    def test_invalidation(self):
+        # Create students
+        students = self._create_students(2)
+        s1 = students[0]
+        s2 = students[1]
+
+        # Generate certificates
+        for s in students:
+            GeneratedCertificateFactory.create(
+                user=s,
+                course_id=self.course.id,
+                status=CertificateStatuses.downloadable,
+                mode='verified'
+            )
+
+        # Whitelist a student
+        CertificateWhitelistFactory.create(user=s1, course_id=self.course.id)
+
+        statuses = [CertificateStatuses.downloadable]
+        _invalidate_generated_certificates(self.course.id, students, statuses)
+
+        certs = GeneratedCertificate.objects.filter(user=s1, course_id=self.course.id)
+        assert certs.count() == 1
+        downloadable_cert = certs.first()
+        assert downloadable_cert.status == CertificateStatuses.downloadable
+
+        certs = GeneratedCertificate.objects.filter(user=s2, course_id=self.course.id)
+        assert certs.count() == 1
+        invalidated_cert = certs.first()
+        assert invalidated_cert.status == CertificateStatuses.unavailable
+
     def assertCertificatesGenerated(self, task_input, expected_results):
         """
         Generate certificates for the given task_input and compare with expected_results.