diff --git a/aleksis/apps/kolego/migrations/0005_fix_overlapping_absences.py b/aleksis/apps/kolego/migrations/0005_fix_overlapping_absences.py
new file mode 100644
index 0000000000000000000000000000000000000000..244567acd1a253c385493dc7da5ff06f40765945
--- /dev/null
+++ b/aleksis/apps/kolego/migrations/0005_fix_overlapping_absences.py
@@ -0,0 +1,96 @@
+# Generated by Django 4.2.13 on 2024-07-16 11:01
+from datetime import datetime, time
+
+from django.db import migrations, models
+from django.db.models import Q
+
+
+def _run_forward(apps, schema_editor):
+ Absence = apps.get_model("kolego", "Absence")
+ for absence in Absence.objects.order_by("-datetime_start", "-date_start"):
+ # Convert dates of new event to datetimes in case dates are used
+ new_datetime_start = (
+ absence.datetime_start
+ if absence.datetime_start
+ else datetime.combine(absence.date_start, time.min)
+ ).astimezone(absence.timezone)
+ new_datetime_end = (
+ absence.datetime_end
+ if absence.datetime_end
+ else datetime.combine(absence.date_end, time.max)
+ ).astimezone(absence.timezone)
+
+ events_within = Absence.objects.filter(
+ person=absence.person,
+ ).filter(
+ Q(datetime_start__lte=new_datetime_end, datetime_end__gte=new_datetime_start)
+ | Q(date_start__lte=new_datetime_end.date(), date_end__gte=new_datetime_start.date())
+ )
+
+ for event_within in events_within:
+ event_within_datetime_start = (
+ event_within.datetime_start
+ if event_within.datetime_start
+ else datetime.combine(event_within.date_start, time.min)
+ )
+ event_within_datetime_end = (
+ event_within.datetime_end
+ if event_within.datetime_end
+ else datetime.combine(event_within.date_end, time.max)
+ )
+
+ # If overlapping absence has the same reason, just extend it
+ if event_within.reason == absence.reason:
+ event_within.datetime_start = min(new_datetime_start, event_within_datetime_start)
+ event_within.datetime_end = max(new_datetime_end, event_within_datetime_end)
+ event_within.save()
+ else:
+ if (
+ new_datetime_start > event_within_datetime_start
+ and new_datetime_end < event_within_datetime_end
+ ):
+ # Cut existing event in two parts
+ # First, cut end date of existing one
+ event_within.datetime_end = new_datetime_start
+ event_within.save()
+ # Then, create new event based on existing one filling up the remaining time
+ end_filler_event = event_within
+ end_filler_event.pk = None
+ end_filler_event.id = None
+ end_filler_event.calendarevent_ptr_id = None
+ end_filler_event.freebusy_ptr_id = None
+ end_filler_event._state.adding = True
+ end_filler_event.datetime_start = new_datetime_end
+ end_filler_event.datetime_end = event_within_datetime_end
+
+ end_filler_event.save()
+ elif (
+ new_datetime_start <= event_within_datetime_start
+ and new_datetime_end >= event_within_datetime_end
+ ):
+ # Delete existing event
+ event_within.delete()
+ elif (
+ new_datetime_start > event_within_datetime_start
+ and new_datetime_start < event_within_datetime_end
+ and new_datetime_end >= event_within_datetime_end
+ ):
+ # Cut end of existing event
+ event_within.datetime_end = new_datetime_start
+ event_within.save()
+ elif (
+ new_datetime_start <= event_within_datetime_start
+ and new_datetime_end < event_within_datetime_end
+ and new_datetime_end > event_within_datetime_start
+ ):
+ # Cut start of existing event
+ event_within.datetime_start = new_datetime_end
+ event_within.save()
+
+
+class Migration(migrations.Migration):
+ dependencies = [
+ ("kolego", "0004_absencereasontag_absencereason_tags"),
+ ]
+
+ operations = [migrations.RunPython(_run_forward)]
diff --git a/aleksis/apps/kolego/models/absence.py b/aleksis/apps/kolego/models/absence.py
index a2c58927324c462dbbdfc6f9a1658924a8974d78..da1326d07909f3833171ea9e3d468637f2aaa3ba 100644
--- a/aleksis/apps/kolego/models/absence.py
+++ b/aleksis/apps/kolego/models/absence.py
@@ -1,3 +1,5 @@
+from datetime import date, datetime, time
+
from django.db import models
from django.db.models import Q, QuerySet
from django.http import HttpRequest
@@ -9,7 +11,7 @@ from aleksis.core.managers import (
RecurrencePolymorphicManager,
)
from aleksis.core.mixins import ExtensibleModel
-from aleksis.core.models import FreeBusy
+from aleksis.core.models import FreeBusy, Person
from ..managers import AbsenceQuerySet
@@ -145,6 +147,136 @@ class Absence(FreeBusy):
def __str__(self):
return f"{self.person} ({self.datetime_start} - {self.datetime_end})"
+ def save(self, *args, skip_overlap_handling: bool = False, **kwargs):
+ extended_absence = None
+ modified_absences = []
+
+ if not skip_overlap_handling:
+ # Convert dates of new event to datetimes in case dates are used
+ new_datetime_start = (
+ self.datetime_start
+ if self.datetime_start
+ else datetime.combine(self.date_start, time.min)
+ ).astimezone(self.timezone)
+ new_datetime_end = (
+ self.datetime_end
+ if self.datetime_end
+ else datetime.combine(self.date_end, time.max)
+ ).astimezone(self.timezone)
+
+ events_within = Absence.get_objects(
+ None,
+ {"person": self.person.pk},
+ start=new_datetime_start,
+ end=new_datetime_end,
+ )
+
+ for event_within in events_within:
+ event_within_datetime_start = (
+ event_within.datetime_start
+ if event_within.datetime_start
+ else datetime.combine(event_within.date_start, time.min)
+ )
+ event_within_datetime_end = (
+ event_within.datetime_end
+ if event_within.datetime_end
+ else datetime.combine(event_within.date_end, time.max)
+ )
+
+ # If overlapping absence has the same reason, just extend it
+ if event_within.reason == self.reason:
+ event_within.datetime_start = min(
+ new_datetime_start, event_within_datetime_start
+ )
+ event_within.datetime_end = max(new_datetime_end, event_within_datetime_end)
+ event_within.save(skip_overlap_handling=True)
+ extended_absence = event_within
+ modified_absences.append(event_within)
+ else:
+ if (
+ new_datetime_start > event_within_datetime_start
+ and new_datetime_end < event_within_datetime_end
+ ):
+ # Cut existing event in two parts
+ # First, cut end date of existing one
+ event_within.datetime_end = new_datetime_start
+ event_within.save(skip_overlap_handling=True)
+ modified_absences.append(event_within)
+ # Then, create new event based on existing one filling up the remaining time
+ end_filler_event = event_within
+ end_filler_event.pk = None
+ end_filler_event.id = None
+ end_filler_event.calendarevent_ptr_id = None
+ end_filler_event.freebusy_ptr_id = None
+ end_filler_event._state.adding = True
+ end_filler_event.datetime_start = new_datetime_end
+ end_filler_event.datetime_end = event_within_datetime_end
+
+ end_filler_event.save(skip_overlap_handling=True)
+ modified_absences.append(end_filler_event)
+ elif (
+ new_datetime_start <= event_within_datetime_start
+ and new_datetime_end >= event_within_datetime_end
+ ):
+ # Delete existing event
+ if event_within in modified_absences:
+ modified_absences.remove(event_within)
+ event_within.delete()
+ elif (
+ new_datetime_start > event_within_datetime_start
+ and new_datetime_start < event_within_datetime_end
+ and new_datetime_end >= event_within_datetime_end
+ ):
+ # Cut end of existing event
+ event_within.datetime_end = new_datetime_start
+ event_within.save(skip_overlap_handling=True)
+ modified_absences.append(event_within)
+ elif (
+ new_datetime_start <= event_within_datetime_start
+ and new_datetime_end < event_within_datetime_end
+ and new_datetime_end > event_within_datetime_start
+ ):
+ # Cut start of existing event
+ event_within.datetime_start = new_datetime_end
+ event_within.save(skip_overlap_handling=True)
+ modified_absences.append(event_within)
+
+ modified_absences.append(self)
+
+ if extended_absence is not None:
+ self._extended_absence = extended_absence
+ else:
+ super().save(*args, **kwargs)
+ self._modified_absences = modified_absences
+
+ @classmethod
+ def get_for_person_by_datetimes(
+ cls,
+ person: Person,
+ datetime_start: datetime | None = None,
+ datetime_end: datetime | None = None,
+ date_start: date | None = None,
+ date_end: date | None = None,
+ defaults: dict | None = None,
+ **kwargs,
+ ) -> "Absence":
+ data = {"person": person, **kwargs}
+
+ if date_start:
+ data["date_start"] = date_start
+ data["date_end"] = date_end
+ else:
+ data["datetime_start"] = datetime_start
+ data["datetime_end"] = datetime_end
+
+ absence, created = cls.objects.get_or_create(**data, defaults=defaults)
+
+ if created:
+ if hasattr(absence, "_extended_absence"):
+ return absence._extended_absence
+ return absence
+ return absence
+
class Meta:
verbose_name = _("Absence")
verbose_name_plural = _("Absences")
diff --git a/aleksis/apps/kolego/schema/absence.py b/aleksis/apps/kolego/schema/absence.py
index 775725f6218ec41df9a480a41365b0c8705333ed..bb5f2988e51cd263d90e653192f1b8b7df2ca2e6 100644
--- a/aleksis/apps/kolego/schema/absence.py
+++ b/aleksis/apps/kolego/schema/absence.py
@@ -87,6 +87,15 @@ class AbsenceBatchCreateMutation(BaseBatchCreateMutation):
optional_fields = ("comment", "reason")
permissions = ("kolego.create_absence_rule",)
+ @classmethod
+ def before_save(cls, root, info, input, created_objs): # noqa
+ modified_absences = []
+
+ for absence in created_objs:
+ modified_absences += getattr(absence, "_modified_absences", [])
+
+ return modified_absences
+
class AbsenceBatchDeleteMutation(BaseBatchDeleteMutation):
class Meta:
@@ -109,6 +118,16 @@ class AbsenceBatchPatchMutation(BaseBatchPatchMutation):
)
permissions = ("kolego.edit_absence_rule",)
+ @classmethod
+ def after_mutate(cls, root, info, input, updated_objs, return_data): # noqa
+ modified_absences = []
+
+ for absence in updated_objs:
+ modified_absences += getattr(absence, "_modified_absences", [])
+
+ return_data[cls._meta.return_field_name] = modified_absences
+ return return_data
+
class AbsenceReasonBatchCreateMutation(BaseBatchCreateMutation):
class Meta:
diff --git a/aleksis/apps/kolego/tests/models/test_absence.py b/aleksis/apps/kolego/tests/models/test_absence.py
new file mode 100644
index 0000000000000000000000000000000000000000..8ec8660c7687f5008cf60f652c29b1bf495bae2e
--- /dev/null
+++ b/aleksis/apps/kolego/tests/models/test_absence.py
@@ -0,0 +1,148 @@
+from datetime import datetime, timezone
+
+import pytest
+from aleksis.apps.kolego.models import Absence, AbsenceReason
+from aleksis.core.models import Person
+
+from zoneinfo import ZoneInfo
+
+pytestmark = pytest.mark.django_db
+
+
+@pytest.fixture
+def absences_test_data():
+ datetime_start_existing_1 = datetime(2024, 1, 1, 12, 0, tzinfo=timezone.utc).astimezone(ZoneInfo("Europe/Berlin"))
+ datetime_end_existing_1 = datetime(2024, 1, 2, 10, 0, tzinfo=timezone.utc).astimezone(ZoneInfo("Europe/Berlin"))
+ datetime_start_existing_2 = datetime(2024, 1, 2, 12, 0, tzinfo=timezone.utc).astimezone(ZoneInfo("Europe/Berlin"))
+ datetime_end_existing_2 = datetime(2024, 1, 3, 19, 43, tzinfo=timezone.utc).astimezone(ZoneInfo("Europe/Berlin"))
+
+ reason_1 = AbsenceReason.objects.create(short_name="i", name="ill")
+ reason_2 = AbsenceReason.objects.create(short_name="e", name="excused")
+ reason_1.refresh_from_db()
+ reason_2.refresh_from_db()
+
+ person_1 = Person.objects.create(first_name="Maria", last_name="Montessori")
+ person_2 = Person.objects.create(first_name="Karl Moritz", last_name="Fleischer")
+ person_1.refresh_from_db()
+ person_2.refresh_from_db()
+
+ # Create existing absencess
+ existing_absence_1 = Absence.objects.create(datetime_start=datetime_start_existing_1, datetime_end=datetime_end_existing_1, person=person_1, reason=reason_1)
+ existing_absence_2 = Absence.objects.create(datetime_start=datetime_start_existing_2, datetime_end=datetime_end_existing_2, person=person_1, reason=reason_1)
+ existing_absence_1.refresh_from_db()
+ existing_absence_2.refresh_from_db()
+
+ return {
+ "datetime_start_existing_1": datetime_start_existing_1,
+ "datetime_end_existing_1": datetime_end_existing_1,
+ "datetime_start_existing_2": datetime_start_existing_2,
+ "datetime_end_existing_2": datetime_end_existing_2,
+ "reason_1": reason_1,
+ "reason_2": reason_2,
+ "person_1": person_1,
+ "person_2": person_2,
+ "existing_absence_1": existing_absence_1,
+ "existing_absence_2": existing_absence_2,
+ }
+
+
+# Create new absences in six scenarios in relation to existing absence 1 with same person and different absence reason
+
+# 1: new absence is fully inside exiting absence
+def test_overlapping_single_absence_different_reason_inside(absences_test_data):
+ datetime_start_new = datetime(2024, 1, 1, 14, 0, tzinfo=timezone.utc).astimezone(ZoneInfo("Europe/Berlin"))
+ datetime_end_new = datetime(2024, 1, 2, 8, 0, tzinfo=timezone.utc).astimezone(ZoneInfo("Europe/Berlin"))
+ new_absence = Absence.objects.create(datetime_start=datetime_start_new, datetime_end=datetime_end_new, person=absences_test_data["person_1"], reason=absences_test_data["reason_2"])
+ new_absence.refresh_from_db()
+
+ absences_test_data["existing_absence_1"].refresh_from_db()
+
+ assert absences_test_data["existing_absence_1"].datetime_start == datetime(2024, 1, 1, 12, 0, tzinfo=timezone.utc)
+ assert absences_test_data["existing_absence_1"].datetime_end == datetime(2024, 1, 1, 14, 0, tzinfo=timezone.utc)
+
+ assert new_absence.datetime_start == datetime(2024, 1, 1, 14, 0, tzinfo=timezone.utc)
+ assert new_absence.datetime_end == datetime(2024, 1, 2, 8, 0, tzinfo=timezone.utc)
+
+ assert len(Absence.objects.all()) == 4
+ assert Absence.objects.filter(datetime_start=datetime(2024, 1, 2, 8, 0, tzinfo=timezone.utc)).exists()
+
+ existing_absence_1_part_2 = Absence.objects.get(datetime_start=datetime(2024, 1, 2, 8, 0, tzinfo=timezone.utc))
+
+ assert existing_absence_1_part_2.datetime_start == datetime(2024, 1, 2, 8, 0, tzinfo=timezone.utc)
+ assert existing_absence_1_part_2.datetime_end == datetime(2024, 1, 2, 10, 0, tzinfo=timezone.utc)
+
+
+# 2: new absence covers the same time span as the existing one
+def test_overlapping_single_absence_different_reason_same(absences_test_data):
+ datetime_start_new = absences_test_data["datetime_start_existing_1"]
+ datetime_end_new = absences_test_data["datetime_end_existing_1"]
+ new_absence = Absence.objects.create(datetime_start=datetime_start_new, datetime_end=datetime_end_new, person=absences_test_data["person_1"], reason=absences_test_data["reason_2"])
+ new_absence.refresh_from_db()
+
+ assert len(Absence.objects.all()) == 2
+ assert not Absence.objects.filter(datetime_start=datetime(2024, 1, 1, 12, 0, tzinfo=timezone.utc), datetime_end=datetime(2024, 1, 2, 10, 0, tzinfo=timezone.utc)).exclude(pk=new_absence.pk).exists()
+
+ assert new_absence.datetime_start == datetime(2024, 1, 1, 12, 0, tzinfo=timezone.utc)
+ assert new_absence.datetime_end == datetime(2024, 1, 2, 10, 0, tzinfo=timezone.utc)
+
+
+# 3: new absence starts earlier and ends later than the existing absence
+def test_overlapping_single_absence_different_reason_exceed(absences_test_data):
+ datetime_start_new = datetime(2024, 1, 1, 10, 0, tzinfo=timezone.utc).astimezone(ZoneInfo("Europe/Berlin"))
+ datetime_end_new = datetime(2024, 1, 2, 12, 0, tzinfo=timezone.utc).astimezone(ZoneInfo("Europe/Berlin"))
+ new_absence = Absence.objects.create(datetime_start=datetime_start_new, datetime_end=datetime_end_new, person=absences_test_data["person_1"], reason=absences_test_data["reason_2"])
+ new_absence.refresh_from_db()
+
+ assert len(Absence.objects.all()) == 2
+ assert not Absence.objects.filter(datetime_start=datetime(2024, 1, 1, 12, 0, tzinfo=timezone.utc), datetime_end=datetime(2024, 1, 2, 10, 0, tzinfo=timezone.utc)).exists()
+
+ assert new_absence.datetime_start == datetime(2024, 1, 1, 10, 0, tzinfo=timezone.utc)
+ assert new_absence.datetime_end == datetime(2024, 1, 2, 12, 0, tzinfo=timezone.utc)
+
+
+# 4: new absence starts before existing absence and ends within the existing absence
+def test_overlapping_single_absence_different_reason_cut_start(absences_test_data):
+ datetime_start_new = datetime(2024, 1, 1, 10, 0, tzinfo=timezone.utc).astimezone(ZoneInfo("Europe/Berlin"))
+ datetime_end_new = datetime(2024, 1, 1, 14, 0, tzinfo=timezone.utc).astimezone(ZoneInfo("Europe/Berlin"))
+ new_absence = Absence.objects.create(datetime_start=datetime_start_new, datetime_end=datetime_end_new, person=absences_test_data["person_1"], reason=absences_test_data["reason_2"])
+ new_absence.refresh_from_db()
+
+ absences_test_data["existing_absence_1"].refresh_from_db()
+
+ assert absences_test_data["existing_absence_1"].datetime_start == datetime(2024, 1, 1, 14, 0, tzinfo=timezone.utc)
+ assert absences_test_data["existing_absence_1"].datetime_end == datetime(2024, 1, 2, 10, 0, tzinfo=timezone.utc)
+
+ assert new_absence.datetime_start == datetime(2024, 1, 1, 10, 0, tzinfo=timezone.utc)
+ assert new_absence.datetime_end == datetime(2024, 1, 1, 14, 0, tzinfo=timezone.utc)
+
+
+# 5: new absence starts within existing absence and ends after the existing absence
+def test_overlapping_single_absence_different_reason_cut_end(absences_test_data):
+ datetime_start_new = datetime(2024, 1, 2, 8, 0, tzinfo=timezone.utc).astimezone(ZoneInfo("Europe/Berlin"))
+ datetime_end_new = datetime(2024, 1, 2, 12, 0, tzinfo=timezone.utc).astimezone(ZoneInfo("Europe/Berlin"))
+ new_absence = Absence.objects.create(datetime_start=datetime_start_new, datetime_end=datetime_end_new, person=absences_test_data["person_1"], reason=absences_test_data["reason_2"])
+ new_absence.refresh_from_db()
+
+ absences_test_data["existing_absence_1"].refresh_from_db()
+
+ assert absences_test_data["existing_absence_1"].datetime_start == datetime(2024, 1, 1, 12, 0, tzinfo=timezone.utc)
+ assert absences_test_data["existing_absence_1"].datetime_end == datetime(2024, 1, 2, 8, 0, tzinfo=timezone.utc)
+
+ assert new_absence.datetime_start == datetime(2024, 1, 2, 8, 0, tzinfo=timezone.utc)
+ assert new_absence.datetime_end == datetime(2024, 1, 2, 12, 0, tzinfo=timezone.utc)
+
+
+# 6: new absence is completely outside of existing absence
+def test_overlapping_single_absence_different_reason_outside(absences_test_data):
+ datetime_start_new = datetime(2024, 1, 1, 8, 0, tzinfo=timezone.utc).astimezone(ZoneInfo("Europe/Berlin"))
+ datetime_end_new = datetime(2024, 1, 1, 10, 0, tzinfo=timezone.utc).astimezone(ZoneInfo("Europe/Berlin"))
+ new_absence = Absence.objects.create(datetime_start=datetime_start_new, datetime_end=datetime_end_new, person=absences_test_data["person_1"], reason=absences_test_data["reason_2"])
+ new_absence.refresh_from_db()
+
+ absences_test_data["existing_absence_1"].refresh_from_db()
+
+ assert absences_test_data["existing_absence_1"].datetime_start == datetime(2024, 1, 1, 12, 0, tzinfo=timezone.utc)
+ assert absences_test_data["existing_absence_1"].datetime_end == datetime(2024, 1, 2, 10, 0, tzinfo=timezone.utc)
+
+ assert new_absence.datetime_start == datetime(2024, 1, 1, 8, 0, tzinfo=timezone.utc)
+ assert new_absence.datetime_end == datetime(2024, 1, 1, 10, 0, tzinfo=timezone.utc)