summary refs log tree commit diff
path: root/tests/rest/client/test_redactions.py
blob: fe66e397c4fcd8eda15549e9265bcb5c5dab3c40 (plain) (blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
# -*- coding: utf-8 -*-
# Copyright 2019 The Matrix.org Foundation C.I.C.
#
# 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.

from synapse.rest import admin
from synapse.rest.client.v1 import login, room
from synapse.rest.client.v2_alpha import sync

from tests.unittest import HomeserverTestCase


class RedactionsTestCase(HomeserverTestCase):
    """Tests that various redaction events are handled correctly"""

    servlets = [
        admin.register_servlets,
        room.register_servlets,
        login.register_servlets,
        sync.register_servlets,
    ]

    def prepare(self, reactor, clock, hs):
        # register a couple of users
        self.mod_user_id = self.register_user("user1", "pass")
        self.mod_access_token = self.login("user1", "pass")
        self.other_user_id = self.register_user("otheruser", "pass")
        self.other_access_token = self.login("otheruser", "pass")

        # Create a room
        self.room_id = self.helper.create_room_as(
            self.mod_user_id, tok=self.mod_access_token
        )

        # Invite the other user
        self.helper.invite(
            room=self.room_id,
            src=self.mod_user_id,
            tok=self.mod_access_token,
            targ=self.other_user_id,
        )
        # The other user joins
        self.helper.join(
            room=self.room_id, user=self.other_user_id, tok=self.other_access_token
        )

    def _redact_event(self, access_token, room_id, event_id, expect_code=200):
        """Helper function to send a redaction event.

        Returns the json body.
        """
        path = "/_matrix/client/r0/rooms/%s/redact/%s" % (room_id, event_id)

        request, channel = self.make_request(
            "POST", path, content={}, access_token=access_token
        )
        self.render(request)
        self.assertEqual(int(channel.result["code"]), expect_code)
        return channel.json_body

    def _sync_room_timeline(self, access_token, room_id):
        request, channel = self.make_request(
            "GET", "sync", access_token=self.mod_access_token
        )
        self.render(request)
        self.assertEqual(channel.result["code"], b"200")
        room_sync = channel.json_body["rooms"]["join"][room_id]
        return room_sync["timeline"]["events"]

    def test_redact_event_as_moderator(self):
        # as a regular user, send a message to redact
        b = self.helper.send(room_id=self.room_id, tok=self.other_access_token)
        msg_id = b["event_id"]

        # as the moderator, send a redaction
        b = self._redact_event(self.mod_access_token, self.room_id, msg_id)
        redaction_id = b["event_id"]

        # now sync
        timeline = self._sync_room_timeline(self.mod_access_token, self.room_id)

        # the last event should be the redaction
        self.assertEqual(timeline[-1]["event_id"], redaction_id)
        self.assertEqual(timeline[-1]["redacts"], msg_id)

        # and the penultimate should be the redacted original
        self.assertEqual(timeline[-2]["event_id"], msg_id)
        self.assertEqual(timeline[-2]["unsigned"]["redacted_by"], redaction_id)
        self.assertEqual(timeline[-2]["content"], {})

    def test_redact_event_as_normal(self):
        # as a regular user, send a message to redact
        b = self.helper.send(room_id=self.room_id, tok=self.other_access_token)
        normal_msg_id = b["event_id"]

        # also send one as the admin
        b = self.helper.send(room_id=self.room_id, tok=self.mod_access_token)
        admin_msg_id = b["event_id"]

        # as a normal, try to redact the admin's event
        self._redact_event(
            self.other_access_token, self.room_id, admin_msg_id, expect_code=403
        )

        # now try to redact our own event
        b = self._redact_event(self.other_access_token, self.room_id, normal_msg_id)
        redaction_id = b["event_id"]

        # now sync
        timeline = self._sync_room_timeline(self.other_access_token, self.room_id)

        # the last event should be the redaction of the normal event
        self.assertEqual(timeline[-1]["event_id"], redaction_id)
        self.assertEqual(timeline[-1]["redacts"], normal_msg_id)

        # the penultimate should be the unredacted one from the admin
        self.assertEqual(timeline[-2]["event_id"], admin_msg_id)
        self.assertNotIn("redacted_by", timeline[-2]["unsigned"])
        self.assertTrue(timeline[-2]["content"]["body"], {})

        # and the antepenultimate should be the redacted normal
        self.assertEqual(timeline[-3]["event_id"], normal_msg_id)
        self.assertEqual(timeline[-3]["unsigned"]["redacted_by"], redaction_id)
        self.assertEqual(timeline[-3]["content"], {})

    def test_redact_nonexistent_event(self):
        # control case: an existing event
        b = self.helper.send(room_id=self.room_id, tok=self.other_access_token)
        msg_id = b["event_id"]
        b = self._redact_event(self.other_access_token, self.room_id, msg_id)
        redaction_id = b["event_id"]

        # room moderators can send redactions for non-existent events
        self._redact_event(self.mod_access_token, self.room_id, "$zzz")

        # ... but normals cannot
        self._redact_event(
            self.other_access_token, self.room_id, "$zzz", expect_code=404
        )

        # when we sync, we should see only the valid redaction
        timeline = self._sync_room_timeline(self.other_access_token, self.room_id)
        self.assertEqual(timeline[-1]["event_id"], redaction_id)
        self.assertEqual(timeline[-1]["redacts"], msg_id)

        # and the penultimate should be the redacted original
        self.assertEqual(timeline[-2]["event_id"], msg_id)
        self.assertEqual(timeline[-2]["unsigned"]["redacted_by"], redaction_id)
        self.assertEqual(timeline[-2]["content"], {})

    def test_redact_create_event(self):
        # control case: an existing event
        b = self.helper.send(room_id=self.room_id, tok=self.mod_access_token)
        msg_id = b["event_id"]
        self._redact_event(self.mod_access_token, self.room_id, msg_id)

        # sync the room, to get the id of the create event
        timeline = self._sync_room_timeline(self.other_access_token, self.room_id)
        create_event_id = timeline[0]["event_id"]

        # room moderators cannot send redactions for create events
        self._redact_event(
            self.mod_access_token, self.room_id, create_event_id, expect_code=403
        )

        # and nor can normals
        self._redact_event(
            self.other_access_token, self.room_id, create_event_id, expect_code=403
        )