test_retention.py 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379
  1. # Copyright 2019 New Vector Ltd
  2. #
  3. # Licensed under the Apache License, Version 2.0 (the "License");
  4. # you may not use this file except in compliance with the License.
  5. # You may obtain a copy of the License at
  6. #
  7. # http://www.apache.org/licenses/LICENSE-2.0
  8. #
  9. # Unless required by applicable law or agreed to in writing, software
  10. # distributed under the License is distributed on an "AS IS" BASIS,
  11. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. # See the License for the specific language governing permissions and
  13. # limitations under the License.
  14. from typing import Any, Dict
  15. from unittest.mock import Mock
  16. from twisted.test.proto_helpers import MemoryReactor
  17. from synapse.api.constants import EventTypes
  18. from synapse.rest import admin
  19. from synapse.rest.client import login, room
  20. from synapse.server import HomeServer
  21. from synapse.types import JsonDict, create_requester
  22. from synapse.util import Clock
  23. from synapse.visibility import filter_events_for_client
  24. from tests import unittest
  25. from tests.unittest import override_config
  26. one_hour_ms = 3600000
  27. one_day_ms = one_hour_ms * 24
  28. class RetentionTestCase(unittest.HomeserverTestCase):
  29. servlets = [
  30. admin.register_servlets,
  31. login.register_servlets,
  32. room.register_servlets,
  33. ]
  34. def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
  35. config = self.default_config()
  36. # merge this default retention config with anything that was specified in
  37. # @override_config
  38. retention_config = {
  39. "enabled": True,
  40. "default_policy": {
  41. "min_lifetime": one_day_ms,
  42. "max_lifetime": one_day_ms * 3,
  43. },
  44. "allowed_lifetime_min": one_day_ms,
  45. "allowed_lifetime_max": one_day_ms * 3,
  46. }
  47. retention_config.update(config.get("retention", {}))
  48. config["retention"] = retention_config
  49. self.hs = self.setup_test_homeserver(config=config)
  50. return self.hs
  51. def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
  52. self.user_id = self.register_user("user", "password")
  53. self.token = self.login("user", "password")
  54. self.store = self.hs.get_datastores().main
  55. self.serializer = self.hs.get_event_client_serializer()
  56. self.clock = self.hs.get_clock()
  57. def test_retention_event_purged_with_state_event(self) -> None:
  58. """Tests that expired events are correctly purged when the room's retention policy
  59. is defined by a state event.
  60. """
  61. room_id = self.helper.create_room_as(self.user_id, tok=self.token)
  62. # Set the room's retention period to 2 days.
  63. lifetime = one_day_ms * 2
  64. self.helper.send_state(
  65. room_id=room_id,
  66. event_type=EventTypes.Retention,
  67. body={"max_lifetime": lifetime},
  68. tok=self.token,
  69. )
  70. self._test_retention_event_purged(room_id, one_day_ms * 1.5)
  71. def test_retention_event_purged_with_state_event_outside_allowed(self) -> None:
  72. """Tests that the server configuration can override the policy for a room when
  73. running the purge jobs.
  74. """
  75. room_id = self.helper.create_room_as(self.user_id, tok=self.token)
  76. # Set a max_lifetime higher than the maximum allowed value.
  77. self.helper.send_state(
  78. room_id=room_id,
  79. event_type=EventTypes.Retention,
  80. body={"max_lifetime": one_day_ms * 4},
  81. tok=self.token,
  82. )
  83. # Check that the event is purged after waiting for the maximum allowed duration
  84. # instead of the one specified in the room's policy.
  85. self._test_retention_event_purged(room_id, one_day_ms * 1.5)
  86. # Set a max_lifetime lower than the minimum allowed value.
  87. self.helper.send_state(
  88. room_id=room_id,
  89. event_type=EventTypes.Retention,
  90. body={"max_lifetime": one_hour_ms},
  91. tok=self.token,
  92. )
  93. # Check that the event is purged after waiting for the minimum allowed duration
  94. # instead of the one specified in the room's policy.
  95. self._test_retention_event_purged(room_id, one_day_ms * 0.5)
  96. def test_retention_event_purged_without_state_event(self) -> None:
  97. """Tests that expired events are correctly purged when the room's retention policy
  98. is defined by the server's configuration's default retention policy.
  99. """
  100. room_id = self.helper.create_room_as(self.user_id, tok=self.token)
  101. self._test_retention_event_purged(room_id, one_day_ms * 2)
  102. @override_config({"retention": {"purge_jobs": [{"interval": "5d"}]}})
  103. def test_visibility(self) -> None:
  104. """Tests that synapse.visibility.filter_events_for_client correctly filters out
  105. outdated events, even if the purge job hasn't got to them yet.
  106. We do this by setting a very long time between purge jobs.
  107. """
  108. store = self.hs.get_datastores().main
  109. storage_controllers = self.hs.get_storage_controllers()
  110. room_id = self.helper.create_room_as(self.user_id, tok=self.token)
  111. # Send a first event, which should be filtered out at the end of the test.
  112. resp = self.helper.send(room_id=room_id, body="1", tok=self.token)
  113. first_event_id = resp.get("event_id")
  114. assert isinstance(first_event_id, str)
  115. # Advance the time by 2 days. We're using the default retention policy, therefore
  116. # after this the first event will still be valid.
  117. self.reactor.advance(one_day_ms * 2 / 1000)
  118. # Send another event, which shouldn't get filtered out.
  119. resp = self.helper.send(room_id=room_id, body="2", tok=self.token)
  120. valid_event_id = resp.get("event_id")
  121. assert isinstance(valid_event_id, str)
  122. # Advance the time by another 2 days. After this, the first event should be
  123. # outdated but not the second one.
  124. self.reactor.advance(one_day_ms * 2 / 1000)
  125. # Fetch the events, and run filter_events_for_client on them
  126. events = self.get_success(
  127. store.get_events_as_list([first_event_id, valid_event_id])
  128. )
  129. self.assertEqual(2, len(events), "events retrieved from database")
  130. filtered_events = self.get_success(
  131. filter_events_for_client(storage_controllers, self.user_id, events)
  132. )
  133. # We should only get one event back.
  134. self.assertEqual(len(filtered_events), 1, filtered_events)
  135. # That event should be the second, not outdated event.
  136. self.assertEqual(filtered_events[0].event_id, valid_event_id, filtered_events)
  137. def _test_retention_event_purged(self, room_id: str, increment: float) -> None:
  138. """Run the following test scenario to test the message retention policy support:
  139. 1. Send event 1
  140. 2. Increment time by `increment`
  141. 3. Send event 2
  142. 4. Increment time by `increment`
  143. 5. Check that event 1 has been purged
  144. 6. Check that event 2 has not been purged
  145. 7. Check that state events that were sent before event 1 aren't purged.
  146. The main reason for sending a second event is because currently Synapse won't
  147. purge the latest message in a room because it would otherwise result in a lack of
  148. forward extremities for this room. It's also a good thing to ensure the purge jobs
  149. aren't too greedy and purge messages they shouldn't.
  150. Args:
  151. room_id: The ID of the room to test retention in.
  152. increment: The number of milliseconds to advance the clock each time. Must be
  153. defined so that events in the room aren't purged if they are `increment`
  154. old but are purged if they are `increment * 2` old.
  155. """
  156. # Get the create event to, later, check that we can still access it.
  157. message_handler = self.hs.get_message_handler()
  158. create_event = self.get_success(
  159. message_handler.get_room_data(
  160. create_requester(self.user_id), room_id, EventTypes.Create, state_key=""
  161. )
  162. )
  163. # Send a first event to the room. This is the event we'll want to be purged at the
  164. # end of the test.
  165. resp = self.helper.send(room_id=room_id, body="1", tok=self.token)
  166. expired_event_id = resp.get("event_id")
  167. assert expired_event_id is not None
  168. # Check that we can retrieve the event.
  169. expired_event = self.get_event(expired_event_id)
  170. self.assertEqual(
  171. expired_event.get("content", {}).get("body"), "1", expired_event
  172. )
  173. # Advance the time.
  174. self.reactor.advance(increment / 1000)
  175. # Send another event. We need this because the purge job won't purge the most
  176. # recent event in the room.
  177. resp = self.helper.send(room_id=room_id, body="2", tok=self.token)
  178. valid_event_id = resp.get("event_id")
  179. assert valid_event_id is not None
  180. # Advance the time again. Now our first event should have expired but our second
  181. # one should still be kept.
  182. self.reactor.advance(increment / 1000)
  183. # Check that the first event has been purged from the database, i.e. that we
  184. # can't retrieve it anymore, because it has expired.
  185. self.get_event(expired_event_id, expect_none=True)
  186. # Check that the event that hasn't expired can still be retrieved.
  187. valid_event = self.get_event(valid_event_id)
  188. self.assertEqual(valid_event.get("content", {}).get("body"), "2", valid_event)
  189. # Check that we can still access state events that were sent before the event that
  190. # has been purged.
  191. self.get_event(room_id, bool(create_event))
  192. def get_event(self, event_id: str, expect_none: bool = False) -> JsonDict:
  193. event = self.get_success(self.store.get_event(event_id, allow_none=True))
  194. if expect_none:
  195. self.assertIsNone(event)
  196. return {}
  197. assert event is not None
  198. time_now = self.clock.time_msec()
  199. serialized = self.serializer.serialize_event(event, time_now)
  200. return serialized
  201. class RetentionNoDefaultPolicyTestCase(unittest.HomeserverTestCase):
  202. servlets = [
  203. admin.register_servlets,
  204. login.register_servlets,
  205. room.register_servlets,
  206. ]
  207. def default_config(self) -> Dict[str, Any]:
  208. config = super().default_config()
  209. retention_config = {
  210. "enabled": True,
  211. }
  212. # Update this config with what's in the default config so that
  213. # override_config works as expected.
  214. retention_config.update(config.get("retention", {}))
  215. config["retention"] = retention_config
  216. return config
  217. def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
  218. mock_federation_client = Mock(spec=["backfill"])
  219. self.hs = self.setup_test_homeserver(
  220. federation_client=mock_federation_client,
  221. )
  222. return self.hs
  223. def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
  224. self.user_id = self.register_user("user", "password")
  225. self.token = self.login("user", "password")
  226. def test_no_default_policy(self) -> None:
  227. """Tests that an event doesn't get expired if there is neither a default retention
  228. policy nor a policy specific to the room.
  229. """
  230. room_id = self.helper.create_room_as(self.user_id, tok=self.token)
  231. self._test_retention(room_id)
  232. def test_state_policy(self) -> None:
  233. """Tests that an event gets correctly expired if there is no default retention
  234. policy but there's a policy specific to the room.
  235. """
  236. room_id = self.helper.create_room_as(self.user_id, tok=self.token)
  237. # Set the maximum lifetime to 35 days so that the first event gets expired but not
  238. # the second one.
  239. self.helper.send_state(
  240. room_id=room_id,
  241. event_type=EventTypes.Retention,
  242. body={"max_lifetime": one_day_ms * 35},
  243. tok=self.token,
  244. )
  245. self._test_retention(room_id, expected_code_for_first_event=404)
  246. @unittest.override_config({"retention": {"enabled": False}})
  247. def test_visibility_when_disabled(self) -> None:
  248. """Retention policies should be ignored when the retention feature is disabled."""
  249. room_id = self.helper.create_room_as(self.user_id, tok=self.token)
  250. self.helper.send_state(
  251. room_id=room_id,
  252. event_type=EventTypes.Retention,
  253. body={"max_lifetime": one_day_ms},
  254. tok=self.token,
  255. )
  256. resp = self.helper.send(room_id=room_id, body="test", tok=self.token)
  257. self.reactor.advance(one_day_ms * 2 / 1000)
  258. self.get_event(room_id, resp["event_id"])
  259. def _test_retention(
  260. self, room_id: str, expected_code_for_first_event: int = 200
  261. ) -> None:
  262. # Send a first event to the room. This is the event we'll want to be purged at the
  263. # end of the test.
  264. resp = self.helper.send(room_id=room_id, body="1", tok=self.token)
  265. first_event_id = resp.get("event_id")
  266. assert first_event_id is not None
  267. # Check that we can retrieve the event.
  268. expired_event = self.get_event(room_id, first_event_id)
  269. self.assertEqual(
  270. expired_event.get("content", {}).get("body"), "1", expired_event
  271. )
  272. # Advance the time by a month.
  273. self.reactor.advance(one_day_ms * 30 / 1000)
  274. # Send another event. We need this because the purge job won't purge the most
  275. # recent event in the room.
  276. resp = self.helper.send(room_id=room_id, body="2", tok=self.token)
  277. second_event_id = resp.get("event_id")
  278. assert second_event_id is not None
  279. # Advance the time by another month.
  280. self.reactor.advance(one_day_ms * 30 / 1000)
  281. # Check if the event has been purged from the database.
  282. first_event = self.get_event(
  283. room_id, first_event_id, expected_code=expected_code_for_first_event
  284. )
  285. if expected_code_for_first_event == 200:
  286. self.assertEqual(
  287. first_event.get("content", {}).get("body"), "1", first_event
  288. )
  289. # Check that the event that hasn't been purged can still be retrieved.
  290. second_event = self.get_event(room_id, second_event_id)
  291. self.assertEqual(second_event.get("content", {}).get("body"), "2", second_event)
  292. def get_event(
  293. self, room_id: str, event_id: str, expected_code: int = 200
  294. ) -> JsonDict:
  295. url = "/_matrix/client/r0/rooms/%s/event/%s" % (room_id, event_id)
  296. channel = self.make_request("GET", url, access_token=self.token)
  297. self.assertEqual(channel.code, expected_code, channel.result)
  298. return channel.json_body