Make redirects query compatible with multi-site setups (#380)

* Renamed `RedirectType` to `RedirectObjectType`
* Simplified `old_url` handling (by using the `Redirect.link` property)
* Changed `RedirectQuery` to return a result for each related Site
* Added `site` field to `RedirectObjectType`.

CHANGELOG.md

@@ -1,9 +1,17 @@
 ## Unreleased
 
+### Changed
+
+-   Renamed `RedirectType` to `RedirectObjectType` ([#380](https://github.com/torchbox/wagtail-grapple/pull/380)) @JakubMastalerz
+-   `RedirectsQuery` now returns a `Redirect` for each associated `Site` ([#380](https://github.com/torchbox/wagtail-grapple/pull/380)) @JakubMastalerz
+    -   Added `site` field on `RedirectObjectType`
+    -   Changed `new_url` field on `RedirectObjectType` to be optional
+
 ### Fixed
 
 - `test_src_set_invalid_format` not working with Wagtail 5.2 and above. ([#378](https://github.com/torchbox/wagtail-grapple/pull/378)) @JakubMastalerz
 
+
 ## [0.23.0] - 2023-09-29
 
 ### Added

grapple/types/redirects.py

@@ -1,38 +1,77 @@
+import copy
+
+from typing import List, Optional
+
 import graphene
 
-from django.conf import settings
 from wagtail.contrib.redirects.models import Redirect
+from wagtail.models import Page, Site
+
+from grapple.types.sites import SiteObjectType
 
 from .pages import get_page_interface
 
 
-class RedirectType(graphene.ObjectType):
+class RedirectObjectType(graphene.ObjectType):
     old_path = graphene.String(required=True)
     old_url = graphene.String(required=True)
-    new_url = graphene.String(required=True)
+    new_url = graphene.String(required=False)
     page = graphene.Field(get_page_interface())
+    site = graphene.Field(
+        SiteObjectType, required=True
+    )  # Required because `RedirectsQuery` always adds a value.
     is_permanent = graphene.Boolean(required=True)
 
-    # Give old_path with BASE_URL attached.
-    def resolve_old_url(self, info, **kwargs):
-        return settings.BASE_URL + self.old_path
+    class Meta:
+        name = "Redirect"
+
+    def resolve_old_url(self, info, **kwargs) -> str:
+        """
+        Resolve the value of `old_url` using the `root_url` of the associated
+        site and `old_path`.
+        """
 
-    # Get new url
-    def resolve_new_url(self, info, **kwargs):
-        if self.redirect_page is None:
-            return self.link
+        return self.site.root_url + self.old_path
 
-        return self.redirect_page.url
+    def resolve_new_url(self, info, **kwargs) -> Optional[str]:
+        """
+        Resolve the value of `new_url`. If `redirect_page` is specified then its
+        URL is prioritised.
+        """
+
+        return self.link
 
     # Return the page that's being redirected to, if at all.
-    def resolve_page(self, info, **kwargs):
+    def resolve_page(self, info, **kwargs) -> Optional[Page]:
         if self.redirect_page is not None:
             return self.redirect_page.specific
 
 
 class RedirectsQuery:
-    redirects = graphene.List(graphene.NonNull(RedirectType), required=True)
+    redirects = graphene.List(graphene.NonNull(RedirectObjectType), required=True)
 
     # Return all redirects.
-    def resolve_redirects(self, info, **kwargs):
-        return Redirect.objects.select_related("redirect_page")
+    def resolve_redirects(self, info, **kwargs) -> List[Redirect]:
+        """
+        Resolve the query set of redirects. If `site` is None, a redirect works
+        for all sites. To show this, a new redirect object is created for each
+        of the sites.
+        """
+
+        redirects_qs = (
+            Redirect.objects.select_related("redirect_page")
+            .select_related("site")
+            .all()
+        )
+        finalised_redirects: list[Redirect] = []  # Redirects to return in API.
+
+        for redirect in redirects_qs:
+            if redirect.site is None:
+                # Duplicate Redirect for each Site as it applies to all Sites.
+                for site in Site.objects.all():
+                    _new_redirect = copy.deepcopy(redirect)
+                    _new_redirect.site = site
+                    finalised_redirects.append(_new_redirect)
+            else:
+                finalised_redirects.append(redirect)
+        return finalised_redirects

tests/test_redirects.py

@@ -0,0 +1,279 @@
+from test_grapple import BaseGrappleTest
+from testapp.factories import RedirectFactory
+from testapp.models import BlogPage
+from wagtail.contrib.redirects.models import Redirect
+from wagtail.models import Site
+from wagtail_factories import SiteFactory
+
+
+class TestRedirectQueries(BaseGrappleTest):
+    @classmethod
+    def setUpTestData(cls) -> None:
+        super().setUpTestData()
+        cls.page = BlogPage(title="Test page", slug="test-page-url", date="2020-01-01")
+        cls.home.add_child(instance=cls.page)
+
+    def test_basic_query(self):
+        """
+        Test that Redirect fields can be queried through graphql.
+        """
+
+        # Create a basic redirect without a specified `redirect_page` or `site`.
+        RedirectFactory(
+            old_path="/test/old",
+            redirect_link="http://localhost:8000/test/new",
+            is_permanent=True,
+            redirect_page=None,
+            site=None,
+        )
+
+        query = """
+        {
+            redirects {
+                oldPath
+                newUrl
+                isPermanent
+            }
+        }
+        """
+
+        result = self.client.execute(query)
+
+        # Verify that the structure of the results is correct.
+        self.assertEqual(type(result["data"]), dict)
+        self.assertEqual(type(result["data"]["redirects"]), list)
+        self.assertEqual(type(result["data"]["redirects"][0]), dict)
+
+        result_data = result["data"]["redirects"][0]
+
+        self.assertEqual(result_data["oldPath"], "/test/old")
+        self.assertEqual(result_data["newUrl"], "http://localhost:8000/test/new")
+        self.assertEqual(result_data["isPermanent"], True)
+
+    def test_sub_type_query(self):
+        """
+        Test that `Page` and `Site` fields on Redirects can be queried through graphql.
+        """
+
+        # Create a redirect with specified `redirect_page` and `site`.
+        RedirectFactory(
+            redirect_page=self.page,
+            site=SiteFactory(
+                hostname="test-site",
+                port=81,
+            ),
+        )
+
+        query = """
+        {
+            redirects {
+                page {
+                    title
+                    url
+                }
+                site {
+                    hostname
+                    port
+                }
+            }
+        }
+        """
+
+        result = self.client.execute(query)
+
+        # Verify the structure of the query for `page` and `site`.
+        self.assertEqual(type(result["data"]["redirects"][0]), dict)
+        self.assertEqual(type(result["data"]["redirects"][0]["page"]), dict)
+        self.assertEqual(type(result["data"]["redirects"][0]["site"]), dict)
+
+        result_data = result["data"]["redirects"][0]
+
+        self.assertEqual(result_data["page"]["title"], "Test page")
+        self.assertEqual(result_data["page"]["url"], "http://localhost/test-page-url/")
+        self.assertEqual(result_data["site"]["hostname"], "test-site")
+        self.assertEqual(result_data["site"]["port"], 81)
+
+    def test_new_url_sources(self):
+        """
+        Test that when redirects are queried, the right source for `newUrl` is chosen.
+        When both are provided: `newUrl` is taken from `redirect page` rather than `redirect link`.
+        When just one is provided, use that.
+        """
+
+        # Create a redirect with both `redirect_link` and `redirect_page`.
+        RedirectFactory(redirect_link="/test/incorrect", redirect_page=self.page)
+        # Create a redirect with just `redirect_page`.
+        RedirectFactory(
+            redirect_link="",
+            is_permanent=True,
+            redirect_page=self.page,
+        )
+        # Create a redirect with just `redirect_link`.
+        RedirectFactory(redirect_link="http://test/just-link", redirect_page=None)
+
+        query = """
+        {
+            redirects {
+                newUrl
+            }
+        }
+        """
+
+        result = self.client.execute(query)["data"]["redirects"]
+
+        # Sanity check to ensure only those three redirect objects are in the database.
+        self.assertEqual(Redirect.objects.count(), 3)
+
+        # Redirects with a specified `redirect_page` should return its url.
+        self.assertEqual(result[0]["newUrl"], "http://localhost/test-page-url/")
+        self.assertEqual(result[1]["newUrl"], "http://localhost/test-page-url/")
+        # Redirects without a specified `redirect_page` should return the
+        # `redirect_link` instead.
+        self.assertEqual(result[2]["newUrl"], "http://test/just-link")
+
+    def test_no_new_url_query(self):
+        """
+        Test that a redirect can be created without a `redirect_link` or
+        `redirect_page` and queried. This can be used for producing a HTTP 410
+        for any pages/URLs that are no longer in existence.
+        """
+
+        # Create a redirect with no source for `new_url` generation.
+        RedirectFactory(
+            redirect_link="",
+            redirect_page=None,
+        )
+
+        query = """
+        {
+            redirects {
+                newUrl
+            }
+        }
+        """
+
+        result = self.client.execute(query)["data"]["redirects"][0]["newUrl"]
+
+        self.assertEqual(result, None)
+
+    def test_specified_site_url(self):
+        """
+        Test that a redirect with a specified site returns the correct `old_url`.
+        """
+        # Create a redirect with `site` using port 81, which should result in a
+        # standard `old_url` generation.
+        RedirectFactory(
+            old_path="old-path",
+            site=SiteFactory(
+                hostname="test-site",
+                port=81,
+            ),
+        )
+        # Create a redirect with `site` using port 80, which should be resolved
+        # to "http" without a specified port in `old_url`.
+        RedirectFactory(
+            old_path="old-path",
+            site=SiteFactory(
+                hostname="test-site-default-port",
+                port=80,
+            ),
+        )
+        # Create a redirect with `site` using port 443, which should be resolved
+        # to "https" without a specified port in `old_url`.
+        RedirectFactory(
+            old_path="old-path",
+            site=SiteFactory(
+                hostname="test-site-secure",
+                port=443,
+            ),
+        )
+
+        query = """
+        {
+            redirects {
+                oldUrl
+            }
+        }
+        """
+
+        result = self.client.execute(query)["data"]["redirects"]
+
+        # Sanity check to ensure only those three redirect objects are in the database.
+        self.assertEqual(Redirect.objects.count(), 3)
+
+        self.assertEqual(result[0]["oldUrl"], "http://test-site:81/old-path")
+        self.assertEqual(result[1]["oldUrl"], "http://test-site-default-port/old-path")
+        self.assertEqual(result[2]["oldUrl"], "https://test-site-secure/old-path")
+
+    def test_all_sites_url(self):
+        """
+        Test that when no site is specified on a redirect, that redirect is
+        shown for each existing site.
+        """
+
+        # Create two additional sites (on top of the default `Localhost` one).
+        SiteFactory(hostname="test-site", port=81)
+        SiteFactory(hostname="another-test-site", port=82)
+
+        # Create a `Redirect` without a `site` (treated as active for all sites).
+        RedirectFactory(
+            old_path="old-path",
+            site=None,
+        )
+
+        query = """
+        {
+            redirects {
+                oldUrl
+            }
+        }
+        """
+
+        result = self.client.execute(query)["data"]["redirects"]
+
+        # Sanity check to ensure there are three `Site`s.
+        self.assertEqual(Site.objects.count(), 3)
+
+        # Sanity check to ensure only one `Redirect` is in the database.
+        self.assertEqual(Redirect.objects.count(), 1)
+
+        # Ensure there are three `Redirect`s returned despite only 1 object
+        # in the database.
+        self.assertEqual(len(result), 3)
+
+        self.assertEqual(result[0]["oldUrl"], "http://another-test-site:82/old-path")
+        self.assertEqual(result[1]["oldUrl"], "http://localhost/old-path")
+        self.assertEqual(result[2]["oldUrl"], "http://test-site:81/old-path")
+
+    def test_query_efficiency(self):
+        """
+        Verify the number of queries when querying Redirects is constant to
+        prevent n+1 queries.
+        """
+
+        # Number of queries should remain constant for any N of redirects.
+        RedirectFactory()
+        RedirectFactory()
+        RedirectFactory()
+
+        query = """
+        {
+            redirects {
+                oldPath
+                newUrl
+                isPermanent
+                page {
+                    title
+                    url
+                }
+                site {
+                    hostname
+                    port
+                }
+            }
+        }
+        """
+
+        # There should be one SELECT query for Redirects and one for Sites.
+        with self.assertNumQueries(2):
+            self.client.execute(query)