1
0

test_wheel_timer.py 2.5 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374
  1. # -*- coding: utf-8 -*-
  2. # Copyright 2016 OpenMarket Ltd
  3. #
  4. # Licensed under the Apache License, Version 2.0 (the "License");
  5. # you may not use this file except in compliance with the License.
  6. # You may obtain a copy of the License at
  7. #
  8. # http://www.apache.org/licenses/LICENSE-2.0
  9. #
  10. # Unless required by applicable law or agreed to in writing, software
  11. # distributed under the License is distributed on an "AS IS" BASIS,
  12. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13. # See the License for the specific language governing permissions and
  14. # limitations under the License.
  15. from synapse.util.wheel_timer import WheelTimer
  16. from .. import unittest
  17. class WheelTimerTestCase(unittest.TestCase):
  18. def test_single_insert_fetch(self):
  19. wheel = WheelTimer(bucket_size=5)
  20. obj = object()
  21. wheel.insert(100, obj, 150)
  22. self.assertListEqual(wheel.fetch(101), [])
  23. self.assertListEqual(wheel.fetch(110), [])
  24. self.assertListEqual(wheel.fetch(120), [])
  25. self.assertListEqual(wheel.fetch(130), [])
  26. self.assertListEqual(wheel.fetch(149), [])
  27. self.assertListEqual(wheel.fetch(156), [obj])
  28. self.assertListEqual(wheel.fetch(170), [])
  29. def test_multi_insert(self):
  30. wheel = WheelTimer(bucket_size=5)
  31. obj1 = object()
  32. obj2 = object()
  33. obj3 = object()
  34. wheel.insert(100, obj1, 150)
  35. wheel.insert(105, obj2, 130)
  36. wheel.insert(106, obj3, 160)
  37. self.assertListEqual(wheel.fetch(110), [])
  38. self.assertListEqual(wheel.fetch(135), [obj2])
  39. self.assertListEqual(wheel.fetch(149), [])
  40. self.assertListEqual(wheel.fetch(158), [obj1])
  41. self.assertListEqual(wheel.fetch(160), [])
  42. self.assertListEqual(wheel.fetch(200), [obj3])
  43. self.assertListEqual(wheel.fetch(210), [])
  44. def test_insert_past(self):
  45. wheel = WheelTimer(bucket_size=5)
  46. obj = object()
  47. wheel.insert(100, obj, 50)
  48. self.assertListEqual(wheel.fetch(120), [obj])
  49. def test_insert_past_multi(self):
  50. wheel = WheelTimer(bucket_size=5)
  51. obj1 = object()
  52. obj2 = object()
  53. obj3 = object()
  54. wheel.insert(100, obj1, 150)
  55. wheel.insert(100, obj2, 140)
  56. wheel.insert(100, obj3, 50)
  57. self.assertListEqual(wheel.fetch(110), [obj3])
  58. self.assertListEqual(wheel.fetch(120), [])
  59. self.assertListEqual(wheel.fetch(147), [obj2])
  60. self.assertListEqual(wheel.fetch(200), [obj1])
  61. self.assertListEqual(wheel.fetch(240), [])