TestRateLimit.py 文件源码

def testCall(self):
        obj1 = ExampleClass()
        obj2 = ExampleClass()

        s = time.time()
        assert RateLimit.call("counting", allowed_again=0.1, func=obj1.count) == "counted"
        assert around(time.time() - s, 0.0)  # First allow to call instantly
        assert obj1.counted == 1

        # Call again
        assert not RateLimit.isAllowed("counting", 0.1)
        assert RateLimit.isAllowed("something else", 0.1)
        assert RateLimit.call("counting", allowed_again=0.1, func=obj1.count) == "counted"
        assert around(time.time() - s, 0.1)  # Delays second call within interval
        assert obj1.counted == 2

        # Call 3 times async
        s = time.time()
        assert obj2.counted == 0
        threads = [
            gevent.spawn(lambda: RateLimit.call("counting", allowed_again=0.1, func=obj2.count)),  # Instant
            gevent.spawn(lambda: RateLimit.call("counting", allowed_again=0.1, func=obj2.count)),  # 0.1s delay
            gevent.spawn(lambda: RateLimit.call("counting", allowed_again=0.1, func=obj2.count))   # 0.2s delay
        ]
        gevent.joinall(threads)
        assert [thread.value for thread in threads] == ["counted", "counted", "counted"]
        assert around(time.time() - s, 0.2)

        # No queue = instant again
        s = time.time()
        assert RateLimit.isAllowed("counting", 0.1)
        assert RateLimit.call("counting", allowed_again=0.1, func=obj2.count) == "counted"
        assert around(time.time() - s, 0.0)

        assert obj2.counted == 4