test_async_wait() — langchain Function Reference

Architecture documentation for the test_async_wait() function in test_in_memory_rate_limiter.py from the langchain codebase.

Function python LangChainCore ApiManagement

Entity Profile

LangChainCore→ ApiManagement→ test_async_wait() — langchain Function Reference

Dependency Diagram

graph TD
  edf66d6f_d1f7_6350_0ced_cf7280663e2a["test_async_wait()"]
  a2ad4c1d_ffde_beed_c251_817f2cc8edf6["test_in_memory_rate_limiter.py"]
  edf66d6f_d1f7_6350_0ced_cf7280663e2a -->|defined in| a2ad4c1d_ffde_beed_c251_817f2cc8edf6
  style edf66d6f_d1f7_6350_0ced_cf7280663e2a fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/rate_limiters/test_in_memory_rate_limiter.py lines 52–72

async def test_async_wait(rate_limiter: InMemoryRateLimiter) -> None:
    with freeze_time("2023-01-01 00:00:00") as frozen_time:
        rate_limiter.last = time.time()
        assert not await rate_limiter.aacquire(blocking=False)
        frozen_time.tick(0.1)  # Increment by 0.1 seconds
        assert rate_limiter.available_tokens == 0
        assert not await rate_limiter.aacquire(blocking=False)
        frozen_time.tick(0.1)  # Increment by 0.1 seconds
        assert rate_limiter.available_tokens == 0
        assert not await rate_limiter.aacquire(blocking=False)
        frozen_time.tick(1.8)
        assert await rate_limiter.aacquire(blocking=False)
        assert rate_limiter.available_tokens == 1.0
        assert await rate_limiter.aacquire(blocking=False)
        assert rate_limiter.available_tokens == 0
        frozen_time.tick(2.1)
        assert await rate_limiter.aacquire(blocking=False)
        assert rate_limiter.available_tokens == 1
        frozen_time.tick(0.9)
        assert await rate_limiter.aacquire(blocking=False)
        assert rate_limiter.available_tokens == 1