Add synced unpublish, delete and move for pages

wagtail_localize/tests/test_synctree.py

@@ -1,6 +1,9 @@
+import unittest
+
 from django.contrib.contenttypes.models import ContentType
-from django.test import TestCase
+from django.test import TestCase, override_settings
 from django.urls import reverse
+from wagtail import VERSION as WAGTAIL_VERSION
 from wagtail.core.models import Locale, Page
 from wagtail.tests.utils import WagtailTestUtils
 
@@ -9,6 +12,12 @@
 from wagtail_localize.test.models import TestHomePage, TestPage
 
 
+try:
+    from wagtail import hooks
+except ImportError:
+    from wagtail.core import hooks
+
+
 class TestPageIndex(TestCase):
     def setUp(self):
         self.en_locale = Locale.objects.get(language_code="en")
@@ -93,7 +102,7 @@ def test_from_database(self):
         self.assertEqual(canadaonlypage_entry.aliased_locales, [])
 
 
-class TestSignalsAndHooks(TestCase, WagtailTestUtils):
+class SyncTreeTestsSetupBase(TestCase):
     def setUp(self):
         self.en_locale = Locale.objects.get(language_code="en")
         self.fr_locale = Locale.objects.create(language_code="fr")
@@ -126,6 +135,11 @@ def setUp(self):
             )
         )
 
+
+class TestSignalsAndHooks(SyncTreeTestsSetupBase, WagtailTestUtils):
+    def setUp(self):
+        super().setUp()
+
         LocaleSynchronization.objects.create(
             locale=self.fr_locale,
             sync_from=self.en_locale,
@@ -263,3 +277,89 @@ def test_create_homepage_in_sync_source_locale(self):
         self.assertTrue(new_en_homepage.has_translation(self.fr_locale))
         self.assertTrue(new_en_homepage.has_translation(self.fr_ca_locale))
         self.assertTrue(new_en_homepage.has_translation(self.es_locale))
+
+
+@unittest.skipUnless(
+    WAGTAIL_VERSION >= (4, 0),
+    "construct_translated_pages_to_cascade_actions was added starting with Wagtail 3.0",
+)
+class TestConstructSyncedPageTreeListHook(SyncTreeTestsSetupBase):
+    def _get_hook_function(self):
+        the_hooks = hooks.get_hooks("construct_translated_pages_to_cascade_actions")
+        return the_hooks[0]
+
+    def setup_locale_synchronisation(self, locale, sync_from_locale):
+        LocaleSynchronization.objects.create(
+            locale=locale,
+            sync_from=sync_from_locale,
+        )
+
+    def test_hook(self):
+        the_hooks = hooks.get_hooks("construct_translated_pages_to_cascade_actions")
+        self.assertEqual(len(the_hooks), 1)
+
+    def test_hook_returns_nothing_without_locale_synchronisation(self):
+        hook = self._get_hook_function()
+        for action in ["unpublish", "delete", "move"]:
+            with self.subTest(
+                f"Calling construct_translated_pages_to_cascade_actions with {action}"
+            ):
+                results = hook([self.en_aboutpage], action)
+                self.assertDictEqual(results, {})
+
+    def test_hook_returns_nothing_without_explicit_setting(self):
+        self.setup_locale_synchronisation(self.fr_locale, self.en_locale)
+        hook = self._get_hook_function()
+        for action in ["unpublish", "delete", "move"]:
+            with self.subTest(
+                f"Calling construct_translated_pages_to_cascade_actions with {action} "
+                f"without WAGTAILLOCALIZE_CASCADE_PAGE_ACTIONS set"
+            ):
+                results = hook([self.en_aboutpage], action)
+                self.assertDictEqual(results, {})
+
+        with override_settings(WAGTAILLOCALIZE_CASCADE_PAGE_ACTIONS=False):
+            for action in ["unpublish", "delete", "move"]:
+                with self.subTest(
+                    f"Calling construct_translated_pages_to_cascade_actions with {action} "
+                    f"and WAGTAILLOCALIZE_CASCADE_PAGE_ACTIONS=False"
+                ):
+                    results = hook([self.en_aboutpage], action)
+                    self.assertDictEqual(results, {})
+
+    @override_settings(WAGTAILLOCALIZE_CASCADE_PAGE_ACTIONS=True)
+    def test_hook_returns_relevant_pages_from_synced_locale_on_unpublish_action(self):
+        self.setup_locale_synchronisation(self.fr_locale, self.en_locale)
+        hook = self._get_hook_function()
+        results = hook([self.en_aboutpage], "unpublish")
+        self.assertIsNotNone(results.get(self.en_aboutpage))
+        self.assertQuerysetEqual(
+            results[self.en_aboutpage], Page.objects.filter(pk=self.fr_aboutpage.pk)
+        )
+
+        # unpublish should not include alias pages as they follow the parent
+        self.fr_aboutpage.alias_of = self.en_aboutpage
+        self.fr_aboutpage.save()
+        results = hook([self.en_aboutpage], "unpublish")
+        self.assertIsNotNone(results.get(self.en_aboutpage))
+        self.assertQuerysetEqual(results[self.en_aboutpage], Page.objects.none())
+
+    @override_settings(WAGTAILLOCALIZE_CASCADE_PAGE_ACTIONS=True)
+    def test_hook_returns_relevant_pages_from_synced_locale_on_move_action(self):
+        self.setup_locale_synchronisation(self.fr_locale, self.en_locale)
+        hook = self._get_hook_function()
+        results = hook([self.en_aboutpage], "move")
+        self.assertIsNotNone(results.get(self.en_aboutpage))
+        self.assertQuerysetEqual(
+            results[self.en_aboutpage], Page.objects.filter(pk=self.fr_aboutpage.pk)
+        )
+
+    @override_settings(WAGTAILLOCALIZE_CASCADE_PAGE_ACTIONS=True)
+    def test_hook_returns_relevant_pages_from_synced_locale_on_delete_action(self):
+        self.setup_locale_synchronisation(self.fr_locale, self.en_locale)
+        hook = self._get_hook_function()
+        results = hook([self.en_aboutpage], "move")
+        self.assertIsNotNone(results.get(self.en_aboutpage))
+        self.assertQuerysetEqual(
+            results[self.en_aboutpage], Page.objects.filter(pk=self.fr_aboutpage.pk)
+        )

wagtail_localize/wagtail_hooks.py

@@ -1,3 +1,4 @@
+from typing import List
 from urllib.parse import urlencode
 
 from django.contrib.admin.utils import quote
@@ -466,3 +467,49 @@ def convert_to_alias_message(data):
 def register_icons(icons):
     # icon id "wagtail-localize-convert" (which translates to `.icon-wagtail-localize-convert`)
     return icons + ["wagtail_localize/icons/wagtail-localize-convert.svg"]
+
+
+if WAGTAIL_VERSION >= (4, 0):
+    from django.conf import settings
+
+    from .models import LocaleSynchronization
+
+    @hooks.register("construct_translated_pages_to_cascade_actions")
+    def construct_synced_page_tree_list(pages: List[Page], action: str) -> dict:
+        if not getattr(settings, "WAGTAILLOCALIZE_CASCADE_PAGE_ACTIONS", False):
+            return {}
+
+        locale_sync_map = {}
+        for page in pages:
+            # TODO: what about locale C follows B which follows A, when we come in from A?
+            if page.locale_id not in locale_sync_map:
+                locale_sync_map[page.locale_id] = list(
+                    LocaleSynchronization.objects.filter(
+                        sync_from=page.locale_id
+                    ).values_list("locale", flat=True)
+                )
+
+        page_list = {}
+        if action == "unpublish":
+            for page in pages:
+                if not locale_sync_map[page.locale_id]:
+                    # no locales sync from this page locale
+                    continue
+
+                page_list[page] = Page.objects.translation_of(
+                    page, inclusive=False
+                ).filter(
+                    alias_of__isnull=True, locale__in=locale_sync_map[page.locale_id]
+                )
+        elif action in ["move", "delete"]:
+            for page in pages:
+                if not locale_sync_map[page.locale_id]:
+                    # no locales sync from this page locale
+                    continue
+
+                # only include those translations or aliases from locales that sync from this locale
+                page_list[page] = Page.objects.translation_of(
+                    page, inclusive=False
+                ).filter(locale__in=locale_sync_map[page.locale_id])
+
+        return page_list