test_sync_wait() — langchain Function Reference

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

Function python LangChainCore MessageInterface

Entity Profile

LangChainCore→ MessageInterface→ test_sync_wait() — langchain Function Reference

Dependency Diagram

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

Relationship Graph

Source Code

libs/core/tests/unit_tests/rate_limiters/test_in_memory_rate_limiter.py lines 24–49

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

        # Check max bucket size
        frozen_time.tick(100)
        assert rate_limiter.acquire(blocking=False)
        assert rate_limiter.available_tokens == 1