Do not propagate typing notifications from shadow-banned users. (#8176)

1 files changed, 48 insertions, 0 deletions

diff --git a/tests/rest/client/test_shadow_banned.py b/tests/rest/client/test_shadow_banned.py
index 3eb9aeaa9e..0c48a9fd5e 100644
--- a/tests/rest/client/test_shadow_banned.py
+++ b/tests/rest/client/test_shadow_banned.py
@@ -179,6 +179,54 @@ class RoomTestCase(_ShadowBannedBase):
         # The summary should be empty since the room doesn't exist.
         self.assertEqual(summary, {})
 
+    def test_typing(self):
+        """Typing notifications should not be propagated into the room."""
+        # The create works fine.
+        room_id = self.helper.create_room_as(
+            self.banned_user_id, tok=self.banned_access_token
+        )
+
+        request, channel = self.make_request(
+            "PUT",
+            "/rooms/%s/typing/%s" % (room_id, self.banned_user_id),
+            {"typing": True, "timeout": 30000},
+            access_token=self.banned_access_token,
+        )
+        self.render(request)
+        self.assertEquals(200, channel.code)
+
+        # There should be no typing events.
+        event_source = self.hs.get_event_sources().sources["typing"]
+        self.assertEquals(event_source.get_current_key(), 0)
+
+        # The other user can join and send typing events.
+        self.helper.join(room_id, self.other_user_id, tok=self.other_access_token)
+
+        request, channel = self.make_request(
+            "PUT",
+            "/rooms/%s/typing/%s" % (room_id, self.other_user_id),
+            {"typing": True, "timeout": 30000},
+            access_token=self.other_access_token,
+        )
+        self.render(request)
+        self.assertEquals(200, channel.code)
+
+        # These appear in the room.
+        self.assertEquals(event_source.get_current_key(), 1)
+        events = self.get_success(
+            event_source.get_new_events(from_key=0, room_ids=[room_id])
+        )
+        self.assertEquals(
+            events[0],
+            [
+                {
+                    "type": "m.typing",
+                    "room_id": room_id,
+                    "content": {"user_ids": [self.other_user_id]},
+                }
+            ],
+        )
+
 
 # To avoid the tests timing out don't add a delay to "annoy the requester".
 @patch("random.randint", new=lambda a, b: 0)