(self)
| 65 | assert obj2.counted == 4 |
| 66 | |
| 67 | def testCallAsync(self): |
| 68 | obj1 = ExampleClass() |
| 69 | obj2 = ExampleClass() |
| 70 | |
| 71 | s = time.time() |
| 72 | RateLimit.callAsync("counting async", allowed_again=0.1, func=obj1.count, back="call #1").join() |
| 73 | assert obj1.counted == 1 # First instant |
| 74 | assert around(time.time() - s, 0.0) |
| 75 | |
| 76 | # After that the calls delayed |
| 77 | s = time.time() |
| 78 | t1 = RateLimit.callAsync("counting async", allowed_again=0.1, func=obj1.count, back="call #2") # Dumped by the next call |
| 79 | time.sleep(0.03) |
| 80 | t2 = RateLimit.callAsync("counting async", allowed_again=0.1, func=obj1.count, back="call #3") # Dumped by the next call |
| 81 | time.sleep(0.03) |
| 82 | t3 = RateLimit.callAsync("counting async", allowed_again=0.1, func=obj1.count, back="call #4") # Will be called |
| 83 | assert obj1.counted == 1 # Delay still in progress: Not called yet |
| 84 | t3.join() |
| 85 | assert t3.value == "call #4" |
| 86 | assert around(time.time() - s, 0.1) |
| 87 | |
| 88 | # Only the last one called |
| 89 | assert obj1.counted == 2 |
| 90 | assert obj1.last_called == "call #4" |
| 91 | |
| 92 | # Just called, not allowed again |
| 93 | assert not RateLimit.isAllowed("counting async", 0.1) |
| 94 | s = time.time() |
| 95 | t4 = RateLimit.callAsync("counting async", allowed_again=0.1, func=obj1.count, back="call #5").join() |
| 96 | assert obj1.counted == 3 |
| 97 | assert around(time.time() - s, 0.1) |
| 98 | assert not RateLimit.isAllowed("counting async", 0.1) |
| 99 | time.sleep(0.11) |
| 100 | assert RateLimit.isAllowed("counting async", 0.1) |
nothing calls this directly
no test coverage detected