mirror of
https://github.com/element-hq/synapse.git
synced 2025-03-21 23:32:15 +00:00
23740eaa3d
During the migration the automated script to update the copyright headers accidentally got rid of some of the existing copyright lines. Reinstate them.
153 lines
5.7 KiB
Python
153 lines
5.7 KiB
Python
#
|
|
# This file is licensed under the Affero General Public License (AGPL) version 3.
|
|
#
|
|
# Copyright 2020 Half-Shot
|
|
# Copyright (C) 2023 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>.
|
|
#
|
|
# Originally licensed under the Apache License, Version 2.0:
|
|
# <http://www.apache.org/licenses/LICENSE-2.0>.
|
|
#
|
|
# [This file includes modifications made by New Vector Limited]
|
|
#
|
|
#
|
|
from urllib.parse import quote
|
|
|
|
from twisted.test.proto_helpers import MemoryReactor
|
|
|
|
import synapse.rest.admin
|
|
from synapse.rest.client import login, mutual_rooms, room
|
|
from synapse.server import HomeServer
|
|
from synapse.util import Clock
|
|
|
|
from tests import unittest
|
|
from tests.server import FakeChannel
|
|
|
|
|
|
class UserMutualRoomsTest(unittest.HomeserverTestCase):
|
|
"""
|
|
Tests the UserMutualRoomsServlet.
|
|
"""
|
|
|
|
servlets = [
|
|
login.register_servlets,
|
|
synapse.rest.admin.register_servlets_for_client_rest_resource,
|
|
room.register_servlets,
|
|
mutual_rooms.register_servlets,
|
|
]
|
|
|
|
def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
|
|
config = self.default_config()
|
|
return self.setup_test_homeserver(config=config)
|
|
|
|
def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
|
|
self.store = hs.get_datastores().main
|
|
|
|
def _get_mutual_rooms(self, token: str, other_user: str) -> FakeChannel:
|
|
return self.make_request(
|
|
"GET",
|
|
"/_matrix/client/unstable/uk.half-shot.msc2666/user/mutual_rooms"
|
|
f"?user_id={quote(other_user)}",
|
|
access_token=token,
|
|
)
|
|
|
|
def test_shared_room_list_public(self) -> None:
|
|
"""
|
|
A room should show up in the shared list of rooms between two users
|
|
if it is public.
|
|
"""
|
|
self._check_mutual_rooms_with(room_one_is_public=True, room_two_is_public=True)
|
|
|
|
def test_shared_room_list_private(self) -> None:
|
|
"""
|
|
A room should show up in the shared list of rooms between two users
|
|
if it is private.
|
|
"""
|
|
self._check_mutual_rooms_with(
|
|
room_one_is_public=False, room_two_is_public=False
|
|
)
|
|
|
|
def test_shared_room_list_mixed(self) -> None:
|
|
"""
|
|
The shared room list between two users should contain both public and private
|
|
rooms.
|
|
"""
|
|
self._check_mutual_rooms_with(room_one_is_public=True, room_two_is_public=False)
|
|
|
|
def _check_mutual_rooms_with(
|
|
self, room_one_is_public: bool, room_two_is_public: bool
|
|
) -> None:
|
|
"""Checks that shared public or private rooms between two users appear in
|
|
their shared room lists
|
|
"""
|
|
u1 = self.register_user("user1", "pass")
|
|
u1_token = self.login(u1, "pass")
|
|
u2 = self.register_user("user2", "pass")
|
|
u2_token = self.login(u2, "pass")
|
|
|
|
# Create a room. user1 invites user2, who joins
|
|
room_id_one = self.helper.create_room_as(
|
|
u1, is_public=room_one_is_public, tok=u1_token
|
|
)
|
|
self.helper.invite(room_id_one, src=u1, targ=u2, tok=u1_token)
|
|
self.helper.join(room_id_one, user=u2, tok=u2_token)
|
|
|
|
# Check shared rooms from user1's perspective.
|
|
# We should see the one room in common
|
|
channel = self._get_mutual_rooms(u1_token, u2)
|
|
self.assertEqual(200, channel.code, channel.result)
|
|
self.assertEqual(len(channel.json_body["joined"]), 1)
|
|
self.assertEqual(channel.json_body["joined"][0], room_id_one)
|
|
|
|
# Create another room and invite user2 to it
|
|
room_id_two = self.helper.create_room_as(
|
|
u1, is_public=room_two_is_public, tok=u1_token
|
|
)
|
|
self.helper.invite(room_id_two, src=u1, targ=u2, tok=u1_token)
|
|
self.helper.join(room_id_two, user=u2, tok=u2_token)
|
|
|
|
# Check shared rooms again. We should now see both rooms.
|
|
channel = self._get_mutual_rooms(u1_token, u2)
|
|
self.assertEqual(200, channel.code, channel.result)
|
|
self.assertEqual(len(channel.json_body["joined"]), 2)
|
|
for room_id_id in channel.json_body["joined"]:
|
|
self.assertIn(room_id_id, [room_id_one, room_id_two])
|
|
|
|
def test_shared_room_list_after_leave(self) -> None:
|
|
"""
|
|
A room should no longer be considered shared if the other
|
|
user has left it.
|
|
"""
|
|
u1 = self.register_user("user1", "pass")
|
|
u1_token = self.login(u1, "pass")
|
|
u2 = self.register_user("user2", "pass")
|
|
u2_token = self.login(u2, "pass")
|
|
|
|
room = self.helper.create_room_as(u1, is_public=True, tok=u1_token)
|
|
self.helper.invite(room, src=u1, targ=u2, tok=u1_token)
|
|
self.helper.join(room, user=u2, tok=u2_token)
|
|
|
|
# Assert user directory is not empty
|
|
channel = self._get_mutual_rooms(u1_token, u2)
|
|
self.assertEqual(200, channel.code, channel.result)
|
|
self.assertEqual(len(channel.json_body["joined"]), 1)
|
|
self.assertEqual(channel.json_body["joined"][0], room)
|
|
|
|
self.helper.leave(room, user=u1, tok=u1_token)
|
|
|
|
# Check user1's view of shared rooms with user2
|
|
channel = self._get_mutual_rooms(u1_token, u2)
|
|
self.assertEqual(200, channel.code, channel.result)
|
|
self.assertEqual(len(channel.json_body["joined"]), 0)
|
|
|
|
# Check user2's view of shared rooms with user1
|
|
channel = self._get_mutual_rooms(u2_token, u1)
|
|
self.assertEqual(200, channel.code, channel.result)
|
|
self.assertEqual(len(channel.json_body["joined"]), 0)
|