85 lines
2.5 KiB
Python
85 lines
2.5 KiB
Python
"""unit test: execute_pagination honours a caller-supplied shared semaphore
|
|
|
|
run from the repo root:
|
|
uv run pytest tests/test_pagination_semaphore.py
|
|
|
|
offline — no network. A fake page func records peak concurrency so we can
|
|
assert the fan-out limit actually holds, both for the shared-semaphore path
|
|
(one instance capping many categories at once) and a single call bounding one
|
|
category with its own semaphore.
|
|
"""
|
|
|
|
import asyncio
|
|
|
|
from monitor_utils.runner import execute_pagination
|
|
|
|
|
|
def _make_tracking_func(state: dict):
|
|
"""Return a page func that tracks peak simultaneous in-flight calls."""
|
|
|
|
async def func(session, offset, **kwargs):
|
|
state["live"] += 1
|
|
state["peak"] = max(state["peak"], state["live"])
|
|
await asyncio.sleep(0.01)
|
|
state["live"] -= 1
|
|
|
|
return func
|
|
|
|
|
|
def test_shared_semaphore_caps_total_concurrency_across_categories():
|
|
"""One shared semaphore bounds combined fan-out over concurrent categories.
|
|
|
|
Two categories paginate at once through the same semaphore(limit); without
|
|
sharing, each would build its own semaphore and the peak could reach twice
|
|
the limit. Peak must stay at or below the shared limit.
|
|
"""
|
|
|
|
limit = 3
|
|
shared = asyncio.Semaphore(limit)
|
|
state = {"live": 0, "peak": 0}
|
|
func = _make_tracking_func(state)
|
|
|
|
async def run():
|
|
await asyncio.gather(
|
|
execute_pagination(
|
|
"catA",
|
|
total_products=300,
|
|
products_per_page=30,
|
|
func=func,
|
|
session=None,
|
|
semaphore=shared,
|
|
),
|
|
execute_pagination(
|
|
"catB",
|
|
total_products=300,
|
|
products_per_page=30,
|
|
func=func,
|
|
session=None,
|
|
semaphore=shared,
|
|
),
|
|
)
|
|
|
|
asyncio.run(run())
|
|
assert state["peak"] <= limit, f"peak {state['peak']} exceeded shared limit {limit}"
|
|
|
|
|
|
def test_single_call_respects_its_semaphore():
|
|
"""A single call bounds its own fan-out to the supplied semaphore limit."""
|
|
|
|
limit = 2
|
|
state = {"live": 0, "peak": 0}
|
|
func = _make_tracking_func(state)
|
|
|
|
async def run():
|
|
await execute_pagination(
|
|
"cat",
|
|
total_products=300,
|
|
products_per_page=30,
|
|
func=func,
|
|
session=None,
|
|
semaphore=asyncio.Semaphore(limit),
|
|
)
|
|
|
|
asyncio.run(run())
|
|
assert state["peak"] <= limit, f"peak {state['peak']} over limit {limit}"
|