fix: clean up periodic tasks when deleting course configurations

openedx_certificates/models.py

@@ -13,6 +13,8 @@
 from django.contrib.auth import get_user_model
 from django.core.exceptions import ValidationError
 from django.db import models
+from django.db.models.signals import post_delete
+from django.dispatch import receiver
 from django.utils.translation import gettext_lazy as _
 from django_celery_beat.models import IntervalSchedule, PeriodicTask
 from edx_ace import Message, Recipient, ace
@@ -231,6 +233,14 @@ def generate_certificate_for_user(self, user_id: int, celery_task_id: int = 0):
                 certificate.send_email()
 
 
+# noinspection PyUnusedLocal
+@receiver(post_delete, sender=ExternalCertificateCourseConfiguration)
+def post_delete_periodic_task(sender, instance, *_args, **_kwargs):  # noqa: ANN001, ARG001
+    """Delete the associated periodic task when the object is deleted."""
+    if instance.periodic_task:
+        instance.periodic_task.delete()
+
+
 class ExternalCertificate(TimeStampedModel):
     """
     Model to represent each individual certificate awarded to a user for a course.

pyproject.toml

@@ -88,6 +88,7 @@ target-version = 'py38'
     'ANN205', # missing-return-type-static-method
     'INP001', # implicit-namespace-package
     'SLF001', # private-member-access
+    'RUF018', # assignment-in-assert
 ]
 
 [tool.ruff.flake8-annotations]

tests/test_models.py

@@ -8,6 +8,7 @@
 import pytest
 from django.core.exceptions import ValidationError
 from django.db import IntegrityError
+from django_celery_beat.models import PeriodicTask
 
 from openedx_certificates.exceptions import CertificateGenerationError
 from openedx_certificates.models import (
@@ -18,6 +19,7 @@
 from test_utils.factories import UserFactory
 
 if TYPE_CHECKING:
+    from django.db.models import Model
     from django.contrib.auth.models import User
     from opaque_keys.edx.keys import CourseKey
 
@@ -102,6 +104,48 @@ def test_periodic_task_is_auto_created(self):
         assert periodic_task.args == f'[{self.course_config.id}]'
         assert periodic_task.task == 'openedx_certificates.tasks.generate_certificates_for_course_task'
 
+    @pytest.mark.django_db()
+    def test_periodic_task_is_deleted_on_deletion(self):
+        """Test that the periodic task is deleted when the configuration is deleted."""
+        self.certificate_type.save()
+        self.course_config.save()
+        assert PeriodicTask.objects.count() == 1
+
+        self.course_config.delete()
+        assert not PeriodicTask.objects.exists()
+
+    @pytest.mark.django_db()
+    def test_periodic_task_deletion_removes_the_configuration(self):
+        """Test that the configuration is deleted when the periodic task is deleted."""
+        self.certificate_type.save()
+        self.course_config.save()
+        assert PeriodicTask.objects.count() == 1
+
+        self.course_config.periodic_task.delete()
+        assert not ExternalCertificateCourseConfiguration.objects.exists()
+
+    @pytest.mark.django_db()
+    @pytest.mark.parametrize(
+        ("deleted_model", "verified_model"),
+        [
+            (ExternalCertificateCourseConfiguration, PeriodicTask),  # `post_delete` signal.
+            (PeriodicTask, ExternalCertificateCourseConfiguration),  # Cascade deletion of the `OneToOneField`.
+        ],
+    )
+    def test_bulk_delete(self, deleted_model: type[Model], verified_model: type[Model]):
+        """Test that the bulk deletion of configurations removes the periodic tasks (and vice versa)."""
+        self.certificate_type.save()
+        self.course_config.save()
+
+        ExternalCertificateCourseConfiguration(
+            course_id="course-v1:TestX+T101+2024",
+            certificate_type=self.certificate_type,
+        ).save()
+        assert PeriodicTask.objects.count() == 2
+
+        deleted_model.objects.all().delete()
+        assert not verified_model.objects.exists()
+
     def test_str_representation(self):
         """Test the string representation of the model."""
         assert str(self.course_config) == f'{self.certificate_type.name} in course-v1:TestX+T101+2023'