44 lines
1.2 KiB
Python
44 lines
1.2 KiB
Python
"""unit test: fetch_all_pages result accounting for any iterable input
|
|
|
|
run from the repo root:
|
|
uv run pytest tests/test_fetch_all_pages.py
|
|
|
|
pure and offline — no HTTP. Regression: a generator page_range was exhausted
|
|
building the tasks, so the result pairing dropped every failure and swallowed
|
|
unexpected exceptions.
|
|
"""
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
from aioweb import AiowebError
|
|
|
|
from monitor_utils.runner import fetch_all_pages
|
|
|
|
|
|
async def _failing_fetch(page: int):
|
|
raise AiowebError(f"page {page} failed")
|
|
|
|
|
|
async def _crashing_fetch(page: int):
|
|
raise ValueError(f"page {page} crashed")
|
|
|
|
|
|
def test_generator_input_still_reports_failed_pages():
|
|
failed = asyncio.run(
|
|
fetch_all_pages((p for p in [2, 3]), _failing_fetch, "cat", "page")
|
|
)
|
|
assert failed == [2, 3]
|
|
|
|
|
|
def test_generator_input_still_raises_unexpected_errors():
|
|
with pytest.raises(ValueError):
|
|
asyncio.run(fetch_all_pages((p for p in [2]), _crashing_fetch, "cat", "page"))
|
|
|
|
|
|
def test_range_input_reports_failed_pages():
|
|
"""The re-iterable path the in-repo caller uses keeps working."""
|
|
|
|
failed = asyncio.run(fetch_all_pages(range(2, 4), _failing_fetch, "cat", "page"))
|
|
assert failed == [2, 3]
|