# -*- coding: utf-8 -*- # Copyright 2014 matrix.org # # 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 twisted.trial import unittest from twisted.internet import defer from mock import Mock import logging from ..utils import MockHttpServer from synapse.api.constants import PresenceState from synapse.server import HomeServer logging.getLogger().addHandler(logging.NullHandler()) OFFLINE = PresenceState.OFFLINE UNAVAILABLE = PresenceState.UNAVAILABLE ONLINE = PresenceState.ONLINE myid = "@apple:test" PATH_PREFIX = "/matrix/client/api/v1" class PresenceStateTestCase(unittest.TestCase): def setUp(self): self.mock_server = MockHttpServer(prefix=PATH_PREFIX) self.mock_handler = Mock(spec=[ "get_state", "set_state", ]) hs = HomeServer("test", db_pool=None, http_client=None, resource_for_client=self.mock_server, resource_for_federation=self.mock_server, ) def _get_user_by_token(token=None): return hs.parse_userid(myid) hs.get_auth().get_user_by_token = _get_user_by_token hs.get_handlers().presence_handler = self.mock_handler hs.register_servlets() self.u_apple = hs.parse_userid(myid) @defer.inlineCallbacks def test_get_my_status(self): mocked_get = self.mock_handler.get_state mocked_get.return_value = defer.succeed( {"state": ONLINE, "status_msg": "Available"}) (code, response) = yield self.mock_server.trigger("GET", "/presence/%s/status" % (myid), None) self.assertEquals(200, code) self.assertEquals({"state": ONLINE, "status_msg": "Available"}, response) mocked_get.assert_called_with(target_user=self.u_apple, auth_user=self.u_apple) @defer.inlineCallbacks def test_set_my_status(self): mocked_set = self.mock_handler.set_state mocked_set.return_value = defer.succeed(()) (code, response) = yield self.mock_server.trigger("PUT", "/presence/%s/status" % (myid), '{"state": "unavailable", "status_msg": "Away"}') self.assertEquals(200, code) mocked_set.assert_called_with(target_user=self.u_apple, auth_user=self.u_apple, state={"state": UNAVAILABLE, "status_msg": "Away"}) class PresenceListTestCase(unittest.TestCase): def setUp(self): self.mock_server = MockHttpServer(prefix=PATH_PREFIX) self.mock_handler = Mock(spec=[ "get_presence_list", "send_invite", "drop", ]) hs = HomeServer("test", db_pool=None, http_client=None, resource_for_client=self.mock_server, resource_for_federation=self.mock_server ) def _get_user_by_token(token=None): return hs.parse_userid(myid) hs.get_auth().get_user_by_token = _get_user_by_token hs.get_handlers().presence_handler = self.mock_handler 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.mock_handler.get_presence_list.return_value = defer.succeed( [{"observed_user": self.u_banana}] ) (code, response) = yield self.mock_server.trigger("GET", "/presence_list/%s" % (myid), None) self.assertEquals(200, code) self.assertEquals([{"user_id": "@banana:test"}], response) @defer.inlineCallbacks def test_invite(self): self.mock_handler.send_invite.return_value = defer.succeed(()) (code, response) = yield self.mock_server.trigger("POST", "/presence_list/%s" % (myid), """{ "invite": ["@banana:test"] }""") self.assertEquals(200, code) self.mock_handler.send_invite.assert_called_with( observer_user=self.u_apple, observed_user=self.u_banana) @defer.inlineCallbacks def test_drop(self): self.mock_handler.drop.return_value = defer.succeed(()) (code, response) = yield self.mock_server.trigger("POST", "/presence_list/%s" % (myid), """{ "drop": ["@banana:test"] }""") self.assertEquals(200, code) self.mock_handler.drop.assert_called_with( observer_user=self.u_apple, observed_user=self.u_banana) class PresenceEventStreamTestCase(unittest.TestCase): def setUp(self): self.mock_server = MockHttpServer(prefix=PATH_PREFIX) # TODO: mocked data store # HIDEOUS HACKERY # TODO(paul): This should be injected in via the HomeServer DI system from synapse.handlers.events import EventStreamHandler from synapse.handlers.presence import PresenceStreamData EventStreamHandler.stream_data_classes = [ PresenceStreamData ] hs = HomeServer("test", db_pool=None, http_client=None, resource_for_client=self.mock_server, resource_for_federation=self.mock_server, datastore=Mock(spec=[ "set_presence_state", "get_presence_list", ]), clock=Mock(spec=[ "call_later", "cancel_call_later", "time_msec", ]), ) 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=[ "get_rooms_for_user", ]) hs.handlers.room_member_handler.get_rooms_for_user = ( lambda u: defer.succeed([])) self.mock_datastore = hs.get_datastore() 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.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_server.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", "end": "1", "chunk": [ {"type": "m.presence", "content": { "user_id": "@apple:test", "state": ONLINE, "mtime_age": 0, }}, ]}, 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={"state": ONLINE}) (code, response) = yield self.mock_server.trigger("GET", "/events?from=1&timeout=0", None) self.assertEquals(200, code) self.assertEquals({"start": "1", "end": "2", "chunk": [ {"type": "m.presence", "content": { "user_id": "@banana:test", "state": ONLINE, "mtime_age": 0, }}, ]}, response)