synapse/tests/rest/client/test_matrixrtc.py

#
# This file is licensed under the Affero General Public License (AGPL) version 3.
#
# Copyright (C) 2025 New Vector, 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>.
#
# [This file includes modifications made by New Vector Limited]
#
#

"""Tests REST events for /rtc/endpoints path."""

import unittest as stdlib_unittest

from pydantic import ValidationError

from twisted.internet.testing import MemoryReactor

from synapse.config.matrixrtc import TransportConfigModel
from synapse.rest import admin
from synapse.rest.client import login, matrixrtc, register, room, versions
from synapse.server import HomeServer
from synapse.synapse_rust.http_client import HttpClient
from synapse.util.clock import Clock

from tests import unittest
from tests.unittest import HomeserverTestCase, override_config

PATH_PREFIX = "/_matrix/client/unstable/org.matrix.msc4143"
RTC_ENDPOINT = {"type": "focusA", "required_field": "theField"}
LIVEKIT_TRANSPORT = {
    "type": "livekit",
    "url": "wss://livekit.example.com",
}
BACKWARDS_COMPATIBLE_LIVEKIT_TRANSPORT = {
    "type": "livekit",
    "url": "wss://livekit.example.com",
    "livekit_service_url": "https://livekit.example.com",
}
LEGACY_LIVEKIT_TRANSPORT = {
    "type": "livekit",
    "livekit_service_url": "https://livekit.example.com",
}


class MatrixRtcTestCase(HomeserverTestCase):
    """Tests /rtc/transports Client-Server REST API."""

    servlets = [
        admin.register_servlets,
        room.register_servlets,
        login.register_servlets,
        register.register_servlets,
        matrixrtc.register_servlets,
    ]

    def prepare(
        self, reactor: MemoryReactor, clock: Clock, homeserver: HomeServer
    ) -> None:
        self.register_user("alice", "password")
        self._alice_tok = self.login("alice", "password")

    def test_matrixrtc_endpoint_not_enabled(self) -> None:
        channel = self.make_request(
            "GET", f"{PATH_PREFIX}/rtc/transports", access_token=self._alice_tok
        )
        self.assertEqual(404, channel.code, channel.json_body)
        self.assertEqual(
            "M_UNRECOGNIZED", channel.json_body["errcode"], channel.json_body
        )

    @override_config({"experimental_features": {"msc4143_enabled": True}})
    def test_matrixrtc_endpoint_requires_authentication(self) -> None:
        channel = self.make_request("GET", f"{PATH_PREFIX}/rtc/transports")
        self.assertEqual(401, channel.code, channel.json_body)

    @override_config(
        {
            "experimental_features": {"msc4143_enabled": True},
            "matrix_rtc": {"transports": [RTC_ENDPOINT]},
        }
    )
    def test_matrixrtc_endpoint_contains_expected_transport(self) -> None:
        channel = self.make_request(
            "GET", f"{PATH_PREFIX}/rtc/transports", access_token=self._alice_tok
        )
        self.assertEqual(200, channel.code, channel.json_body)
        self.assert_dict({"rtc_transports": [RTC_ENDPOINT]}, channel.json_body)

    @override_config(
        {
            "experimental_features": {"msc4143_enabled": True},
            "matrix_rtc": {"transports": []},
        }
    )
    def test_matrixrtc_endpoint_no_transports_configured(self) -> None:
        channel = self.make_request(
            "GET", f"{PATH_PREFIX}/rtc/transports", access_token=self._alice_tok
        )
        self.assertEqual(200, channel.code, channel.json_body)
        self.assert_dict({}, channel.json_body)

    @override_config(
        {
            "experimental_features": {"msc4143_enabled": True},
            "matrix_rtc": {"transports": [LIVEKIT_TRANSPORT]},
        }
    )
    def test_matrixrtc_endpoint_livekit_transport(self) -> None:
        channel = self.make_request(
            "GET", f"{PATH_PREFIX}/rtc/transports", access_token=self._alice_tok
        )
        self.assertEqual(200, channel.code, channel.json_body)
        self.assert_dict({"rtc_transports": [LIVEKIT_TRANSPORT]}, channel.json_body)

    @override_config(
        {
            "experimental_features": {"msc4143_enabled": True},
            "matrix_rtc": {"transports": [BACKWARDS_COMPATIBLE_LIVEKIT_TRANSPORT]},
        }
    )
    def test_matrixrtc_endpoint_backwards_compatible_livekit_transport(self) -> None:
        channel = self.make_request(
            "GET", f"{PATH_PREFIX}/rtc/transports", access_token=self._alice_tok
        )
        self.assertEqual(200, channel.code, channel.json_body)
        self.assert_dict(
            {"rtc_transports": [BACKWARDS_COMPATIBLE_LIVEKIT_TRANSPORT]},
            channel.json_body,
        )

    @override_config(
        {
            "experimental_features": {"msc4143_enabled": True},
            "matrix_rtc": {"transports": [LEGACY_LIVEKIT_TRANSPORT]},
        }
    )
    def test_matrixrtc_endpoint_legacy_livekit_transport(self) -> None:
        channel = self.make_request(
            "GET", f"{PATH_PREFIX}/rtc/transports", access_token=self._alice_tok
        )
        self.assertEqual(200, channel.code, channel.json_body)
        self.assert_dict(
            {"rtc_transports": [LEGACY_LIVEKIT_TRANSPORT]}, channel.json_body
        )


class MatrixRtcVersionsTestCase(HomeserverTestCase):
    """Tests that org.matrix.msc4143 is correctly advertised in /versions."""

    servlets = [versions.register_servlets]

    def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
        hs = self.setup_test_homeserver()

        # XXX: We must create the Rust HTTP client before we call `reactor.run()` below.
        # Twisted's `MemoryReactor` doesn't invoke `callWhenRunning` callbacks if it's
        # already running and we rely on that to start the Tokio thread pool in Rust. In
        # the future, this may not matter, see https://github.com/twisted/twisted/pull/12514
        self._http_client = hs.get_proxied_http_client()
        _ = HttpClient(
            reactor=hs.get_reactor(),
            user_agent=self._http_client.user_agent.decode("utf8"),
        )

        # This triggers the server startup hooks, which starts the Tokio thread pool
        reactor.run()

        return hs

    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 test_msc4143_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"]["org.matrix.msc4143"])

    @unittest.override_config({"experimental_features": {"msc4143_enabled": True}})
    def test_msc4143_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"]["org.matrix.msc4143"])


class TransportConfigModelTestCase(stdlib_unittest.TestCase):
    """Tests validation of the `TransportConfigModel` pydantic model."""

    def test_livekit_transport_requires_url_or_livekit_service_url(self) -> None:
        with self.assertRaises(ValidationError):
            TransportConfigModel(type="livekit")

    def test_livekit_transport_with_only_url(self) -> None:
        TransportConfigModel(type="livekit", url="wss://livekit.example.com")

    def test_livekit_transport_with_only_livekit_service_url(self) -> None:
        TransportConfigModel(
            type="livekit", livekit_service_url="https://livekit.example.com"
        )

    def test_invalid_field_type(self) -> None:
        with self.assertRaises(ValidationError):
            TransportConfigModel(type="livekit", url=1234)  # type: ignore[arg-type]