test_sync_wait_max_bucket_size() — langchain Function Reference

Source Code

libs/core/tests/unit_tests/rate_limiters/test_in_memory_rate_limiter.py lines 75–91

def test_sync_wait_max_bucket_size() -> None:
    with freeze_time("2023-01-01 00:00:00") as frozen_time:
        rate_limiter = InMemoryRateLimiter(
            requests_per_second=2, check_every_n_seconds=0.1, max_bucket_size=500
        )
        rate_limiter.last = time.time()
        frozen_time.tick(100)  # Increment by 100 seconds
        assert rate_limiter.acquire(blocking=False)
        # After 100 seconds we manage to refill the bucket with 200 tokens
        # After consuming 1 token, we should have 199 tokens left
        assert rate_limiter.available_tokens == 199.0
        frozen_time.tick(10000)
        assert rate_limiter.acquire(blocking=False)
        assert rate_limiter.available_tokens == 499.0
        # Assert that sync wait can proceed without blocking
        # since we have enough tokens
        rate_limiter.acquire(blocking=True)