test_account_data.py 4.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114
  1. # Copyright 2020 The Matrix.org Foundation C.I.C.
  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 synapse.replication.tcp.streams._base import (
  15. _STREAM_UPDATE_TARGET_ROW_COUNT,
  16. AccountDataStream,
  17. )
  18. from tests.replication._base import BaseStreamTestCase
  19. class AccountDataStreamTestCase(BaseStreamTestCase):
  20. def test_update_function_room_account_data_limit(self) -> None:
  21. """Test replication with many room account data updates"""
  22. store = self.hs.get_datastores().main
  23. # generate lots of account data updates
  24. updates = []
  25. for i in range(_STREAM_UPDATE_TARGET_ROW_COUNT + 5):
  26. update = "m.test_type.%i" % (i,)
  27. self.get_success(
  28. store.add_account_data_to_room("test_user", "test_room", update, {})
  29. )
  30. updates.append(update)
  31. # also one global update
  32. self.get_success(store.add_account_data_for_user("test_user", "m.global", {}))
  33. # tell the notifier to catch up to avoid duplicate rows.
  34. # workaround for https://github.com/matrix-org/synapse/issues/7360
  35. # FIXME remove this when the above is fixed
  36. self.replicate()
  37. # check we're testing what we think we are: no rows should yet have been
  38. # received
  39. self.assertEqual([], self.test_handler.received_rdata_rows)
  40. # now reconnect to pull the updates
  41. self.reconnect()
  42. self.replicate()
  43. # we should have received all the expected rows in the right order
  44. received_rows = self.test_handler.received_rdata_rows
  45. for t in updates:
  46. (stream_name, token, row) = received_rows.pop(0)
  47. self.assertEqual(stream_name, AccountDataStream.NAME)
  48. self.assertIsInstance(row, AccountDataStream.AccountDataStreamRow)
  49. self.assertEqual(row.data_type, t)
  50. self.assertEqual(row.room_id, "test_room")
  51. (stream_name, token, row) = received_rows.pop(0)
  52. self.assertIsInstance(row, AccountDataStream.AccountDataStreamRow)
  53. self.assertEqual(row.data_type, "m.global")
  54. self.assertIsNone(row.room_id)
  55. self.assertEqual([], received_rows)
  56. def test_update_function_global_account_data_limit(self) -> None:
  57. """Test replication with many global account data updates"""
  58. store = self.hs.get_datastores().main
  59. # generate lots of account data updates
  60. updates = []
  61. for i in range(_STREAM_UPDATE_TARGET_ROW_COUNT + 5):
  62. update = "m.test_type.%i" % (i,)
  63. self.get_success(store.add_account_data_for_user("test_user", update, {}))
  64. updates.append(update)
  65. # also one per-room update
  66. self.get_success(
  67. store.add_account_data_to_room("test_user", "test_room", "m.per_room", {})
  68. )
  69. # tell the notifier to catch up to avoid duplicate rows.
  70. # workaround for https://github.com/matrix-org/synapse/issues/7360
  71. # FIXME remove this when the above is fixed
  72. self.replicate()
  73. # check we're testing what we think we are: no rows should yet have been
  74. # received
  75. self.assertEqual([], self.test_handler.received_rdata_rows)
  76. # now reconnect to pull the updates
  77. self.reconnect()
  78. self.replicate()
  79. # we should have received all the expected rows in the right order
  80. received_rows = self.test_handler.received_rdata_rows
  81. for t in updates:
  82. (stream_name, token, row) = received_rows.pop(0)
  83. self.assertEqual(stream_name, AccountDataStream.NAME)
  84. self.assertIsInstance(row, AccountDataStream.AccountDataStreamRow)
  85. self.assertEqual(row.data_type, t)
  86. self.assertIsNone(row.room_id)
  87. (stream_name, token, row) = received_rows.pop(0)
  88. self.assertIsInstance(row, AccountDataStream.AccountDataStreamRow)
  89. self.assertEqual(row.data_type, "m.per_room")
  90. self.assertEqual(row.room_id, "test_room")
  91. self.assertEqual([], received_rows)