49 lines
1.9 KiB
Python
49 lines
1.9 KiB
Python
|
"""Smoke tests for the reusable scraper building blocks in :mod:`app.scraper.base`.
|
||
|
|
|
||
|
|
The generic paginator is exercised end-to-end by the WeChat friend-scraper
|
||
|
|
tests (it is the only caller today). Here we only sanity-check the
|
||
|
|
:class:`RateLimiter`'s spacing behaviour so the helper is not shipped blind.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import asyncio
|
||
|
|
import time
|
||
|
|
|
||
|
|
from app.scraper.base import RateLimiter
|
||
|
|
|
||
|
|
|
||
|
|
async def test_rate_limiter_no_interval_never_waits():
|
||
|
|
"""``min_interval=0`` is a passthrough — acquire returns immediately."""
|
||
|
|
limiter = RateLimiter(min_interval=0.0)
|
||
|
|
start = time.monotonic()
|
||
|
|
for _ in range(5):
|
||
|
|
await limiter.acquire()
|
||
|
|
elapsed = time.monotonic() - start
|
||
|
|
# 5 acquires with no spacing should complete well under a real interval.
|
||
|
|
assert elapsed < 0.1
|
||
|
|
|
||
|
|
|
||
|
|
async def test_rate_limiter_enforces_min_interval():
|
||
|
|
"""Two acquires are spaced by at least ``min_interval`` seconds."""
|
||
|
|
limiter = RateLimiter(min_interval=0.05)
|
||
|
|
await limiter.acquire() # first call never waits; primes ``_last``
|
||
|
|
start = time.monotonic()
|
||
|
|
await limiter.acquire() # must wait >= 0.05s since the prior call
|
||
|
|
elapsed = time.monotonic() - start
|
||
|
|
assert elapsed >= 0.04 # allow a small scheduler slack margin
|
||
|
|
|
||
|
|
|
||
|
|
async def test_rate_limiter_concurrent_acquire_is_safe():
|
||
|
|
"""Concurrent acquires do not raise; each just waits its turn."""
|
||
|
|
limiter = RateLimiter(min_interval=0.01)
|
||
|
|
await asyncio.gather(*(limiter.acquire() for _ in range(8)))
|
||
|
|
# No assertion on timing here — the contract is "no crash, polite spacing".
|
||
|
|
# We only assert the gather completed without raising.
|
||
|
|
|
||
|
|
|
||
|
|
def test_rate_limiter_is_instantiable_with_negative_interval():
|
||
|
|
"""Negative intervals collapse to the passthrough branch (no surprising raise)."""
|
||
|
|
limiter = RateLimiter(min_interval=-1.0)
|
||
|
|
assert limiter._min_interval <= 0 # sanity check the guard
|