feat!: Enable v2 of course certificates for all course runs (#27817)

This moves all course runs that use certificates over to V2 of course certificates, and disables the option for a course run to use V1 of course certificates.

MICROBA-1082
This commit is contained in:
Christie Rice
2021-06-15 15:08:40 -04:00
committed by GitHub
parent 4f5f829911
commit 59dc0a4a39
10 changed files with 12 additions and 429 deletions

View File

@@ -30,7 +30,3 @@ be true at the time the certificate is generated:
* The user must have passed the course run
* The user must not be a beta tester in the course run
* The course run must not be a CCX (custom edX course)
Note: the above requirements were written for V2 of course certificates, which
assumes the CourseWaffleFlag *certificates_revamp.use_updated* has been enabled
for the course run. If it has not been enabled, the prior logic will apply.

View File

@@ -342,11 +342,14 @@ def _can_set_cert_status_common(user, course_key):
return True
def is_using_v2_course_certificates(course_key):
def is_using_v2_course_certificates(course_key): # pylint: disable=unused-argument
"""
Return True if the course run is using v2 course certificates
Note: this currently always returns True. This is an interim step as we roll out the feature to all course runs,
and the method will be removed entirely in MICROBA-1083.
"""
return CERTIFICATES_USE_UPDATED.is_enabled(course_key)
return True
def is_on_certificate_allowlist(user, course_key):

View File

@@ -6,17 +6,13 @@ from unittest.mock import patch
import ddt
from django.core.management import call_command
from django.core.management.base import CommandError
from django.test.utils import override_settings
from common.djangoapps.course_modes.models import CourseMode
from common.djangoapps.student.tests.factories import CourseEnrollmentFactory, UserFactory
from lms.djangoapps.badges.events.course_complete import get_completion_badge
from lms.djangoapps.badges.models import BadgeAssertion
from lms.djangoapps.badges.tests.factories import BadgeAssertionFactory, CourseCompleteImageConfigurationFactory
from lms.djangoapps.badges.tests.factories import CourseCompleteImageConfigurationFactory
from lms.djangoapps.certificates.data import CertificateStatuses
from lms.djangoapps.certificates.models import GeneratedCertificate
from lms.djangoapps.certificates.tests.factories import GeneratedCertificateFactory
from lms.djangoapps.grades.tests.utils import mock_passing_grade
from openedx.core.djangoapps.content.course_overviews.tests.factories import CourseOverviewFactory
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
from xmodule.modulestore.tests.factories import CourseFactory, ItemFactory
@@ -68,56 +64,6 @@ class ResubmitErrorCertificatesTest(CertificateManagementTest):
"""Tests for the resubmit_error_certificates management command. """
ENABLED_SIGNALS = ['course_published']
@ddt.data(CourseMode.HONOR, CourseMode.VERIFIED)
def test_resubmit_error_certificate(self, mode):
# Create a certificate with status 'error'
self._create_cert(self.courses[0].id, self.user, CertificateStatuses.error, mode)
# Re-submit all certificates with status 'error'
call_command(self.command)
# Expect that the certificate was re-submitted
self._assert_cert_status(self.courses[0].id, self.user, CertificateStatuses.notpassing)
def test_resubmit_error_certificate_in_a_course(self):
# Create a certificate with status 'error'
# in three courses.
for idx in range(3):
self._create_cert(self.courses[idx].id, self.user, CertificateStatuses.error)
# Re-submit certificates for two of the courses
call_command(self.command, course_key_list=[
str(self.courses[0].id),
str(self.courses[1].id)
])
# Expect that the first two courses have been re-submitted,
# but not the third course.
self._assert_cert_status(self.courses[0].id, self.user, CertificateStatuses.notpassing)
self._assert_cert_status(self.courses[1].id, self.user, CertificateStatuses.notpassing)
self._assert_cert_status(self.courses[2].id, self.user, CertificateStatuses.error)
@ddt.data(
CertificateStatuses.deleted,
CertificateStatuses.deleting,
CertificateStatuses.downloadable,
CertificateStatuses.generating,
CertificateStatuses.notpassing,
CertificateStatuses.restricted,
CertificateStatuses.unavailable,
)
def test_resubmit_error_certificate_skips_non_error_certificates(self, other_status):
# Create certificates with an error status and some other status
self._create_cert(self.courses[0].id, self.user, CertificateStatuses.error)
self._create_cert(self.courses[1].id, self.user, other_status)
# Re-submit certificates for all courses
call_command(self.command)
# Only the certificate with status "error" should have been re-submitted
self._assert_cert_status(self.courses[0].id, self.user, CertificateStatuses.notpassing)
self._assert_cert_status(self.courses[1].id, self.user, other_status)
def test_resubmit_error_certificate_none_found(self):
self._create_cert(self.courses[0].id, self.user, CertificateStatuses.downloadable)
call_command(self.command)
@@ -146,111 +92,3 @@ class ResubmitErrorCertificatesTest(CertificateManagementTest):
invalid_key = "invalid/"
with self.assertRaisesRegex(CommandError, invalid_key):
call_command(self.command, course_key_list=[invalid_key])
@ddt.ddt
class RegenerateCertificatesTest(CertificateManagementTest):
"""
Tests for regenerating certificates.
"""
command = 'regenerate_user'
def setUp(self):
"""
We just need one course here.
"""
super().setUp()
self.course = self.courses[0]
@ddt.data(True, False)
@override_settings(CERT_QUEUE='test-queue')
@patch.dict('django.conf.settings.FEATURES', {'ENABLE_OPENBADGES': True})
@patch('lms.djangoapps.certificates.generation_handler.XQueueCertInterface', spec=True)
def test_clear_badge(self, issue_badges, xqueue):
"""
Given that I have a user with a badge
If I run regeneration for a user
Then certificate generation will be requested
And the badge will be deleted if badge issuing is enabled
"""
key = self.course.location.course_key
self._create_cert(key, self.user, CertificateStatuses.downloadable)
badge_class = get_completion_badge(key, self.user)
BadgeAssertionFactory(badge_class=badge_class, user=self.user)
assert BadgeAssertion.objects.filter(user=self.user, badge_class=badge_class)
self.course.issue_badges = issue_badges
self.store.update_item(self.course, None)
args = f'-u {self.user.email} -c {str(key)}'
call_command(self.command, *args.split(' '))
assert xqueue.return_value.regen_cert.call_args.args == (
self.user,
key,
)
regen_cert_call_kwargs = xqueue.return_value.regen_cert.call_args.kwargs
assert regen_cert_call_kwargs == {
'forced_grade': None,
'template_file': None,
'generate_pdf': True,
}
assert bool(BadgeAssertion.objects.filter(user=self.user, badge_class=badge_class)) == (not issue_badges)
@override_settings(CERT_QUEUE='test-queue')
@patch('capa.xqueue_interface.XQueueInterface.send_to_queue', spec=True)
def test_regenerating_certificate(self, mock_send_to_queue):
"""
Given that I have a user who has not passed course
If I run regeneration for that user
Then certificate generation will be not be requested
"""
key = self.course.location.course_key
self._create_cert(key, self.user, CertificateStatuses.downloadable)
args = f'-u {self.user.email} -c {str(key)} --insecure'
call_command(self.command, *args.split(' '))
certificate = GeneratedCertificate.eligible_certificates.get(
user=self.user,
course_id=key
)
assert certificate.status == CertificateStatuses.notpassing
assert not mock_send_to_queue.called
class UngenerateCertificatesTest(CertificateManagementTest):
"""
Tests for generating certificates.
"""
command = 'ungenerated_certs'
def setUp(self):
"""
We just need one course here.
"""
super().setUp()
self.course = self.courses[0]
@override_settings(CERT_QUEUE='test-queue')
@patch('capa.xqueue_interface.XQueueInterface.send_to_queue', spec=True)
def test_ungenerated_certificate(self, mock_send_to_queue):
"""
Given that I have ended course
If I run ungenerated certs command
Then certificates should be generated for all users who passed course
"""
mock_send_to_queue.return_value = (0, "Successfully queued")
key = self.course.location.course_key
self._create_cert(key, self.user, CertificateStatuses.unavailable)
with mock_passing_grade():
args = f'-c {str(key)} --insecure'
call_command(self.command, *args.split(' '))
assert mock_send_to_queue.called
certificate = GeneratedCertificate.eligible_certificates.get(
user=self.user,
course_id=key
)
assert certificate.status == CertificateStatuses.generating

View File

@@ -212,11 +212,6 @@ class CertificateDownloadableStatusTests(WebCertificateTestMixin, ModuleStoreTes
@patch.dict(settings.FEATURES, {'CERTIFICATES_HTML_VIEW': True})
def test_with_downloadable_web_cert(self):
CourseEnrollment.enroll(self.student, self.course.id, mode='honor')
self._setup_course_certificate()
with mock_passing_grade():
generate_user_certificates(self.student, self.course.id)
cert_status = certificate_status_for_student(self.student, self.course.id)
assert certificate_downloadable_status(self.student, self.course.id) ==\
{'is_downloadable': True,
@@ -242,10 +237,7 @@ class CertificateDownloadableStatusTests(WebCertificateTestMixin, ModuleStoreTes
self.course.certificate_available_date = cert_avail_date
self.course.save()
CourseEnrollment.enroll(self.student, self.course.id, mode='honor')
self._setup_course_certificate()
with mock_passing_grade():
generate_user_certificates(self.student, self.course.id)
downloadable_status = certificate_downloadable_status(self.student, self.course.id)
assert downloadable_status['is_downloadable'] == cert_downloadable_status
@@ -558,66 +550,17 @@ class GenerateUserCertificatesTest(EventTestMixin, WebCertificateTestMixin, Modu
self.enrollment = CourseEnrollment.enroll(self.student, self.course.id, mode='honor')
self.request_factory = RequestFactory()
def test_new_cert_requests_into_xqueue_returns_generating(self):
with mock_passing_grade():
with self._mock_queue():
generate_user_certificates(self.student, self.course.id)
# Verify that the certificate has status 'generating'
cert = GeneratedCertificate.eligible_certificates.get(user=self.student, course_id=self.course.id)
assert cert.status == CertificateStatuses.generating
self.assert_event_emitted(
'edx.certificate.created',
user_id=self.student.id,
course_id=str(self.course.id),
certificate_url=get_certificate_url(self.student.id, self.course.id),
certificate_id=cert.verify_uuid,
enrollment_mode=cert.mode,
generation_mode='batch'
)
def test_xqueue_submit_task_error(self):
with mock_passing_grade():
with self._mock_queue(is_successful=False):
generate_user_certificates(self.student, self.course.id)
# Verify that the certificate has been marked with status error
cert = GeneratedCertificate.eligible_certificates.get(user=self.student, course_id=self.course.id)
assert cert.status == CertificateStatuses.error
assert self.ERROR_REASON in cert.error_reason
def test_generate_user_certificates_with_unverified_cert_status(self):
"""
Generate user certificate when the certificate is unverified
will trigger an update to the certificate if the user has since
verified.
"""
self._setup_course_certificate()
# generate certificate with unverified status.
GeneratedCertificateFactory.create(
user=self.student,
course_id=self.course.id,
status=CertificateStatuses.unverified,
mode='verified'
)
with mock_passing_grade():
with self._mock_queue():
status = generate_user_certificates(self.student, self.course.id)
assert status == CertificateStatuses.generating
@patch.dict(settings.FEATURES, {'CERTIFICATES_HTML_VIEW': True})
def test_new_cert_requests_returns_generating_for_html_certificate(self):
def test_new_cert_request_for_html_certificate(self):
"""
Test no message sent to Xqueue if HTML certificate view is enabled
Test generate_user_certificates with HTML certificates
"""
self._setup_course_certificate()
with mock_passing_grade():
generate_user_certificates(self.student, self.course.id)
# Verify that the certificate has status 'downloadable'
cert = GeneratedCertificate.eligible_certificates.get(user=self.student, course_id=self.course.id)
assert cert.status == CertificateStatuses.downloadable
assert cert.status == CertificateStatuses.unverified
@patch.dict(settings.FEATURES, {'CERTIFICATES_HTML_VIEW': False})
def test_cert_url_empty_with_invalid_certificate(self):

View File

@@ -133,7 +133,6 @@ class AllowlistTests(ModuleStoreTestCase):
assert not _can_generate_allowlist_certificate(u, self.course_run_key)
assert not generate_allowlist_certificate_task(u, self.course_run_key)
assert not can_generate_certificate_task(u, self.course_run_key)
assert not generate_certificate_task(u, self.course_run_key)
assert _set_allowlist_cert_status(u, self.course_run_key) is None
@@ -321,7 +320,6 @@ class CertificateTests(ModuleStoreTestCase):
"""
other_user = UserFactory()
assert not _can_generate_v2_certificate(other_user, self.course_run_key)
assert not can_generate_certificate_task(other_user, self.course_run_key)
assert not generate_certificate_task(other_user, self.course_run_key)
assert not generate_regular_certificate_task(other_user, self.course_run_key)
@@ -331,13 +329,6 @@ class CertificateTests(ModuleStoreTestCase):
"""
assert is_using_v2_course_certificates(self.course_run_key)
@override_waffle_flag(CERTIFICATES_USE_UPDATED, active=False)
def test_is_using_updated_false(self):
"""
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),
@@ -479,13 +470,6 @@ class CertificateTests(ModuleStoreTestCase):
assert not _can_generate_v2_certificate(self.user, self.course_run_key)
assert _set_v2_cert_status(self.user, self.course_run_key) is None
@override_waffle_flag(CERTIFICATES_USE_UPDATED, active=False)
def test_cert_status_v1(self):
"""
Test cert status with V1 of course certs
"""
assert _set_v2_cert_status(self.user, self.course_run_key) is None
def test_cert_status_downloadable(self):
"""
Test cert status when status is already downloadable

View File

@@ -19,7 +19,6 @@ from lms.djangoapps.certificates.models import (
GeneratedCertificate
)
from lms.djangoapps.certificates.signals import _fire_ungenerated_certificate_task
from lms.djangoapps.certificates.tasks import CERTIFICATE_DELAY_SECONDS
from lms.djangoapps.certificates.tests.factories import CertificateAllowlistFactory, GeneratedCertificateFactory
from lms.djangoapps.grades.course_grade_factory import CourseGradeFactory
from lms.djangoapps.grades.tests.utils import mock_passing_grade
@@ -147,27 +146,6 @@ class PassingGradeCertsTest(ModuleStoreTestCase):
)
attempt.approve()
def test_cert_generation_on_passing_v1(self):
with mock.patch(
'lms.djangoapps.certificates.signals.generate_certificate.apply_async',
return_value=None
) as mock_generate_certificate_apply_async:
with override_waffle_switch(AUTO_CERTIFICATE_GENERATION_SWITCH, active=True):
grade_factory = CourseGradeFactory()
# Not passing
grade_factory.update(self.user, self.ip_course)
mock_generate_certificate_apply_async.assert_not_called()
# Certs fired after passing
with mock_passing_grade():
grade_factory.update(self.user, self.ip_course)
mock_generate_certificate_apply_async.assert_called_with(
countdown=CERTIFICATE_DELAY_SECONDS,
kwargs={
'student': str(self.user.id),
'course_key': str(self.ip_course.id),
}
)
def test_cert_already_generated(self):
with mock.patch(
'lms.djangoapps.certificates.signals.generate_certificate.apply_async',
@@ -199,7 +177,7 @@ class PassingGradeCertsTest(ModuleStoreTestCase):
return_value=None
) as mock_cert_task:
CourseGradeFactory().update(self.user, self.course)
mock_cert_task.assert_not_called()
mock_cert_task.assert_called_with(self.user, self.course.id)
# User who is on the allowlist
u = UserFactory.create()
@@ -398,26 +376,6 @@ class LearnerIdVerificationTest(ModuleStoreTestCase):
attempt.approve()
mock_cert_task.assert_called_with(self.user_two, self.course_two.id)
def test_cert_generation_on_photo_verification_v1(self):
with mock.patch(
'lms.djangoapps.certificates.signals.generate_certificate.apply_async',
return_value=None
) as mock_cert_task:
with override_waffle_switch(AUTO_CERTIFICATE_GENERATION_SWITCH, active=True):
attempt = SoftwareSecurePhotoVerification.objects.create(
user=self.user_two,
status='submitted'
)
attempt.approve()
mock_cert_task.assert_called_with(
countdown=CERTIFICATE_DELAY_SECONDS,
kwargs={
'student': str(self.user_two.id),
'course_key': str(self.course_two.id),
'expected_verification_status': 'approved'
}
)
def test_id_verification_allowlist(self):
# User is not on the allowlist
with mock.patch(

View File

@@ -5,7 +5,6 @@ Tests for certificate app views used by the support team.
import json
from unittest import mock
from unittest.mock import patch
from uuid import uuid4
import ddt
@@ -17,10 +16,8 @@ from opaque_keys.edx.keys import CourseKey
from common.djangoapps.student.models import CourseEnrollment
from common.djangoapps.student.roles import GlobalStaff, SupportStaffRole
from common.djangoapps.student.tests.factories import UserFactory
from lms.djangoapps.certificates.api import regenerate_user_certificates
from lms.djangoapps.certificates.models import CertificateInvalidation, CertificateStatuses, GeneratedCertificate
from lms.djangoapps.certificates.tests.factories import CertificateInvalidationFactory, GeneratedCertificateFactory
from lms.djangoapps.grades.tests.utils import mock_passing_grade
from openedx.core.djangoapps.content.course_overviews.tests.factories import CourseOverviewFactory
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
from xmodule.modulestore.tests.factories import CourseFactory
@@ -285,38 +282,6 @@ class CertificateRegenerateTests(CertificateSupportTestCase):
else:
assert response.status_code == 403
def test_regenerate_certificate(self):
"""Test web certificate regeneration."""
self.cert.download_url = ''
self.cert.save()
response = self._regenerate(
course_key=self.course_key,
username=self.STUDENT_USERNAME,
)
assert response.status_code == 200
# Check that the user's certificate was updated
# Since the student hasn't actually passed the course,
# we'd expect that the certificate status will be "notpassing"
cert = GeneratedCertificate.eligible_certificates.get(user=self.student)
assert cert.status == CertificateStatuses.notpassing
@patch('lms.djangoapps.certificates.queue.XQueueCertInterface._generate_cert')
def test_regenerate_certificate_for_honor_mode(self, mock_generate_cert):
"""Test web certificate regeneration for the users who have earned the
certificate in honor mode
"""
self.cert.mode = 'honor'
self.cert.download_url = ''
self.cert.save()
with mock_passing_grade(percent=0.75):
with patch('common.djangoapps.course_modes.models.CourseMode.mode_for_course') as mock_mode_for_course:
mock_mode_for_course.return_value = 'honor'
regenerate_user_certificates(self.student, self.course_key)
mock_generate_cert.assert_called()
def test_regenerate_certificate_missing_params(self):
# Missing username
response = self._regenerate(course_key=self.CERT_COURSE_KEY)
@@ -367,33 +332,6 @@ class CertificateRegenerateTests(CertificateSupportTestCase):
num_certs = GeneratedCertificate.eligible_certificates.filter(user=self.student).count()
assert num_certs == 1
@mock.patch(CAN_GENERATE_METHOD, mock.Mock(return_value=True))
def test_regenerate_cert_with_invalidated_record(self):
""" If the certificate is marked as invalid, regenerate the certificate. """
# mark certificate as invalid
self._invalidate_certificate(self.cert)
self.assertCertInvalidationExists()
# after invalidation certificate status become un-available.
self.assertGeneratedCertExists(
user=self.student, status=CertificateStatuses.unavailable
)
# Should be able to regenerate
response = self._regenerate(
course_key=self.CERT_COURSE_KEY,
username=self.STUDENT_USERNAME
)
assert response.status_code == 200
self.assertCertInvalidationExists()
# Check that the user's certificate was updated
# Since the student hasn't actually passed the course,
# we'd expect that the certificate status will be "notpassing"
self.assertGeneratedCertExists(
user=self.student, status=CertificateStatuses.notpassing
)
def _regenerate(self, course_key=None, username=None):
"""Call the regeneration end-point and return the response. """
url = reverse("certificates:regenerate_certificate_for_user")

View File

@@ -41,7 +41,6 @@ from lms.djangoapps.certificates.tests.factories import (
LinkedInAddToProfileConfigurationFactory
)
from lms.djangoapps.certificates.utils import get_certificate_url
from lms.djangoapps.grades.tests.utils import mock_passing_grade
from openedx.core.djangoapps.certificates.config import waffle
from openedx.core.djangoapps.dark_lang.models import DarkLangConfig
from openedx.core.djangoapps.site_configuration.tests.test_util import (
@@ -966,21 +965,7 @@ class CertificatesViewsTests(CommonCertificatesTestCase, CacheIsolationTestCase)
response = self.client.post(request_certificate_url, {'course_id': str(self.course.id)})
assert response.status_code == 200
response_json = json.loads(response.content.decode('utf-8'))
assert CertificateStatuses.notpassing == response_json['add_status']
@override_settings(FEATURES=FEATURES_WITH_CERTS_DISABLED)
@override_settings(CERT_QUEUE='test-queue')
def test_request_certificate_after_passing(self):
self.cert.status = CertificateStatuses.unavailable
self.cert.save()
request_certificate_url = reverse('request_certificate')
with patch('capa.xqueue_interface.XQueueInterface.send_to_queue') as mock_queue:
mock_queue.return_value = (0, "Successfully queued")
with mock_passing_grade():
response = self.client.post(request_certificate_url, {'course_id': str(self.course.id)})
assert response.status_code == 200
response_json = json.loads(response.content.decode('utf-8'))
assert CertificateStatuses.generating == response_json['add_status']
assert CertificateStatuses.unavailable == response_json['add_status']
# TEMPLATES WITHOUT LANGUAGE TESTS
@override_settings(FEATURES=FEATURES_WITH_CUSTOM_CERTS_ENABLED)

View File

@@ -2271,15 +2271,6 @@ class GenerateUserCertTests(ModuleStoreTestCase):
resp = self.client.post(self.url)
assert resp.status_code == 200
# Verify Google Analytics event fired after generating certificate
mock_tracker.track.assert_called_once_with(
self.student.id,
'edx.bi.user.certificate.generate',
{
'category': 'certificates',
'label': str(self.course.id)
},
)
mock_tracker.reset_mock()
def test_user_with_passing_existing_generating_cert(self):

View File

@@ -2034,7 +2034,7 @@ class TestCertificateGeneration(InstructorTaskModuleTestCase):
'failed': 0,
'skipped': 2
}
with self.assertNumQueries(114):
with self.assertNumQueries(74):
self.assertCertificatesGenerated(task_input, expected_results)
@ddt.data(
@@ -2433,59 +2433,6 @@ class TestCertificateGeneration(InstructorTaskModuleTestCase):
self.assertCertificatesGenerated(task_input, expected_results)
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'
)
# Allowlist a student
CertificateAllowlistFactory.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
@override_waffle_flag(CERTIFICATES_USE_UPDATED, active=False)
def test_invalidation_v2_certificates_disabled(self):
"""
Test that ensures the bulk invalidation step (as part of bulk certificate regeneration) continues to occur when
the v2 certificates feature is disabled for a course run.
"""
students = self._create_students(2)
for s in students:
GeneratedCertificateFactory.create(
user=s,
course_id=self.course.id,
status=CertificateStatuses.downloadable,
mode='verified'
)
_invalidate_generated_certificates(self.course.id, students, [CertificateStatuses.downloadable])
for s in students:
cert = GeneratedCertificate.objects.get(user=s, course_id=self.course.id)
assert cert.status == CertificateStatuses.unavailable
@override_waffle_flag(CERTIFICATES_USE_UPDATED, active=True)
def test_invalidation_v2_certificates_enabled(self):
"""