Supression des inscriptions archivées depuis plus de 6 mois

mesads/app/crons.py

@@ -89,3 +89,19 @@ def do(self):
         with redirect_stdout(buf), redirect_stderr(buf):
             call_command("notify_prefectures_gestionnaires")
         return buf.getvalue()
+
+
+class SupressionInscriptionsArchivees(CronJobBase):
+    # Run every week
+    schedule = Schedule(run_every_mins=60 * 24 * 7)
+
+    code = "supression_inscriptions_archivees"  # unique code to represent this cron job
+
+    @sentry_exceptions
+    def do(self):
+        # Redirect stdout and stderr to a buffer to capture the output of the
+        # command. By returning it, django-cron will log it in the database.
+        buf = io.StringIO()
+        with redirect_stdout(buf), redirect_stderr(buf):
+            call_command("supression_inscriptions_archivees")
+        return buf.getvalue()

mesads/app/management/commands/supression_inscriptions_archivees.py

@@ -0,0 +1,15 @@
+from django.core.management.base import BaseCommand
+
+from mesads.app.services.liste_attente import supression_inscriptions_archivees
+
+
+class Command(BaseCommand):
+    help = (
+        "Supprime les inscriptions à la liste d'attente archivées depuis plus de 6 mois"
+    )
+
+    def handle(self, *args, **options):
+        inscriptions_count = supression_inscriptions_archivees()
+        self.stdout.write(
+            f"{inscriptions_count} inscriptions archivées ont été supprimées."
+        )

mesads/app/services/liste_attente.py

@@ -4,6 +4,7 @@
 from django.conf import settings
 from django.core.mail import send_mail
 from django.template.loader import render_to_string
+from django.utils import timezone
 
 from mesads.app.models import ADSManagerRequest, InscriptionListeAttente
 
@@ -75,3 +76,20 @@ def check_and_notify_duplicated(inscription: InscriptionListeAttente):
 
     for inscription_a_notifier in inscriptions_a_notifier:
         _notification_doublon(inscription_a_notifier)
+
+
+def supression_inscriptions_archivees():
+    """
+    Fonction qui va supprimer les inscriptions à la liste d'attente
+    archivées depuis plus de 6 mois.
+    """
+
+    date_limite = timezone.now() - relativedelta(months=6)
+
+    inscriptions = InscriptionListeAttente.with_deleted.filter(
+        deleted_at__isnull=False, deleted_at__lt=date_limite
+    )
+    inscriptions_count = inscriptions.count()
+    inscriptions.delete()
+
+    return inscriptions_count

mesads/app/tests/test_service_exports.py → mesads/app/tests/services/test_exports.py

@@ -2,7 +2,7 @@
 
 from mesads.app.services.export import get_prefectures_data_listes_attente
 
-from .factories import (
+from ..factories import (
     ADSManagerAdministratorFactory,
     ADSManagerFactory,
     InscriptionListeAttenteFactory,

mesads/app/tests/test_service_import_ads.py → mesads/app/tests/services/test_import_ads.py

@@ -7,7 +7,7 @@
 from mesads.app.services.import_ads import import_ads_from_excel
 from mesads.users.tests.factories import UserFactory
 
-from .factories import (
+from ..factories import (
     ADSManagerAdministratorFactory,
     ADSManagerFactory,
 )

mesads/app/tests/services/test_liste_attente.py

@@ -0,0 +1,83 @@
+import pytest
+from dateutil.relativedelta import relativedelta
+from django.core import mail
+from django.utils import timezone
+
+from mesads.app.models import InscriptionListeAttente
+from mesads.app.services.liste_attente import (
+    check_and_notify_duplicated,
+    supression_inscriptions_archivees,
+)
+from mesads.users.tests.factories import UserFactory
+
+from ..factories import (
+    ADSManagerAdministratorFactory,
+    ADSManagerFactory,
+    ADSManagerRequestFactory,
+    InscriptionListeAttenteFactory,
+)
+
+pytestmark = pytest.mark.django_db
+
+
+def test_get_export_data_liste_attente():
+    administrator = ADSManagerAdministratorFactory()
+    commune = ADSManagerFactory(administrator=administrator, for_commune=True)
+
+    inscription_1 = InscriptionListeAttenteFactory(
+        ads_manager=commune
+    )  # ne doit pas être supprimée
+    inscription_2 = InscriptionListeAttenteFactory(
+        ads_manager=commune, deleted_at=timezone.now() - relativedelta(months=5)
+    )  # ne doit pas être supprimée
+    inscription_3 = InscriptionListeAttenteFactory(
+        ads_manager=commune, deleted_at=timezone.now() - relativedelta(months=6)
+    )  # doit être supprimée
+    inscription_4 = InscriptionListeAttenteFactory(
+        ads_manager=commune, deleted_at=timezone.now() - relativedelta(months=8)
+    )  # doit être supprimée
+
+    inscriptions_supprimees = supression_inscriptions_archivees()
+    assert inscriptions_supprimees == 2
+    assert InscriptionListeAttente.with_deleted.count() == 2
+    inscription_ids = InscriptionListeAttente.with_deleted.values_list("pk", flat=True)
+    assert inscription_1.pk in inscription_ids
+    assert inscription_2.pk in inscription_ids
+    assert inscription_3.pk not in inscription_ids
+    assert inscription_4.pk not in inscription_ids
+
+
+def test_check_and_notify_duplicated_pas_de_doublon():
+    administrator = ADSManagerAdministratorFactory()
+    commune = ADSManagerFactory(administrator=administrator, for_commune=True)
+
+    inscription = InscriptionListeAttenteFactory(ads_manager=commune)
+
+    check_and_notify_duplicated(inscription)
+    assert len(mail.outbox) == 0
+
+
+def test_check_and_notify_duplicated_doublon_detecte():
+    administrator = ADSManagerAdministratorFactory()
+    commune_1 = ADSManagerFactory(administrator=administrator, for_commune=True)
+    commune_2 = ADSManagerFactory(administrator=administrator, for_commune=True)
+    request_commune_1 = ADSManagerRequestFactory(
+        ads_manager=commune_1, user=UserFactory()
+    )
+    request_commune_2 = ADSManagerRequestFactory(
+        ads_manager=commune_2, user=UserFactory()
+    )
+
+    inscription_1 = InscriptionListeAttenteFactory(
+        ads_manager=commune_1, numero_licence="12345"
+    )
+    InscriptionListeAttenteFactory(ads_manager=commune_2, numero_licence="12345")
+
+    check_and_notify_duplicated(inscription_1)
+    assert len(mail.outbox) == 2
+    emails = []
+    for email in mail.outbox:
+        emails = emails + email.to
+
+    assert request_commune_1.user.email in emails
+    assert request_commune_2.user.email in emails

mesads/app/tests/test_service_notifications.py → mesads/app/tests/services/test_notifications.py

@@ -6,8 +6,8 @@
 from mesads.app.services.notifications import notify_prefectures_gestionnaires
 from mesads.users.tests.factories import UserFactory
 
-from ..models import ADS, ADSUpdateLog, ADSUser
-from .factories import (
+from ...models import ADS, ADSUpdateLog, ADSUser
+from ..factories import (
     ADSManagerAdministratorFactory,
     ADSManagerFactory,
     ADSManagerRequestFactory,

mesads/common/mixins.py

@@ -44,6 +44,13 @@ def delete(self, using=None, keep_parents=False):
         self.deleted_at = timezone.now()
         self.save()
 
+    def hard_delete(self, using=None, keep_parents=False):
+        """Supprime réellement l'objet."""
+        return super().delete(
+            using=using,
+            keep_parents=keep_parents,
+        )
+
 
 class SmartValidationMixin:
     """Override clean() to only validate fields that have changed."""

mesads/settings.py

@@ -159,6 +159,7 @@ def parse_env_bool(key, default):
     "mesads.app.crons.DeleteOldUsers",
     "mesads.app.crons.NotificationListeAttente",
     "mesads.app.crons.NotificationVerification",
+    "mesads.app.crons.SupressionInscriptionsArchivees",
 ]
 
 AUTH_USER_MODEL = "users.User"