From 8903be2685bba6672cc4fada93b3c77de8ecce1f Mon Sep 17 00:00:00 2001 From: TPV Date: Thu, 30 Jul 2026 14:13:26 -0700 Subject: [PATCH] abstracted out bounded cycles --- src/monitor_utils/runner.py | 78 +++++++++++++++- tests/test_build_product_details.py | 19 ++++ tests/test_run_bounded_cycle.py | 135 ++++++++++++++++++++++++++++ 3 files changed, 229 insertions(+), 3 deletions(-) create mode 100644 tests/test_run_bounded_cycle.py diff --git a/src/monitor_utils/runner.py b/src/monitor_utils/runner.py index 9bc4c60..2affe1c 100644 --- a/src/monitor_utils/runner.py +++ b/src/monitor_utils/runner.py @@ -15,8 +15,15 @@ from monitor_utils.config import ( is_production, set_logger_levels, ) -from monitor_utils.constants import EXPECTED_TASK_ERRORS, LOGGING_LEVEL +from monitor_utils.constants import ( + CYCLE_TIMEOUT_SECONDS, + EXPECTED_TASK_ERRORS, + LOGGING_LEVEL, + MAX_CURL_HANDLES, + TIMEOUT_SECONDS, +) from monitor_utils.database import get_collection +from monitor_utils.sessions import get_default_session from monitor_utils.models import ( AvailabilityInfo, CycleResult, @@ -233,8 +240,10 @@ async def run_monitor_cycle( concurrently, logs expected failures, and raises unexpected ones. The caller owns the loop, the session lifetime, and the inter-cycle delay: wrap this in a `while True` that opens a fresh session per cycle and sleeps - between cycles. The returned CycleResult is pure counts for the caller to - log or act on; this function applies no policy to it. + between cycles — or use run_bounded_cycle, which owns the session and a + hard cycle budget and leaves only the loop and the sleep to the caller. + The returned CycleResult is pure counts for the caller to log or act on; + this function applies no policy to it. session and collection are the two per-cycle resources, passed to every callback (pre_task_func, concurrent_pre_task_func, and each task_func) just @@ -329,6 +338,69 @@ async def run_monitor_cycle( return CycleResult(successes=successes, failures=failures) +async def run_bounded_cycle( + category_identifiers: list[str] | dict[str, str], + task_func: Callable[..., Awaitable[None]], + proxies: AioProxies, + collection: AsyncIOMotorCollection, + *, + session_timeout: int = TIMEOUT_SECONDS, + max_clients: int = MAX_CURL_HANDLES, + cycle_timeout: int = CYCLE_TIMEOUT_SECONDS, + **kwargs, +) -> CycleResult | None: + """Run one monitor cycle on a fresh session under a hard time budget. + + The session is opened and closed inside the call, so curl multi-handle + state cannot outlive one cycle: a wedged transfer dies with its session + instead of parking the process. The budget turns an engine-level lost + wakeup into a caught TimeoutError and an abandoned cycle, never a + permanent wedge; the session's own close is bounded by + get_default_session, so the exit cannot park past the budget either. + + The caller still owns the loop and the inter-cycle delay: + + while True: + await run_bounded_cycle(ids, task_func, proxies, collection) + await asyncio.sleep(CYCLE_DELAY) + + Args: + category_identifiers: Categories, passed through to run_monitor_cycle. + task_func: Awaitable invoked per category. + proxies: Pool bound to each cycle's fresh session. + collection: Mongo collection forwarded to every callback. + session_timeout: Per-request timeout on the cycle's session. + max_clients: Curl handle pool size. Keep at or above the monitor's + whole fan-out; see get_default_session for why. + cycle_timeout: Hard budget in seconds for the whole cycle. + **kwargs: Forwarded to run_monitor_cycle (pre_task_func, + concurrent_pre_task_func, and per-task kwargs). + + Returns: + The cycle's CycleResult, or None when the budget expired and the + cycle was abandoned. + """ + + async with get_default_session( + proxies, timeout=session_timeout, max_clients=max_clients + ) as session: + try: + async with asyncio.timeout(cycle_timeout): + return await run_monitor_cycle( + category_identifiers, + task_func, + session=session, + collection=collection, + **kwargs, + ) + except TimeoutError: + LOGGER.error( + "cycle exceeded budget, abandoning cycle", + extra={"timeout": cycle_timeout}, + ) + return None + + async def execute_pagination( category_identifier: str, total_products: int, diff --git a/tests/test_build_product_details.py b/tests/test_build_product_details.py index 69ad098..aac1220 100644 --- a/tests/test_build_product_details.py +++ b/tests/test_build_product_details.py @@ -30,6 +30,9 @@ def _doc(**overrides) -> dict: "endDate": None, "priceBeforeDrop": None, "eduPrice": None, + "pastPrices": [ + {"price": 80.0, "date": datetime(2026, 7, 26, tzinfo=timezone.utc)} + ], "available": True, "condition": None, "appExclusive": None, @@ -103,3 +106,19 @@ def test_dollar_code_does_not_affect_threshold(): promoMinSpend=100, ) assert build_product_details(doc, min_percentage=40) is None + + +# ----------------------------------------------------------------- pastPrices + + +def test_past_prices_read_through(): + """Price history passes through unchanged.""" + + history = [ + {"price": 95.0, "date": datetime(2026, 7, 20, tzinfo=timezone.utc)}, + {"price": 90.0, "date": datetime(2026, 7, 24, tzinfo=timezone.utc)}, + ] + + product = build_product_details(_doc(pastPrices=history), min_percentage=0) + + assert product.pricing.past_prices == history diff --git a/tests/test_run_bounded_cycle.py b/tests/test_run_bounded_cycle.py new file mode 100644 index 0000000..833bd9a --- /dev/null +++ b/tests/test_run_bounded_cycle.py @@ -0,0 +1,135 @@ +"""unit test: run_bounded_cycle session lifetime, budget, and pass-through + +run from the repo root: + uv run pytest tests/test_run_bounded_cycle.py + +offline — get_default_session is faked at the runner's namespace so no proxies +or network are touched; run_monitor_cycle underneath runs real. Gap: the real +session wiring (timeout/max_clients reaching curl) is covered by the monitors' +own pool-size tests, not here. +""" + +import asyncio +import logging + +import aiohttp +import pytest + +from monitor_utils import runner +from monitor_utils.models import CycleResult +from monitor_utils.runner import run_bounded_cycle + + +class FakeSession: + def __init__(self): + self.closed = False + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + self.closed = True + + +def stub_session(monkeypatch) -> dict: + """Fake get_default_session, recording each call's kwargs and session.""" + calls = [] + + def fake(proxies, timeout=None, max_clients=None): + session = FakeSession() + calls.append( + {"timeout": timeout, "max_clients": max_clients, "session": session} + ) + return session + + monkeypatch.setattr(runner, "get_default_session", fake) + return calls + + +def test_returns_cycle_result_and_closes_session(monkeypatch): + calls = stub_session(monkeypatch) + + async def task(category, session, collection, **kwargs): + pass + + result = asyncio.run(run_bounded_cycle(["a", "b"], task, None, None)) + + assert result == CycleResult(successes=2, failures=0) + assert calls[0]["session"].closed + + +def test_session_params_reach_the_factory(monkeypatch): + calls = stub_session(monkeypatch) + + async def task(category, session, collection, **kwargs): + pass + + asyncio.run( + run_bounded_cycle( + ["a"], task, None, None, session_timeout=8, max_clients=81 + ) + ) + + assert calls[0]["timeout"] == 8 + assert calls[0]["max_clients"] == 81 + + +def test_expired_budget_returns_none_not_raise(monkeypatch, caplog): + """A cycle past its budget is abandoned and logged, and the loop lives on.""" + + calls = stub_session(monkeypatch) + + async def parked(category, session, collection, **kwargs): + await asyncio.Event().wait() + + with caplog.at_level(logging.ERROR, logger="monitor_utils.runner"): + result = asyncio.run( + run_bounded_cycle(["a"], parked, None, None, cycle_timeout=0.05) + ) + + assert result is None + assert calls[0]["session"].closed + assert any( + r.message == "cycle exceeded budget, abandoning cycle" for r in caplog.records + ) + + +def test_expected_task_error_counts_as_failure(monkeypatch): + stub_session(monkeypatch) + + async def flaky(category, session, collection, **kwargs): + raise aiohttp.ClientError("routine") + + result = asyncio.run(run_bounded_cycle(["a"], flaky, None, None)) + + assert result == CycleResult(successes=0, failures=1) + + +def test_unexpected_error_propagates_and_closes_session(monkeypatch): + """fail loud: a real bug must escape the helper, not vanish into a log.""" + + calls = stub_session(monkeypatch) + + async def broken(category, session, collection, **kwargs): + raise RuntimeError("bug") + + with pytest.raises(RuntimeError): + asyncio.run(run_bounded_cycle(["a"], broken, None, None)) + + assert calls[0]["session"].closed + + +def test_each_call_opens_a_fresh_session(monkeypatch): + calls = stub_session(monkeypatch) + + async def task(category, session, collection, **kwargs): + pass + + async def two_cycles(): + await run_bounded_cycle(["a"], task, None, None) + await run_bounded_cycle(["a"], task, None, None) + + asyncio.run(two_cycles()) + + assert len(calls) == 2 + assert calls[0]["session"] is not calls[1]["session"]