test_wheel_timer.py 2.6 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273
  1. # Copyright 2016 OpenMarket 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 synapse.util.wheel_timer import WheelTimer
  15. from .. import unittest
  16. class WheelTimerTestCase(unittest.TestCase):
  17. def test_single_insert_fetch(self) -> None:
  18. wheel: WheelTimer[object] = WheelTimer(bucket_size=5)
  19. obj = object()
  20. wheel.insert(100, obj, 150)
  21. self.assertListEqual(wheel.fetch(101), [])
  22. self.assertListEqual(wheel.fetch(110), [])
  23. self.assertListEqual(wheel.fetch(120), [])
  24. self.assertListEqual(wheel.fetch(130), [])
  25. self.assertListEqual(wheel.fetch(149), [])
  26. self.assertListEqual(wheel.fetch(156), [obj])
  27. self.assertListEqual(wheel.fetch(170), [])
  28. def test_multi_insert(self) -> None:
  29. wheel: WheelTimer[object] = WheelTimer(bucket_size=5)
  30. obj1 = object()
  31. obj2 = object()
  32. obj3 = object()
  33. wheel.insert(100, obj1, 150)
  34. wheel.insert(105, obj2, 130)
  35. wheel.insert(106, obj3, 160)
  36. self.assertListEqual(wheel.fetch(110), [])
  37. self.assertListEqual(wheel.fetch(135), [obj2])
  38. self.assertListEqual(wheel.fetch(149), [])
  39. self.assertListEqual(wheel.fetch(158), [obj1])
  40. self.assertListEqual(wheel.fetch(160), [])
  41. self.assertListEqual(wheel.fetch(200), [obj3])
  42. self.assertListEqual(wheel.fetch(210), [])
  43. def test_insert_past(self) -> None:
  44. wheel: WheelTimer[object] = WheelTimer(bucket_size=5)
  45. obj = object()
  46. wheel.insert(100, obj, 50)
  47. self.assertListEqual(wheel.fetch(120), [obj])
  48. def test_insert_past_multi(self) -> None:
  49. wheel: WheelTimer[object] = WheelTimer(bucket_size=5)
  50. obj1 = object()
  51. obj2 = object()
  52. obj3 = object()
  53. wheel.insert(100, obj1, 150)
  54. wheel.insert(100, obj2, 140)
  55. wheel.insert(100, obj3, 50)
  56. self.assertListEqual(wheel.fetch(110), [obj3])
  57. self.assertListEqual(wheel.fetch(120), [])
  58. self.assertListEqual(wheel.fetch(147), [obj2])
  59. self.assertListEqual(wheel.fetch(200), [obj1])
  60. self.assertListEqual(wheel.fetch(240), [])