Method testCall

src/Test/TestRateLimit.py:25–65

(self)

class TestRateLimit:
    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
        time.sleep(0.1)  # Wait the cooldown time

        # 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)

        # Wait 0.1s cooldown
        assert not RateLimit.isAllowed("counting", 0.1)
        time.sleep(0.11)
        assert RateLimit.isAllowed("counting", 0.1)

        # 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

    def testCallAsync(self):
        obj1 = ExampleClass()