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