diff --git a/CHANGELOG.md b/CHANGELOG.md index 12683e7..ad0c740 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -71,6 +71,23 @@ All notable changes to the OilPriceAPI Python SDK will be documented in this fil ### Fixed +- **`subscriptions.list()` and `subscriptions.events()` no longer report a + malformed success as "nothing there" (#142), sync and async.** A 200 without + a `data.subscriptions` list returned `[]`, and one without `data.events` / + `data.cursor` returned an empty page with `cursor=None`. Fed back as + `events(since=page.cursor)`, that `None` dropped `since`, and the API reads a + missing `since` as `0`, so the poller replayed the account's whole event + history. Both now raise `OilPriceAPIError(code="MALFORMED_RESPONSE")` with the + raw body when the collection is missing or mistyped, a record is invalid, + `cursor` is not a non-negative integer, `has_more` is not a boolean, or the + cursor is behind `since` or behind an event in the page. A genuinely empty + list or page is still an empty success, and `page.cursor` is now always an + `int`. +- **`subscriptions.events(since=...)` refuses a cursor the API would read as + `0`.** The API parses `since` with `to_i`, so `"abc"`, `""` and `-1` replay + every event and `1.5` becomes `1` (verified against production on + 2026-09-13). Anything but a non-negative `int` or `None` now raises + `ValidationError(field="since", status_code=None)` before a request is sent. - **`subscriptions.create()` no longer turns a malformed success into a half-built record.** It fell back to treating the whole `data` object as the subscription when `data.subscription` was missing, and leaked a raw pydantic diff --git a/oilpriceapi/_subscriptions_common.py b/oilpriceapi/_subscriptions_common.py index 2c96b1a..be2cf55 100644 --- a/oilpriceapi/_subscriptions_common.py +++ b/oilpriceapi/_subscriptions_common.py @@ -14,6 +14,7 @@ if TYPE_CHECKING: from .models import Subscription + from .resources.subscriptions import SubscriptionEventsPage # Default attribution source stamped on subscriptions created via this SDK. DEFAULT_SOURCE = "sdk-python" @@ -251,6 +252,152 @@ def unwrap_subscription(response: Any, *, subject: str) -> "Subscription": ) from error +def _field_errors(error: Any) -> str: + return ", ".join(".".join(str(part) for part in item["loc"]) for item in error.errors()) + + +def unwrap_subscription_list(response: Any, *, subject: str) -> List["Subscription"]: + """Return the typed ``data.subscriptions`` list from a success body. + + ``GET /v1/subscriptions`` always answers + ``{"status": "success", "data": {"subscriptions": [...]}}``. An empty list is + returned only when the API sent an empty list. + + Raises: + OilPriceAPIError: ``code="MALFORMED_RESPONSE"`` when ``data.subscriptions`` + is missing or not a list, or a record in it is not a valid + subscription. A malformed success is never reported as "no + subscriptions". + """ + from pydantic import ValidationError as PydanticValidationError + + from ._fuel_surcharge_common import _malformed + from .models import Subscription + + data = response.get("data") if isinstance(response, dict) else None + records = data.get("subscriptions") if isinstance(data, dict) else None + if not isinstance(records, list): + raise _malformed(subject, "expected data.subscriptions to be a list", response) + + subscriptions: List[Subscription] = [] + for index, record in enumerate(records): + if not isinstance(record, dict): + raise _malformed( + subject, f"expected data.subscriptions[{index}] to be an object", response + ) + try: + subscriptions.append(Subscription(**record)) + except PydanticValidationError as error: + raise _malformed( + subject, + f"data.subscriptions[{index}] has {error.error_count()} invalid or missing " + f"field(s): {_field_errors(error)}", + response, + ) from error + return subscriptions + + +def validate_since(since: Any) -> Optional[int]: + """Return ``since`` if the API will read it as the cursor it is. + + ``GET /v1/subscriptions/events`` reads ``params[:since].to_i``: a blank, + non-numeric or negative value becomes ``0`` and replays every event the + account has, and ``1.5`` becomes ``1``. ``None`` (omitted) is the first + poll; anything else must be a previous page's integer ``cursor``, or ``0``. + + Raises: + ValidationError: ``field="since"``, ``status_code=None``. Nothing is sent. + """ + if since is None: + return None + if isinstance(since, bool) or not isinstance(since, int) or since < 0: + raise _refuse( + f"Invalid events cursor since={since!r}: pass page.cursor from the previous " + f"events() call, 0 to start from the first event, or omit it on the first " + f"poll. The API reads any other value as 0 and replays every event.", + "since", + since, + ) + return since + + +def unwrap_events_page( + response: Any, *, since: Optional[int], subject: str +) -> "SubscriptionEventsPage": + """Return the typed events page from a ``GET /v1/subscriptions/events`` body. + + The API always sends ``data.cursor`` (an integer: the last event's ``seq``, + or ``since`` when there are none), ``data.has_more`` and ``data.events``. + The cursor is what the caller sends as ``since`` next, so a missing or + wrong-typed cursor is refused rather than defaulted: ``cursor=None`` would + make the next poll omit ``since`` and restart from the first event. + + Raises: + OilPriceAPIError: ``code="MALFORMED_RESPONSE"`` when ``data.events`` is + not a list of events, ``data.cursor`` is not a non-negative integer, + ``data.has_more`` is not a boolean, or the cursor is behind ``since`` + or behind an event in the page (following it would replay events). + """ + from pydantic import ValidationError as PydanticValidationError + + from ._fuel_surcharge_common import _malformed + from .models import SubscriptionEvent + from .resources.subscriptions import SubscriptionEventsPage + + data = response.get("data") if isinstance(response, dict) else None + if not isinstance(data, dict): + raise _malformed(subject, "expected a 'data' object", response) + + records = data.get("events") + if not isinstance(records, list): + raise _malformed(subject, "expected data.events to be a list", response) + + cursor = data.get("cursor") + if isinstance(cursor, bool) or not isinstance(cursor, int) or cursor < 0: + raise _malformed( + subject, + f"expected data.cursor to be a non-negative integer, got {cursor!r}", + response, + ) + + has_more = data.get("has_more") + if not isinstance(has_more, bool): + raise _malformed( + subject, f"expected data.has_more to be true or false, got {has_more!r}", response + ) + + if since is not None and cursor < since: + raise _malformed( + subject, + f"data.cursor {cursor} is behind since={since}; following it would replay events", + response, + ) + + events: List[SubscriptionEvent] = [] + for index, record in enumerate(records): + if not isinstance(record, dict): + raise _malformed(subject, f"expected data.events[{index}] to be an object", response) + try: + event = SubscriptionEvent(**record) + except PydanticValidationError as error: + raise _malformed( + subject, + f"data.events[{index}] has {error.error_count()} invalid field(s): " + f"{_field_errors(error)}", + response, + ) from error + if event.seq is not None and event.seq > cursor: + raise _malformed( + subject, + f"data.cursor {cursor} is behind data.events[{index}].seq {event.seq}; " + f"following it would replay events", + response, + ) + events.append(event) + + return SubscriptionEventsPage(events=events, cursor=cursor, has_more=has_more) + + def unwrap_data(response: Any) -> Dict[str, Any]: """Return the ``data`` object from a ``{status, data}`` envelope.""" if isinstance(response, dict) and "data" in response: diff --git a/oilpriceapi/async_resources.py b/oilpriceapi/async_resources.py index bd146d5..817fa8c 100644 --- a/oilpriceapi/async_resources.py +++ b/oilpriceapi/async_resources.py @@ -8,8 +8,10 @@ build_attribution_headers, build_create_body, build_update_body, - unwrap_data, + unwrap_events_page, unwrap_subscription, + unwrap_subscription_list, + validate_since, validate_subscription_id, ) from .exceptions import ValidationError @@ -42,7 +44,6 @@ ParcelFuelSurchargeCarrier, PriceAlert, Subscription, - SubscriptionEvent, ) from .resource_validators import ( VALID_OPERATORS, @@ -1600,11 +1601,13 @@ def __init__(self, client: Any) -> None: self.client = client async def list(self) -> List[Subscription]: - """List all subscriptions for the authenticated user.""" + """List all subscriptions. See ``SubscriptionsResource.list``. + + Raises ``OilPriceAPIError(code="MALFORMED_RESPONSE")`` when a success + body has no ``data.subscriptions`` list; empty only when the API sent []. + """ response = await self.client.request(method="GET", path="/v1/subscriptions") - data = unwrap_data(response) - subs = data.get("subscriptions", []) - return [Subscription(**s) for s in subs] + return unwrap_subscription_list(response, subject="subscriptions.list") async def create( self, @@ -1706,8 +1709,12 @@ async def events( ) -> SubscriptionEventsPage: """Poll for subscription events newer than a cursor. - Returns a SubscriptionEventsPage with events, cursor, and has_more. + See ``SubscriptionsResource.events``. ``since`` must be a non-negative + int (``ValidationError``, nothing sent, otherwise), and a success body + without an integer cursor raises ``MALFORMED_RESPONSE`` rather than + returning ``cursor=None``, which would restart polling from event 1. """ + since = validate_since(since) params: Dict[str, Any] = {} if since is not None: params["since"] = since @@ -1721,11 +1728,7 @@ async def events( path="/v1/subscriptions/events", params=params, ) - data = unwrap_data(response) - events = [SubscriptionEvent(**e) for e in data.get("events", [])] - cursor = data.get("cursor") - has_more = bool(data.get("has_more", False)) - return SubscriptionEventsPage(events=events, cursor=cursor, has_more=has_more) + return unwrap_events_page(response, since=since, subject="subscriptions.events") class AsyncFuelSurchargeResource: diff --git a/oilpriceapi/resources/subscriptions.py b/oilpriceapi/resources/subscriptions.py index eca4316..c60d4c7 100644 --- a/oilpriceapi/resources/subscriptions.py +++ b/oilpriceapi/resources/subscriptions.py @@ -5,14 +5,16 @@ periodically evaluate commodity codes and emit events an agent can poll for. """ -from typing import Any, Dict, List, Optional, Union, cast +from typing import Any, Dict, List, Optional, Union from .._subscriptions_common import ( build_attribution_headers, build_create_body, build_update_body, - unwrap_data, + unwrap_events_page, unwrap_subscription, + unwrap_subscription_list, + validate_since, validate_subscription_id, ) from ..models import Subscription, SubscriptionEvent @@ -57,16 +59,19 @@ def list(self) -> List[Subscription]: """List all subscriptions for the authenticated user. Returns: - List of Subscription models. + List of Subscription models. Empty only when the API sent an empty + list. + + Raises: + OilPriceAPIError: ``code="MALFORMED_RESPONSE"`` when a success body + has no ``data.subscriptions`` list or a record in it is invalid. Example: >>> for sub in client.subscriptions.list(): ... print(sub.name, sub.codes) """ response = self.client.request(method="GET", path="/v1/subscriptions") - data = unwrap_data(response) - subs = data.get("subscriptions", []) - return [Subscription(**s) for s in subs] + return unwrap_subscription_list(response, subject="subscriptions.list") def create( self, @@ -254,19 +259,32 @@ def events( """Poll for subscription events newer than a cursor. Args: - since: Sequence cursor; only events with seq > since are returned. + since: ``page.cursor`` from the previous call; only events with + ``seq > since`` are returned. Omit it (or pass ``0``) only to + start from the first event. Must be a non-negative ``int``: the + API reads any other value as ``0`` and replays every event. limit: Max events to return (server clamps to its own max). watch_id: Restrict to a single subscription. Returns: - A SubscriptionEventsPage with events, cursor, and has_more. + A SubscriptionEventsPage with events, cursor, and has_more. The + cursor is always an ``int``, so ``events(since=page.cursor)`` never + restarts from the beginning. + + Raises: + ValidationError: ``field="since"``, ``status_code=None``, if ``since`` + is not a non-negative int. Nothing is sent. + OilPriceAPIError: ``code="MALFORMED_RESPONSE"`` when a success body + lacks the ``events`` list, an integer ``cursor`` or a boolean + ``has_more``, or its cursor would move polling backwards. Example: >>> page = client.subscriptions.events(since=0) >>> for event in page: - ... print(event.type, event.code) + ... print(event.seq, event.watch_id) >>> next_page = client.subscriptions.events(since=page.cursor) """ + since = validate_since(since) params: Dict[str, Any] = {} if since is not None: params["since"] = since @@ -280,8 +298,4 @@ def events( path="/v1/subscriptions/events", params=params, ) - data = unwrap_data(response) - events = [SubscriptionEvent(**e) for e in data.get("events", [])] - cursor = cast(Optional[int], data.get("cursor")) - has_more = bool(data.get("has_more", False)) - return SubscriptionEventsPage(events=events, cursor=cursor, has_more=has_more) + return unwrap_events_page(response, since=since, subject="subscriptions.events") diff --git a/tests/unit/test_subscriptions_list_events_strict.py b/tests/unit/test_subscriptions_list_events_strict.py new file mode 100644 index 0000000..e5f3662 --- /dev/null +++ b/tests/unit/test_subscriptions_list_events_strict.py @@ -0,0 +1,274 @@ +"""#142 -- ``subscriptions.list()`` / ``events()`` never turn a malformed 200 into success. + +``V1::SubscriptionsController`` (oilpriceapi-api origin/main, 2026-09-13): + +* ``#index`` always renders ``{"status": "success", "data": {"subscriptions": [...]}}``. +* ``#events`` always renders ``{"cursor", "has_more", "events"}`` under ``data``, + with ``cursor = events.last&.seq || since``, so the cursor is an integer that + never moves backwards. +* ``#events`` reads ``since = params[:since].to_i``. Missing, ``""``, ``"abc"`` + and ``-1`` all become ``0`` and replay the account's whole event history; + ``1.5`` becomes ``1``. Confirmed live on 2026-09-13 against an account with + 101 events: every one of those returned ``first_seq=1, cursor=100``. + +Before this fix a body without those keys returned ``[]`` / an empty page with +``cursor=None``, and the documented loop ``events(since=page.cursor)`` then sent +no ``since`` at all -- restarting the poller from event 1. + +Every test drives the REAL sync and async clients over a mocked transport. +""" + +import asyncio +import json +from unittest.mock import AsyncMock, Mock, patch + +import pytest + +from oilpriceapi import AsyncOilPriceAPI, OilPriceAPI, Subscription, SubscriptionEventsPage +from oilpriceapi.exceptions import OilPriceAPIError, ValidationError + +# Not a credential: a fixture string. Every request here is mocked. +FIXTURE_KEY = "-".join(["fixture", "not", "a", "real", "key"]) + +MODES = ["sync", "async"] + +# live: GET /v1/subscriptions, 2026-09-13 +WIRE_SUBSCRIPTION = { + "id": "b84b24a0-2b28-4eac-835e-db92bab5c0cb", + "name": "mcp-live-contract-1788524538194", + "codes": ["BRENT_CRUDE_USD"], + "interval_seconds": 3600, + "status": "active", + "deliver_webhook": False, + "source": "api", + "tool_name": "opa_create_price_subscription", + "last_evaluated_at": "2026-09-13T20:32:18Z", + "next_run_at": "2026-09-13T21:32:18Z", + "created_at": "2026-09-04T12:22:19Z", +} + +# live: GET /v1/subscriptions/events?since=99&limit=2, 2026-09-13 +WIRE_EVENTS_PAGE = { + "status": "success", + "data": { + "cursor": 101, + "has_more": True, + "events": [ + { + "id": "d73c7b25-875b-403e-a429-0d7049245c30", + "seq": 100, + "watch_id": "b84b24a0-2b28-4eac-835e-db92bab5c0cb", + "observed_at": "2026-09-08T16:20:42Z", + "snapshot": {"BRENT_CRUDE_USD": {"as_of": "2026-09-08T16:16:14Z", "price": 97.27, "currency": "USD", "change_24h_pct": -0.04}}, + "deltas": {"BRENT_CRUDE_USD": {"pct_change": -0.53, "price_change": -0.52}}, + "source": "api", + "tool_name": "opa_create_price_subscription", + }, + { + "id": "a835a930-f34f-4001-80b1-38fb3cde3797", + "seq": 101, + "watch_id": "b84b24a0-2b28-4eac-835e-db92bab5c0cb", + "observed_at": "2026-09-08T17:21:19Z", + "snapshot": {"BRENT_CRUDE_USD": {"as_of": "2026-09-08T17:20:37Z", "price": 97.02, "currency": "USD", "change_24h_pct": -0.08}}, + "deltas": {"BRENT_CRUDE_USD": {"pct_change": -0.26, "price_change": -0.25}}, + "source": "api", + "tool_name": "opa_create_price_subscription", + }, + ], + }, +} + + +def _response(payload): + response = Mock() + response.status_code = 200 + response.headers = {} + text = json.dumps(payload) + response.content = text.encode() + response.text = text + response.json.return_value = payload + return response + + +def _run(mode, call, payload): + """Run ``call(client)`` over a mocked transport; return ``(result, transport)``.""" + if mode == "sync": + client = OilPriceAPI(api_key=FIXTURE_KEY, max_retries=1) + with patch("httpx.Client.request", return_value=_response(payload)) as transport: + return call(client), transport + + async def go(): + client = AsyncOilPriceAPI(api_key=FIXTURE_KEY, max_retries=1) + with patch("httpx.AsyncClient.request", new=AsyncMock(return_value=_response(payload))) as transport: + return await call(client), transport + + return asyncio.run(go()) + + +def _raises(mode, call, payload, exc_type=OilPriceAPIError): + with pytest.raises(exc_type) as info: + _run(mode, call, payload) + return info.value + + +def _list(client): + return client.subscriptions.list() + + +def _events(**kwargs): + return lambda client: client.subscriptions.events(**kwargs) + + +def _page(**data): + return {"status": "success", "data": data} + + +# --- list() -------------------------------------------------------------------- + +MALFORMED_LIST_BODIES = { + "data without subscriptions": {"status": "success", "data": {}}, + "data is a list": {"status": "success", "data": []}, + "no data at all": {"message": "maintenance"}, + "subscriptions is null": {"status": "success", "data": {"subscriptions": None}}, + "subscriptions is an object": {"status": "success", "data": {"subscriptions": {}}}, + "a record is not an object": {"status": "success", "data": {"subscriptions": ["b84b24a0"]}}, + "a record lacks id": { + "status": "success", + "data": {"subscriptions": [{k: v for k, v in WIRE_SUBSCRIPTION.items() if k != "id"}]}, + }, +} + + +@pytest.mark.parametrize("mode", MODES) +@pytest.mark.parametrize("case", sorted(MALFORMED_LIST_BODIES)) +def test_list_malformed_success_raises(mode, case): + payload = MALFORMED_LIST_BODIES[case] + + error = _raises(mode, _list, payload) + + assert error.code == "MALFORMED_RESPONSE" + assert error.raw_body == payload + assert "subscriptions.list" in error.message + + +@pytest.mark.parametrize("mode", MODES) +def test_list_genuinely_empty_is_an_empty_success(mode): + result, _ = _run(mode, _list, {"status": "success", "data": {"subscriptions": []}}) + + assert result == [] + + +@pytest.mark.parametrize("mode", MODES) +def test_list_live_body_parses(mode): + result, _ = _run(mode, _list, {"status": "success", "data": {"subscriptions": [WIRE_SUBSCRIPTION]}}) + + assert len(result) == 1 + assert isinstance(result[0], Subscription) + assert result[0].id == WIRE_SUBSCRIPTION["id"] + assert result[0].codes == ["BRENT_CRUDE_USD"] + + +# --- events(): the response ---------------------------------------------------- + +MALFORMED_EVENT_BODIES = { + "data without events or cursor": {"status": "success", "data": {}}, + "empty object": {}, + "events missing": _page(cursor=41, has_more=False), + "events is an object": _page(cursor=41, has_more=False, events={}), + "an event is not an object": _page(cursor=42, has_more=False, events=[42]), + "cursor missing": _page(has_more=False, events=[]), + "cursor is null": _page(cursor=None, has_more=False, events=[]), + "cursor is a string": _page(cursor="41", has_more=False, events=[]), + "cursor is a float": _page(cursor=41.5, has_more=False, events=[]), + "cursor is a bool": _page(cursor=True, has_more=False, events=[]), + "cursor is negative": _page(cursor=-1, has_more=False, events=[]), + "has_more missing": _page(cursor=41, events=[]), + "has_more is a string": _page(cursor=41, has_more="false", events=[]), + "cursor behind since": _page(cursor=0, has_more=False, events=[]), + "cursor behind the last event": _page(cursor=41, has_more=False, events=[{"seq": 43}]), +} + + +@pytest.mark.parametrize("mode", MODES) +@pytest.mark.parametrize("case", sorted(MALFORMED_EVENT_BODIES)) +def test_events_malformed_success_raises(mode, case): + payload = MALFORMED_EVENT_BODIES[case] + + error = _raises(mode, _events(since=41), payload) + + assert error.code == "MALFORMED_RESPONSE" + assert error.raw_body == payload + assert "subscriptions.events" in error.message + + +@pytest.mark.parametrize("mode", MODES) +def test_events_empty_page_keeps_the_cursor(mode): + page, _ = _run(mode, _events(since=41), _page(cursor=41, has_more=False, events=[])) + + assert isinstance(page, SubscriptionEventsPage) + assert len(page) == 0 + assert page.cursor == 41 + assert page.has_more is False + + +@pytest.mark.parametrize("mode", MODES) +def test_events_live_body_parses_and_advances(mode): + page, transport = _run(mode, _events(since=99, limit=2), WIRE_EVENTS_PAGE) + + assert page.cursor == 101 + assert page.has_more is True + assert [event.seq for event in page] == [100, 101] + assert page.events[0].watch_id == "b84b24a0-2b28-4eac-835e-db92bab5c0cb" + assert transport.call_args.kwargs["params"] == {"since": 99, "limit": 2} + + +@pytest.mark.parametrize("mode", MODES) +def test_first_poll_without_since_is_allowed(mode): + page, transport = _run(mode, _events(), _page(cursor=0, has_more=False, events=[])) + + assert page.cursor == 0 + assert "since" not in (transport.call_args.kwargs.get("params") or {}) + + +# --- events(): the cursor the caller sends -------------------------------------- + + +@pytest.mark.parametrize("mode", MODES) +@pytest.mark.parametrize("since", ["abc", "", "41", -1, 1.5, True, [41]], ids=repr) +def test_events_refuses_a_cursor_the_api_would_read_as_zero(mode, since): + error = _raises(mode, _events(since=since), _page(cursor=0, has_more=False, events=[]), ValidationError) + + assert error.status_code is None + assert error.field == "since" + assert error.value == since + + +@pytest.mark.parametrize("mode", MODES) +def test_refused_cursor_sends_nothing(mode): + if mode == "sync": + client = OilPriceAPI(api_key=FIXTURE_KEY, max_retries=1) + with patch("httpx.Client.request") as transport: + with pytest.raises(ValidationError): + client.subscriptions.events(since="abc") + assert transport.call_count == 0 + return + + async def go(): + client = AsyncOilPriceAPI(api_key=FIXTURE_KEY, max_retries=1) + with patch("httpx.AsyncClient.request", new=AsyncMock()) as transport: + with pytest.raises(ValidationError): + await client.subscriptions.events(since="abc") + assert transport.call_count == 0 + + asyncio.run(go()) + + +@pytest.mark.parametrize("mode", MODES) +def test_poll_loop_never_resends_from_the_start(mode): + """The documented loop, fed a malformed page, stops instead of replaying.""" + first, _ = _run(mode, _events(since=99, limit=2), WIRE_EVENTS_PAGE) + assert first.cursor == 101 + + error = _raises(mode, _events(since=first.cursor), {"status": "success", "data": {}}) + + assert error.code == "MALFORMED_RESPONSE"