Matrix-Mirrors/anonymousland-synapse
1
0
Fork 0
mirror of https://git.anonymousland.org/anonymousland/synapse.git synced 2025-09-25 02:38:30 -04:00
Code Issues Projects Releases Packages Wiki Activity

Move client v1 api rest servlets into a "client/v1" directory

Browse source
This commit is contained in:
Mark Haines 2015-01-22 14:59:08 +00:00
parent 16bfabb9c5
commit 1d2016b4a8
23 changed files with 52 additions and 24 deletions
Download patch file Download diff file Expand all files Collapse all files

15
tests/client/v1/__init__.py Normal file
View file

@ -0,0 +1,15 @@
# -*- coding: utf-8 -*-
# Copyright 2014 OpenMarket Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

225
tests/client/v1/test_events.py Normal file
View file

@ -0,0 +1,225 @@
# -*- coding: utf-8 -*-
# Copyright 2014 OpenMarket Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
""" Tests REST events for /events paths."""
from tests import unittest
# twisted imports
from twisted.internet import defer
import synapse.client.v1.events
import synapse.client.v1.register
import synapse.client.v1.room
from synapse.server import HomeServer
from ...utils import MockHttpResource, SQLiteMemoryDbPool, MockKey
from .utils import RestTestCase
from mock import Mock, NonCallableMock
PATH_PREFIX = "/_matrix/client/api/v1"
class EventStreamPaginationApiTestCase(unittest.TestCase):
""" Tests event streaming query parameters and start/end keys used in the
Pagination stream API. """
user_id = "sid1"
def setUp(self):
# configure stream and inject items
pass
def tearDown(self):
pass
def TODO_test_long_poll(self):
# stream from 'end' key, send (self+other) message, expect message.
# stream from 'END', send (self+other) message, expect message.
# stream from 'end' key, send (self+other) topic, expect topic.
# stream from 'END', send (self+other) topic, expect topic.
# stream from 'end' key, send (self+other) invite, expect invite.
# stream from 'END', send (self+other) invite, expect invite.
pass
def TODO_test_stream_forward(self):
# stream from START, expect injected items
# stream from 'start' key, expect same content
# stream from 'end' key, expect nothing
# stream from 'END', expect nothing
# The following is needed for cases where content is removed e.g. you
# left a room, so the token you're streaming from is > the one that
# would be returned naturally from START>END.
# stream from very new token (higher than end key), expect same token
# returned as end key
pass
def TODO_test_limits(self):
# stream from a key, expect limit_num items
# stream from START, expect limit_num items
pass
def TODO_test_range(self):
# stream from key to key, expect X items
# stream from key to END, expect X items
# stream from START to key, expect X items
# stream from START to END, expect all items
pass
def TODO_test_direction(self):
# stream from END to START and fwds, expect newest first
# stream from END to START and bwds, expect oldest first
# stream from START to END and fwds, expect oldest first
# stream from START to END and bwds, expect newest first
pass
class EventStreamPermissionsTestCase(RestTestCase):
""" Tests event streaming (GET /events). """
@defer.inlineCallbacks
def setUp(self):
self.mock_resource = MockHttpResource(prefix=PATH_PREFIX)
self.mock_config = NonCallableMock()
self.mock_config.signing_key = [MockKey()]
db_pool = SQLiteMemoryDbPool()
yield db_pool.prepare()
hs = HomeServer(
"test",
db_pool=db_pool,
http_client=None,
replication_layer=Mock(),
clock=Mock(spec=[
"call_later",
"cancel_call_later",
"time_msec",
"time"
]),
ratelimiter=NonCallableMock(spec_set=[
"send_message",
]),
config=self.mock_config,
)
self.ratelimiter = hs.get_ratelimiter()
self.ratelimiter.send_message.return_value = (True, 0)
hs.config.enable_registration_captcha = False
hs.get_handlers().federation_handler = Mock()
hs.get_clock().time_msec.return_value = 1000000
hs.get_clock().time.return_value = 1000
synapse.client.v1.register.register_servlets(hs, self.mock_resource)
synapse.client.v1.events.register_servlets(hs, self.mock_resource)
synapse.client.v1.room.register_servlets(hs, self.mock_resource)
# register an account
self.user_id = "sid1"
response = yield self.register(self.user_id)
self.token = response["access_token"]
self.user_id = response["user_id"]
# register a 2nd account
self.other_user = "other1"
response = yield self.register(self.other_user)
self.other_token = response["access_token"]
self.other_user = response["user_id"]
def tearDown(self):
pass
@defer.inlineCallbacks
def test_stream_basic_permissions(self):
# invalid token, expect 403
(code, response) = yield self.mock_resource.trigger_get(
"/events?access_token=%s" % ("invalid" + self.token, )
)
self.assertEquals(403, code, msg=str(response))
# valid token, expect content
(code, response) = yield self.mock_resource.trigger_get(
"/events?access_token=%s&timeout=0" % (self.token,)
)
self.assertEquals(200, code, msg=str(response))
self.assertTrue("chunk" in response)
self.assertTrue("start" in response)
self.assertTrue("end" in response)
@defer.inlineCallbacks
def test_stream_room_permissions(self):
room_id = yield self.create_room_as(
self.other_user,
tok=self.other_token
)
yield self.send(room_id, tok=self.other_token)
# invited to room (expect no content for room)
yield self.invite(
room_id,
src=self.other_user,
targ=self.user_id,
tok=self.other_token
)
(code, response) = yield self.mock_resource.trigger_get(
"/events?access_token=%s&timeout=0" % (self.token,)
)
self.assertEquals(200, code, msg=str(response))
self.assertEquals(0, len(response["chunk"]))
# joined room (expect all content for room)
yield self.join(room=room_id, user=self.user_id, tok=self.token)
# left to room (expect no content for room)
def TODO_test_stream_items(self):
# new user, no content
# join room, expect 1 item (join)
# send message, expect 2 items (join,send)
# set topic, expect 3 items (join,send,topic)
# someone else join room, expect 4 (join,send,topic,join)
# someone else send message, expect 5 (join,send.topic,join,send)
# someone else set topic, expect 6 (join,send,topic,join,send,topic)
pass

372
tests/client/v1/test_presence.py Normal file
View file

@ -0,0 +1,372 @@
# -*- coding: utf-8 -*-
# Copyright 2014 OpenMarket Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Tests REST events for /presence paths."""
from tests import unittest
from twisted.internet import defer
from mock import Mock
from ...utils import MockHttpResource, MockKey
from synapse.api.constants import PresenceState
from synapse.handlers.presence import PresenceHandler
from synapse.server import HomeServer
OFFLINE = PresenceState.OFFLINE
UNAVAILABLE = PresenceState.UNAVAILABLE
ONLINE = PresenceState.ONLINE
myid = "@apple:test"
PATH_PREFIX = "/_matrix/client/api/v1"
class JustPresenceHandlers(object):
def __init__(self, hs):
self.presence_handler = PresenceHandler(hs)
class PresenceStateTestCase(unittest.TestCase):
def setUp(self):
self.mock_resource = MockHttpResource(prefix=PATH_PREFIX)
self.mock_config = Mock()
self.mock_config.signing_key = [MockKey()]
hs = HomeServer("test",
db_pool=None,
datastore=Mock(spec=[
"get_presence_state",
"set_presence_state",
"insert_client_ip",
]),
http_client=None,
resource_for_client=self.mock_resource,
resource_for_federation=self.mock_resource,
config=self.mock_config,
)
hs.handlers = JustPresenceHandlers(hs)
self.datastore = hs.get_datastore()
def get_presence_list(*a, **kw):
return defer.succeed([])
self.datastore.get_presence_list = get_presence_list
def _get_user_by_token(token=None):
return {
"user": hs.parse_userid(myid),
"admin": False,
"device_id": None,
}
hs.get_auth().get_user_by_token = _get_user_by_token
room_member_handler = hs.handlers.room_member_handler = Mock(
spec=[
"get_rooms_for_user",
]
)
def get_rooms_for_user(user):
return defer.succeed([])
room_member_handler.get_rooms_for_user = get_rooms_for_user
hs.register_servlets()
self.u_apple = hs.parse_userid(myid)
@defer.inlineCallbacks
def test_get_my_status(self):
mocked_get = self.datastore.get_presence_state
mocked_get.return_value = defer.succeed(
{"state": ONLINE, "status_msg": "Available"}
)
(code, response) = yield self.mock_resource.trigger("GET",
"/presence/%s/status" % (myid), None)
self.assertEquals(200, code)
self.assertEquals(
{"presence": ONLINE, "status_msg": "Available"},
response
)
mocked_get.assert_called_with("apple")
@defer.inlineCallbacks
def test_set_my_status(self):
mocked_set = self.datastore.set_presence_state
mocked_set.return_value = defer.succeed({"state": OFFLINE})
(code, response) = yield self.mock_resource.trigger("PUT",
"/presence/%s/status" % (myid),
'{"presence": "unavailable", "status_msg": "Away"}')
self.assertEquals(200, code)
mocked_set.assert_called_with("apple",
{"state": UNAVAILABLE, "status_msg": "Away"}
)
class PresenceListTestCase(unittest.TestCase):
def setUp(self):
self.mock_resource = MockHttpResource(prefix=PATH_PREFIX)
self.mock_config = Mock()
self.mock_config.signing_key = [MockKey()]
hs = HomeServer("test",
db_pool=None,
datastore=Mock(spec=[
"has_presence_state",
"get_presence_state",
"allow_presence_visible",
"is_presence_visible",
"add_presence_list_pending",
"set_presence_list_accepted",
"del_presence_list",
"get_presence_list",
"insert_client_ip",
]),
http_client=None,
resource_for_client=self.mock_resource,
resource_for_federation=self.mock_resource,
config=self.mock_config,
)
hs.handlers = JustPresenceHandlers(hs)
self.datastore = hs.get_datastore()
def has_presence_state(user_localpart):
return defer.succeed(
user_localpart in ("apple", "banana",)
)
self.datastore.has_presence_state = has_presence_state
def _get_user_by_token(token=None):
return {
"user": hs.parse_userid(myid),
"admin": False,
"device_id": None,
}
room_member_handler = hs.handlers.room_member_handler = Mock(
spec=[
"get_rooms_for_user",
]
)
hs.get_auth().get_user_by_token = _get_user_by_token
hs.register_servlets()
self.u_apple = hs.parse_userid("@apple:test")
self.u_banana = hs.parse_userid("@banana:test")
@defer.inlineCallbacks
def test_get_my_list(self):
self.datastore.get_presence_list.return_value = defer.succeed(
[{"observed_user_id": "@banana:test"}],
)
(code, response) = yield self.mock_resource.trigger("GET",
"/presence/list/%s" % (myid), None)
self.assertEquals(200, code)
self.assertEquals([
{"user_id": "@banana:test", "presence": OFFLINE},
], response)
self.datastore.get_presence_list.assert_called_with(
"apple", accepted=True
)
@defer.inlineCallbacks
def test_invite(self):
self.datastore.add_presence_list_pending.return_value = (
defer.succeed(())
)
self.datastore.is_presence_visible.return_value = defer.succeed(
True
)
(code, response) = yield self.mock_resource.trigger("POST",
"/presence/list/%s" % (myid),
"""{"invite": ["@banana:test"]}"""
)
self.assertEquals(200, code)
self.datastore.add_presence_list_pending.assert_called_with(
"apple", "@banana:test"
)
self.datastore.set_presence_list_accepted.assert_called_with(
"apple", "@banana:test"
)
@defer.inlineCallbacks
def test_drop(self):
self.datastore.del_presence_list.return_value = (
defer.succeed(())
)
(code, response) = yield self.mock_resource.trigger("POST",
"/presence/list/%s" % (myid),
"""{"drop": ["@banana:test"]}"""
)
self.assertEquals(200, code)
self.datastore.del_presence_list.assert_called_with(
"apple", "@banana:test"
)
class PresenceEventStreamTestCase(unittest.TestCase):
def setUp(self):
self.mock_resource = MockHttpResource(prefix=PATH_PREFIX)
self.mock_config = Mock()
self.mock_config.signing_key = [MockKey()]
# HIDEOUS HACKERY
# TODO(paul): This should be injected in via the HomeServer DI system
from synapse.streams.events import (
PresenceEventSource, NullSource, EventSources
)
old_SOURCE_TYPES = EventSources.SOURCE_TYPES
def tearDown():
EventSources.SOURCE_TYPES = old_SOURCE_TYPES
self.tearDown = tearDown
EventSources.SOURCE_TYPES = {
k: NullSource for k in old_SOURCE_TYPES.keys()
}
EventSources.SOURCE_TYPES["presence"] = PresenceEventSource
hs = HomeServer("test",
db_pool=None,
http_client=None,
resource_for_client=self.mock_resource,
resource_for_federation=self.mock_resource,
datastore=Mock(spec=[
"set_presence_state",
"get_presence_list",
]),
clock=Mock(spec=[
"call_later",
"cancel_call_later",
"time_msec",
]),
config=self.mock_config,
)
hs.get_clock().time_msec.return_value = 1000000
def _get_user_by_req(req=None):
return hs.parse_userid(myid)
hs.get_auth().get_user_by_req = _get_user_by_req
hs.register_servlets()
hs.handlers.room_member_handler = Mock(spec=[])
self.room_members = []
def get_rooms_for_user(user):
if user in self.room_members:
return ["a-room"]
else:
return []
hs.handlers.room_member_handler.get_rooms_for_user = get_rooms_for_user
self.mock_datastore = hs.get_datastore()
def get_profile_displayname(user_id):
return defer.succeed("Frank")
self.mock_datastore.get_profile_displayname = get_profile_displayname
def get_profile_avatar_url(user_id):
return defer.succeed(None)
self.mock_datastore.get_profile_avatar_url = get_profile_avatar_url
def user_rooms_intersect(user_list):
room_member_ids = map(lambda u: u.to_string(), self.room_members)
shared = all(map(lambda i: i in room_member_ids, user_list))
return defer.succeed(shared)
self.mock_datastore.user_rooms_intersect = user_rooms_intersect
def get_joined_hosts_for_room(room_id):
return []
self.mock_datastore.get_joined_hosts_for_room = get_joined_hosts_for_room
self.presence = hs.get_handlers().presence_handler
self.u_apple = hs.parse_userid("@apple:test")
self.u_banana = hs.parse_userid("@banana:test")
@defer.inlineCallbacks
def test_shortpoll(self):
self.room_members = [self.u_apple, self.u_banana]
self.mock_datastore.set_presence_state.return_value = defer.succeed(
{"state": ONLINE}
)
self.mock_datastore.get_presence_list.return_value = defer.succeed(
[]
)
(code, response) = yield self.mock_resource.trigger("GET",
"/events?timeout=0", None)
self.assertEquals(200, code)
# We've forced there to be only one data stream so the tokens will
# all be ours
# I'll already get my own presence state change
self.assertEquals({"start": "0_1_0", "end": "0_1_0", "chunk": []},
response
)
self.mock_datastore.set_presence_state.return_value = defer.succeed(
{"state": ONLINE}
)
self.mock_datastore.get_presence_list.return_value = defer.succeed(
[]
)
yield self.presence.set_state(self.u_banana, self.u_banana,
state={"presence": ONLINE}
)
(code, response) = yield self.mock_resource.trigger("GET",
"/events?from=0_1_0&timeout=0", None)
self.assertEquals(200, code)
self.assertEquals({"start": "0_1_0", "end": "0_2_0", "chunk": [
{"type": "m.presence",
"content": {
"user_id": "@banana:test",
"presence": ONLINE,
"displayname": "Frank",
"last_active_ago": 0,
}},
]}, response)

150
tests/client/v1/test_profile.py Normal file
View file

@ -0,0 +1,150 @@
# -*- coding: utf-8 -*-
# Copyright 2014 OpenMarket Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Tests REST events for /profile paths."""
from tests import unittest
from twisted.internet import defer
from mock import Mock, NonCallableMock
from ...utils import MockHttpResource, MockKey
from synapse.api.errors import SynapseError, AuthError
from synapse.server import HomeServer
myid = "@1234ABCD:test"
PATH_PREFIX = "/_matrix/client/api/v1"
class ProfileTestCase(unittest.TestCase):
""" Tests profile management. """
def setUp(self):
self.mock_resource = MockHttpResource(prefix=PATH_PREFIX)
self.mock_handler = Mock(spec=[
"get_displayname",
"set_displayname",
"get_avatar_url",
"set_avatar_url",
])
self.mock_config = NonCallableMock()
self.mock_config.signing_key = [MockKey()]
hs = HomeServer("test",
db_pool=None,
http_client=None,
resource_for_client=self.mock_resource,
federation=Mock(),
replication_layer=Mock(),
datastore=None,
config=self.mock_config,
)
def _get_user_by_req(request=None):
return hs.parse_userid(myid)
hs.get_auth().get_user_by_req = _get_user_by_req
hs.get_handlers().profile_handler = self.mock_handler
hs.register_servlets()
@defer.inlineCallbacks
def test_get_my_name(self):
mocked_get = self.mock_handler.get_displayname
mocked_get.return_value = defer.succeed("Frank")
(code, response) = yield self.mock_resource.trigger("GET",
"/profile/%s/displayname" % (myid), None)
self.assertEquals(200, code)
self.assertEquals({"displayname": "Frank"}, response)
self.assertEquals(mocked_get.call_args[0][0].localpart, "1234ABCD")
@defer.inlineCallbacks
def test_set_my_name(self):
mocked_set = self.mock_handler.set_displayname
mocked_set.return_value = defer.succeed(())
(code, response) = yield self.mock_resource.trigger("PUT",
"/profile/%s/displayname" % (myid),
'{"displayname": "Frank Jr."}')
self.assertEquals(200, code)
self.assertEquals(mocked_set.call_args[0][0].localpart, "1234ABCD")
self.assertEquals(mocked_set.call_args[0][1].localpart, "1234ABCD")
self.assertEquals(mocked_set.call_args[0][2], "Frank Jr.")
@defer.inlineCallbacks
def test_set_my_name_noauth(self):
mocked_set = self.mock_handler.set_displayname
mocked_set.side_effect = AuthError(400, "message")
(code, response) = yield self.mock_resource.trigger("PUT",
"/profile/%s/displayname" % ("@4567:test"), '"Frank Jr."')
self.assertTrue(400 <= code < 499,
msg="code %d is in the 4xx range" % (code))
@defer.inlineCallbacks
def test_get_other_name(self):
mocked_get = self.mock_handler.get_displayname
mocked_get.return_value = defer.succeed("Bob")
(code, response) = yield self.mock_resource.trigger("GET",
"/profile/%s/displayname" % ("@opaque:elsewhere"), None)
self.assertEquals(200, code)
self.assertEquals({"displayname": "Bob"}, response)
@defer.inlineCallbacks
def test_set_other_name(self):
mocked_set = self.mock_handler.set_displayname
mocked_set.side_effect = SynapseError(400, "message")
(code, response) = yield self.mock_resource.trigger("PUT",
"/profile/%s/displayname" % ("@opaque:elsewhere"), None)
self.assertTrue(400 <= code <= 499,
msg="code %d is in the 4xx range" % (code))
@defer.inlineCallbacks
def test_get_my_avatar(self):
mocked_get = self.mock_handler.get_avatar_url
mocked_get.return_value = defer.succeed("http://my.server/me.png")
(code, response) = yield self.mock_resource.trigger("GET",
"/profile/%s/avatar_url" % (myid), None)
self.assertEquals(200, code)
self.assertEquals({"avatar_url": "http://my.server/me.png"}, response)
self.assertEquals(mocked_get.call_args[0][0].localpart, "1234ABCD")
@defer.inlineCallbacks
def test_set_my_avatar(self):
mocked_set = self.mock_handler.set_avatar_url
mocked_set.return_value = defer.succeed(())
(code, response) = yield self.mock_resource.trigger("PUT",
"/profile/%s/avatar_url" % (myid),
'{"avatar_url": "http://my.server/pic.gif"}')
self.assertEquals(200, code)
self.assertEquals(mocked_set.call_args[0][0].localpart, "1234ABCD")
self.assertEquals(mocked_set.call_args[0][1].localpart, "1234ABCD")
self.assertEquals(mocked_set.call_args[0][2],
"http://my.server/pic.gif")

1068
tests/client/v1/test_rooms.py Normal file
View file

File diff suppressed because it is too large Load diff

164
tests/client/v1/test_typing.py Normal file
View file

@ -0,0 +1,164 @@
# -*- coding: utf-8 -*-
# Copyright 2014 OpenMarket Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Tests REST events for /rooms paths."""
# twisted imports
from twisted.internet import defer
import synapse.client.v1.room
from synapse.server import HomeServer
from ...utils import MockHttpResource, MockClock, SQLiteMemoryDbPool, MockKey
from .utils import RestTestCase
from mock import Mock, NonCallableMock
PATH_PREFIX = "/_matrix/client/api/v1"
class RoomTypingTestCase(RestTestCase):
""" Tests /rooms/$room_id/typing/$user_id REST API. """
user_id = "@sid:red"
@defer.inlineCallbacks
def setUp(self):
self.clock = MockClock()
self.mock_resource = MockHttpResource(prefix=PATH_PREFIX)
self.auth_user_id = self.user_id
self.mock_config = NonCallableMock()
self.mock_config.signing_key = [MockKey()]
db_pool = SQLiteMemoryDbPool()
yield db_pool.prepare()
hs = HomeServer(
"red",
clock=self.clock,
db_pool=db_pool,
http_client=None,
replication_layer=Mock(),
ratelimiter=NonCallableMock(spec_set=[
"send_message",
]),
config=self.mock_config,
)
self.hs = hs
self.event_source = hs.get_event_sources().sources["typing"]
self.ratelimiter = hs.get_ratelimiter()
self.ratelimiter.send_message.return_value = (True, 0)
hs.get_handlers().federation_handler = Mock()
def _get_user_by_token(token=None):
return {
"user": hs.parse_userid(self.auth_user_id),
"admin": False,
"device_id": None,
}
hs.get_auth().get_user_by_token = _get_user_by_token
def _insert_client_ip(*args, **kwargs):
return defer.succeed(None)
hs.get_datastore().insert_client_ip = _insert_client_ip
def get_room_members(room_id):
if room_id == self.room_id:
return defer.succeed([hs.parse_userid(self.user_id)])
else:
return defer.succeed([])
@defer.inlineCallbacks
def fetch_room_distributions_into(room_id, localusers=None,
remotedomains=None, ignore_user=None):
members = yield get_room_members(room_id)
for member in members:
if ignore_user is not None and member == ignore_user:
continue
if hs.is_mine(member):
if localusers is not None:
localusers.add(member)
else:
if remotedomains is not None:
remotedomains.add(member.domain)
hs.get_handlers().room_member_handler.fetch_room_distributions_into = (
fetch_room_distributions_into)
synapse.client.v1.room.register_servlets(hs, self.mock_resource)
self.room_id = yield self.create_room_as(self.user_id)
# Need another user to make notifications actually work
yield self.join(self.room_id, user="@jim:red")
def tearDown(self):
self.hs.get_handlers().typing_notification_handler.tearDown()
@defer.inlineCallbacks
def test_set_typing(self):
(code, _) = yield self.mock_resource.trigger("PUT",
"/rooms/%s/typing/%s" % (self.room_id, self.user_id),
'{"typing": true, "timeout": 30000}'
)
self.assertEquals(200, code)
self.assertEquals(self.event_source.get_current_key(), 1)
self.assertEquals(
self.event_source.get_new_events_for_user(self.user_id, 0, None)[0],
[
{"type": "m.typing",
"room_id": self.room_id,
"content": {
"user_ids": [self.user_id],
}},
]
)
@defer.inlineCallbacks
def test_set_not_typing(self):
(code, _) = yield self.mock_resource.trigger("PUT",
"/rooms/%s/typing/%s" % (self.room_id, self.user_id),
'{"typing": false}'
)
self.assertEquals(200, code)
@defer.inlineCallbacks
def test_typing_timeout(self):
(code, _) = yield self.mock_resource.trigger("PUT",
"/rooms/%s/typing/%s" % (self.room_id, self.user_id),
'{"typing": true, "timeout": 30000}'
)
self.assertEquals(200, code)
self.assertEquals(self.event_source.get_current_key(), 1)
self.clock.advance_time(31);
self.assertEquals(self.event_source.get_current_key(), 2)
(code, _) = yield self.mock_resource.trigger("PUT",
"/rooms/%s/typing/%s" % (self.room_id, self.user_id),
'{"typing": true, "timeout": 30000}'
)
self.assertEquals(200, code)
self.assertEquals(self.event_source.get_current_key(), 3)

134
tests/client/v1/utils.py Normal file
View file

@ -0,0 +1,134 @@
# -*- coding: utf-8 -*-
# Copyright 2014 OpenMarket Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# twisted imports
from twisted.internet import defer
# trial imports
from tests import unittest
from synapse.api.constants import Membership
import json
import time
class RestTestCase(unittest.TestCase):
"""Contains extra helper functions to quickly and clearly perform a given
REST action, which isn't the focus of the test.
This subclass assumes there are mock_resource and auth_user_id attributes.
"""
def __init__(self, *args, **kwargs):
super(RestTestCase, self).__init__(*args, **kwargs)
self.mock_resource = None
self.auth_user_id = None
def mock_get_user_by_token(self, token=None):
return self.auth_user_id
@defer.inlineCallbacks
def create_room_as(self, room_creator, is_public=True, tok=None):
temp_id = self.auth_user_id
self.auth_user_id = room_creator
path = "/createRoom"
content = "{}"
if not is_public:
content = '{"visibility":"private"}'
if tok:
path = path + "?access_token=%s" % tok
(code, response) = yield self.mock_resource.trigger("POST", path, content)
self.assertEquals(200, code, msg=str(response))
self.auth_user_id = temp_id
defer.returnValue(response["room_id"])
@defer.inlineCallbacks
def invite(self, room=None, src=None, targ=None, expect_code=200, tok=None):
yield self.change_membership(room=room, src=src, targ=targ, tok=tok,
membership=Membership.INVITE,
expect_code=expect_code)
@defer.inlineCallbacks
def join(self, room=None, user=None, expect_code=200, tok=None):
yield self.change_membership(room=room, src=user, targ=user, tok=tok,
membership=Membership.JOIN,
expect_code=expect_code)
@defer.inlineCallbacks
def leave(self, room=None, user=None, expect_code=200, tok=None):
yield self.change_membership(room=room, src=user, targ=user, tok=tok,
membership=Membership.LEAVE,
expect_code=expect_code)
@defer.inlineCallbacks
def change_membership(self, room, src, targ, membership, tok=None,
expect_code=200):
temp_id = self.auth_user_id
self.auth_user_id = src
path = "/rooms/%s/state/m.room.member/%s" % (room, targ)
if tok:
path = path + "?access_token=%s" % tok
data = {
"membership": membership
}
(code, response) = yield self.mock_resource.trigger("PUT", path,
json.dumps(data))
self.assertEquals(expect_code, code, msg=str(response))
self.auth_user_id = temp_id
@defer.inlineCallbacks
def register(self, user_id):
(code, response) = yield self.mock_resource.trigger(
"POST",
"/register",
json.dumps({
"user": user_id,
"password": "test",
"type": "m.login.password"
}))
self.assertEquals(200, code)
defer.returnValue(response)
@defer.inlineCallbacks
def send(self, room_id, body=None, txn_id=None, tok=None,
expect_code=200):
if txn_id is None:
txn_id = "m%s" % (str(time.time()))
if body is None:
body = "body_text_here"
path = "/rooms/%s/send/m.room.message/%s" % (room_id, txn_id)
content = '{"msgtype":"m.text","body":"%s"}' % body
if tok:
path = path + "?access_token=%s" % tok
(code, response) = yield self.mock_resource.trigger("PUT", path, content)
self.assertEquals(expect_code, code, msg=str(response))
def assert_dict(self, required, actual):
"""Does a partial assert of a dict.
Args:
required (dict): The keys and value which MUST be in 'actual'.
actual (dict): The test result. Extra keys will not be checked.
"""
for key in required:
self.assertEquals(required[key], actual[key],
msg="%s mismatch. %s" % (key, actual))