url_decoding: Add NearLinkHandler class.

This prep commit adds the NearLinkHandler class, it simplifies handling
and cleaning up "near links" by providing structure for near link
validation and modification (splitting, reassmebling, etc).

This can be used anywhere if one needs to work with local near links,
such as to remap near links in exported messages during import.

zerver/lib/url_decoding.py

@@ -1,4 +1,4 @@
-from urllib.parse import urlsplit
+from urllib.parse import SplitResult, urlsplit
 
 from django.conf import settings
 
@@ -66,3 +66,56 @@ def is_same_server_message_link(url: str) -> bool:
     ends_with_near_message_id = fragment_parts[-2] == "near" and fragment_parts[-1].isdigit()
 
     return ends_with_near_message_id
+
+
+class NearLinkHandler:
+    """
+    The NearLinkHandler is a helper class for editing
+    and cleaning up near links.
+
+    It can do basic operations such as splitting, fetching
+    link parts, and reassembling. It also applies some near
+    link related validations and clean-up operations.
+
+    See `test_near_link_variations.json` for examples of
+    links this class is intended to handle.
+    """
+
+    def __init__(self, near_link: str) -> None:
+        if not check_near_link_base(near_link):
+            raise AssertionError("This near link is either invalid or not from this server.")
+        self.split_result: SplitResult
+        self.patch_near_link(urlsplit(near_link))
+
+    def clean_near_link(self, split_result: SplitResult) -> SplitResult:
+        """
+        This function fixes legacy near links (uses "stream"),
+        and makes sure relative links starts with "/".
+        """
+        fragment_parts = split_result.fragment.split("/")
+        changed_parts = {}
+
+        if fragment_parts[1] == "stream":
+            fragment_parts[1] = "channel"
+        if split_result.hostname is None and split_result.path == "":
+            # Makes sure a relative near link starts with "/"
+            changed_parts["path"] = "/"
+
+        fragments = "/".join(fragment_parts)
+        changed_parts["fragment"] = fragments
+        cleaned_split_result = split_result._replace(**changed_parts)
+        return cleaned_split_result
+
+    def get_url(self) -> str:
+        return self.split_result.geturl()
+
+    def patch_near_link(self, split_result: SplitResult) -> None:
+        self.split_result = self.clean_near_link(split_result)
+
+    def get_near_link_fragment_parts(self) -> list[str]:
+        return self.split_result.fragment.split("/")
+
+    def patch_near_link_fragment_parts(self, fragment_parts: list[str]) -> None:
+        fragments = "/".join(fragment_parts)
+        patched_split_result = self.split_result._replace(fragment=fragments)
+        self.patch_near_link(patched_split_result)

zerver/tests/fixtures/test_near_link_variations.json

@@ -0,0 +1,90 @@
+{
+    "valid":
+        [
+            {
+                "name": "channel_link",
+                "near_link": "http://testserver/#narrow/channel/13-Denmark",
+                "expected_output": "http://testserver/#narrow/channel/13-Denmark"
+            },
+            {
+                "name": "topic_link",
+                "near_link": "http://testserver/#narrow/channel/13-Denmark/topic/desktop",
+                "expected_output": "http://testserver/#narrow/channel/13-Denmark/topic/desktop"
+            },
+            {
+                "name": "channel_message_link",
+                "near_link": "http://testserver/#narrow/channel/13-Denmark/topic/desktop/near/555",
+                "expected_output": "http://testserver/#narrow/channel/13-Denmark/topic/desktop/near/555"
+            },
+            {
+                "name": "dm_link",
+                "near_link": "http://testserver/#narrow/dm/15-John",
+                "expected_output": "http://testserver/#narrow/dm/15-John"
+            },
+            {
+                "name": "near_dm_link",
+                "near_link": "http://testserver/#narrow/dm/15-John/near/19",
+                "expected_output": "http://testserver/#narrow/dm/15-John/near/19"
+            },
+            {
+                "name": "group_link",
+                "near_link": "http://testserver/#narrow/dm/15,12,13-group",
+                "expected_output": "http://testserver/#narrow/dm/15,12,13-group"
+            },
+            {
+                "name": "near_group_message_link",
+                "near_link": "http://testserver/#narrow/dm/15,12,13-group/near/19",
+                "expected_output": "http://testserver/#narrow/dm/15,12,13-group/near/19"
+            },
+            {
+                "name": "old_near_link_using_stream",
+                "near_link": "http://testserver/#narrow/stream/13-Denmark",
+                "expected_output": "http://testserver/#narrow/channel/13-Denmark"
+            },
+            {
+                "name": "relative_near_link",
+                "near_link": "/#narrow/channel/13-Denmark/topic/desktop/near/555",
+                "expected_output": "/#narrow/channel/13-Denmark/topic/desktop/near/555"
+            },
+            {
+                "name": "broken_relative_near_link",
+                "near_link": "#narrow/channel/13-Denmark/topic/desktop/near/555",
+                "expected_output": "/#narrow/channel/13-Denmark/topic/desktop/near/555"
+            }
+        ],
+    "invalid":
+        [
+            {
+                "name": "other_narrow_link",
+                "near_link": "http://testserver/#narrow/is/starred"
+            },
+            {
+                "name": "different_server_message_link",
+                "near_link": "https://fakechat.zulip.org/#narrow/dm/8,1848,2369-group/near/1717378"
+            },
+            {
+                "name": "empty_fragment",
+                "near_link": "http://testserver/#"
+            },
+            {
+                "name": "incomplete_near_link",
+                "near_link": "http://testserver/#narrow/channel/near/555"
+            },
+            {
+                "name": "recipient_not_encoded",
+                "near_link": "http://testserver/#narrow/channel/Denmark/near/555"
+            },
+            {
+                "name": "recipient_not_encoded_2",
+                "near_link": "http://testserver/#narrow/channel/This is not encoded properly/near/555"
+            },
+            {
+                "name": "broken_recipient_encoding",
+                "near_link": "http://testserver/#narrow/channel/asd-3-asd/near/555"
+            },
+            {
+                "name": "broken_recipient_encoding_2",
+                "near_link": "http://testserver/#narrow/dm/a,3,b-3d/near/555"
+            }
+        ]
+}

zerver/tests/test_url_decoding.py

@@ -1,7 +1,9 @@
 import orjson
 
 from zerver.lib.test_classes import ZulipTestCase
-from zerver.lib.url_decoding import is_same_server_message_link
+from zerver.lib.url_decoding import NearLinkHandler, is_same_server_message_link
+from zerver.lib.url_encoding import near_message_url
+from zerver.models.realms import get_realm
 
 
 class URLDecodeTest(ZulipTestCase):
@@ -11,3 +13,58 @@ class URLDecodeTest(ZulipTestCase):
             self.assertEqual(
                 is_same_server_message_link(test["message_link"]), test["expected_output"]
             )
+
+
+class NearLinkHandlerTest(ZulipTestCase):
+    def build_test_message_near_link(self) -> str:
+        realm = get_realm("zulip")
+        message = dict(
+            type="personal",
+            id=555,
+            display_recipient=[
+                dict(id=77),
+                dict(id=80),
+            ],
+        )
+        url = near_message_url(
+            realm=realm,
+            message=message,
+        )
+        return url
+
+    def test_initialize_near_link(self) -> None:
+        url = self.build_test_message_near_link()
+
+        with self.settings(EXTERNAL_HOST_WITHOUT_PORT="zulip.testserver"):
+            handler = NearLinkHandler(url)
+        self.assertEqual(handler.get_url(), url)
+
+    def test_initialize_various_near_links(self) -> None:
+        tests = orjson.loads(self.fixture_data("test_near_link_variations.json"))
+        for test in tests["valid"]:
+            url = test["near_link"]
+            handler = NearLinkHandler(url)
+            self.assertEqual(handler.get_url(), test["expected_output"], msg=test["name"])
+
+    def test_initialize_invalid_near_links(self) -> None:
+        tests = orjson.loads(self.fixture_data("test_near_link_variations.json"))
+        error_message = "This near link is either invalid or not from this server."
+        for test in tests["invalid"]:
+            url = test["near_link"]
+            with self.assertRaises(AssertionError) as e:
+                NearLinkHandler(url)
+            self.assertEqual(str(e.exception), error_message, msg=test["name"])
+
+    def test_patch_near_link_fragment(self) -> None:
+        old_url = "http://testserver/#narrow/stream/13-Denmark/topic/desktop/near/555"
+        handler = NearLinkHandler(old_url)
+
+        fragment_parts: list[str] = handler.get_near_link_fragment_parts()
+        new_message_id = "444"
+        fragment_parts[-1] = new_message_id
+        handler.patch_near_link_fragment_parts(fragment_parts)
+
+        new_url = (
+            f"http://testserver/#narrow/channel/13-Denmark/topic/desktop/near/{new_message_id}"
+        )
+        self.assertEqual(handler.get_url(), new_url)