synapse/tests/rest/client/test_versions.py

# 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.types import JsonDict
from synapse.util.clock import Clock

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 tearDown(self) -> None:
        # MemoryReactor doesn't trigger the shutdown phases, and we want the
        # Tokio thread pool to be stopped
        # XXX: This logic should probably get moved somewhere else
        shutdown_triggers = self.reactor.triggers.get("shutdown", {})
        for phase in ["before", "during", "after"]:
            triggers = shutdown_triggers.get(phase, [])
            for callbable, args, kwargs in triggers:
                callbable(*args, **kwargs)

    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",
        )
        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",
            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",
            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, features={"msc3881": True}
        )

        # The experimental feature should be enabled for this user
        channel = self.make_request(
            "GET",
            "/_matrix/client/versions",
            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",
            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 test_msc4446_false_by_default(self) -> None:
        channel = self.make_request("GET", "/_matrix/client/versions")
        self.assertEqual(channel.code, 200, channel.result)
        self.assertFalse(channel.json_body["unstable_features"]["com.beeper.msc4446"])

    @unittest.override_config({"experimental_features": {"msc4446_enabled": True}})
    def test_msc4446_true_if_enabled(self) -> None:
        channel = self.make_request("GET", "/_matrix/client/versions")
        self.assertEqual(channel.code, 200, channel.result)
        self.assertTrue(channel.json_body["unstable_features"]["com.beeper.msc4446"])

    def test_msc4502_false_by_default(self) -> None:
        channel = self.make_request("GET", "/_matrix/client/versions")
        self.assertEqual(channel.code, 200, channel.result)
        self.assertFalse(channel.json_body["unstable_features"]["io.element.msc4502"])

    @unittest.override_config({"experimental_features": {"msc4502_enabled": True}})
    def test_msc4502_true_if_enabled(self) -> None:
        channel = self.make_request("GET", "/_matrix/client/versions")
        self.assertEqual(channel.code, 200, channel.result)
        self.assertTrue(channel.json_body["unstable_features"]["io.element.msc4502"])

    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, features: dict[str, bool]
    ) -> 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": features,
            },
            access_token=self.admin_user_tok,
        )
        self.assertEqual(channel.code, 200)