instructor task for auto cert creation

lms/djangoapps/certificates/config/waffle.py

+"""
+This module contains various configuration settings via
+waffle switches for the Certificates app.
+"""
+from openedx.core.djangoapps.waffle_utils import WaffleSwitchNamespace
+
+# Namespace
+WAFFLE_NAMESPACE = u'certificates'
+
+# Switches
+SELF_PACED_ONLY = u'self_paced_only'
+INSTRUCTOR_PACED_ONLY = u'instructor_paced_only'
+
+
+def waffle():
+    """
+    Returns the namespaced, cached, audited Waffle class for Certificates.
+    """
+    return WaffleSwitchNamespace(name=WAFFLE_NAMESPACE, log_prefix=u'Certificates: ')

lms/djangoapps/certificates/tasks.py

+from celery import task
+from logging import getLogger
+
+from celery_utils.logged_task import LoggedTask
+from celery_utils.persist_on_failure import PersistOnFailureTask
+
+from .api import generate_user_certificates
+
+logger = getLogger(__name__)
+
+
+class _BaseCertificateTask(PersistOnFailureTask, LoggedTask):  # pylint: disable=abstract-method
+    """
+    Include persistence features, as well as logging of task invocation.
+    """
+    abstract = True
+
+
+@task(base=_BaseCertificateTask)
+def generate_certificate(**kwargs):
+    """
+    Generates a certificate for a single user.
+    """
+    student = kwargs.pop('student')
+    course_key = kwargs.pop('course_key')
+    generate_user_certificates(student=student, course_key=course_key, **kwargs)

lms/djangoapps/certificates/tests/test_tasks.py

+from unittest import TestCase
+
+import ddt
+from mock import patch
+
+from lms.djangoapps.certificates.tasks import generate_certificate
+
+
+@ddt.ddt
+class GenerateUserCertificateTest(TestCase):
+    @patch('lms.djangoapps.certificates.tasks.generate_user_certificates')
+    def test_cert_task(self, generate_user_certs_mock):
+        generate_certificate(student='a', course_key='b', otherarg='c', otherotherarg='d')
+        generate_user_certs_mock.assert_called_with(student='a', course_key='b', otherarg='c', otherotherarg='d')
+
+    @ddt.data('student', 'course_key')
+    def test_cert_task_missing_args(self, missing_arg):
+        kwargs = {'student': 'a', 'course_key': 'b', 'otherarg': 'c'}
+        del kwargs[missing_arg]
+        with self.assertRaisesRegexp(KeyError, missing_arg):
+            generate_certificate(**kwargs)