Andal.LND/edx-platform
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #28245 from edx/jhynes/microba-1227_cleanup

Browse Source 
fix: remove disabled v1 course certificate queue code
This commit is contained in:
Justin Hynes
2021-07-26 10:31:27 -04:00
committed by GitHub
parent 19b45069fa 0069bd724f
commit 49bcdfe481
5 changed files with 27 additions and 1053 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
lms/djangoapps/certificates/api.py
View File

@@ -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):

582
lms/djangoapps/certificates/queue.py
View File

@@ -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
)

82
lms/djangoapps/certificates/tests/test_api.py
View File

@@ -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. """

366
lms/djangoapps/certificates/tests/test_queue.py
View File

@@ -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

20
lms/djangoapps/instructor/tests/test_certificates.py
View File

@@ -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')