Add unit test for /purge_room API

This commit is contained in:
Erik Johnston 2019-10-31 11:30:25 +00:00
parent c6bcd38841
commit 97c60ccaa3
2 changed files with 83 additions and 1 deletions

View File

@ -561,3 +561,81 @@ class DeleteGroupTestCase(unittest.HomeserverTestCase):
self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"]) self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
return channel.json_body["groups"] return channel.json_body["groups"]
class PurgeRoomTestCase(unittest.HomeserverTestCase):
"""Test /purge_room admin API.
"""
servlets = [
synapse.rest.admin.register_servlets,
login.register_servlets,
room.register_servlets,
]
def prepare(self, reactor, clock, hs):
self.store = hs.get_datastore()
self.admin_user = self.register_user("admin", "pass", admin=True)
self.admin_user_tok = self.login("admin", "pass")
def test_purge_room(self):
room_id = self.helper.create_room_as(self.admin_user, tok=self.admin_user_tok)
# All users have to have left the room.
self.helper.leave(room_id, user=self.admin_user, tok=self.admin_user_tok)
url = "/_synapse/admin/v1/purge_room"
request, channel = self.make_request(
"POST",
url.encode("ascii"),
{"room_id": room_id},
access_token=self.admin_user_tok,
)
self.render(request)
self.assertEqual(200, int(channel.result["code"]), msg=channel.result["body"])
# Test that the following tables have been purged of all rows related to the room.
for table in (
"current_state_events",
"event_backward_extremities",
"event_forward_extremities",
"event_json",
"event_push_actions",
"event_search",
"events",
"group_rooms",
"public_room_list_stream",
"receipts_graph",
"receipts_linearized",
"room_aliases",
"room_depth",
"room_memberships",
"room_stats_state",
"room_stats_current",
"room_stats_historical",
"room_stats_earliest_token",
"rooms",
"stream_ordering_to_exterm",
"users_in_public_rooms",
"users_who_share_private_rooms",
"appservice_room_list",
"e2e_room_keys",
"event_push_summary",
"pusher_throttle",
"group_summary_rooms",
"local_invites",
"room_account_data",
"room_tags",
):
count = self.get_success(
self.store._simple_select_one_onecol(
table="events",
keyvalues={"room_id": room_id},
retcol="COUNT(*)",
desc="test_purge_room",
)
)
self.assertEqual(count, 0, msg="Rows not purged in {}".format(table))

View File

@ -161,7 +161,11 @@ def make_request(
path = path.encode("ascii") path = path.encode("ascii")
# Decorate it to be the full path, if we're using shorthand # Decorate it to be the full path, if we're using shorthand
if shorthand and not path.startswith(b"/_matrix"): if (
shorthand
and not path.startswith(b"/_matrix")
and not path.startswith(b"/_synapse")
):
path = b"/_matrix/client/r0/" + path path = b"/_matrix/client/r0/" + path
path = path.replace(b"//", b"/") path = path.replace(b"//", b"/")