# # 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: # . from parameterized import parameterized_class from twisted.internet.testing import MemoryReactor import synapse.rest.admin from synapse.api.constants import EventTypes, RelationTypes from synapse.config.server_announcements import ServerAnnouncementRoom from synapse.rest.client import login, relations, room from synapse.server import HomeServer from synapse.types import JsonDict from synapse.util.clock import Clock from tests import unittest @parameterized_class( ("reader_type",), [("never_joined",), ("removed_sender",)], ) class ServerAnnouncementsHistoryTestCase(unittest.HomeserverTestCase): """Tests that a local user who is not a member of a server announcement room can read its history through the endpoints clients use outside of `/sync`. The reader has either never been in the room, or is a former announcement sender who was removed from the room before the announcement was sent. Both scenarios are tested because they reach the history through different paths: - a user who never joined relies on the room being world-readable - a removed sender has a leave membership, which would otherwise hide every event sent after they left. """ reader_type: str servlets = [ synapse.rest.admin.register_servlets, login.register_servlets, room.register_servlets, room.register_deprecated_servlets, relations.register_servlets, ] def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: self.announcer = self.register_user("announcer", "pass") self.announcer_tok = self.login("announcer", "pass") reader = self.register_user("reader", "pass") self.reader_tok = self.login("reader", "pass") room_config = ServerAnnouncementRoom( config_id="news", room_name="Server News", room_topic="News from the server", room_avatar_url=None, room_avatar_path=None, allowed_user_ids=[self.announcer], ) config = hs.config.serverannouncements handler = hs.get_server_announcements_handler() # Make the reader initially join the room as an announcement sender. # The second reconciliation below then removes them, so their leave # comes before the announcement is sent. if self.reader_type == "removed_sender": config.rooms = [ ServerAnnouncementRoom( config_id=room_config.config_id, room_name=room_config.room_name, room_topic=room_config.room_topic, room_avatar_url=room_config.room_avatar_url, room_avatar_path=room_config.room_avatar_path, allowed_user_ids=[self.announcer, reader], ) ] self.get_success(handler.reconcile_rooms()) config.rooms = [room_config] self.get_success(handler.reconcile_rooms()) room_id = self.get_success( hs.get_datastores().main.get_server_announcement_room("news") ) assert room_id is not None self.room_id = room_id self.announcement_id = self.helper.send( self.room_id, "hello", tok=self.announcer_tok )["event_id"] def _get(self, path: str) -> JsonDict: channel = self.make_request("GET", path, access_token=self.reader_tok) self.assertEqual(channel.code, 200, channel.json_body) return channel.json_body def _send_relation(self, rel_type: str, content: JsonDict) -> str: """Send a relation as the announcer user.""" return self.helper.send_event( self.room_id, EventTypes.Reaction if rel_type == RelationTypes.ANNOTATION else EventTypes.Message, content={ **content, "m.relates_to": { **content.get("m.relates_to", {}), "rel_type": rel_type, "event_id": self.announcement_id, }, }, tok=self.announcer_tok, )["event_id"] def test_messages(self) -> None: """The reader can paginate back to the announcement.""" body = self._get(f"/rooms/{self.room_id}/messages?dir=b") self.assertIn( self.announcement_id, [event["event_id"] for event in body["chunk"]] ) def test_context(self) -> None: """The reader can load the announcement with its surrounding events.""" body = self._get(f"/rooms/{self.room_id}/context/{self.announcement_id}") self.assertEqual(body["event"]["event_id"], self.announcement_id) def test_event(self) -> None: """The reader can fetch the announcement by its ID.""" body = self._get(f"/rooms/{self.room_id}/event/{self.announcement_id}") self.assertEqual(body["event_id"], self.announcement_id) def test_relations(self) -> None: """The reader can list the reactions to the announcement.""" reaction_id = self._send_relation( RelationTypes.ANNOTATION, {"m.relates_to": {"key": "👍"}} ) body = self._get( f"/_matrix/client/v1/rooms/{self.room_id}/relations/{self.announcement_id}" ) self.assertEqual([event["event_id"] for event in body["chunk"]], [reaction_id]) def test_threads(self) -> None: """The reader can list the announcement as a thread root.""" self._send_relation(RelationTypes.THREAD, {"msgtype": "m.text", "body": "a"}) body = self._get(f"/_matrix/client/v1/rooms/{self.room_id}/threads") self.assertEqual( [event["event_id"] for event in body["chunk"]], [self.announcement_id] ) def test_room_initial_sync(self) -> None: """The reader can load a snapshot of the room including the announcement.""" body = self._get(f"/rooms/{self.room_id}/initialSync") self.assertIn( self.announcement_id, [event["event_id"] for event in body["messages"]["chunk"]], ) def test_messages_after_room_is_unconfigured(self) -> None: """Once the room is removed from the configuration it is treated like any other world-readable room: a reader who never joined can still paginate back to the announcement, but a removed sender only sees the history up to their leave.""" self.hs.config.serverannouncements.rooms = [] store = self.hs.get_datastores().main store.get_server_announcement_config_ids_by_room_id.invalidate_all() body = self._get(f"/rooms/{self.room_id}/messages?dir=b") event_ids = [event["event_id"] for event in body["chunk"]] if self.reader_type == "never_joined": self.assertIn(self.announcement_id, event_ids) else: self.assertNotIn(self.announcement_id, event_ids)