| 12345678910111213141516171819202122232425262728293031323334353637383940414243 |
- """跨线程异步限速单元测试。"""
- from __future__ import annotations
- import asyncio
- import threading
- import time
- from agents.find_agent.support.rate_limit import make_async_interval_limiter
- def test_async_interval_limiter_is_safe_across_threads() -> None:
- wait = make_async_interval_limiter(0.05)
- hits: list[float] = []
- lock = threading.Lock()
- errors: list[BaseException] = []
- def worker() -> None:
- loop = asyncio.new_event_loop()
- asyncio.set_event_loop(loop)
- try:
- for _ in range(2):
- loop.run_until_complete(wait())
- with lock:
- hits.append(time.monotonic())
- except BaseException as exc: # noqa: BLE001 - collect for assertion
- with lock:
- errors.append(exc)
- finally:
- loop.close()
- asyncio.set_event_loop(None)
- threads = [threading.Thread(target=worker) for _ in range(3)]
- for thread in threads:
- thread.start()
- for thread in threads:
- thread.join(timeout=5)
- assert not thread.is_alive()
- assert not errors
- assert len(hits) == 6
- ordered = sorted(hits)
- for prev, curr in zip(ordered, ordered[1:]):
- assert curr - prev >= 0.04
|
