[#1715] Added management command to delete expired contact invitations

src/open_inwoner/accounts/management/commands/delete_invitations.py

@@ -0,0 +1,38 @@
+from collections import defaultdict
+from datetime import timedelta
+
+from django.conf import settings
+from django.core.management.base import BaseCommand
+from django.utils import timezone
+
+from open_inwoner.utils.logentry import system_action
+
+from ...models import Invite
+
+
+class Command(BaseCommand):
+    help = (
+        "Delete contact invitations which are older than a predefined period "
+        "(the default is defined in `settings.INVITE_EXPIRY_DAYS`)"
+    )
+
+    def handle(self, *args, **kwargs):
+        now = timezone.now()
+        expired_invitations = Invite.objects.select_related("inviter").filter(
+            created_on__lt=now - timedelta(days=settings.INVITE_EXPIRY_DAYS)
+        )
+
+        grouped = defaultdict(list)
+
+        for invite in expired_invitations:
+            grouped[invite.inviter].append(invite)
+
+        for inviter, invites in grouped.items():
+            invites_info = [
+                f"{invite.invitee_email} (invited on {invite.created_on.strftime('%Y-%m-%d')})"
+                for invite in invites
+            ]
+            message = f"{len(invites)} expired contact invitations from {inviter.get_full_name()} deleted"
+            system_action(message, user=inviter, deleted_invitations=invites_info)
+
+        expired_invitations.delete()

src/open_inwoner/accounts/models.py

@@ -807,5 +807,5 @@ def get_absolute_url(self) -> str:
         return reverse("profile:invite_accept", kwargs={"key": self.key})
 
     def expired(self) -> bool:
-        expiration_date = self.created_on + timedelta(days=settings.INVITE_EXPIRY)
+        expiration_date = self.created_on + timedelta(days=settings.INVITE_EXPIRY_DAYS)
         return expiration_date <= timezone.now()

src/open_inwoner/accounts/tests/test_commands.py

@@ -0,0 +1,52 @@
+from django.core.management import call_command
+from django.test import TestCase
+
+from freezegun import freeze_time
+from timeline_logger.models import TimelineLog
+
+from ..models import Invite
+from .factories import InviteFactory, UserFactory
+
+
+class DeleteContactInvitationsTest(TestCase):
+    @freeze_time("2023-09-26", as_arg=True)
+    def test_delete_expired_invitations(frozen_time, self):
+        user = UserFactory(
+            first_name="Johann Maria Salvadore",
+            infix="van de",
+            last_name="Eenzaameiland",
+        )
+
+        invite1 = InviteFactory.create(inviter=user)
+        invite2 = InviteFactory.create(inviter=user)
+        invite3 = InviteFactory.create()
+
+        frozen_time.move_to("2023-10-01")
+
+        invite4 = InviteFactory.create()
+
+        frozen_time.move_to("2023-10-27")
+
+        call_command("delete_invitations")
+
+        # check remaining invites
+        invitations = Invite.objects.all()
+        self.assertEqual(len(invitations), 1)
+        self.assertEqual(invitations[0].invitee_email, invite4.invitee_email)
+
+        # check logs
+        logs = TimelineLog.objects.all()
+        self.assertEqual(len(logs), 2)
+
+        self.assertEqual(
+            logs[0].extra_data["deleted_invitations"][0],
+            f"{invite1.invitee_email} (invited on {invite1.created_on.strftime('%Y-%m-%d')})",
+        )
+        self.assertEqual(
+            logs[0].extra_data["deleted_invitations"][1],
+            f"{invite2.invitee_email} (invited on {invite2.created_on.strftime('%Y-%m-%d')})",
+        )
+        self.assertEqual(
+            logs[1].extra_data["deleted_invitations"][0],
+            f"{invite3.invitee_email} (invited on {invite3.created_on.strftime('%Y-%m-%d')})",
+        )

src/open_inwoner/conf/base.py

@@ -784,9 +784,8 @@
     }
 }
 
-
 # invite expires in X days after sending
-INVITE_EXPIRY = 14
+INVITE_EXPIRY_DAYS = config("INVITE_EXPIRY_DAYS", default=30)
 
 # zgw-consumers
 ZGW_CONSUMERS_TEST_SCHEMA_DIRS = [

src/open_inwoner/utils/logentry.py

@@ -142,7 +142,12 @@ def user_action(request, object, message):
 
 
 def system_action(
-    message, *, content_object=None, user=None, log_level=logging.INFO, exc_info=None
+    message,
+    content_object=None,
+    user=None,
+    log_level=logging.INFO,
+    exc_info=None,
+    **kwargs,
 ):
     """
     Log a generic action done by business logic.
@@ -163,5 +168,6 @@ def system_action(
             "action_flag": LOG_ACTIONS[5],
             "message": message,
             "log_level": log_level,
+            **kwargs,
         },
     )