✨(core) create a management command to generate certificates

Create a management command which aims to be called by a cron task to generate
certificate at a regular interval. This command accepts three options (course,
product and order) to restrict review to those resources.

CHANGELOG.md

@@ -10,6 +10,7 @@ and this project adheres to
 
 ### Added
 
+- Add a management command to generate certificate for eligible orders
 - Add `is_passed` cached property to enrollment model
 - Add an api endpoint to retrieve and download certificates
 - Add a `get_grades` method to LMSHandler

src/backend/joanie/core/management/commands/create_certificates.py

@@ -0,0 +1,223 @@
+"""Management command to create all pending certificates."""
+import logging
+
+from django.core.exceptions import ValidationError
+from django.core.management import BaseCommand
+from django.utils import timezone
+
+from joanie.core import enums, models
+
+logger = logging.getLogger("joanie.core.create_certificates")
+
+
+class Command(BaseCommand):
+    """
+    A command to generate all pending certificates.
+    It browses all certifying products and check if related orders are eligible for
+    certification then generate the certificate if it is.
+
+    Through options, you are able to restrict this command
+    to a list of courses (-c), products (-p) or orders (-o).
+    """
+
+    help = __doc__
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "-c",
+            "--courses",
+            "--course",
+            help=(
+                "Accept a single or a list of course code to restrict review to "
+                "this/those course(s)."
+            ),
+        )
+        parser.add_argument(
+            "-p",
+            "--products",
+            "--product",
+            help=(
+                "Accept a single or a list of product uuid to restrict review to "
+                "this/those product(s)."
+            ),
+        )
+        parser.add_argument(
+            "-o",
+            "--orders",
+            "--order",
+            help=(
+                "Accept a single or a list of order uuid to restrict review to "
+                "this/those order(s)."
+            ),
+        )
+
+    # pylint: disable=too-many-locals
+    def handle(self, *args, **options):
+        """
+        Retrieve all certifying products then for each of them check eligibility for
+        certification of all related orders.
+        If `order` option is used, this order is directly retrieve.
+        """
+        order_uids = None
+        product_uids = None
+        course_codes = None
+
+        if options["orders"]:
+            order_uids = (
+                options["orders"]
+                if isinstance(options["orders"], list)
+                else [options["orders"]]
+            )
+
+        total = 0
+        if order_uids:
+            filters = {
+                "is_canceled": False,
+                "certificate__isnull": True,
+                "product__type__in": enums.PRODUCT_TYPE_CERTIFICATE_ALLOWED,
+                "uid__in": order_uids,
+            }
+
+            try:
+                orders = [
+                    order
+                    for order in models.Order.objects.select_related(
+                        "course__organization"
+                    ).filter(**filters)
+                    if order.state == enums.ORDER_STATE_VALIDATED
+                ]
+            except models.Order.DoesNotExist:
+                return str(total)
+
+            for order in orders:
+                result = self._generate_certificate_for_order(order)
+                total += result
+
+            return str(total)
+
+        if options["courses"]:
+            course_codes = (
+                options["courses"]
+                if isinstance(options["courses"], list)
+                else [options["courses"]]
+            )
+
+        if options["products"]:
+            product_uids = (
+                options["products"]
+                if isinstance(options["products"], list)
+                else [options["products"]]
+            )
+
+        filters = {"type__in": enums.PRODUCT_TYPE_CERTIFICATE_ALLOWED}
+        if course_codes:
+            filters.update({"courses__code__in": course_codes})
+        if product_uids:
+            filters.update({"uid__in": product_uids})
+
+        products = models.Product.objects.filter(**filters)
+
+        for product in products:
+            result = self._generate_certificate_for_product(product, course_codes)
+            total += result
+
+        return str(total)
+
+    def _generate_certificate_for_product(self, product, course_codes):
+        """
+        Retrieve orders related to the product and course code then check if
+        it is eligible for certification.
+        """
+        filters = {"is_canceled": False, "certificate__isnull": True}
+        if course_codes:
+            filters.update({"course__code__in": course_codes})
+
+        orders = [
+            order
+            for order in product.orders.filter(**filters).select_related(
+                "course__organization"
+            )
+            if order.state == enums.ORDER_STATE_VALIDATED
+        ]
+
+        certificate_counter = 0
+        for order in orders:
+            result = self._generate_certificate_for_order(order)
+            certificate_counter += result
+
+        if certificate_counter > 0:
+            logger.info(
+                '%s certificate(s) for "%s" have been generated.',
+                certificate_counter,
+                product.title,
+            )
+
+        return certificate_counter
+
+    @staticmethod
+    def _generate_certificate_for_order(order):
+        """
+        Check if order is eligible for certification then generate certificate is it is.
+
+        Eligibility means that order contains
+        one passed enrollment per graded courses.
+
+        Return:
+            0 : if the order is not eligible to certification
+            1: if a certificate has been generated for the current order
+        """
+        graded_courses = order.target_courses.filter(
+            order_relations__is_graded=True
+        ).order_by("order_relations__position")
+        graded_courses_count = len(graded_courses)
+
+        if graded_courses_count == 0:
+            return 0
+
+        course_runs = models.CourseRun.objects.filter(
+            course__in=graded_courses,
+            is_gradable=True,
+            start__lte=timezone.now(),
+        )
+
+        enrollments = order.enrollments.filter(
+            course_run__in=course_runs, is_active=True
+        ).select_related("user", "course_run")
+
+        # Cross graded courses and enrollments to check there is an active enrollment
+        # for each graded courses, if not it is useless to go further
+        course_enrollments = []
+        for course in graded_courses:
+            for enrollment in enrollments:
+                # Check if the enrollment relies on course by crossing
+                # all course runs implied
+                intersections = len(
+                    {enrollment.course_run} & set(course.course_runs.all())
+                )
+                if intersections == 1:
+                    course_enrollments.append(enrollment)
+                    break
+
+        # If we do not have one enrollment per graded course, there is no need to
+        # continue, we are sure that order is not eligible for certification.
+        if len(course_enrollments) != graded_courses_count:
+            return 0
+
+        # Otherwise, we now need to know if each enrollment has been passed
+        passed_enrollment_count = 0
+        for enrollment in course_enrollments:
+            if enrollment.is_passed is False:
+                # If one enrollment has not been passed, no need to continue,
+                # We are sure that order is not eligible for certification.
+                break
+            passed_enrollment_count += 1
+
+        if passed_enrollment_count != graded_courses_count:
+            return 0
+
+        try:
+            order.create_certificate()
+        except ValidationError:
+            return 0
+
+        return 1

src/backend/joanie/tests/test_admin_product.py

@@ -243,7 +243,7 @@ def test_admin_product_should_display_related_course_links(self):
         """
         Product admin view should display a read only field "related courses"
         in charge of listing related courses with a link to the course admin
-        change view.
+        change view
         """
 
         # Create courses