Merge pull request #11080 from edx/saleem-latif/SOL-1529

SOL-1529: Allow PMs to Invalidate Certificates

lms/djangoapps/instructor/tests/test_certificates.py

@@ -13,9 +13,10 @@ from xmodule.modulestore.tests.django_utils import SharedModuleStoreTestCase
 from xmodule.modulestore.tests.factories import CourseFactory
 from config_models.models import cache
 from courseware.tests.factories import GlobalStaffFactory, InstructorFactory, UserFactory
-from certificates.tests.factories import GeneratedCertificateFactory, CertificateWhitelistFactory
+from certificates.tests.factories import GeneratedCertificateFactory, CertificateWhitelistFactory, \
+    CertificateInvalidationFactory
 from certificates.models import CertificateGenerationConfiguration, CertificateStatuses, CertificateWhitelist, \
-    GeneratedCertificate
+    GeneratedCertificate, CertificateInvalidation
 from certificates import api as certs_api
 from student.models import CourseEnrollment
 from django.core.files.uploadedfile import SimpleUploadedFile
@@ -923,3 +924,299 @@ class TestCertificatesInstructorApiBulkWhiteListExceptions(SharedModuleStoreTest
         self.assertEqual(response.status_code, 200)
         data = json.loads(response.content)
         return data
+
+
+@attr('shard_1')
+@ddt.ddt
+class CertificateInvalidationViewTests(SharedModuleStoreTestCase):
+    """
+    Test certificate invalidation view.
+    """
+    @classmethod
+    def setUpClass(cls):
+        super(CertificateInvalidationViewTests, cls).setUpClass()
+        cls.course = CourseFactory.create()
+        cls.url = reverse('certificate_invalidation_view',
+                          kwargs={'course_id': cls.course.id})
+        cls.notes = "Test notes."
+
+    def setUp(self):
+        super(CertificateInvalidationViewTests, self).setUp()
+        self.global_staff = GlobalStaffFactory()
+        self.enrolled_user_1 = UserFactory(
+            username='TestStudent1',
+            email='test_student1@example.com',
+            first_name='Enrolled',
+            last_name='Student',
+        )
+        self.enrolled_user_2 = UserFactory(
+            username='TestStudent2',
+            email='test_student2@example.com',
+            first_name='Enrolled',
+            last_name='Student',
+        )
+
+        self.not_enrolled_student = UserFactory(
+            username='NotEnrolledStudent',
+            email='nonenrolled@test.com',
+            first_name='NotEnrolled',
+            last_name='Student',
+        )
+        CourseEnrollment.enroll(self.enrolled_user_1, self.course.id)
+        CourseEnrollment.enroll(self.enrolled_user_2, self.course.id)
+
+        self.generated_certificate = GeneratedCertificateFactory.create(
+            user=self.enrolled_user_1,
+            course_id=self.course.id,
+            status=CertificateStatuses.downloadable,
+            mode='honor',
+        )
+
+        self.certificate_invalidation_data = dict(
+            user=self.enrolled_user_1.username,
+            notes=self.notes,
+        )
+
+        # Global staff can see the certificates section
+        self.client.login(username=self.global_staff.username, password="test")
+
+    def test_invalidate_certificate(self):
+        """
+        Test user can invalidate a generated certificate.
+        """
+        response = self.client.post(
+            self.url,
+            data=json.dumps(self.certificate_invalidation_data),
+            content_type='application/json',
+        )
+        # Assert successful request processing
+        self.assertEqual(response.status_code, 200)
+        result = json.loads(response.content)
+
+        # Assert Certificate Exception Updated data
+        self.assertEqual(result['user'], self.enrolled_user_1.username)
+        self.assertEqual(result['invalidated_by'], self.global_staff.username)
+        self.assertEqual(result['notes'], self.notes)
+
+        # Verify that CertificateInvalidation record has been created in the database i.e. no DoesNotExist error
+        try:
+            CertificateInvalidation.objects.get(
+                generated_certificate=self.generated_certificate,
+                invalidated_by=self.global_staff,
+                notes=self.notes,
+                active=True,
+            )
+        except ObjectDoesNotExist:
+            self.fail("The certificate is not invalidated.")
+
+        # Validate generated certificate was invalidated
+        generated_certificate = GeneratedCertificate.objects.get(
+            user=self.enrolled_user_1,
+            course_id=self.course.id,
+        )
+        self.assertFalse(generated_certificate.is_valid())
+
+    def test_missing_username_and_email_error(self):
+        """
+        Test error message if user name or email is missing.
+        """
+        self.certificate_invalidation_data.update({'user': ''})
+        response = self.client.post(
+            self.url,
+            data=json.dumps(self.certificate_invalidation_data),
+            content_type='application/json',
+        )
+
+        # Assert 400 status code in response
+        self.assertEqual(response.status_code, 400)
+        res_json = json.loads(response.content)
+
+        # Assert Error Message
+        self.assertEqual(
+            res_json['message'],
+            u'Student username/email field is required and can not be empty. '
+            u'Kindly fill in username/email and then press "Invalidate Certificate" button.',
+        )
+
+    def test_invalid_user_name_error(self):
+        """
+        Test error message if invalid user name is given.
+        """
+        invalid_user = "test_invalid_user_name"
+
+        self.certificate_invalidation_data.update({"user": invalid_user})
+
+        response = self.client.post(
+            self.url,
+            data=json.dumps(self.certificate_invalidation_data),
+            content_type='application/json',
+        )
+
+        # Assert 400 status code in response
+        self.assertEqual(response.status_code, 400)
+        res_json = json.loads(response.content)
+
+        # Assert Error Message
+        self.assertEqual(
+            res_json['message'],
+            u"{user} does not exist in the LMS. Please check your spelling and retry.".format(user=invalid_user),
+        )
+
+    def test_user_not_enrolled_error(self):
+        """
+        Test error message if user is not enrolled in the course.
+        """
+        self.certificate_invalidation_data.update({"user": self.not_enrolled_student.username})
+
+        response = self.client.post(
+            self.url,
+            data=json.dumps(self.certificate_invalidation_data),
+            content_type='application/json',
+        )
+
+        # Assert 400 status code in response
+        self.assertEqual(response.status_code, 400)
+        res_json = json.loads(response.content)
+
+        # Assert Error Message
+        self.assertEqual(
+            res_json['message'],
+            u"{user} is not enrolled in this course. Please check your spelling and retry.".format(
+                user=self.not_enrolled_student.username,
+            ),
+        )
+
+    def test_no_generated_certificate_error(self):
+        """
+        Test error message if there is no generated certificate for the student.
+        """
+        self.certificate_invalidation_data.update({"user": self.enrolled_user_2.username})
+
+        response = self.client.post(
+            self.url,
+            data=json.dumps(self.certificate_invalidation_data),
+            content_type='application/json',
+        )
+
+        # Assert 400 status code in response
+        self.assertEqual(response.status_code, 400)
+        res_json = json.loads(response.content)
+
+        # Assert Error Message
+        self.assertEqual(
+            res_json['message'],
+            u"The student {student} does not have certificate for the course {course}. "
+            u"Kindly verify student username/email and the selected course are correct and try again.".format(
+                student=self.enrolled_user_2.username,
+                course=self.course.number,
+            ),
+        )
+
+    def test_certificate_already_invalid_error(self):
+        """
+        Test error message if certificate for the student is already invalid.
+        """
+        # Invalidate user certificate
+        self.generated_certificate.invalidate()
+
+        response = self.client.post(
+            self.url,
+            data=json.dumps(self.certificate_invalidation_data),
+            content_type='application/json',
+        )
+
+        # Assert 400 status code in response
+        self.assertEqual(response.status_code, 400)
+        res_json = json.loads(response.content)
+
+        # Assert Error Message
+        self.assertEqual(
+            res_json['message'],
+            u"Certificate for student {user} is already invalid, kindly verify that certificate "
+            u"was generated for this student and then proceed.".format(
+                user=self.enrolled_user_1.username,
+            ),
+        )
+
+    def test_duplicate_certificate_invalidation_error(self):
+        """
+        Test error message if certificate invalidation for the student is already present.
+        """
+        CertificateInvalidationFactory.create(
+            generated_certificate=self.generated_certificate,
+            invalidated_by=self.global_staff,
+        )
+        # Invalidate user certificate
+        self.generated_certificate.invalidate()
+
+        response = self.client.post(
+            self.url,
+            data=json.dumps(self.certificate_invalidation_data),
+            content_type='application/json',
+        )
+
+        # Assert 400 status code in response
+        self.assertEqual(response.status_code, 400)
+        res_json = json.loads(response.content)
+
+        # Assert Error Message
+        self.assertEqual(
+            res_json['message'],
+            u"Certificate of {user} has already been invalidated. Please check your spelling and retry.".format(
+                user=self.enrolled_user_1.username,
+            ),
+        )
+
+    def test_remove_certificate_invalidation(self):
+        """
+        Test that user can remove certificate invalidation.
+        """
+        # Invalidate user certificate
+        self.generated_certificate.invalidate()
+
+        CertificateInvalidationFactory.create(
+            generated_certificate=self.generated_certificate,
+            invalidated_by=self.global_staff,
+        )
+
+        response = self.client.post(
+            self.url,
+            data=json.dumps(self.certificate_invalidation_data),
+            content_type='application/json',
+            REQUEST_METHOD='DELETE'
+        )
+
+        # Assert 204 status code in response
+        self.assertEqual(response.status_code, 204)
+
+        # Verify that certificate invalidation successfully removed from database
+        with self.assertRaises(ObjectDoesNotExist):
+            CertificateInvalidation.objects.get(
+                generated_certificate=self.generated_certificate,
+                invalidated_by=self.global_staff,
+                active=True,
+            )
+
+    def test_remove_certificate_invalidation_error(self):
+        """
+        Test error message if certificate invalidation does not exists.
+        """
+        # Invalidate user certificate
+        self.generated_certificate.invalidate()
+
+        response = self.client.post(
+            self.url,
+            data=json.dumps(self.certificate_invalidation_data),
+            content_type='application/json',
+            REQUEST_METHOD='DELETE'
+        )
+
+        # Assert 400 status code in response
+        self.assertEqual(response.status_code, 400)
+        res_json = json.loads(response.content)
+
+        # Assert Error Message
+        self.assertEqual(
+            res_json['message'],
+            u"Certificate Invalidation does not exist, Please refresh the page and try again.",
+        )

lms/djangoapps/instructor/views/api.py

@@ -89,7 +89,7 @@ from instructor.views import INVOICE_KEY
 from submissions import api as sub_api  # installed from the edx-submissions repository
 
 from certificates import api as certs_api
-from certificates.models import CertificateWhitelist, GeneratedCertificate, CertificateStatuses
+from certificates.models import CertificateWhitelist, GeneratedCertificate, CertificateStatuses, CertificateInvalidation
 
 from bulk_email.models import CourseEmail
 from student.models import get_user_by_username_or_email
@@ -2839,28 +2839,55 @@ def parse_request_data_and_get_user(request, course_key):
     :param course_key: Course Identifier of the course for whom to process certificate exception
     :return: key-value pairs containing certificate exception data and User object
     """
-    try:
-        certificate_exception = json.loads(request.body or '{}')
-    except ValueError:
-        raise ValueError(_('The record is not in the correct format. Please add a valid username or email address.'))
+    certificate_exception = parse_request_data(request)
 
     user = certificate_exception.get('user_name', '') or certificate_exception.get('user_email', '')
     if not user:
         raise ValueError(_('Student username/email field is required and can not be empty. '
                            'Kindly fill in username/email and then press "Add to Exception List" button.'))
-    try:
-        db_user = get_user_by_username_or_email(user)
-    except ObjectDoesNotExist:
-        raise ValueError(_("{user} does not exist in the LMS. Please check your spelling and retry.").format(user=user))
-
-    # Make Sure the given student is enrolled in the course
-    if not CourseEnrollment.is_enrolled(db_user, course_key):
-        raise ValueError(_("{user} is not enrolled in this course. Please check your spelling and retry.")
-                         .format(user=user))
+    db_user = get_student(user, course_key)
 
     return certificate_exception, db_user
 
 
+def parse_request_data(request):
+    """
+    Parse and return request data, raise ValueError in case of invalid JSON data.
+
+    :param request: HttpRequest request object.
+    :return: dict object containing parsed json data.
+    """
+    try:
+        data = json.loads(request.body or '{}')
+    except ValueError:
+        raise ValueError(_('The record is not in the correct format. Please add a valid username or email address.'))
+
+    return data
+
+
+def get_student(username_or_email, course_key):
+    """
+    Retrieve and return User object from db, raise ValueError
+    if user is does not exists or is not enrolled in the given course.
+
+    :param username_or_email: String containing either user name or email of the student.
+    :param course_key: CourseKey object identifying the current course.
+    :return: User object
+    """
+    try:
+        student = get_user_by_username_or_email(username_or_email)
+    except ObjectDoesNotExist:
+        raise ValueError(_("{user} does not exist in the LMS. Please check your spelling and retry.").format(
+            user=username_or_email
+        ))
+
+    # Make Sure the given student is enrolled in the course
+    if not CourseEnrollment.is_enrolled(student, course_key):
+        raise ValueError(_("{user} is not enrolled in this course. Please check your spelling and retry.")
+                         .format(user=username_or_email))
+    return student
+
+
 @transaction.non_atomic_requests
 @ensure_csrf_cookie
 @cache_control(no_cache=True, no_store=True, must_revalidate=True)
@@ -3014,3 +3041,142 @@ def generate_bulk_certificate_exceptions(request, course_id):  # pylint: disable
     }
 
     return JsonResponse(results)
+
+
+@transaction.non_atomic_requests
+@ensure_csrf_cookie
+@cache_control(no_cache=True, no_store=True, must_revalidate=True)
+@require_global_staff
+@require_http_methods(['POST', 'DELETE'])
+def certificate_invalidation_view(request, course_id):
+    """
+    Invalidate/Re-Validate students to/from certificate.
+
+    :param request: HttpRequest object
+    :param course_id: course identifier of the course for whom to add/remove certificates exception.
+    :return: JsonResponse object with success/error message or certificate invalidation data.
+    """
+    course_key = CourseKey.from_string(course_id)
+    # Validate request data and return error response in case of invalid data
+    try:
+        certificate_invalidation_data = parse_request_data(request)
+        certificate = validate_request_data_and_get_certificate(certificate_invalidation_data, course_key)
+    except ValueError as error:
+        return JsonResponse({'message': error.message}, status=400)
+
+    # Invalidate certificate of the given student for the course course
+    if request.method == 'POST':
+        try:
+            certificate_invalidation = invalidate_certificate(request, certificate, certificate_invalidation_data)
+        except ValueError as error:
+            return JsonResponse({'message': error.message}, status=400)
+        return JsonResponse(certificate_invalidation)
+
+    # Re-Validate student certificate for the course course
+    elif request.method == 'DELETE':
+        try:
+            re_validate_certificate(request, course_key, certificate)
+        except ValueError as error:
+            return JsonResponse({'message': error.message}, status=400)
+
+        return JsonResponse({}, status=204)
+
+
+def invalidate_certificate(request, generated_certificate, certificate_invalidation_data):
+    """
+    Invalidate given GeneratedCertificate and add CertificateInvalidation record for future reference or re-validation.
+
+    :param request: HttpRequest object
+    :param generated_certificate: GeneratedCertificate object, the certificate we want to invalidate
+    :param certificate_invalidation_data: dict object containing data for CertificateInvalidation.
+    :return: dict object containing updated certificate invalidation data.
+    """
+    if len(CertificateInvalidation.get_certificate_invalidations(
+            generated_certificate.course_id,
+            generated_certificate.user,
+    )) > 0:
+        raise ValueError(
+            _("Certificate of {user} has already been invalidated. Please check your spelling and retry.").format(
+                user=generated_certificate.user.username,
+            )
+        )
+
+    # Verify that certificate user wants to invalidate is a valid one.
+    if not generated_certificate.is_valid():
+        raise ValueError(
+            _("Certificate for student {user} is already invalid, kindly verify that certificate was generated "
+              "for this student and then proceed.").format(user=generated_certificate.user.username)
+        )
+
+    # Add CertificateInvalidation record for future reference or re-validation
+    certificate_invalidation, __ = CertificateInvalidation.objects.update_or_create(
+        generated_certificate=generated_certificate,
+        defaults={
+            'invalidated_by': request.user,
+            'notes': certificate_invalidation_data.get("notes", ""),
+            'active': True,
+        }
+    )
+
+    # Invalidate GeneratedCertificate
+    generated_certificate.invalidate()
+    return {
+        'id': certificate_invalidation.id,
+        'user': certificate_invalidation.generated_certificate.user.username,
+        'invalidated_by': certificate_invalidation.invalidated_by.username,
+        'created': certificate_invalidation.created.strftime("%B %d, %Y"),
+        'notes': certificate_invalidation.notes,
+    }
+
+
+def re_validate_certificate(request, course_key, generated_certificate):
+    """
+    Remove certificate invalidation from db and start certificate generation task for this student.
+    Raises ValueError if certificate invalidation is present.
+
+    :param request: HttpRequest object
+    :param course_key: CourseKey object identifying the current course.
+    :param generated_certificate: GeneratedCertificate object of the student for the given course
+    """
+    try:
+        # Fetch CertificateInvalidation object
+        certificate_invalidation = CertificateInvalidation.objects.get(generated_certificate=generated_certificate)
+    except ObjectDoesNotExist:
+        raise ValueError(_("Certificate Invalidation does not exist, Please refresh the page and try again."))
+    else:
+        # Deactivate certificate invalidation if it was fetched successfully.
+        certificate_invalidation.deactivate()
+
+    # We need to generate certificate only for a single student here
+    students = [certificate_invalidation.generated_certificate.user]
+    instructor_task.api.generate_certificates_for_students(request, course_key, students=students)
+
+
+def validate_request_data_and_get_certificate(certificate_invalidation, course_key):
+    """
+    Fetch and return GeneratedCertificate of the student passed in request data for the given course.
+
+    Raises ValueError in case of missing student username/email or
+    if student does not have certificate for the given course.
+
+    :param certificate_invalidation: dict containing certificate invalidation data
+    :param course_key: CourseKey object identifying the current course.
+    :return: GeneratedCertificate object of the student for the given course
+    """
+    user = certificate_invalidation.get("user")
+
+    if not user:
+        raise ValueError(
+            _('Student username/email field is required and can not be empty. '
+              'Kindly fill in username/email and then press "Invalidate Certificate" button.')
+        )
+
+    student = get_student(user, course_key)
+
+    certificate = GeneratedCertificate.certificate_for_student(student, course_key)
+    if not certificate:
+        raise ValueError(_(
+            "The student {student} does not have certificate for the course {course}. Kindly verify student "
+            "username/email and the selected course are correct and try again."
+        ).format(student=student.username, course=course_key.course))
+    return certificate

lms/djangoapps/instructor/views/api_urls.py

@@ -161,4 +161,8 @@ urlpatterns = patterns(
     url(r'^generate_bulk_certificate_exceptions',
         'instructor.views.api.generate_bulk_certificate_exceptions',
         name='generate_bulk_certificate_exceptions'),
+
+    url(r'^certificate_invalidation_view/$',
+        'instructor.views.api.certificate_invalidation_view',
+        name='certificate_invalidation_view'),
 )

lms/djangoapps/instructor/views/instructor_dashboard.py

@@ -43,6 +43,7 @@ from certificates.models import (
     GeneratedCertificate,
     CertificateStatuses,
     CertificateGenerationHistory,
+    CertificateInvalidation,
 )
 from certificates import api as certs_api
 from util.date_utils import get_default_time_display
@@ -184,6 +185,13 @@ def instructor_dashboard_2(request, course_id):
         kwargs={'course_id': unicode(course_key)}
     )
 
+    certificate_invalidation_view_url = reverse(  # pylint: disable=invalid-name
+        'certificate_invalidation_view',
+        kwargs={'course_id': unicode(course_key)}
+    )
+
+    certificate_invalidations = CertificateInvalidation.get_certificate_invalidations(course_key)
+
     context = {
         'course': course,
         'studio_url': get_studio_url(course, 'course'),
@@ -191,9 +199,11 @@ def instructor_dashboard_2(request, course_id):
         'disable_buttons': disable_buttons,
         'analytics_dashboard_message': analytics_dashboard_message,
         'certificate_white_list': certificate_white_list,
+        'certificate_invalidations': certificate_invalidations,
         'generate_certificate_exceptions_url': generate_certificate_exceptions_url,
         'generate_bulk_certificate_exceptions_url': generate_bulk_certificate_exceptions_url,
-        'certificate_exception_view_url': certificate_exception_view_url
+        'certificate_exception_view_url': certificate_exception_view_url,
+        'certificate_invalidation_view_url': certificate_invalidation_view_url,
     }
 
     return render_to_response('instructor/instructor_dashboard_2/instructor_dashboard_2.html', context)