pyproject.toml

@@ -8,7 +8,8 @@ version = "0.1.0"
 requires-python = ">=3.10"
 dependencies = [
     "aiohttp>=3.14.1",
-    "aioproxies",
+    "aioproxies @ git+https://git.rethinkstudios.io/rethink-public/aioproxies.git",
+    "aioweb-tls[curl] @ git+https://git.rethinkstudios.io/rethink-public/aioweb_tls.git",
     "curl-cffi>=0.15.0",
     "fake-useragent>=2.2.0",
     "motor>=3.7.1",
@@ -17,11 +18,30 @@ dependencies = [
     "requests>=2.34.2",
     "tenacity>=9.1.4",
     "log-setup @ git+https://git.rethinkstudios.io/rethink-public/log_setup.git",
+    "httpx>=0.28.1",
 ]
 
 [tool.setuptools.packages.find]
 where = ["src"]
 include = ["monitor_utils*"]
 
-[tool.uv.sources]
+[dependency-groups]
-aioproxies = { git = "https://git.rethinkstudios.io/rethink-public/aioproxies.git" }
+dev = [
+    "pytest>=8.0",
+    "ruff>=0.6",
+]
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+
+[tool.ruff]
+line-length = 88
+target-version = "py310"
+# dsql_shortlinks.py is vendored third-party code; leave it untouched.
+extend-exclude = ["src/monitor_utils/dsql_shortlinks.py"]
+
+[tool.ruff.lint]
+select = ["E", "F", "I", "W", "UP", "B"]
+# B905: zip(strict=) is opinionated; the zips here are equal-length by
+# construction and adding strict= would change working code for no gain.
+ignore = ["B905"]

src/monitor_utils/constants.py

@@ -11,6 +11,7 @@ from json import JSONDecodeError
 
 import aiohttp
 import requests
+from aioweb import AiowebError
 from curl_cffi.requests.errors import RequestsError
 from pymongo.errors import PyMongoError
 
@@ -27,6 +28,7 @@ LOGGERS = [
 
 RETRY_LIMIT = 5
 RETRY_WAIT = 3
+CYCLE_DELAY = 0.30
 MIN_BACKOFF_DELAY = 0.1
 MAX_BACKOFF_DELAY = 2.0
 FAILURE_DELAY = 3
@@ -44,6 +46,7 @@ EXPECTED_TASK_ERRORS = (
     requests.HTTPError,
     PyMongoError,
     JSONDecodeError,
+    AiowebError,
 )
 
 

src/monitor_utils/database.py

@@ -99,7 +99,8 @@ async def add_products_to_database(
     price is lower (a drop) and cleared otherwise, so any price_before_drop on
     the incoming product is ignored. Prices are recorded in a pastPrices array
     of price/date entries; a price already present in the array is never
-    recorded again, so each distinct price keeps only its first-seen timestamp. New products and never-before-recorded prices
+    recorded again, so each distinct price keeps only its first-seen timestamp.
+    New products and never-before-recorded prices
     upsert an alert referencing the product; a price returning to a value
     already in pastPrices emits nothing, so per-source price oscillation
     (member vs non-member quotes for the same SKU) alerts once per distinct
@@ -119,6 +120,12 @@ async def add_products_to_database(
             product_id.
     """
 
+    # Callers are expected to filter empty batches, but guard here too: an empty
+    # products list would make collection.bulk_write([]) raise InvalidOperation,
+    # which callers would then log as a spurious category failure every cycle.
+    if not products:
+        return
+
     doc_ids = [
         build_doc_id(p.product_id, category_identifier, use_composite) for p in products
     ]
@@ -311,8 +318,10 @@ def build_doc_id(
 def convert_product_to_mongo(product: Product) -> dict:
     """Flatten a Product into its MongoDB document representation.
 
-    Coerces string prices to floats and stamps lastUpdated with the current
+    Stamps lastUpdated with the current UTC time. Prices are already floats via
-    UTC time. The _id is assigned by the caller via build_doc_id.
+    the Product's pydantic validation (numeric strings coerced, comma strings
+    rejected at construction), so no coercion happens here. The _id is assigned
+    by the caller via build_doc_id.
 
     Args:
         product: Product to convert.
@@ -328,9 +337,6 @@ def convert_product_to_mongo(product: Product) -> dict:
     availability = data["availability"]
     current_price = pricing["current_price"]
     original_price = pricing["original_price"]
-    if isinstance(current_price, str):
-        current_price = float(current_price.replace(",", ""))
-        original_price = float(original_price.replace(",", ""))
 
     return {
         "title": data["title"],

src/monitor_utils/dsql_shortlinks.py

@@ -1,3 +1,4 @@
+# Idk who made this i was provided it by louis
 import asyncio
 from typing import Dict
 

src/monitor_utils/models.py

@@ -6,6 +6,24 @@ from datetime import datetime
 from pydantic.dataclasses import dataclass
 
 
+@dataclass
+class CycleResult:
+    """Tallied outcome of one monitor cycle.
+
+    Pure counts for the caller to log or act on as it sees fit; no policy is
+    baked in.
+
+    Attributes:
+        successes: Category tasks that completed without an expected error.
+        failures: Category tasks that raised an expected error.
+        skipped: True when pre_task_func failed and no category tasks ran.
+    """
+
+    successes: int = 0
+    failures: int = 0
+    skipped: bool = False
+
+
 @dataclass
 class ConsumerConfig:
     """Configuration for a single retailer's consumer.

src/monitor_utils/runner.py

@@ -1,6 +1,6 @@
 import asyncio
 import logging
-from typing import Awaitable, Callable, Iterable
+from collections.abc import Awaitable, Callable, Iterable
 
 import aiohttp
 from aioproxies import AioProxies
@@ -18,12 +18,12 @@ from monitor_utils.constants import EXPECTED_TASK_ERRORS, LOGGING_LEVEL
 from monitor_utils.database import get_collection
 from monitor_utils.models import (
     AvailabilityInfo,
+    CycleResult,
     PricingInfo,
     Product,
     ProductMetadata,
     ProductTiming,
 )
-from monitor_utils.sessions import get_base_session, get_tls_session
 
 LOGGER = logging.getLogger(__name__)
 
@@ -186,87 +186,97 @@ def build_product_details(
     return built
 
 
-async def execute_monitor_logic(
+async def run_monitor_cycle(
     category_identifiers: list[str] | dict[str, str],
     task_func: Callable[..., Awaitable[None]],
+    session: aiohttp.ClientSession | AsyncSession,
+    collection: AsyncIOMotorCollection,
     *,
-    proxies: AioProxies,
-    headers: dict | None = None,
-    cookies: dict | None = None,
     pre_task_func: Callable[..., Awaitable[None]] | None = None,
     concurrent_pre_task_func: Callable[..., Awaitable[None]] | None = None,
-    use_tls: bool = False,
     **kwargs,
-):
+) -> CycleResult:
-    """Run the monitor's category tasks in a loop with a fresh session per cycle.
+    """Run one monitor cycle over a caller-supplied session and report the outcome.
 
-    Each cycle opens a TLS or base session, awaits pre_task_func if given,
+    Awaits pre_task_func if given, dispatches task_func per category
-    dispatches task_func per category concurrently, logs expected failures,
+    concurrently, logs expected failures, and raises unexpected ones. The
-    raises unexpected ones, then sleeps briefly before the next cycle.
+    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.
+
+    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
+    like each other. A callback that does not need one still receives it, so
+    every callback must accept **kwargs to absorb the ones it ignores.
 
     Args:
         category_identifiers: List of category ids, or a mapping of category
             id to query id; mapping entries pass query_category_id to
             task_func.
-        task_func: Awaitable invoked per category with the identifier,
+        task_func: Awaitable invoked per category with the identifier, session,
-            session, and **kwargs.
+            collection, and **kwargs.
-        proxies: Proxies applied to the session.
+        session: Open HTTP session for this cycle, built and closed by the
-        headers: Headers applied to the session.
+            caller. Proxies are the caller's responsibility, bound to the
-        cookies: Cookies applied to the base session; ignored for TLS.
+            session (see get_default_session).
-        pre_task_func: Optional awaitable run once per cycle, awaited before any
+        collection: Mongo collection the tasks persist products to; forwarded to
-            category tasks are dispatched and receiving session and **kwargs.
+            every callback so any that writes has it.
-            Use it for setup the category tasks depend on (session headers,
+        pre_task_func: Optional awaitable run once at the start of the cycle,
-            cookies, fetched state); if it raises an expected task error the
+            awaited before any category tasks are dispatched and receiving
-            cycle is skipped and retried after the usual delay.
+            session, collection, and **kwargs. Use it for setup the category
+            tasks depend on (session headers, cookies, fetched state); if it
+            raises an expected task error the cycle is skipped and a skipped
+            CycleResult returned.
         concurrent_pre_task_func: Optional awaitable run once per cycle, gathered
-            alongside the category tasks and receiving session and **kwargs.
+            alongside the category tasks and receiving session, collection, and
-            Category tasks must not depend on its side effects (session
+            **kwargs. Category tasks must not depend on its side effects
-            headers, cookies, fetched state) since no ordering is guaranteed.
+            (session headers, cookies, fetched state) since no ordering is
-        use_tls: Use the TLS fingerprint session instead of the base session.
+            guaranteed.
         **kwargs: Forwarded verbatim to pre_task_func, concurrent_pre_task_func,
             and every task_func call.
+
+    Returns:
+        A CycleResult tallying task successes and expected-error failures, or a
+        skipped result when pre_task_func failed.
+
+    Raises:
+        Exception: Any unexpected (non-EXPECTED_TASK_ERRORS) error raised by a
+            task or pre_task_func propagates, failing loud so the caller's loop
+            crashes rather than silently retrying a broken monitor.
     """
 
-    while True:
-        session_ctx = (
-            get_tls_session(
-                headers=headers,
-                proxies=proxies,
-            )
-            if use_tls
-            else get_base_session(headers=headers, cookies=cookies, proxies=proxies)
-        )
-        async with session_ctx as session:
-            skip_cycle = False
     if pre_task_func:
         try:
-                    await pre_task_func(session=session, **kwargs)
+            await pre_task_func(session=session, collection=collection, **kwargs)
         except EXPECTED_TASK_ERRORS as error:
             LOGGER.warning("pre task failed, skipping cycle", exc_info=error)
-                    skip_cycle = True
+            return CycleResult(skipped=True)
 
-            if not skip_cycle:
     LOGGER.info("starting tasks")
     if isinstance(category_identifiers, dict):
         tasks = [
             task_func(
-                            category_id, session, query_category_id=query, **kwargs
+                category_id, session, collection, query_category_id=query, **kwargs
             )
             for category_id, query in category_identifiers.items()
         ]
         category_keys = list(category_identifiers.keys())
     else:
         tasks = [
-                        task_func(category_identifier, session, **kwargs)
+            task_func(category_identifier, session, collection, **kwargs)
             for category_identifier in category_identifiers
         ]
         category_keys = list(category_identifiers)
 
     if concurrent_pre_task_func:
-                    tasks.append(concurrent_pre_task_func(session=session, **kwargs))
+        tasks.append(
+            concurrent_pre_task_func(session=session, collection=collection, **kwargs)
+        )
         category_keys.append("pre_task")
 
     results = await asyncio.gather(*tasks, return_exceptions=True)
+    successes = 0
+    failures = 0
     for category_identifier, result in zip(category_keys, results):
         if isinstance(result, EXPECTED_TASK_ERRORS):
             LOGGER.warning(
@@ -274,6 +284,7 @@ async def execute_monitor_logic(
                 extra={"category": category_identifier},
                 exc_info=result,
             )
+            failures += 1
         elif isinstance(result, Exception):
             LOGGER.error(
                 "unexpected error",
@@ -281,9 +292,10 @@ async def execute_monitor_logic(
                 exc_info=result,
             )
             raise result
+        else:
+            successes += 1
 
-        LOGGER.info("sleeping before next tasks", extra={"delay": 0.30})
+    return CycleResult(successes=successes, failures=failures)
-        await asyncio.sleep(0.30)
 
 
 async def execute_pagination(
@@ -311,11 +323,10 @@ async def execute_pagination(
         func: Awaitable task invoked per page; receives session plus either
             offset or page_number depending on pagination_mode.
         session: HTTP session forwarded to each func call.
-        max_concurrent_tasks: Semaphore limit for concurrent page fetches, used
+        semaphore: Required bound on concurrent page fetches. Pass one shared
-            only when semaphore is not supplied. Each execute_pagination call
+            semaphore across all categories so the limit caps total fan-out; a
-            builds its own semaphore, so this bounds one category in isolation;
+            per-call semaphore would instead bound each category in isolation,
-            with many categories running concurrently the effective limit is
+            making the effective limit this value times the category count.
-            this value times the category count.
         start_offset: Additive base applied to every offset, for APIs whose
             offset space does not begin at zero. Page N requests offset
             start_offset + (N - 1) * products_per_page.

src/monitor_utils/sessions.py

@@ -1,225 +1,105 @@
 """Shared HTTP session helpers for all monitors.
 
-Known gap: UA/TLS fingerprint does not always match headers/header order,
+A single aioweb_tls TLSSession (curl_cffi backend, random client fingerprint per
-but targeted antibots do not currently seem to care.
+session) replaces the old aiohttp/curl_cffi split. send_request wraps aioweb's
+request_with_retries, adding per-request proxy rotation and old-style JSON/text
+return values so monitor call sites change as little as possible.
 """
 
-import asyncio
-import json
-import logging
 import random
-from contextlib import asynccontextmanager
-from typing import AsyncGenerator
 
-import aiohttp
-import requests
 from aioproxies import AioProxies
-from curl_cffi.requests import AsyncSession
+from aioweb import Response
-from curl_cffi.requests.errors import RequestsError
+from aioweb_tls import CurlCffi, TLSSession
-from curl_cffi.requests.exceptions import Timeout as CurlTimeout
-from fake_useragent import FakeUserAgent
-from tenacity import (
-    after_log,
-    before_log,
-    retry,
-    retry_if_exception_type,
-    stop_after_attempt,
-    wait_fixed,
-)
 
-from monitor_utils.constants import (
+from monitor_utils.constants import CLIENT_IDENTIFIERS, RETRY_LIMIT
-    CLIENT_IDENTIFIERS,
-    CLIENT_TIMEOUT,
-    RETRY_LIMIT,
-    RETRY_WAIT,
-    TIMEOUT_SECONDS,
-)
-
-LOGGER = logging.getLogger(__name__)
-UA = FakeUserAgent()
 
 
-@asynccontextmanager
+def get_default_session(
-async def get_base_session(
+    proxies: AioProxies, headers: dict | None = None
-    headers: dict | None = None,
+) -> TLSSession:
-    cookies: dict | None = None,
+    """Return a TLSSession impersonating a random client fingerprint.
-    connector: aiohttp.TCPConnector | None = None,
-    timeout: aiohttp.ClientTimeout = CLIENT_TIMEOUT,
-    proxies: AioProxies | None = None,
-) -> AsyncGenerator[aiohttp.ClientSession, None]:
-    """Yield an aiohttp session with a random User-Agent and attached proxy pool.
 
-    The proxy pool is stored on the session as proxy_pool for send_request to
+    Fingerprint rotation lives here so every monitor rotates the same way; the
-    rotate per request. The session is always closed on exit.
+    caller owns the session lifetime via ``async with``.
+
+    Cookies are not set here: aioweb has no constructor cookie path, and
+    monitors usually obtain them at runtime, so call session.set_cookie(name,
+    value, domain=...) where they are fetched (e.g. a pre_task_func) instead.
 
     Args:
-        headers: Headers applied to the session; User-Agent is defaulted if
+        proxies: Pool attached to the session as proxy_pool, which send_request
-            absent.
+            rotates through per request. Required so a session can never send
-        cookies: Cookies applied to the session.
+            un-proxied.
-        connector: Optional TCP connector.
+        headers: Default headers merged into every request by aioweb. Use for
-        timeout: Client timeout applied to the session.
+            monitor-wide static headers; per-request headers still go through
-        proxies: Proxy pool attached to the session.
+            send_request. The curl_cffi impersonation already supplies a
+            matching User-Agent, so one is not defaulted here.
 
-    Yields:
+    Returns:
-        The configured aiohttp session.
+        A TLSSession over a curl_cffi backend with a random impersonation, with
+        the proxy pool attached as proxy_pool.
     """
 
-    headers = headers if headers else {}
+    impersonate = random.choice(CLIENT_IDENTIFIERS)
-    headers.setdefault("User-Agent", UA.random)
+    session = TLSSession(backend=CurlCffi(impersonate=impersonate), headers=headers)
-    session = aiohttp.ClientSession(
-        headers=headers, cookies=cookies, connector=connector, timeout=timeout
-    )
     session.proxy_pool = proxies
-    try:
+    return session
-        yield session
-    finally:
-        await session.close()
 
 
-@asynccontextmanager
-async def get_tls_session(
-    headers: dict | None = None,
-    timeout: int = TIMEOUT_SECONDS,
-    proxies: AioProxies | None = None,
-) -> AsyncGenerator[AsyncSession, None]:
-    """Yield a curl_cffi session impersonating a random client fingerprint.
-
-    The proxy pool is stored on the session as proxy_pool for send_request to
-    rotate per request.
-
-    Args:
-        headers: Headers applied to the session; a Chrome User-Agent is
-            defaulted if absent.
-        timeout: Request timeout in seconds.
-        proxies: Proxy pool attached to the session.
-
-    Yields:
-        The configured curl_cffi session.
-    """
-
-    headers = headers if headers else {}
-    headers.setdefault("User-Agent", UA.chrome)
-    async with AsyncSession(
-        impersonate=random.choice(CLIENT_IDENTIFIERS),
-        timeout=timeout,
-        headers=headers,
-    ) as session:
-        session.proxy_pool = proxies
-        yield session
-
-
-def build_session_details(
-    session: aiohttp.ClientSession | AsyncSession, headers: dict = None
-):
-    """Merge headers into an existing session, defaulting a random User-Agent.
-
-    Args:
-        session: Session whose headers are updated in place.
-        headers: Headers to merge; User-Agent is defaulted if absent.
-    """
-
-    headers = headers or {}
-    headers.setdefault("User-Agent", UA.random)
-    session.headers.update(headers)
-
-
-@retry(
-    stop=stop_after_attempt(RETRY_LIMIT),
-    wait=wait_fixed(RETRY_WAIT),
-    retry=retry_if_exception_type(
-        (
-            aiohttp.ClientError,
-            asyncio.TimeoutError,
-            RequestsError,
-            requests.Timeout,
-            CurlTimeout,
-        )
-    ),
-    before=before_log(LOGGER, logging.DEBUG),
-    after=after_log(LOGGER, logging.DEBUG),
-    reraise=True,
-)
 async def send_request(
-    session: aiohttp.ClientSession | AsyncSession,
+    session: TLSSession,
     url: str,
     method: str,
+    *,
     params: dict | None = None,
     json_data: dict | None = None,
     data: dict | None = None,
     headers: dict | None = None,
-    reraise_json_error: bool = False,
+) -> Response:
-    development_proxy: str | None = None,
+    """Send a request through a TLSSession with retries and proxy rotation.
-) -> dict | str:
-    """Send a request through either session type with retries and proxy rotation.
 
-    Retries transport errors and timeouts up to RETRY_LIMIT with a fixed wait.
+    Wraps aioweb's request_with_retries, which retries transport errors,
-    When development_proxy is set it overrides the session's proxy pool and
+    timeouts, and retryable statuses (429/5xx) up to RETRY_LIMIT attempts with
-    disables TLS verification for local traffic inspection; otherwise the next
+    exponential backoff. The next proxy is drawn from the pool per call and
-    proxy is drawn from the session's proxy_pool when present. Responses are
+    applied to both schemes. A non-2xx response or a fully failed request raises
-    parsed as JSON when possible.
+    AiowebError, an expected task error, so a failed fetch surfaces the same way
+    the old HTTPError path did. On success the aioweb Response is returned for
+    the caller to read via response.json(), response.text(), etc.
 
     Args:
-        session: aiohttp or curl_cffi session to send through.
+        session: The TLSSession to send through.
         url: Request URL.
         method: HTTP method.
         params: Query parameters.
-        json_data: JSON body.
+        json_data: JSON body. A dict here is sent as a JSON body.
-        data: Form body.
+        data: Alternative body. Note aioweb routes any dict or list to a JSON
-        headers: Per-request headers merged over the session headers,
+            body, so true form-encoding requires a pre-encoded string; a form
-            avoiding mutation of shared session state from concurrent tasks.
+            dict passed here is sent as JSON.
-        reraise_json_error: Raise JSONDecodeError instead of returning raw
+        headers: Per-request headers merged over the session defaults.
-            text when the response is not valid JSON.
-        development_proxy: Local proxy URL that overrides the pool and
-            disables verification.
 
     Returns:
-        Parsed JSON dict, or raw response text when parsing fails and
+        The aioweb Response, guaranteed 2xx (non-2xx raises). Read the body with
-        reraise_json_error is False.
+        response.json() (dict/list, or None if not valid JSON) or
+        response.text().
 
     Raises:
-        json.JSONDecodeError: When parsing fails and reraise_json_error is
+        AiowebError: On a non-2xx response or after all retries fail.
-            True.
     """
 
-    verify = True
+    proxy_url = session.proxy_pool.next().url()
-    if development_proxy:
+    proxy = {"http": proxy_url, "https": proxy_url}
-        proxy = development_proxy
-        verify = False
-    else:
-        manager = getattr(session, "proxy_pool", None)
-        proxy = manager.next().url() if manager else None
 
-    if isinstance(session, aiohttp.ClientSession):
+    body = json_data if json_data is not None else data
-        async with session.request(
+    response = await session.request_with_retries(
-            method=method,
+        method.upper(),
-            url=url,
+        url,
         params=params,
-            json=json_data,
+        data=body,
-            data=data,
         headers=headers,
-            proxy=proxy,
+        proxies=proxy,
-            ssl=verify,
+        attempts=RETRY_LIMIT,
-        ) as response:
-            response.raise_for_status()
-            text = await response.text()
-    else:
-        response = await session.request(
-            method=method.upper(),
-            url=url,
-            params=params,
-            json=json_data,
-            data=data,
-            headers=headers,
-            proxy=proxy,
-            verify=verify,
     )
     response.raise_for_status()
-        text = response.text
 
-    try:
+    return response
-        return json.loads(text)
-    except json.JSONDecodeError:
-        if reraise_json_error:
-            raise
-
-        return text

tests/test_convert_product_to_mongo.py

@@ -0,0 +1,48 @@
+"""unit test: convert_product_to_mongo flattens a Product to its Mongo document
+
+run from the repo root:
+    uv run pytest tests/test_convert_product_to_mongo.py
+
+pure and offline — no Mongo. Also pins the fact that prices are already floats
+by the time they reach the flattener (pydantic coerces numeric strings on the
+Product), which is why convert_product_to_mongo no longer coerces them itself.
+"""
+
+from datetime import datetime
+
+from monitor_utils.database import convert_product_to_mongo
+from monitor_utils.models import PricingInfo, Product
+
+
+def _product(**pricing) -> Product:
+    return Product(
+        product_id="p1",
+        title="Widget",
+        url="http://x/p1",
+        pricing=PricingInfo(**pricing),
+    )
+
+
+def test_numeric_string_prices_arrive_as_floats():
+    """a numeric-string price is a float in the doc (pydantic coerced it on init)"""
+
+    doc = convert_product_to_mongo(_product(original_price="200", current_price=150.0))
+    assert doc["originalPrice"] == 200.0
+    assert isinstance(doc["originalPrice"], float)
+    assert doc["currentPrice"] == 150.0
+    assert isinstance(doc["currentPrice"], float)
+
+
+def test_lastupdated_is_stamped():
+    """every document gets a UTC lastUpdated timestamp"""
+
+    doc = convert_product_to_mongo(_product(original_price=200.0, current_price=150.0))
+    assert isinstance(doc["lastUpdated"], datetime)
+
+
+def test_core_fields_map_through():
+    """title and url pass straight through to the document"""
+
+    doc = convert_product_to_mongo(_product(original_price=200.0, current_price=150.0))
+    assert doc["title"] == "Widget"
+    assert doc["url"] == "http://x/p1"

tests/test_pagination_semaphore.py

@@ -1,11 +1,12 @@
 """unit test: execute_pagination honours a caller-supplied shared semaphore
 
 run from the repo root:
-    uv run python tests/test_pagination_semaphore.py
+    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 the default per-call path.
+(one instance capping many categories at once) and a single call bounding one
+category with its own semaphore.
 """
 
 import asyncio
@@ -62,9 +63,10 @@ def test_shared_semaphore_caps_total_concurrency_across_categories():
     assert state["peak"] <= limit, f"peak {state['peak']} exceeded shared limit {limit}"
 
 
-def test_default_falls_back_to_max_concurrent_tasks():
+def test_single_call_respects_its_semaphore():
-    """Without a semaphore, max_concurrent_tasks still bounds a single call."""
+    """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)
 
@@ -75,13 +77,8 @@ def test_default_falls_back_to_max_concurrent_tasks():
             products_per_page=30,
             func=func,
             session=None,
-            max_concurrent_tasks=2,
+            semaphore=asyncio.Semaphore(limit),
         )
 
     asyncio.run(run())
-    assert state["peak"] <= 2, f"peak {state['peak']} exceeded max_concurrent_tasks 2"
+    assert state["peak"] <= limit, f"peak {state['peak']} over limit {limit}"
-
-
-test_shared_semaphore_caps_total_concurrency_across_categories()
-test_default_falls_back_to_max_concurrent_tasks()
-print("pagination semaphore test passed")

tests/test_price_before_drop.py

@@ -1,7 +1,7 @@
 """unit test: resolve_price_before_drop detects a drop from the prior stored price
 
 run from the repo root:
-    uv run python tests/test_price_before_drop.py
+    uv run pytest tests/test_price_before_drop.py
 
 pure and offline — no Mongo, no product construction; just the drop rule
 add_products_to_database relies on
@@ -23,8 +23,3 @@ def test_non_drops_return_none():
     assert resolve_price_before_drop(150.0, 150.0) is None  # unchanged
     assert resolve_price_before_drop(150.0, 200.0) is None  # price rose
     assert resolve_price_before_drop(200.0, None) is None  # missing new price
-
-
-test_drop_returns_previous_price()
-test_non_drops_return_none()
-print("price before drop test passed")