diff --git a/lms/djangoapps/certificates/api.py b/lms/djangoapps/certificates/api.py index 33edca00d8..0df1d661c5 100644 --- a/lms/djangoapps/certificates/api.py +++ b/lms/djangoapps/certificates/api.py @@ -38,7 +38,6 @@ from lms.djangoapps.certificates.models import ( ExampleCertificateSet, GeneratedCertificate, ) -from lms.djangoapps.certificates.queue import XQueueCertInterface from lms.djangoapps.certificates.utils import ( get_certificate_url as _get_certificate_url, has_html_certificates_enabled as _has_html_certificates_enabled, @@ -353,31 +352,30 @@ def cert_generation_enabled(course_key): def generate_example_certificates(course_key): - """Generate example certificates for a course. + """Generate example (PDF) certificates for a course. - Example certificates are used to validate that certificates - are configured correctly for the course. Staff members can - view the example certificates before enabling - the self-generated certificates button for students. + Example certificates were used to validate that certificates were configured correctly for the course. Staff + members could view the example certificates before enabling the self-generated certificates button for students. - Several example certificates may be generated for a course. - For example, if a course offers both verified and honor certificates, - examples of both types of certificate will be generated. + [07/20/2021 Update] + This function was updated to remove the references to queue.py, which has been removed as part of MICROBA-1227, and + no longer can fulfill the function it was originally created for. There is further cleanup around PDF certificate + generation code, part of DEPR-155, that will remove this function. See DEPR-155 and MICROBA-1094 for additional + info. - If an error occurs while starting the certificate generation - job, the errors will be recorded in the database and - can be retrieved using `example_certificate_status()`. + It may be important to note that this functionality has been broken since 2018 when the ability to generate PDF + certificates was ripped out of edx-platform. This will be removed as part of MICROBA-1394. Arguments: course_key (CourseKey): The course identifier. Returns: None - """ - xqueue = XQueueCertInterface() - for cert in ExampleCertificateSet.create_example_set(course_key): - xqueue.add_example_cert(cert) + log.warning( + "Generating example certificates is no longer supported. Skipping generation of example certificates for " + f"course {course_key}" + ) def example_certificates_status(course_key): diff --git a/lms/djangoapps/certificates/queue.py b/lms/djangoapps/certificates/queue.py deleted file mode 100644 index dc26b217ee..0000000000 --- a/lms/djangoapps/certificates/queue.py +++ /dev/null @@ -1,582 +0,0 @@ -"""Interface for adding certificate generation tasks to the XQueue. """ - - -import json -import logging -import random -from uuid import uuid4 - -import lxml.html -from django.conf import settings -from django.test.client import RequestFactory -from django.urls import reverse -from django.utils.encoding import python_2_unicode_compatible -from lxml.etree import ParserError, XMLSyntaxError -from requests.auth import HTTPBasicAuth - -from capa.xqueue_interface import XQueueInterface, make_hashkey, make_xheader -from common.djangoapps.course_modes import api as modes_api -from common.djangoapps.course_modes.models import CourseMode -from common.djangoapps.student.models import CourseEnrollment, UserProfile -from lms.djangoapps.certificates.data import CertificateStatuses as status -from lms.djangoapps.certificates.models import ( - CertificateAllowlist, - ExampleCertificate, - GeneratedCertificate, -) -from lms.djangoapps.certificates.utils import certificate_status_for_student -from lms.djangoapps.grades.api import CourseGradeFactory -from lms.djangoapps.verify_student.services import IDVerificationService -from openedx.core.djangoapps.content.course_overviews.api import get_course_overview_or_none - -LOGGER = logging.getLogger(__name__) - - -@python_2_unicode_compatible -class XQueueAddToQueueError(Exception): - """An error occurred when adding a certificate task to the queue. """ - - def __init__(self, error_code, error_msg): - self.error_code = error_code - self.error_msg = error_msg - super().__init__(str(self)) - - def __str__(self): - return ( - "Could not add certificate to the XQueue. " - "The error code was '{code}' and the message was '{msg}'." - ).format( - code=self.error_code, - msg=self.error_msg - ) - - -class XQueueCertInterface: - """ - XQueueCertificateInterface provides an - interface to the xqueue server for - managing student certificates. - - Instantiating an object will create a new - connection to the queue server. - - Summary of methods: - - add_cert: Add a new certificate. Puts a single - request on the queue for the student/course. - Once the certificate is generated a post - will be made to the update_certificate - view which will save the certificate - download URL. - - regen_cert: Regenerate an existing certificate. - For a user that already has a certificate - this will delete the existing one and - generate a new cert. - - - del_cert: Delete an existing certificate - For a user that already has a certificate - this will delete their cert. - - """ - - def __init__(self, request=None): - - # Get basic auth (username/password) for - # xqueue connection if it's in the settings - - if settings.XQUEUE_INTERFACE.get('basic_auth') is not None: - requests_auth = HTTPBasicAuth( - *settings.XQUEUE_INTERFACE['basic_auth']) - else: - requests_auth = None - - if request is None: - factory = RequestFactory() - self.request = factory.get('/') - else: - self.request = request - - self.xqueue_interface = XQueueInterface( - settings.XQUEUE_INTERFACE['url'], - settings.XQUEUE_INTERFACE['django_auth'], - requests_auth, - ) - self.allowlist = CertificateAllowlist.objects.all() - self.use_https = True - - def regen_cert(self, student, course_id, forced_grade=None, template_file=None, generate_pdf=True): - """(Re-)Make certificate for a particular student in a particular course - - Arguments: - student - User.object - course_id - courseenrollment.course_id (string) - - [PDF Certificates only] - WARNING: this command will leave the old certificate, if one exists, - laying around in AWS taking up space. If this is a problem, - take pains to clean up storage before running this command. - - Invalidate the certificate (if it exists) and request a new certificate. - - Return the certificate. - """ - # TODO: when del_cert is implemented and plumbed through certificates - # repo also, do a deletion followed by a creation r/t a simple - # recreation. XXX: this leaves orphan cert files laying around in - # AWS. See note in the docstring too. - try: - certificate = GeneratedCertificate.eligible_certificates.get(user=student, course_id=course_id) - - LOGGER.info( - f"Found an existing certificate entry for student {student.id} in course '{course_id}' with status " - f"'{certificate.status}' while regenerating certificates." - ) - - if certificate.download_url: - self._log_pdf_cert_generation_discontinued_warning( - student.id, course_id, certificate.status, certificate.download_url - ) - return None - - certificate.invalidate(source='certificate_regeneration') - - LOGGER.info( - f"The certificate status for student {student.id} in course '{course_id} has been changed to " - f"'{certificate.status}'." - ) - except GeneratedCertificate.DoesNotExist: - pass - - return self.add_cert( - student, - course_id, - forced_grade=forced_grade, - template_file=template_file, - generate_pdf=generate_pdf - ) - - def del_cert(self, student, course_id): - """ - Arguments: - student - User.object - course_id - courseenrollment.course_id (string) - - Removes certificate for a student, will change - the certificate status to 'deleting'. - - Certificate must be in the 'error' or 'downloadable' state - otherwise it will return the current state - - """ - - raise NotImplementedError - - # pylint: disable=too-many-statements - def add_cert(self, student, course_id, forced_grade=None, template_file=None, generate_pdf=True): - """ - Request a new certificate for a student. - - Arguments: - student - User.object - course_id - courseenrollment.course_id (CourseKey) - forced_grade - a string indicating a grade parameter to pass with - the certificate request. If this is given, grading - will be skipped. - generate_pdf - Boolean should a message be sent in queue to generate certificate PDF - - Will change the certificate status to 'generating' or - `downloadable` in case of web view certificates. - - The course must not be a CCX. - - Certificate must be in the 'unavailable', 'error', - 'deleted' or 'generating' state. - - If a student has a passing grade or is in the allowlist - table for the course a request will be made for a new cert. - - If a student does not have a passing grade the status - will change to status.notpassing - - Returns the newly created certificate instance - """ - - if hasattr(course_id, 'ccx'): - LOGGER.warning( - ( - "Cannot create certificate generation task for user %s " - "in the course '%s'; " - "certificates are not allowed for CCX courses." - ), - student.id, - str(course_id) - ) - return None - - valid_statuses = [ - status.generating, - status.unavailable, - status.deleted, - status.error, - status.notpassing, - status.downloadable, - status.auditing, - status.audit_passing, - status.audit_notpassing, - status.unverified, - ] - - cert_status_dict = certificate_status_for_student(student, course_id) - cert_status = cert_status_dict.get('status') - download_url = cert_status_dict.get('download_url') - cert = None - if download_url: - self._log_pdf_cert_generation_discontinued_warning( - student.id, course_id, cert_status, download_url - ) - return None - - if cert_status not in valid_statuses: - LOGGER.warning( - ( - "Cannot create certificate generation task for user %s " - "in the course '%s'; " - "the certificate status '%s' is not one of %s." - ), - student.id, - str(course_id), - cert_status, - str(valid_statuses) - ) - return None - - profile = UserProfile.objects.get(user=student) - profile_name = profile.name - - # Needed for access control in grading. - self.request.user = student - self.request.session = {} - - is_allowlisted = self.allowlist.filter(user=student, course_id=course_id, allowlist=True).exists() - course_grade = CourseGradeFactory().read(student, course_key=course_id) - enrollment_mode, __ = CourseEnrollment.enrollment_mode_for_user(student, course_id) - mode_is_verified = enrollment_mode in GeneratedCertificate.VERIFIED_CERTS_MODES - user_is_verified = IDVerificationService.user_is_verified(student) - cert_mode = enrollment_mode - - is_eligible_for_certificate = modes_api.is_eligible_for_certificate(enrollment_mode, cert_status) - if is_allowlisted and not is_eligible_for_certificate: - # check if audit certificates are enabled for audit mode - is_eligible_for_certificate = enrollment_mode != CourseMode.AUDIT or \ - not settings.FEATURES['DISABLE_AUDIT_CERTIFICATES'] - - unverified = False - # For credit mode generate verified certificate - if cert_mode in (CourseMode.CREDIT_MODE, CourseMode.MASTERS): - cert_mode = CourseMode.VERIFIED - - if template_file is not None: - template_pdf = template_file - elif mode_is_verified and user_is_verified: - template_pdf = "certificate-template-{id.org}-{id.course}-verified.pdf".format(id=course_id) - elif mode_is_verified and not user_is_verified: - template_pdf = "certificate-template-{id.org}-{id.course}.pdf".format(id=course_id) - if CourseMode.mode_for_course(course_id, CourseMode.HONOR): - cert_mode = GeneratedCertificate.MODES.honor - else: - unverified = True - else: - # honor code and audit students - template_pdf = "certificate-template-{id.org}-{id.course}.pdf".format(id=course_id) - - LOGGER.info( - ( - "Certificate generated for student %s in the course: %s with template: %s. " - "given template: %s, " - "user is verified: %s, " - "mode is verified: %s," - "generate_pdf is: %s" - ), - student.username, - str(course_id), - template_pdf, - template_file, - user_is_verified, - mode_is_verified, - generate_pdf - ) - cert, __ = GeneratedCertificate.objects.get_or_create(user=student, course_id=course_id) - - cert.mode = cert_mode - cert.user = student - cert.grade = course_grade.percent - cert.course_id = course_id - cert.name = profile_name - cert.download_url = '' - - # Strip HTML from grade range label - grade_contents = forced_grade or course_grade.letter_grade - passing = False - try: - grade_contents = lxml.html.fromstring(grade_contents).text_content() - passing = True - except (TypeError, XMLSyntaxError, ParserError) as exc: - LOGGER.info( - ( - "Could not retrieve grade for student %s " - "in the course '%s' " - "because an exception occurred while parsing the " - "grade contents '%s' as HTML. " - "The exception was: '%s'" - ), - student.id, - str(course_id), - grade_contents, - str(exc) - ) - - # Check if the student is on the allowlist - if is_allowlisted: - LOGGER.info( - "Student %s is on the certificate allowlist in '%s'", - student.id, - str(course_id) - ) - passing = True - - # If this user's enrollment is not eligible to receive a - # certificate, mark it as such for reporting and - # analytics. Only do this if the certificate is new, or - # already marked as ineligible -- we don't want to mark - # existing audit certs as ineligible. - if not is_eligible_for_certificate: - cert.status = status.audit_passing if passing else status.audit_notpassing - cert.save() - LOGGER.info( - "Student %s with enrollment mode %s is not eligible for a certificate.", - student.id, - enrollment_mode - ) - return cert - # If they are not passing, short-circuit and don't generate cert - elif not passing: - cert.status = status.notpassing - cert.save() - - LOGGER.info( - ( - "Student %s does not have a grade for '%s', " - "so their certificate status has been set to '%s'. " - "No certificate generation task was sent to the XQueue." - ), - student.id, - str(course_id), - cert.status - ) - return cert - - if unverified: - cert.status = status.unverified - cert.save() - LOGGER.info( - ( - "User %s has a verified enrollment in course %s " - "but is missing ID verification. " - "Certificate status has been set to unverified" - ), - student.id, - str(course_id), - ) - return cert - - # Finally, generate the certificate and send it off. - return self._generate_cert(cert, student, grade_contents, template_pdf, generate_pdf) - - def _generate_cert(self, cert, student, grade_contents, template_pdf, generate_pdf): - """ - Generate a certificate for the student. If `generate_pdf` is True, - sends a request to XQueue. - """ - course_id = str(cert.course_id) - course_overview = get_course_overview_or_none(course_id) - if not course_overview: - LOGGER.warning(f"Skipping cert generation for {student.id} due to missing course overview for {course_id}") - return cert - - key = make_hashkey(random.random()) - cert.key = key - contents = { - 'action': 'create', - 'username': student.username, - 'course_id': course_id, - 'course_name': course_overview.display_name or course_id, - 'name': cert.name, - 'grade': grade_contents, - 'template_pdf': template_pdf, - } - if generate_pdf: - cert.status = status.generating - else: - cert.status = status.downloadable - cert.verify_uuid = uuid4().hex - - cert.save() - logging.info('certificate generated for user: %s with generate_pdf status: %s', - student.username, generate_pdf) - - if generate_pdf: - try: - self._send_to_xqueue(contents, key) - except XQueueAddToQueueError as exc: - cert.status = ExampleCertificate.STATUS_ERROR - cert.error_reason = str(exc) - cert.save() - LOGGER.critical( - ( - "Could not add certificate task to XQueue. " - "The course was '%s' and the student was '%s'." - "The certificate task status has been marked as 'error' " - "and can be re-submitted with a management command." - ), course_id, student.id - ) - else: - LOGGER.info( - ( - "The certificate status has been set to '%s'. " - "Sent a certificate grading task to the XQueue " - "with the key '%s'. " - ), - cert.status, - key - ) - return cert - - def add_example_cert(self, example_cert): - """Add a task to create an example certificate. - - Unlike other certificates, an example certificate is - not associated with any particular user and is never - shown to students. - - If an error occurs when adding the example certificate - to the queue, the example certificate status - will be set to "error". - - Arguments: - example_cert (ExampleCertificate) - - """ - contents = { - 'action': 'create', - 'course_id': str(example_cert.course_key), - 'name': example_cert.full_name, - 'template_pdf': example_cert.template, - - # Example certificates are not associated with a particular user. - # However, we still need to find the example certificate when - # we receive a response from the queue. For this reason, - # we use the example certificate's unique identifier as a username. - # Note that the username is *not* displayed on the certificate; - # it is used only to identify the certificate task in the queue. - 'username': example_cert.uuid, - - # We send this extra parameter to differentiate - # example certificates from other certificates. - # This is not used by the certificates workers or XQueue. - 'example_certificate': True, - } - - # The callback for example certificates is different than the callback - # for other certificates. Although both tasks use the same queue, - # we can distinguish whether the certificate was an example cert based - # on which end-point XQueue uses once the task completes. - callback_url_path = reverse('update_example_certificate') - - try: - self._send_to_xqueue( - contents, - example_cert.access_key, - task_identifier=example_cert.uuid, - callback_url_path=callback_url_path - ) - LOGGER.info("Started generating example certificates for course '%s'.", example_cert.course_key) - except XQueueAddToQueueError as exc: - example_cert.update_status( - ExampleCertificate.STATUS_ERROR, - error_reason=str(exc) - ) - LOGGER.critical( - ( - "Could not add example certificate with uuid '%s' to XQueue. " - "The exception was %s. " - "The example certificate has been marked with status 'error'." - ), example_cert.uuid, str(exc) - ) - - def _send_to_xqueue(self, contents, key, task_identifier=None, callback_url_path='/update_certificate'): - """Create a new task on the XQueue. - - Arguments: - contents (dict): The contents of the XQueue task. - key (str): An access key for the task. This will be sent - to the callback end-point once the task completes, - so that we can validate that the sender is the same - entity that received the task. - - Keyword Arguments: - callback_url_path (str): The path of the callback URL. - If not provided, use the default end-point for student-generated - certificates. - - """ - callback_url = '{protocol}://{base_url}{path}'.format( - protocol=("https" if self.use_https else "http"), - base_url=settings.SITE_NAME, - path=callback_url_path - ) - - # Append the key to the URL - # This is necessary because XQueue assumes that only one - # submission is active for a particular URL. - # If it receives a second submission with the same callback URL, - # it "retires" any other submission with the same URL. - # This was a hack that depended on the URL containing the user ID - # and courseware location; an assumption that does not apply - # to certificate generation. - # XQueue also truncates the callback URL to 128 characters, - # but since our key lengths are shorter than that, this should - # not affect us. - callback_url += "?key={key}".format( - key=( - task_identifier - if task_identifier is not None - else key - ) - ) - - xheader = make_xheader(callback_url, key, settings.CERT_QUEUE) - - (error, msg) = self.xqueue_interface.send_to_queue( - header=xheader, body=json.dumps(contents)) - if error: - exc = XQueueAddToQueueError(error, msg) - LOGGER.critical(str(exc)) - raise exc - - def _log_pdf_cert_generation_discontinued_warning(self, student_id, course_id, cert_status, download_url): - """Logs PDF certificate generation discontinued warning.""" - LOGGER.warning( - ( - "PDF certificate generation discontinued, canceling " - "PDF certificate generation for student %s " - "in course '%s' " - "with status '%s' " - "and download_url '%s'." - ), - student_id, - str(course_id), - cert_status, - download_url - ) diff --git a/lms/djangoapps/certificates/tests/test_api.py b/lms/djangoapps/certificates/tests/test_api.py index b460e41b1b..65858492af 100644 --- a/lms/djangoapps/certificates/tests/test_api.py +++ b/lms/djangoapps/certificates/tests/test_api.py @@ -2,7 +2,6 @@ import uuid -from contextlib import contextmanager from datetime import datetime, timedelta from unittest import mock from unittest.mock import patch @@ -36,9 +35,7 @@ from lms.djangoapps.certificates.api import ( certificate_downloadable_status, create_certificate_invalidation_entry, create_or_update_certificate_allowlist_entry, - example_certificates_status, generate_certificate_task, - generate_example_certificates, get_allowlist_entry, get_allowlisted_users, get_certificate_footer_context, @@ -58,10 +55,8 @@ from lms.djangoapps.certificates.api import ( from lms.djangoapps.certificates.models import ( CertificateGenerationConfiguration, CertificateStatuses, - ExampleCertificate, GeneratedCertificate, ) -from lms.djangoapps.certificates.queue import XQueueAddToQueueError, XQueueCertInterface from lms.djangoapps.certificates.tests.factories import ( CertificateAllowlistFactory, GeneratedCertificateFactory, @@ -80,20 +75,6 @@ class WebCertificateTestMixin: """ Mixin with helpers for testing Web Certificates. """ - @contextmanager - def _mock_queue(self, is_successful=True): - """ - Mock the "send to XQueue" method to return either success or an error. - """ - symbol = 'capa.xqueue_interface.XQueueInterface.send_to_queue' - with patch(symbol) as mock_send_to_queue: - if is_successful: - mock_send_to_queue.return_value = (0, "Successfully queued") - else: - mock_send_to_queue.side_effect = XQueueAddToQueueError(1, self.ERROR_REASON) - - yield mock_send_to_queue - def _setup_course_certificate(self): """ Creates certificate configuration for course @@ -646,69 +627,6 @@ class CertificateGenerationEnabledTest(EventTestMixin, TestCase): assert expect_enabled == actual_enabled -class GenerateExampleCertificatesTest(ModuleStoreTestCase): - """Test generation of example certificates. """ - - COURSE_KEY = CourseLocator(org='test', course='test', run='test') - - def test_generate_example_certs(self): - # Generate certificates for the course - CourseModeFactory.create(course_id=self.COURSE_KEY, mode_slug=CourseMode.HONOR) - with self._mock_xqueue() as mock_queue: - generate_example_certificates(self.COURSE_KEY) - - # Verify that the appropriate certs were added to the queue - self._assert_certs_in_queue(mock_queue, 1) - - # Verify that the certificate status is "started" - self._assert_cert_status({ - 'description': 'honor', - 'status': 'started' - }) - - def test_generate_example_certs_with_verified_mode(self): - # Create verified and honor modes for the course - CourseModeFactory.create(course_id=self.COURSE_KEY, mode_slug='honor') - CourseModeFactory.create(course_id=self.COURSE_KEY, mode_slug='verified') - - # Generate certificates for the course - with self._mock_xqueue() as mock_queue: - generate_example_certificates(self.COURSE_KEY) - - # Verify that the appropriate certs were added to the queue - self._assert_certs_in_queue(mock_queue, 2) - - # Verify that the certificate status is "started" - self._assert_cert_status( - { - 'description': 'verified', - 'status': 'started' - }, - { - 'description': 'honor', - 'status': 'started' - } - ) - - @contextmanager - def _mock_xqueue(self): - """Mock the XQueue method for adding a task to the queue. """ - with patch.object(XQueueCertInterface, 'add_example_cert') as mock_queue: - yield mock_queue - - def _assert_certs_in_queue(self, mock_queue, expected_num): - """Check that the certificate generation task was added to the queue. """ - certs_in_queue = [call_args[0] for (call_args, __) in mock_queue.call_args_list] - assert len(certs_in_queue) == expected_num - for cert in certs_in_queue: - assert isinstance(cert, ExampleCertificate) - - def _assert_cert_status(self, *expected_statuses): - """Check the example certificate status. """ - actual_status = example_certificates_status(self.COURSE_KEY) - assert list(expected_statuses) == actual_status - - @override_settings(FEATURES=FEATURES_WITH_CERTS_ENABLED) class CertificatesBrandingTest(ModuleStoreTestCase): """Test certificates branding. """ diff --git a/lms/djangoapps/certificates/tests/test_queue.py b/lms/djangoapps/certificates/tests/test_queue.py deleted file mode 100644 index 570e986888..0000000000 --- a/lms/djangoapps/certificates/tests/test_queue.py +++ /dev/null @@ -1,366 +0,0 @@ -"""Tests for the XQueue certificates interface. """ - - -import json -from contextlib import contextmanager -from unittest.mock import Mock, patch - -import ddt -from django.conf import settings -from django.test import TestCase -from django.test.utils import override_settings -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 common.djangoapps.course_modes import api as modes_api -from common.djangoapps.course_modes.models import CourseMode -from common.djangoapps.student.tests.factories import CourseEnrollmentFactory, UserFactory -from lms.djangoapps.certificates.models import ( - CertificateStatuses, - ExampleCertificate, - ExampleCertificateSet, - GeneratedCertificate -) -from lms.djangoapps.certificates.queue import LOGGER, XQueueCertInterface -from lms.djangoapps.certificates.tests.factories import CertificateAllowlistFactory, 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 -@override_settings(CERT_QUEUE='certificates') -class XQueueCertInterfaceAddCertificateTest(ModuleStoreTestCase): - """Test the "add to queue" operation of the XQueue interface. """ - - def setUp(self): - super().setUp() - self.user = UserFactory.create() - self.course = CourseFactory.create() - self.enrollment = CourseEnrollmentFactory( - user=self.user, - course_id=self.course.id, - is_active=True, - mode="honor", - ) - self.xqueue = XQueueCertInterface() - self.user_2 = UserFactory.create() - SoftwareSecurePhotoVerificationFactory.create(user=self.user_2, status='approved') - - def test_add_cert_callback_url(self): - - with mock_passing_grade(): - with patch.object(XQueueInterface, 'send_to_queue') as mock_send: - mock_send.return_value = (0, None) - self.xqueue.add_cert(self.user, self.course.id) - - # Verify that the task was sent to the queue with the correct callback URL - assert mock_send.called - __, kwargs = mock_send.call_args_list[0] - actual_header = json.loads(kwargs['header']) - assert 'https://edx.org/update_certificate?key=' in actual_header['lms_callback_url'] - - def test_no_create_action_in_queue_for_html_view_certs(self): - """ - Tests there is no certificate create message in the queue if generate_pdf is False - """ - with mock_passing_grade(): - with patch.object(XQueueInterface, 'send_to_queue') as mock_send: - self.xqueue.add_cert(self.user, self.course.id, generate_pdf=False) - - # Verify that add_cert method does not add message to queue - assert not mock_send.called - certificate = GeneratedCertificate.eligible_certificates.get(user=self.user, course_id=self.course.id) - assert certificate.status == CertificateStatuses.downloadable - assert certificate.verify_uuid is not None - - @ddt.data('honor', 'audit') - def test_add_cert_with_honor_certificates(self, mode): - """Test certificates generations for honor and audit modes.""" - template_name = 'certificate-template-{id.org}-{id.course}.pdf'.format( - id=self.course.id - ) - mock_send = self.add_cert_to_queue(mode) - if modes_api.is_eligible_for_certificate(mode): - self.assert_certificate_generated(mock_send, mode, template_name) - else: - self.assert_ineligible_certificate_generated(mock_send, mode) - - @ddt.data('credit', 'verified') - def test_add_cert_with_verified_certificates(self, mode): - """Test if enrollment mode is verified or credit along with valid - software-secure verification than verified certificate should be generated. - """ - template_name = 'certificate-template-{id.org}-{id.course}-verified.pdf'.format( - id=self.course.id - ) - - mock_send = self.add_cert_to_queue(mode) - self.assert_certificate_generated(mock_send, 'verified', template_name) - - @ddt.data((True, CertificateStatuses.audit_passing), (False, CertificateStatuses.generating)) - @ddt.unpack - def test_ineligible_cert_allowlisted(self, disable_audit_cert, status): - """ - Test that audit mode students receive a certificate if DISABLE_AUDIT_CERTIFICATES - feature is set to false - """ - # Enroll as audit - CourseEnrollmentFactory( - user=self.user_2, - course_id=self.course.id, - is_active=True, - mode='audit' - ) - # Add student to the allowlist - CertificateAllowlistFactory(course_id=self.course.id, user=self.user_2) - - features = settings.FEATURES - features['DISABLE_AUDIT_CERTIFICATES'] = disable_audit_cert - with override_settings(FEATURES=features) and mock_passing_grade(): - with patch.object(XQueueInterface, 'send_to_queue') as mock_send: - mock_send.return_value = (0, None) - self.xqueue.add_cert(self.user_2, self.course.id) - - certificate = GeneratedCertificate.certificate_for_student(self.user_2, self.course.id) - assert certificate is not None - assert certificate.mode == 'audit' - assert certificate.status == status - - def add_cert_to_queue(self, mode): - """ - Dry method for course enrollment and adding request to - queue. Returns a mock object containing information about the - `XQueueInterface.send_to_queue` method, which can be used in other - assertions. - """ - CourseEnrollmentFactory( - user=self.user_2, - course_id=self.course.id, - is_active=True, - mode=mode, - ) - with mock_passing_grade(): - with patch.object(XQueueInterface, 'send_to_queue') as mock_send: - mock_send.return_value = (0, None) - self.xqueue.add_cert(self.user_2, self.course.id) - return mock_send - - def assert_certificate_generated(self, mock_send, expected_mode, expected_template_name): - """ - Assert that a certificate was generated with the correct mode and - template type. - """ - # Verify that the task was sent to the queue with the correct callback URL - assert mock_send.called - __, kwargs = mock_send.call_args_list[0] - - actual_header = json.loads(kwargs['header']) - assert 'https://edx.org/update_certificate?key=' in actual_header['lms_callback_url'] - - body = json.loads(kwargs['body']) - assert expected_template_name in body['template_pdf'] - - certificate = GeneratedCertificate.eligible_certificates.get(user=self.user_2, course_id=self.course.id) - assert certificate.mode == expected_mode - - def assert_ineligible_certificate_generated(self, mock_send, expected_mode): - """ - Assert that an ineligible certificate was generated with the - correct mode. - """ - # Ensure the certificate was not generated - assert not mock_send.called - - certificate = GeneratedCertificate.objects.get( - user=self.user_2, - course_id=self.course.id - ) - - assert certificate.status in (CertificateStatuses.audit_passing, CertificateStatuses.audit_notpassing) - assert certificate.mode == expected_mode - - @ddt.data( - (CertificateStatuses.restricted, False), - (CertificateStatuses.deleting, False), - (CertificateStatuses.generating, True), - (CertificateStatuses.unavailable, True), - (CertificateStatuses.deleted, True), - (CertificateStatuses.error, True), - (CertificateStatuses.notpassing, True), - (CertificateStatuses.downloadable, True), - (CertificateStatuses.auditing, True), - ) - @ddt.unpack - def test_add_cert_statuses(self, status, should_generate): - """ - Test that certificates can or cannot be generated with the given - certificate status. - """ - with patch( - 'lms.djangoapps.certificates.queue.certificate_status_for_student', - Mock(return_value={'status': status}) - ): - mock_send = self.add_cert_to_queue('verified') - if should_generate: - assert mock_send.called - else: - assert not mock_send.called - - def test_regen_cert_with_pdf_certificate(self): - """ - Test that regenerating a PDF certificate logs a warning message and the certificate - status remains unchanged. - """ - download_url = 'http://www.example.com/certificate.pdf' - # Create an existing verified enrollment and certificate - CourseEnrollmentFactory( - user=self.user_2, - course_id=self.course.id, - is_active=True, - mode=CourseMode.VERIFIED, - ) - GeneratedCertificateFactory( - user=self.user_2, - course_id=self.course.id, - grade='1.0', - status=CertificateStatuses.downloadable, - mode=GeneratedCertificate.MODES.verified, - download_url=download_url - ) - - self._assert_pdf_cert_generation_discontinued_logs(download_url) - - def test_add_cert_with_existing_pdf_certificate(self): - """ - Test that adding a certificate for existing PDF certificates logs a warning - message and the certificate status remains unchanged. - """ - download_url = 'http://www.example.com/certificate.pdf' - # Create an existing verified enrollment and certificate - CourseEnrollmentFactory( - user=self.user_2, - course_id=self.course.id, - is_active=True, - mode=CourseMode.VERIFIED, - ) - GeneratedCertificateFactory( - user=self.user_2, - course_id=self.course.id, - grade='1.0', - status=CertificateStatuses.downloadable, - mode=GeneratedCertificate.MODES.verified, - download_url=download_url - ) - - self._assert_pdf_cert_generation_discontinued_logs(download_url, add_cert=True) - - def _assert_pdf_cert_generation_discontinued_logs(self, download_url, add_cert=False): - """Assert PDF certificate generation discontinued logs.""" - with LogCapture(LOGGER.name) as log: - if add_cert: - self.xqueue.add_cert(self.user_2, self.course.id) - else: - self.xqueue.regen_cert(self.user_2, self.course.id) - log.check_present( - ( - LOGGER.name, - 'WARNING', - ( - "PDF certificate generation discontinued, canceling " - "PDF certificate generation for student {student_id} " - "in course '{course_id}' " - "with status '{status}' " - "and download_url '{download_url}'." - ).format( - student_id=self.user_2.id, - course_id=str(self.course.id), - status=CertificateStatuses.downloadable, - download_url=download_url - ) - ) - ) - - -@override_settings(CERT_QUEUE='certificates') -class XQueueCertInterfaceExampleCertificateTest(TestCase): - """Tests for the XQueue interface for certificate generation. """ - - COURSE_KEY = CourseLocator(org='test', course='test', run='test') - - TEMPLATE = 'test.pdf' - DESCRIPTION = 'test' - ERROR_MSG = 'Kaboom!' - - def setUp(self): - super().setUp() - self.xqueue = XQueueCertInterface() - - def test_add_example_cert(self): - cert = self._create_example_cert() - with self._mock_xqueue() as mock_send: - self.xqueue.add_example_cert(cert) - - # Verify that the correct payload was sent to the XQueue - self._assert_queue_task(mock_send, cert) - - # Verify the certificate status - assert cert.status == ExampleCertificate.STATUS_STARTED - - def test_add_example_cert_error(self): - cert = self._create_example_cert() - with self._mock_xqueue(success=False): - self.xqueue.add_example_cert(cert) - - # Verify the error status of the certificate - assert cert.status == ExampleCertificate.STATUS_ERROR - assert self.ERROR_MSG in cert.error_reason - - def _create_example_cert(self): - """Create an example certificate. """ - cert_set = ExampleCertificateSet.objects.create(course_key=self.COURSE_KEY) - return ExampleCertificate.objects.create( - example_cert_set=cert_set, - description=self.DESCRIPTION, - template=self.TEMPLATE - ) - - @contextmanager - def _mock_xqueue(self, success=True): - """Mock the XQueue method for sending a task to the queue. """ - with patch.object(XQueueInterface, 'send_to_queue') as mock_send: - mock_send.return_value = (0, None) if success else (1, self.ERROR_MSG) - yield mock_send - - def _assert_queue_task(self, mock_send, cert): - """Check that the task was added to the queue. """ - expected_header = { - 'lms_key': cert.access_key, - 'lms_callback_url': f'https://edx.org/update_example_certificate?key={cert.uuid}', - 'queue_name': 'certificates' - } - - expected_body = { - 'action': 'create', - 'username': cert.uuid, - 'name': 'John Doë', - 'course_id': str(self.COURSE_KEY), - 'template_pdf': 'test.pdf', - 'example_certificate': True - } - - assert mock_send.called - - __, kwargs = mock_send.call_args_list[0] - actual_header = json.loads(kwargs['header']) - actual_body = json.loads(kwargs['body']) - - assert expected_header == actual_header - assert expected_body == actual_body diff --git a/lms/djangoapps/instructor/tests/test_certificates.py b/lms/djangoapps/instructor/tests/test_certificates.py index 12ba936483..d0bd772564 100644 --- a/lms/djangoapps/instructor/tests/test_certificates.py +++ b/lms/djangoapps/instructor/tests/test_certificates.py @@ -14,6 +14,7 @@ from django.core.exceptions import ObjectDoesNotExist from django.core.files.uploadedfile import SimpleUploadedFile from django.test.utils import override_settings from django.urls import reverse +from testfixtures import LogCapture from common.djangoapps.student.models import CourseEnrollment from common.djangoapps.student.tests.factories import GlobalStaffFactory @@ -243,22 +244,27 @@ class CertificatesInstructorApiTest(SharedModuleStoreTestCase): assert response.status_code == 302 def test_generate_example_certificates(self): + expected_log_message = ( + "Generating example certificates is no longer supported. Skipping generation of example certificates for " + f"course {self.course.id}" + ) + self.client.login(username=self.global_staff.username, password='test') url = reverse( 'generate_example_certificates', kwargs={'course_id': str(self.course.id)} ) - response = self.client.post(url) + logging_messages = None + with LogCapture() as log: + response = self.client.post(url) + logging_messages = [log_msg.getMessage() for log_msg in log.records] + + assert logging_messages is not None + assert expected_log_message in logging_messages # Expect a redirect back to the instructor dashboard self._assert_redirects_to_instructor_dash(response) - # Expect that certificate generation started - # Cert generation will fail here because XQueue isn't configured, - # but the status should at least not be None. - status = certs_api.example_certificates_status(self.course.id) - assert status is not None - @ddt.data(True, False) def test_enable_certificate_generation(self, is_enabled): self.client.login(username=self.global_staff.username, password='test')