1# Copyright 2014-2016 OpenMarket Ltd 2# Copyright 2018 New Vector 3# 4# Licensed under the Apache License, Version 2.0 (the "License"); 5# you may not use this file except in compliance with the License. 6# You may obtain a copy of the License at 7# 8# http://www.apache.org/licenses/LICENSE-2.0 9# 10# Unless required by applicable law or agreed to in writing, software 11# distributed under the License is distributed on an "AS IS" BASIS, 12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 13# See the License for the specific language governing permissions and 14# limitations under the License. 15 16"""Tests REST events for /rooms paths.""" 17 18from unittest.mock import Mock 19 20from synapse.rest.client import room 21from synapse.types import UserID 22 23from tests import unittest 24 25PATH_PREFIX = "/_matrix/client/api/v1" 26 27 28class RoomTypingTestCase(unittest.HomeserverTestCase): 29 """Tests /rooms/$room_id/typing/$user_id REST API.""" 30 31 user_id = "@sid:red" 32 33 user = UserID.from_string(user_id) 34 servlets = [room.register_servlets] 35 36 def make_homeserver(self, reactor, clock): 37 38 hs = self.setup_test_homeserver( 39 "red", 40 federation_http_client=None, 41 federation_client=Mock(), 42 ) 43 44 self.event_source = hs.get_event_sources().sources.typing 45 46 hs.get_federation_handler = Mock() 47 48 async def get_user_by_access_token(token=None, allow_guest=False): 49 return { 50 "user": UserID.from_string(self.auth_user_id), 51 "token_id": 1, 52 "is_guest": False, 53 } 54 55 hs.get_auth().get_user_by_access_token = get_user_by_access_token 56 57 async def _insert_client_ip(*args, **kwargs): 58 return None 59 60 hs.get_datastore().insert_client_ip = _insert_client_ip 61 62 return hs 63 64 def prepare(self, reactor, clock, hs): 65 self.room_id = self.helper.create_room_as(self.user_id) 66 # Need another user to make notifications actually work 67 self.helper.join(self.room_id, user="@jim:red") 68 69 def test_set_typing(self): 70 channel = self.make_request( 71 "PUT", 72 "/rooms/%s/typing/%s" % (self.room_id, self.user_id), 73 b'{"typing": true, "timeout": 30000}', 74 ) 75 self.assertEquals(200, channel.code) 76 77 self.assertEquals(self.event_source.get_current_key(), 1) 78 events = self.get_success( 79 self.event_source.get_new_events( 80 user=UserID.from_string(self.user_id), 81 from_key=0, 82 limit=None, 83 room_ids=[self.room_id], 84 is_guest=False, 85 ) 86 ) 87 self.assertEquals( 88 events[0], 89 [ 90 { 91 "type": "m.typing", 92 "room_id": self.room_id, 93 "content": {"user_ids": [self.user_id]}, 94 } 95 ], 96 ) 97 98 def test_set_not_typing(self): 99 channel = self.make_request( 100 "PUT", 101 "/rooms/%s/typing/%s" % (self.room_id, self.user_id), 102 b'{"typing": false}', 103 ) 104 self.assertEquals(200, channel.code) 105 106 def test_typing_timeout(self): 107 channel = self.make_request( 108 "PUT", 109 "/rooms/%s/typing/%s" % (self.room_id, self.user_id), 110 b'{"typing": true, "timeout": 30000}', 111 ) 112 self.assertEquals(200, channel.code) 113 114 self.assertEquals(self.event_source.get_current_key(), 1) 115 116 self.reactor.advance(36) 117 118 self.assertEquals(self.event_source.get_current_key(), 2) 119 120 channel = self.make_request( 121 "PUT", 122 "/rooms/%s/typing/%s" % (self.room_id, self.user_id), 123 b'{"typing": true, "timeout": 30000}', 124 ) 125 self.assertEquals(200, channel.code) 126 127 self.assertEquals(self.event_source.get_current_key(), 3) 128