add filters for bulk sending

shiftregister/team/forms.py

@@ -1,4 +1,7 @@
 from django import forms
+from django.db.models import Case, Count, ExpressionWrapper, F, When
+from django.db.models.fields import DateTimeField, IntegerField
+from django.utils import timezone
 
 from .models import Helper, ShiftRegistration
 
@@ -17,12 +20,60 @@ class HelperShift(forms.Form):
     helper = NameField(label="Helfi", queryset=Helper.objects.order_by("name"))
 
 
+HELPER_FILTERS = {
+    "all": {"label": "Alle Helfis", "query": lambda base_query: base_query},
+    "checked_in": {
+        "label": "Helfis mit mindestens einem Check-in",
+        "query": lambda base_query: base_query.annotate(
+            shift_count=Count(
+                Case(
+                    When(
+                        shiftregistration__state__in=[
+                            ShiftRegistration.RegState.CHECKED_IN,
+                        ],
+                        then=1,
+                    ),
+                    output_field=IntegerField(),
+                )
+            )
+        )
+        .filter(shift_count__gte=1)
+        .distinct(),
+    },
+    "current": {
+        "label": "Aktuell aktive Helfis",
+        "query": lambda base_query: base_query.annotate(
+            shift_end=ExpressionWrapper(
+                F("shiftregistration__shift__start_at")
+                + F("shiftregistration__shift__duration"),
+                output_field=DateTimeField(),
+            )
+        )
+        .filter(
+            shiftregistration__shift__start_at__lte=timezone.now(),
+            shiftregistration__state=ShiftRegistration.RegState.CHECKED_IN,
+            shift_end__gte=timezone.now(),
+        )
+        .distinct(),
+    },
+    "no_shifts": {
+        "label": "Helfis ohne Schichtanmeldungen",
+        "query": lambda base_query: base_query.annotate(
+            reg_count=Count("shiftregistration")
+        ).filter(reg_count=0),
+    },
+}
+
+
 class BulkMessage(forms.Form):
     message = forms.CharField(
         label="Nachricht", widget=forms.Textarea(attrs={"class": "textarea"})
     )
-    checked_in_only = forms.BooleanField(
+    helper_filter = forms.ChoiceField(
-        label="Nur an Helfis mit mindestens einem Check-in senden", required=False
+        label="Empfänger auswählen",
+        choices=[(k, v["label"]) for k, v in HELPER_FILTERS.items()],
+        initial="all",
+        widget=forms.Select(attrs={"class": "select"}),
     )
 
 

shiftregister/team/tests.py

@@ -1,3 +1,160 @@
-from django.test import TestCase
+from datetime import timedelta
 
-# Create your tests here.
+from django.contrib.auth.models import User
+from django.test import Client, TestCase
+from django.urls import reverse
+from django.utils import timezone
+
+from shiftregister.app.models import (
+    Helper,
+    LoginToken,
+    Message,
+    Room,
+    Shift,
+    ShiftRegistration,
+)
+
+from .forms import HELPER_FILTERS
+
+
+class BulkMessageFilterTests(TestCase):
+    def setUp(self):
+        # Create a test user for login
+        self.user = User.objects.create_user(username="testuser", password="testpass")
+        self.client = Client()
+        self.client.login(username="testuser", password="testpass")
+
+        # Create test room
+        self.room = Room.objects.create(
+            name="Test Room", required_helpers=1, meeting_location="Test Location"
+        )
+
+        # Create test helpers
+        self.helper1 = Helper.objects.create(
+            phone="+491234567890", name="Helper 1", number_validated=True
+        )
+        self.helper2 = Helper.objects.create(
+            phone="+491234567891", name="Helper 2", number_validated=True
+        )
+        self.helper3 = Helper.objects.create(
+            phone="+491234567892", name="Helper 3", number_validated=True
+        )
+        self.helper4 = Helper.objects.create(
+            phone="+491234567893",
+            name="Helper 4",
+            number_validated=False,  # This one should never be included
+        )
+
+        # Create login tokens for each helper so message text replacement works
+        LoginToken.objects.create(helper=self.helper1)
+        LoginToken.objects.create(helper=self.helper2)
+        LoginToken.objects.create(helper=self.helper3)
+        LoginToken.objects.create(helper=self.helper4)
+
+        # Create shifts
+        now = timezone.now()
+        self.current_shift = Shift.objects.create(
+            room=self.room,
+            start_at=now - timedelta(hours=1),
+            duration=timedelta(hours=3),
+            required_helpers=2,
+        )
+        self.past_shift = Shift.objects.create(
+            room=self.room,
+            start_at=now - timedelta(hours=5),
+            duration=timedelta(hours=2),
+            required_helpers=2,
+        )
+
+        # Create registrations
+        # Helper1: Has a checked-in past shift
+        ShiftRegistration.objects.create(
+            shift=self.past_shift,
+            helper=self.helper1,
+            state=ShiftRegistration.RegState.CHECKED_IN,
+        )
+
+        # Helper2: Currently active in a shift
+        ShiftRegistration.objects.create(
+            shift=self.current_shift,
+            helper=self.helper2,
+            state=ShiftRegistration.RegState.CHECKED_IN,
+        )
+
+        # Helper3: Has no shifts
+        # Helper4: Not validated, should never be included
+
+    def test_all_helpers_filter(self):
+        """Test that 'all' filter returns all validated helpers"""
+        base_query = Helper.objects.filter(number_validated=True)
+        helpers = HELPER_FILTERS["all"]["query"](base_query)
+
+        self.assertEqual(helpers.count(), 3)
+        self.assertIn(self.helper1, helpers)
+        self.assertIn(self.helper2, helpers)
+        self.assertIn(self.helper3, helpers)
+        self.assertNotIn(self.helper4, helpers)
+
+    def test_checked_in_helpers_filter(self):
+        """Test that 'checked_in' filter returns only helpers with at least one check-in"""
+        base_query = Helper.objects.filter(number_validated=True)
+        helpers = HELPER_FILTERS["checked_in"]["query"](base_query)
+
+        self.assertEqual(helpers.count(), 2)
+        self.assertIn(self.helper1, helpers)
+        self.assertIn(self.helper2, helpers)
+        self.assertNotIn(self.helper3, helpers)
+        self.assertNotIn(self.helper4, helpers)
+
+    def test_current_helpers_filter(self):
+        """Test that 'current' filter returns only helpers currently in a shift"""
+        base_query = Helper.objects.filter(number_validated=True)
+        helpers = HELPER_FILTERS["current"]["query"](base_query)
+
+        self.assertEqual(helpers.count(), 1)
+        self.assertNotIn(self.helper1, helpers)  # Past shift
+        self.assertIn(self.helper2, helpers)  # Current shift
+        self.assertNotIn(self.helper3, helpers)  # No shifts
+        self.assertNotIn(self.helper4, helpers)  # Not validated
+
+    def test_no_shifts_helpers_filter(self):
+        """Test that 'no_shifts' filter returns only helpers without any shift registrations"""
+        base_query = Helper.objects.filter(number_validated=True)
+        helpers = HELPER_FILTERS["no_shifts"]["query"](base_query)
+
+        self.assertEqual(helpers.count(), 1)
+        self.assertNotIn(self.helper1, helpers)  # Has past shift
+        self.assertNotIn(self.helper2, helpers)  # Has current shift
+        self.assertIn(self.helper3, helpers)  # No shifts
+        self.assertNotIn(self.helper4, helpers)  # Not validated
+
+    def test_bulk_message_view_with_filters(self):
+        """Test that the bulk message view correctly applies filters"""
+        test_message = "Test message"
+
+        for filter_key in HELPER_FILTERS.keys():
+            response = self.client.post(
+                reverse("team:bulk_message"),
+                {"message": test_message, "helper_filter": filter_key},
+            )
+
+            self.assertEqual(response.status_code, 200)
+
+            # Count messages created
+            if filter_key == "all":
+                expected_count = 3
+            elif filter_key == "checked_in":
+                expected_count = 2
+            elif filter_key == "current":
+                expected_count = 1
+            else:  # no_shifts
+                expected_count = 1
+
+            self.assertEqual(
+                Message.objects.filter(text__startswith=test_message).count(),
+                expected_count,
+                f"Expected {expected_count} messages for filter '{filter_key}'",
+            )
+
+            # Clean up for next iteration
+            Message.objects.all().delete()

shiftregister/team/views.py

@@ -12,7 +12,7 @@ from django.utils import timezone
 from django.views.generic import DetailView, ListView
 from django.views.generic.edit import FormMixin
 
-from .forms import BulkMessage, HelperMessage, HelperShift
+from .forms import HELPER_FILTERS, BulkMessage, HelperMessage, HelperShift
 from .models import (
     Helper,
     IncomingMessage,
@@ -135,21 +135,11 @@ def bulk_message(request):
     if request.method == "POST":
         form = BulkMessage(request.POST)
         if form.is_valid():
-            helpers = Helper.objects.filter(number_validated=True)
+            base_query = Helper.objects.filter(number_validated=True)
-            if form.cleaned_data["checked_in_only"]:
+            helper_filter = form.cleaned_data["helper_filter"]
-                helpers = Helper.objects.annotate(
+
-                    shift_count=Count(
+            # Get the query function from HELPER_FILTERS and apply it
-                        Case(
+            helpers = HELPER_FILTERS[helper_filter]["query"](base_query)
-                            When(
-                                shiftregistration__state__in=[
-                                    ShiftRegistration.RegState.CHECKED_IN,
-                                ],
-                                then=1,
-                            ),
-                            output_field=models.IntegerField(),
-                        )
-                    )
-                ).filter(number_validated=True, shift_count__gte=1)
 
             try:
                 outbox = []