Non-working: Add test_versions.py

2026-07-30 07:29:45 +00:00 · 2026-06-17 19:10:39 -05:00
parent 0226a67858
commit 7bf4b6abd3
1 changed files with 146 additions and 0 deletions
@@ -0,0 +1,146 @@
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2026 Element Creations Ltd
+#
+# 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.
+#
+# See the GNU Affero General Public License for more details:
+# <https://www.gnu.org/licenses/agpl-3.0.html>.
+
+import logging
+
+from twisted.internet.testing import MemoryReactor
+
+from synapse.rest import admin
+from synapse.rest.client import login, versions
+from synapse.server import HomeServer
+from synapse.util.clock import Clock
+from synapse.types import JsonDict
+
+from tests import unittest
+
+logger = logging.getLogger(__name__)
+
+
+class VersionsTestCase(unittest.HomeserverTestCase):
+    """
+    Test `VersionsRestServlet`
+    """
+
+    servlets = [
+        admin.register_servlets,
+        login.register_servlets,
+        versions.register_servlets,
+    ]
+
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+        self.admin_user = self.register_user("admin", "pass", admin=True)
+        self.admin_user_tok = self.login("admin", "pass")
+
+    def test_unauthenticated(self) -> None:
+        channel = self.make_request(
+            "GET",
+            "/_matrix/client/versions",
+            content={},
+        )
+        self.assertEqual(channel.code, 200, channel.result)
+        self._sanity_check_versions_response(channel.json_body)
+
+    def test_authenticated(self) -> None:
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+
+        channel = self.make_request(
+            "GET",
+            "/_matrix/client/versions",
+            content={},
+            access_token=user1_tok,
+        )
+        self.assertEqual(channel.code, 200, channel.result)
+        self._sanity_check_versions_response(channel.json_body)
+
+    def test_authenticated_with_per_user_feature(self) -> None:
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+        user2_id = self.register_user("user2", "pass")
+        user2_tok = self.login(user2_id, "pass")
+
+        # Sanity check that the experimental feature should not be enabled yet
+        channel = self.make_request(
+            "GET",
+            "/_matrix/client/versions",
+            content={},
+            access_token=user1_tok,
+        )
+        self.assertEqual(channel.code, 200, channel.result)
+        self._sanity_check_versions_response(channel.json_body)
+        self.assertEqual(
+            channel.json_body["unstable_features"]["org.matrix.msc3881"],
+            False,
+            channel.json_body,
+        )
+
+        # Enable the feature for this specific user
+        self._enable_experimental_feature_for_user(target_user_id=user1_id)
+
+        # The experimental feature should be enabled for this user
+        channel = self.make_request(
+            "GET",
+            "/_matrix/client/versions",
+            content={},
+            access_token=user1_tok,
+        )
+        self.assertEqual(channel.code, 200, channel.result)
+        self._sanity_check_versions_response(channel.json_body)
+        self.assertEqual(
+            channel.json_body["unstable_features"]["org.matrix.msc3881"],
+            True,
+            channel.json_body,
+        )
+
+        # But not for other users
+        channel = self.make_request(
+            "GET",
+            "/_matrix/client/versions",
+            content={},
+            access_token=user2_tok,
+        )
+        self.assertEqual(channel.code, 200, channel.result)
+        self._sanity_check_versions_response(channel.json_body)
+        self.assertEqual(
+            channel.json_body["unstable_features"]["org.matrix.msc3881"],
+            False,
+            channel.json_body,
+        )
+
+    def _sanity_check_versions_response(self, versions_response: JsonDict) -> None:
+        """
+        Make sure this looks like a `/_matrix/client/versions` response
+        """
+        self.assertIsInstance(
+            versions_response["versions"],
+            list,
+            f"Expected `versions` to be a list of strings but saw {versions_response}",
+        )
+        self.assertIsInstance(
+            versions_response["unstable_features"],
+            dict,
+            f"Expected `unstable_features` to be a dict mapping feature name to a bool but saw {versions_response}",
+        )
+
+    def _enable_experimental_feature_for_user(self, *, target_user_id: str) -> None:
+        """
+        Use the admin API to enable an experimental feature for a specific user
+        """
+        channel = self.make_request(
+            "PUT",
+            f"/_synapse/admin/v1/experimental_features/{target_user_id}",
+            content={
+                "features": {"msc3881": True},
+            },
+            access_token=self.admin_user_tok,
+        )
+        self.assertEqual(channel.code, 200)