Purge revisions of non-page models in purge_revisions command

wagtail/management/commands/purge_revisions.py

@@ -13,7 +13,7 @@
 
 
 class Command(BaseCommand):
-    help = "Delete page revisions which are not the latest revision for a page, published or scheduled to be published, or in moderation"
+    help = "Delete revisions which are not the latest revision, published or scheduled to be published, or in moderation"
 
     def add_arguments(self, parser):
         parser.add_argument(
@@ -39,7 +39,7 @@ def handle(self, *args, **options):
 
 def purge_revisions(days=None):
     # exclude revisions which have been submitted for moderation in the old system
-    purgeable_revisions = Revision.page_revisions.exclude(
+    purgeable_revisions = Revision.objects.exclude(
         submitted_for_moderation=True
     ).exclude(
         # and exclude revisions with an approved_go_live_at date
@@ -61,7 +61,7 @@ def purge_revisions(days=None):
     deleted_revisions_count = 0
 
     for revision in purgeable_revisions.iterator():
-        # don't delete the latest revision for any page
+        # don't delete the latest revision
         if not revision.is_latest_revision():
             revision.delete()
             deleted_revisions_count += 1

wagtail/tests/test_management_commands.py

@@ -608,10 +608,11 @@ def test_future_expired_will_not_be_unpublished(self):
         self.assertFalse(self.snippet.expired)
 
 
-class TestPurgeRevisionsCommand(TestCase):
-    fixtures = ["test.json"]
-
+class TestPurgeRevisionsCommandForPages(TestCase):
     def setUp(self):
+        self.object = self.get_object()
+
+    def get_object(self):
         # Find root page
         self.root_page = Page.objects.get(id=2)
         self.page = SimplePage(
@@ -622,6 +623,7 @@ def setUp(self):
         )
         self.root_page.add_child(instance=self.page)
         self.page.refresh_from_db()
+        return self.page
 
     def run_command(self, days=None):
         if days:
@@ -632,32 +634,25 @@ def run_command(self, days=None):
         return management.call_command("purge_revisions", stdout=StringIO())
 
     def test_latest_revision_not_purged(self):
-
-        revision_1 = self.page.save_revision()
-
-        revision_2 = self.page.save_revision()
+        revision_1 = self.object.save_revision()
+        revision_2 = self.object.save_revision()
 
         self.run_command()
 
         # revision 1 should be deleted, revision 2 should not be
-        self.assertNotIn(
-            revision_1, Revision.page_revisions.filter(object_id=self.page.id)
-        )
-        self.assertIn(
-            revision_2, Revision.page_revisions.filter(object_id=self.page.id)
-        )
+        self.assertFalse(Revision.objects.filter(id=revision_1.id).exists())
+        self.assertTrue(Revision.objects.filter(id=revision_2.id).exists())
 
     def test_revisions_in_moderation_not_purged(self):
+        revision = self.object.save_revision(submitted_for_moderation=True)
 
-        self.page.save_revision(submitted_for_moderation=True)
+        # Save a new revision to ensure that the moderated revision
+        # is not the latest one
+        self.object.save_revision()
 
         self.run_command()
 
-        self.assertTrue(
-            Revision.page_revisions.filter(
-                object_id=self.page.id, submitted_for_moderation=True
-            ).exists()
-        )
+        self.assertTrue(Revision.objects.filter(id=revision.id).exists())
 
         try:
             from wagtail.models import Task, Workflow, WorkflowTask
@@ -667,56 +662,56 @@ def test_revisions_in_moderation_not_purged(self):
             user = get_user_model().objects.first()
             WorkflowTask.objects.create(workflow=workflow, task=task_1, sort_order=1)
 
-            snippet = FullFeaturedSnippet.objects.create(text="Initial", live=False)
-            page_revision = self.page.save_revision()
-            snippet_revision = snippet.save_revision()
-            workflow.start(self.page, user)
-            workflow.start(snippet, user)
+            revision = self.object.save_revision()
+            workflow.start(self.object, user)
+
+            # Save a new revision to ensure that the revision in the workflow
+            # is not the latest one
+            self.object.save_revision()
+
             self.run_command()
 
             # even though they're no longer the latest revisions, the old revisions
             # should stay as they are attached to an in progress workflow
-            self.assertTrue(Revision.objects.filter(id=page_revision.id).exists())
-            self.assertTrue(Revision.objects.filter(id=snippet_revision.id).exists())
+            self.assertTrue(Revision.objects.filter(id=revision.id).exists())
         except ImportError:
             pass
 
     def test_revisions_with_approve_go_live_not_purged(self):
-
-        approved_revision = self.page.save_revision(
+        revision = self.object.save_revision(
             approved_go_live_at=timezone.now() + timedelta(days=1)
         )
 
-        self.page.save_revision()
+        # Save a new revision to ensure that the approved revision
+        # is not the latest one
+        self.object.save_revision()
 
         self.run_command()
 
-        self.assertIn(
-            approved_revision, Revision.page_revisions.filter(object_id=self.page.id)
-        )
+        self.assertTrue(Revision.objects.filter(id=revision.id).exists())
 
     def test_purge_revisions_with_date_cutoff(self):
+        old_revision = self.object.save_revision()
 
-        old_revision = self.page.save_revision()
-
-        self.page.save_revision()
+        self.object.save_revision()
 
         self.run_command(days=30)
 
         # revision should not be deleted, as it is younger than 30 days
-        self.assertIn(
-            old_revision, Revision.page_revisions.filter(object_id=self.page.id)
-        )
+        self.assertTrue(Revision.objects.filter(id=old_revision.id).exists())
 
         old_revision.created_at = timezone.now() - timedelta(days=31)
         old_revision.save()
 
         self.run_command(days=30)
 
         # revision is now older than 30 days, so should be deleted
-        self.assertNotIn(
-            old_revision, Revision.page_revisions.filter(object_id=self.page.id)
-        )
+        self.assertFalse(Revision.objects.filter(id=old_revision.id).exists())
+
+
+class TestPurgeRevisionsCommandForSnippets(TestPurgeRevisionsCommandForPages):
+    def get_object(self):
+        return FullFeaturedSnippet.objects.create(text="Hello world!")
 
 
 class TestPurgeEmbedsCommand(TestCase):