Matrix-Mirrors/anonymousland-synapse
1
0
Fork 0
mirror of https://git.anonymousland.org/anonymousland/synapse.git synced 2025-05-03 04:04:54 -04:00
Code Issues Projects Releases Packages Wiki Activity

Add wheeltimer impl

Browse source
This commit is contained in:
Erik Johnston 2016-02-12 16:06:49 +00:00
parent 97d1b3a506
commit 9e7900da1e
2 changed files with 162 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

74
tests/util/test_wheel_timer.py Normal file
View file

@ -0,0 +1,74 @@
# -*- coding: utf-8 -*-
# Copyright 2016 OpenMarket Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from .. import unittest
from synapse.util.wheel_timer import WheelTimer
class WheelTimerTestCase(unittest.TestCase):
def test_single_insert_fetch(self):
wheel = WheelTimer(bucket_size=5)
obj = object()
wheel.insert(100, obj, 150)
self.assertListEqual(wheel.fetch(101), [])
self.assertListEqual(wheel.fetch(110), [])
self.assertListEqual(wheel.fetch(120), [])
self.assertListEqual(wheel.fetch(130), [])
self.assertListEqual(wheel.fetch(149), [])
self.assertListEqual(wheel.fetch(156), [obj])
self.assertListEqual(wheel.fetch(170), [])
def test_mutli_insert(self):
wheel = WheelTimer(bucket_size=5)
obj1 = object()
obj2 = object()
obj3 = object()
wheel.insert(100, obj1, 150)
wheel.insert(105, obj2, 130)
wheel.insert(106, obj3, 160)
self.assertListEqual(wheel.fetch(110), [])
self.assertListEqual(wheel.fetch(135), [obj2])
self.assertListEqual(wheel.fetch(149), [])
self.assertListEqual(wheel.fetch(158), [obj1])
self.assertListEqual(wheel.fetch(160), [])
self.assertListEqual(wheel.fetch(200), [obj3])
self.assertListEqual(wheel.fetch(210), [])
def test_insert_past(self):
wheel = WheelTimer(bucket_size=5)
obj = object()
wheel.insert(100, obj, 50)
self.assertListEqual(wheel.fetch(120), [obj])
def test_insert_past_mutli(self):
wheel = WheelTimer(bucket_size=5)
obj1 = object()
obj2 = object()
obj3 = object()
wheel.insert(100, obj1, 150)
wheel.insert(100, obj2, 140)
wheel.insert(100, obj3, 50)
self.assertListEqual(wheel.fetch(110), [obj3])
self.assertListEqual(wheel.fetch(120), [])
self.assertListEqual(wheel.fetch(147), [obj2])
self.assertListEqual(wheel.fetch(200), [obj1])
self.assertListEqual(wheel.fetch(240), [])