diff --git a/tests/handlers/test_directory.py b/tests/handlers/test_directory.py
index 91c7a17070..00bb776271 100644
--- a/tests/handlers/test_directory.py
+++ b/tests/handlers/test_directory.py
@@ -18,25 +18,20 @@ from mock import Mock
from twisted.internet import defer
+import synapse
+import synapse.api.errors
+from synapse.api.constants import EventTypes
from synapse.config.room_directory import RoomDirectoryConfig
-from synapse.handlers.directory import DirectoryHandler
-from synapse.rest.client.v1 import directory, room
-from synapse.types import RoomAlias
+from synapse.rest.client.v1 import directory, login, room
+from synapse.types import RoomAlias, create_requester
from tests import unittest
-from tests.utils import setup_test_homeserver
-class DirectoryHandlers(object):
- def __init__(self, hs):
- self.directory_handler = DirectoryHandler(hs)
-
-
-class DirectoryTestCase(unittest.TestCase):
+class DirectoryTestCase(unittest.HomeserverTestCase):
""" Tests the directory service. """
- @defer.inlineCallbacks
- def setUp(self):
+ def make_homeserver(self, reactor, clock):
self.mock_federation = Mock()
self.mock_registry = Mock()
@@ -47,14 +42,12 @@ class DirectoryTestCase(unittest.TestCase):
self.mock_registry.register_query_handler = register_query_handler
- hs = yield setup_test_homeserver(
- self.addCleanup,
+ hs = self.setup_test_homeserver(
http_client=None,
resource_for_federation=Mock(),
federation_client=self.mock_federation,
federation_registry=self.mock_registry,
)
- hs.handlers = DirectoryHandlers(hs)
self.handler = hs.get_handlers().directory_handler
@@ -64,23 +57,25 @@ class DirectoryTestCase(unittest.TestCase):
self.your_room = RoomAlias.from_string("#your-room:test")
self.remote_room = RoomAlias.from_string("#another:remote")
- @defer.inlineCallbacks
+ return hs
+
def test_get_local_association(self):
- yield self.store.create_room_alias_association(
- self.my_room, "!8765qwer:test", ["test"]
+ self.get_success(
+ self.store.create_room_alias_association(
+ self.my_room, "!8765qwer:test", ["test"]
+ )
)
- result = yield self.handler.get_association(self.my_room)
+ result = self.get_success(self.handler.get_association(self.my_room))
self.assertEquals({"room_id": "!8765qwer:test", "servers": ["test"]}, result)
- @defer.inlineCallbacks
def test_get_remote_association(self):
self.mock_federation.make_query.return_value = defer.succeed(
{"room_id": "!8765qwer:test", "servers": ["test", "remote"]}
)
- result = yield self.handler.get_association(self.remote_room)
+ result = self.get_success(self.handler.get_association(self.remote_room))
self.assertEquals(
{"room_id": "!8765qwer:test", "servers": ["test", "remote"]}, result
@@ -93,19 +88,303 @@ class DirectoryTestCase(unittest.TestCase):
ignore_backoff=True,
)
- @defer.inlineCallbacks
def test_incoming_fed_query(self):
- yield self.store.create_room_alias_association(
- self.your_room, "!8765asdf:test", ["test"]
+ self.get_success(
+ self.store.create_room_alias_association(
+ self.your_room, "!8765asdf:test", ["test"]
+ )
)
- response = yield self.query_handlers["directory"](
- {"room_alias": "#your-room:test"}
+ response = self.get_success(
+ self.handler.on_directory_query({"room_alias": "#your-room:test"})
)
self.assertEquals({"room_id": "!8765asdf:test", "servers": ["test"]}, response)
+class TestCreateAlias(unittest.HomeserverTestCase):
+ servlets = [
+ synapse.rest.admin.register_servlets,
+ login.register_servlets,
+ room.register_servlets,
+ directory.register_servlets,
+ ]
+
+ def prepare(self, reactor, clock, hs):
+ self.handler = hs.get_handlers().directory_handler
+
+ # Create user
+ self.admin_user = self.register_user("admin", "pass", admin=True)
+ self.admin_user_tok = self.login("admin", "pass")
+
+ # Create a test room
+ self.room_id = self.helper.create_room_as(
+ self.admin_user, tok=self.admin_user_tok
+ )
+
+ self.test_alias = "#test:test"
+ self.room_alias = RoomAlias.from_string(self.test_alias)
+
+ # Create a test user.
+ self.test_user = self.register_user("user", "pass", admin=False)
+ self.test_user_tok = self.login("user", "pass")
+ self.helper.join(room=self.room_id, user=self.test_user, tok=self.test_user_tok)
+
+ def test_create_alias_joined_room(self):
+ """A user can create an alias for a room they're in."""
+ self.get_success(
+ self.handler.create_association(
+ create_requester(self.test_user), self.room_alias, self.room_id,
+ )
+ )
+
+ def test_create_alias_other_room(self):
+ """A user cannot create an alias for a room they're NOT in."""
+ other_room_id = self.helper.create_room_as(
+ self.admin_user, tok=self.admin_user_tok
+ )
+
+ self.get_failure(
+ self.handler.create_association(
+ create_requester(self.test_user), self.room_alias, other_room_id,
+ ),
+ synapse.api.errors.SynapseError,
+ )
+
+ def test_create_alias_admin(self):
+ """An admin can create an alias for a room they're NOT in."""
+ other_room_id = self.helper.create_room_as(
+ self.test_user, tok=self.test_user_tok
+ )
+
+ self.get_success(
+ self.handler.create_association(
+ create_requester(self.admin_user), self.room_alias, other_room_id,
+ )
+ )
+
+
+class TestDeleteAlias(unittest.HomeserverTestCase):
+ servlets = [
+ synapse.rest.admin.register_servlets,
+ login.register_servlets,
+ room.register_servlets,
+ directory.register_servlets,
+ ]
+
+ def prepare(self, reactor, clock, hs):
+ self.store = hs.get_datastore()
+ self.handler = hs.get_handlers().directory_handler
+ self.state_handler = hs.get_state_handler()
+
+ # Create user
+ self.admin_user = self.register_user("admin", "pass", admin=True)
+ self.admin_user_tok = self.login("admin", "pass")
+
+ # Create a test room
+ self.room_id = self.helper.create_room_as(
+ self.admin_user, tok=self.admin_user_tok
+ )
+
+ self.test_alias = "#test:test"
+ self.room_alias = RoomAlias.from_string(self.test_alias)
+
+ # Create a test user.
+ self.test_user = self.register_user("user", "pass", admin=False)
+ self.test_user_tok = self.login("user", "pass")
+ self.helper.join(room=self.room_id, user=self.test_user, tok=self.test_user_tok)
+
+ def _create_alias(self, user):
+ # Create a new alias to this room.
+ self.get_success(
+ self.store.create_room_alias_association(
+ self.room_alias, self.room_id, ["test"], user
+ )
+ )
+
+ def test_delete_alias_not_allowed(self):
+ """A user that doesn't meet the expected guidelines cannot delete an alias."""
+ self._create_alias(self.admin_user)
+ self.get_failure(
+ self.handler.delete_association(
+ create_requester(self.test_user), self.room_alias
+ ),
+ synapse.api.errors.AuthError,
+ )
+
+ def test_delete_alias_creator(self):
+ """An alias creator can delete their own alias."""
+ # Create an alias from a different user.
+ self._create_alias(self.test_user)
+
+ # Delete the user's alias.
+ result = self.get_success(
+ self.handler.delete_association(
+ create_requester(self.test_user), self.room_alias
+ )
+ )
+ self.assertEquals(self.room_id, result)
+
+ # Confirm the alias is gone.
+ self.get_failure(
+ self.handler.get_association(self.room_alias),
+ synapse.api.errors.SynapseError,
+ )
+
+ def test_delete_alias_admin(self):
+ """A server admin can delete an alias created by another user."""
+ # Create an alias from a different user.
+ self._create_alias(self.test_user)
+
+ # Delete the user's alias as the admin.
+ result = self.get_success(
+ self.handler.delete_association(
+ create_requester(self.admin_user), self.room_alias
+ )
+ )
+ self.assertEquals(self.room_id, result)
+
+ # Confirm the alias is gone.
+ self.get_failure(
+ self.handler.get_association(self.room_alias),
+ synapse.api.errors.SynapseError,
+ )
+
+ def test_delete_alias_sufficient_power(self):
+ """A user with a sufficient power level should be able to delete an alias."""
+ self._create_alias(self.admin_user)
+
+ # Increase the user's power level.
+ self.helper.send_state(
+ self.room_id,
+ "m.room.power_levels",
+ {"users": {self.test_user: 100}},
+ tok=self.admin_user_tok,
+ )
+
+ # They can now delete the alias.
+ result = self.get_success(
+ self.handler.delete_association(
+ create_requester(self.test_user), self.room_alias
+ )
+ )
+ self.assertEquals(self.room_id, result)
+
+ # Confirm the alias is gone.
+ self.get_failure(
+ self.handler.get_association(self.room_alias),
+ synapse.api.errors.SynapseError,
+ )
+
+
+class CanonicalAliasTestCase(unittest.HomeserverTestCase):
+ """Test modifications of the canonical alias when delete aliases.
+ """
+
+ servlets = [
+ synapse.rest.admin.register_servlets,
+ login.register_servlets,
+ room.register_servlets,
+ directory.register_servlets,
+ ]
+
+ def prepare(self, reactor, clock, hs):
+ self.store = hs.get_datastore()
+ self.handler = hs.get_handlers().directory_handler
+ self.state_handler = hs.get_state_handler()
+
+ # Create user
+ self.admin_user = self.register_user("admin", "pass", admin=True)
+ self.admin_user_tok = self.login("admin", "pass")
+
+ # Create a test room
+ self.room_id = self.helper.create_room_as(
+ self.admin_user, tok=self.admin_user_tok
+ )
+
+ self.test_alias = "#test:test"
+ self.room_alias = self._add_alias(self.test_alias)
+
+ def _add_alias(self, alias: str) -> RoomAlias:
+ """Add an alias to the test room."""
+ room_alias = RoomAlias.from_string(alias)
+
+ # Create a new alias to this room.
+ self.get_success(
+ self.store.create_room_alias_association(
+ room_alias, self.room_id, ["test"], self.admin_user
+ )
+ )
+ return room_alias
+
+ def _set_canonical_alias(self, content):
+ """Configure the canonical alias state on the room."""
+ self.helper.send_state(
+ self.room_id, "m.room.canonical_alias", content, tok=self.admin_user_tok,
+ )
+
+ def _get_canonical_alias(self):
+ """Get the canonical alias state of the room."""
+ return self.get_success(
+ self.state_handler.get_current_state(
+ self.room_id, EventTypes.CanonicalAlias, ""
+ )
+ )
+
+ def test_remove_alias(self):
+ """Removing an alias that is the canonical alias should remove it there too."""
+ # Set this new alias as the canonical alias for this room
+ self._set_canonical_alias(
+ {"alias": self.test_alias, "alt_aliases": [self.test_alias]}
+ )
+
+ data = self._get_canonical_alias()
+ self.assertEqual(data["content"]["alias"], self.test_alias)
+ self.assertEqual(data["content"]["alt_aliases"], [self.test_alias])
+
+ # Finally, delete the alias.
+ self.get_success(
+ self.handler.delete_association(
+ create_requester(self.admin_user), self.room_alias
+ )
+ )
+
+ data = self._get_canonical_alias()
+ self.assertNotIn("alias", data["content"])
+ self.assertNotIn("alt_aliases", data["content"])
+
+ def test_remove_other_alias(self):
+ """Removing an alias listed as in alt_aliases should remove it there too."""
+ # Create a second alias.
+ other_test_alias = "#test2:test"
+ other_room_alias = self._add_alias(other_test_alias)
+
+ # Set the alias as the canonical alias for this room.
+ self._set_canonical_alias(
+ {
+ "alias": self.test_alias,
+ "alt_aliases": [self.test_alias, other_test_alias],
+ }
+ )
+
+ data = self._get_canonical_alias()
+ self.assertEqual(data["content"]["alias"], self.test_alias)
+ self.assertEqual(
+ data["content"]["alt_aliases"], [self.test_alias, other_test_alias]
+ )
+
+ # Delete the second alias.
+ self.get_success(
+ self.handler.delete_association(
+ create_requester(self.admin_user), other_room_alias
+ )
+ )
+
+ data = self._get_canonical_alias()
+ self.assertEqual(data["content"]["alias"], self.test_alias)
+ self.assertEqual(data["content"]["alt_aliases"], [self.test_alias])
+
+
class TestCreateAliasACL(unittest.HomeserverTestCase):
user_id = "@test:test"
|
