snapshot: command to clear instances from snapshot (#86634)

chrono/apps/snapshot/management/commands/clear_snapshots.py

@@ -0,0 +1,30 @@
+# chrono - agendas system
+# Copyright (C) 2016-2024  Entr'ouvert
+#
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License as published
+# by the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import datetime
+
+from django.core.management.base import BaseCommand
+from django.utils.timezone import now
+
+from chrono.agendas.models import Agenda, Category, EventsType, Resource, UnavailabilityCalendar
+
+
+class Command(BaseCommand):
+    help = 'Clear obsolete snapshot instances'
+
+    def handle(self, **options):
+        for model in [Agenda, Category, EventsType, Resource, UnavailabilityCalendar]:
+            model.snapshots.filter(updated_at__lte=now() - datetime.timedelta(days=1)).delete()

tests/test_snapshot.py

@@ -0,0 +1,85 @@
+import datetime
+
+import pytest
+from django.core.management import call_command
+from django.utils.timezone import now
+
+from chrono.agendas.models import Agenda, Category, Desk, EventsType, Resource, UnavailabilityCalendar
+from chrono.apps.snapshot.models import (
+    AgendaSnapshot,
+    CategorySnapshot,
+    EventsTypeSnapshot,
+    ResourceSnapshot,
+    UnavailabilityCalendarSnapshot,
+)
+
+pytestmark = pytest.mark.django_db
+
+
+def test_clear_snapshot():
+    agenda = Agenda.objects.create(slug='rdv', label='Rdv', kind='meetings')
+    Desk.objects.create(slug='foo', label='Foo', agenda=agenda)
+    category = Category.objects.create(slug='foo', label='Foo')
+    events_type = EventsType.objects.create(slug='foo', label='Foo')
+    resource = Resource.objects.create(slug='foo', label='Foo')
+    unavailability_calendar = UnavailabilityCalendar.objects.create(slug='foo', label='Foo')
+
+    agenda.take_snapshot()
+    category.take_snapshot()
+    events_type.take_snapshot()
+    resource.take_snapshot()
+    unavailability_calendar.take_snapshot()
+
+    snapshot = AgendaSnapshot.objects.get(instance=agenda)
+    snapshot_agenda = snapshot.get_instance()
+    assert snapshot_agenda.snapshot == snapshot
+    assert snapshot_agenda.pk != agenda.pk
+    assert Desk.objects.filter(agenda=agenda).count() == 1
+    assert Desk.objects.filter(agenda=snapshot_agenda).count() == 1
+    snapshot = CategorySnapshot.objects.get(instance=category)
+    snapshot_category = snapshot.get_instance()
+    assert snapshot_category.snapshot == snapshot
+    assert snapshot_category.pk != category.pk
+    snapshot = EventsTypeSnapshot.objects.get(instance=events_type)
+    snapshot_events_type = snapshot.get_instance()
+    assert snapshot_events_type.snapshot == snapshot
+    assert snapshot_events_type.pk != events_type.pk
+    snapshot = ResourceSnapshot.objects.get(instance=resource)
+    snapshot_resource = snapshot.get_instance()
+    assert snapshot_resource.snapshot == snapshot
+    assert snapshot_resource.pk != resource.pk
+    snapshot = UnavailabilityCalendarSnapshot.objects.get(instance=unavailability_calendar)
+    snapshot_unavailability_calendar = snapshot.get_instance()
+    assert snapshot_unavailability_calendar.snapshot == snapshot
+    assert snapshot_unavailability_calendar.pk != unavailability_calendar.pk
+
+    # too soon
+    call_command('clear_snapshots')
+    for model in [Agenda, Category, EventsType, Resource, UnavailabilityCalendar]:
+        assert model.objects.count() == 1
+        assert model.snapshots.count() == 1
+        assert model.get_snapshot_model().objects.count() == 1
+    assert Desk.objects.filter(agenda=agenda).count() == 1
+    assert Desk.objects.filter(agenda=snapshot_agenda).count() == 1
+
+    # still too soon
+    for model in [Agenda, Category, EventsType, Resource, UnavailabilityCalendar]:
+        model.snapshots.update(updated_at=now() - datetime.timedelta(days=1, minutes=-1))
+    call_command('clear_snapshots')
+    for model in [Agenda, Category, EventsType, Resource, UnavailabilityCalendar]:
+        assert model.objects.count() == 1
+        assert model.snapshots.count() == 1
+        assert model.get_snapshot_model().objects.count() == 1
+    assert Desk.objects.filter(agenda=agenda).count() == 1
+    assert Desk.objects.filter(agenda=snapshot_agenda).count() == 1
+
+    # ok, 24H after page snapshot creation
+    for model in [Agenda, Category, EventsType, Resource, UnavailabilityCalendar]:
+        model.snapshots.update(updated_at=now() - datetime.timedelta(days=1))
+    call_command('clear_snapshots')
+    for model in [Agenda, Category, EventsType, Resource, UnavailabilityCalendar]:
+        assert model.objects.count() == 1
+        assert model.snapshots.count() == 0
+        assert model.get_snapshot_model().objects.count() == 1
+    assert Desk.objects.filter(agenda=agenda).count() == 1
+    assert Desk.objects.filter(agenda=snapshot_agenda).count() == 0