diff --git a/py_gasbuddy/parsers.py b/py_gasbuddy/parsers.py
index 85ef4ad..5e7a467 100644
--- a/py_gasbuddy/parsers.py
+++ b/py_gasbuddy/parsers.py
@@ -25,11 +25,18 @@ def parse_distance(value: Any) -> float | None:
return None
-def build_discount_map(offers: list[dict[str, Any]]) -> dict[str, float]:
+def build_discount_map(offers: list[Any]) -> dict[str, float]:
"""Return fuel product key to total pwgbDiscount mapping across all offers."""
discount_map: dict[str, float] = {}
for offer in offers:
- for disc in offer.get("discounts") or []:
+ if not isinstance(offer, dict):
+ continue
+ discounts = offer.get("discounts")
+ if not isinstance(discounts, list):
+ continue
+ for disc in discounts:
+ if not isinstance(disc, dict):
+ continue
raw = disc.get("pwgbDiscount")
if raw is None:
continue
@@ -37,7 +44,12 @@ def build_discount_map(offers: list[dict[str, Any]]) -> dict[str, float]:
amount = float(raw)
except (ValueError, TypeError):
continue
- for grade in disc.get("grades") or []:
+ grades = disc.get("grades")
+ if not isinstance(grades, list):
+ continue
+ for grade in grades:
+ if not isinstance(grade, str):
+ continue
discount_map[grade] = discount_map.get(grade, 0.0) + amount
return discount_map
@@ -46,8 +58,12 @@ def format_price_node(
price_node: dict[str, Any], deal_discount: float | None = None
) -> PriceNode:
"""Format a single price node."""
- credit_data = price_node.get("credit") or {}
- cash_data = price_node.get("cash") or {}
+ credit_data = price_node.get("credit")
+ if not isinstance(credit_data, dict):
+ credit_data = {}
+ cash_data = price_node.get("cash")
+ if not isinstance(cash_data, dict):
+ cash_data = {}
credit_price = credit_data.get("price", 0)
cash_price = cash_data.get("price", 0)
@@ -67,15 +83,39 @@ def format_price_node(
)
+def _stations_block(response: dict[str, Any]) -> dict[str, Any]:
+ """Return the ``stations`` dict from a locationBySearchTerm response.
+
+ Defensive against the GraphQL spec's allowance of partial responses
+ (``data: null``, missing keys, ``stations`` being null or a non-dict).
+ Returns an empty dict when the expected shape isn't present so callers
+ can downgrade to "no results" rather than crash.
+ """
+ data = response.get("data")
+ if not isinstance(data, dict):
+ return {}
+ location = data.get("locationBySearchTerm")
+ if not isinstance(location, dict):
+ return {}
+ stations = location.get("stations")
+ return stations if isinstance(stations, dict) else {}
+
+
def parse_cursor(response: dict[str, Any]) -> str | None:
"""Extract the next-page cursor from a locationBySearchTerm response."""
- stations = response["data"]["locationBySearchTerm"]["stations"]
- return (stations.get("cursor") or {}).get("next")
+ stations = _stations_block(response)
+ cursor = stations.get("cursor")
+ if not isinstance(cursor, dict):
+ return None
+ return cursor.get("next")
def parse_location_results(response: dict[str, Any]) -> LocationSearchResult:
"""Parse location search results into a LocationSearchResult."""
- stations = response["data"]["locationBySearchTerm"]["stations"]
+ stations = _stations_block(response)
+ raw_results = stations.get("results") or []
+ if not isinstance(raw_results, list):
+ raw_results = []
results = [
cast(
StationSummary,
@@ -91,7 +131,8 @@ def parse_location_results(response: dict[str, Any]) -> LocationSearchResult:
"price_unit": r.get("priceUnit"),
},
)
- for r in stations["results"]
+ for r in raw_results
+ if isinstance(r, dict) and "id" in r
]
return cast(
LocationSearchResult,
@@ -102,7 +143,15 @@ def parse_location_results(response: dict[str, Any]) -> LocationSearchResult:
def parse_results(response: dict[str, Any], limit: int) -> list[StationPrice]:
"""Parse price-service API results into a StationPrice list."""
result_list: list[StationPrice] = []
- results = response["data"]["locationBySearchTerm"]["stations"]["results"]
+ raw_results = _stations_block(response).get("results") or []
+ if not isinstance(raw_results, list):
+ raw_results = []
+ required_station_keys = {"id", "priceUnit", "currency", "latitude", "longitude"}
+ results = [
+ r
+ for r in raw_results
+ if isinstance(r, dict) and required_station_keys.issubset(r)
+ ]
for result in results[:limit]:
raw: dict[str, Any] = {
@@ -128,21 +177,26 @@ def parse_results(response: dict[str, Any], limit: int) -> list[StationPrice]:
"phone": result.get("phone") or None,
}
pay_status_obj = result.get("payStatus")
- is_pay_available = (pay_status_obj is None) or bool(
- (pay_status_obj or {}).get("isPayAvailable", False)
+ is_pay_available = (pay_status_obj is None) or (
+ isinstance(pay_status_obj, dict)
+ and bool(pay_status_obj.get("isPayAvailable", False))
)
raw["pay_status"] = is_pay_available
- offers = result.get("offers") or []
+ offers = [o for o in (result.get("offers") or []) if isinstance(o, dict)]
discount_map = build_discount_map(offers) if is_pay_available else {}
for price in result.get("prices") or []:
+ if not isinstance(price, dict) or "fuelProduct" not in price:
+ continue
fuel_key = price["fuelProduct"]
raw[fuel_key] = format_price_node(price, discount_map.get(fuel_key))
result_list.append(cast(StationPrice, raw))
return result_list
-def parse_ev_stations(stations_data: list[dict[str, Any]]) -> list[EvStation]:
+def parse_ev_stations(stations_data: Any) -> list[EvStation]:
"""Parse raw EV station dicts into EvStation list."""
+ if not isinstance(stations_data, list):
+ return []
return [
cast(
EvStation,
@@ -179,16 +233,30 @@ def parse_ev_stations(stations_data: list[dict[str, Any]]) -> list[EvStation]:
},
)
for s in stations_data
+ if isinstance(s, dict) and "id" in s
]
def parse_trends(response: dict[str, Any]) -> list[TrendData]:
"""Parse price-service API results into a TrendData list."""
+ data = response.get("data")
+ if not isinstance(data, dict):
+ return []
+ location = data.get("locationBySearchTerm")
+ if not isinstance(location, dict):
+ return []
+ trends = location.get("trends") or []
+ if not isinstance(trends, list):
+ return []
return [
TrendData(
average_price=trend["today"],
lowest_price=trend["todayLow"],
area=trend["areaName"],
)
- for trend in response["data"]["locationBySearchTerm"]["trends"]
+ for trend in trends
+ if isinstance(trend, dict)
+ and "today" in trend
+ and "todayLow" in trend
+ and "areaName" in trend
]
diff --git a/tests/test_init.py b/tests/test_init.py
index 00af1b1..133c7d7 100644
--- a/tests/test_init.py
+++ b/tests/test_init.py
@@ -338,7 +338,10 @@ async def test_retry_logic(mock_aioclient, caplog):
)
# Patch asyncio.sleep used by backoff so the test doesn't actually
# wait through the exponential delay between retries.
- with caplog.at_level(logging.DEBUG), patch("backoff._async.asyncio.sleep", new=AsyncMock()):
+ with (
+ caplog.at_level(logging.DEBUG),
+ patch("backoff._async.asyncio.sleep", new=AsyncMock()),
+ ):
with pytest.raises(py_gasbuddy.LibraryError):
manager = py_gasbuddy.GasBuddy(station_id=205033)
await manager.price_lookup()
@@ -358,11 +361,14 @@ async def test_retry_succeeds_on_second_attempt(mock_aioclient, caplog):
mock_aioclient.post(
TEST_URL,
status=403,
- body='Just a moment...',
+ body="Just a moment...",
)
mock_aioclient.post(TEST_URL, status=200, body=load_fixture("station.json"))
- with caplog.at_level(logging.DEBUG), patch("backoff._async.asyncio.sleep", new=AsyncMock()):
+ with (
+ caplog.at_level(logging.DEBUG),
+ patch("backoff._async.asyncio.sleep", new=AsyncMock()),
+ ):
manager = py_gasbuddy.GasBuddy(station_id=205033)
data = await manager.price_lookup()
diff --git a/tests/test_parsers.py b/tests/test_parsers.py
new file mode 100644
index 0000000..7aa09d2
--- /dev/null
+++ b/tests/test_parsers.py
@@ -0,0 +1,253 @@
+"""Tests for defensive parser behavior on malformed GraphQL payloads."""
+
+from py_gasbuddy.parsers import (
+ parse_cursor,
+ parse_location_results,
+ parse_results,
+ parse_trends,
+)
+
+
+def test_parse_cursor_data_null() -> None:
+ """parse_cursor returns None when data is null."""
+ assert parse_cursor({"data": None}) is None
+
+
+def test_parse_cursor_location_null() -> None:
+ """parse_cursor returns None when locationBySearchTerm is null."""
+ assert parse_cursor({"data": {"locationBySearchTerm": None}}) is None
+
+
+def test_parse_cursor_stations_not_dict() -> None:
+ """parse_cursor returns None when stations is a list (not a dict)."""
+ payload = {"data": {"locationBySearchTerm": {"stations": []}}}
+ assert parse_cursor(payload) is None
+
+
+def test_parse_cursor_missing_cursor_key() -> None:
+ """parse_cursor returns None when there is no cursor key."""
+ payload = {"data": {"locationBySearchTerm": {"stations": {"results": []}}}}
+ assert parse_cursor(payload) is None
+
+
+def test_parse_cursor_cursor_not_dict() -> None:
+ """parse_cursor returns None when cursor is not a dict."""
+ payload = {"data": {"locationBySearchTerm": {"stations": {"cursor": None}}}}
+ assert parse_cursor(payload) is None
+
+
+def test_parse_location_results_empty() -> None:
+ """parse_location_results returns an empty list when data is null."""
+ result = parse_location_results({"data": None})
+ assert result == {"results": [], "next_cursor": None}
+
+
+def test_parse_location_results_results_not_list() -> None:
+ """parse_location_results tolerates results not being a list."""
+ payload = {"data": {"locationBySearchTerm": {"stations": {"results": None}}}}
+ result = parse_location_results(payload)
+ assert result == {"results": [], "next_cursor": None}
+
+
+def test_parse_location_results_skips_malformed_entries() -> None:
+ """Entries without an id are skipped, not raised."""
+ payload = {
+ "data": {
+ "locationBySearchTerm": {
+ "stations": {
+ "results": [
+ {"id": "1", "name": "Good"},
+ {"name": "no id"},
+ None,
+ ]
+ }
+ }
+ }
+ }
+ result = parse_location_results(payload)
+ assert len(result["results"]) == 1
+ assert result["results"][0]["station_id"] == "1"
+
+
+def test_parse_results_empty() -> None:
+ """parse_results returns an empty list when data is null."""
+ assert parse_results({"data": None}, limit=5) == []
+
+
+def test_parse_results_results_not_list() -> None:
+ """parse_results tolerates results not being a list."""
+ payload = {"data": {"locationBySearchTerm": {"stations": {"results": "oops"}}}}
+ assert parse_results(payload, limit=5) == []
+
+
+def test_parse_trends_empty() -> None:
+ """parse_trends returns an empty list when data is null."""
+ assert parse_trends({"data": None}) == []
+
+
+def test_parse_trends_trends_not_list() -> None:
+ """parse_trends tolerates trends not being a list."""
+ payload = {"data": {"locationBySearchTerm": {"trends": None}}}
+ assert parse_trends(payload) == []
+
+
+def test_parse_trends_skips_malformed_entries() -> None:
+ """Trend entries missing required keys are skipped."""
+ payload = {
+ "data": {
+ "locationBySearchTerm": {
+ "trends": [
+ {"today": 3.5, "todayLow": 3.2, "areaName": "A"},
+ {"today": 3.5}, # missing keys
+ ]
+ }
+ }
+ }
+ result = parse_trends(payload)
+ assert len(result) == 1
+ assert result[0]["area"] == "A"
+
+
+def test_parse_location_results_results_truthy_non_list() -> None:
+ """A truthy non-list 'results' is coerced to an empty list (not just None)."""
+ payload = {"data": {"locationBySearchTerm": {"stations": {"results": "oops"}}}}
+ result = parse_location_results(payload)
+ assert result["results"] == []
+
+
+def test_parse_trends_truthy_non_list() -> None:
+ """A truthy non-list 'trends' returns an empty list (not just None)."""
+ payload = {"data": {"locationBySearchTerm": {"trends": "oops"}}}
+ assert parse_trends(payload) == []
+
+
+def test_parse_results_skips_partially_malformed_entries() -> None:
+ """parse_results skips entries that lack required keys, or contain non-dict nested lists."""
+ payload = {
+ "data": {
+ "locationBySearchTerm": {
+ "stations": {
+ "results": [
+ # 1. Valid station
+ {
+ "id": "1",
+ "priceUnit": "gallon",
+ "currency": "USD",
+ "latitude": 45.0,
+ "longitude": -90.0,
+ "name": "Good Station",
+ "prices": [
+ {
+ "fuelProduct": "regular",
+ "credit": {"price": 3.10},
+ }
+ ],
+ },
+ # 2. Missing required key (priceUnit)
+ {
+ "id": "2",
+ "currency": "USD",
+ "latitude": 45.0,
+ "longitude": -90.0,
+ "name": "Bad Station 1",
+ },
+ # 3. Non-dict prices
+ {
+ "id": "3",
+ "priceUnit": "gallon",
+ "currency": "USD",
+ "latitude": 45.0,
+ "longitude": -90.0,
+ "name": "Bad Station 2",
+ "prices": ["not-a-dict"],
+ },
+ # 4. None/Null entry
+ None,
+ ]
+ }
+ }
+ }
+ }
+ results = parse_results(payload, limit=5)
+ assert len(results) == 2 # Good Station + Bad Station 2
+ assert results[0]["station_id"] == "1"
+ assert "regular" in results[0]
+ assert results[1]["station_id"] == "3"
+ assert "regular" not in results[1] # skipped the malformed price
+
+
+def test_parsers_truthy_non_dict_guards() -> None:
+ """_stations_block and parse_trends return empty values when data or locationBySearchTerm is a truthy non-dict."""
+ # 1. data is a string "oops"
+ payload_str_data = {"data": "oops"}
+ assert parse_cursor(payload_str_data) is None
+ assert parse_location_results(payload_str_data) == {
+ "results": [],
+ "next_cursor": None,
+ }
+ assert parse_results(payload_str_data, limit=5) == []
+ assert parse_trends(payload_str_data) == []
+
+ # 2. locationBySearchTerm is a boolean True
+ payload_bool_loc = {"data": {"locationBySearchTerm": True}}
+ assert parse_cursor(payload_bool_loc) is None
+ assert parse_location_results(payload_bool_loc) == {
+ "results": [],
+ "next_cursor": None,
+ }
+ assert parse_results(payload_bool_loc, limit=5) == []
+ assert parse_trends(payload_bool_loc) == []
+
+
+def test_build_discount_map_non_string_grades() -> None:
+ """build_discount_map ignores non-string grade items gracefully."""
+ from py_gasbuddy.parsers import build_discount_map
+
+ offers = [
+ {"discounts": [{"pwgbDiscount": "0.05", "grades": ["regular", 123, None, {}]}]}
+ ]
+ discount_map = build_discount_map(offers)
+ assert discount_map == {"regular": 0.05}
+
+
+def test_build_discount_map_defensive_coverage() -> None:
+ """build_discount_map covers defensive type check branches."""
+ from py_gasbuddy.parsers import build_discount_map
+
+ offers = [
+ # 1. Non-dict offer (line 33)
+ "not-a-dict",
+ # 2. Non-list discounts (line 36)
+ {"discounts": "not-a-list"},
+ # 3. Non-dict disc in discounts (line 39)
+ {"discounts": ["not-a-dict-disc"]},
+ # 4. Non-list grades in disc (line 49)
+ {
+ "discounts": [
+ {
+ "pwgbDiscount": "0.05",
+ "grades": "not-a-list-grades",
+ }
+ ]
+ },
+ # 5. Valid offer to verify it keeps working
+ {
+ "discounts": [
+ {
+ "pwgbDiscount": "0.10",
+ "grades": ["regular"],
+ }
+ ]
+ },
+ ]
+
+ result = build_discount_map(offers)
+ assert result == {"regular": 0.10}
+
+
+def test_parse_ev_stations_non_list_coverage() -> None:
+ """parse_ev_stations returns empty list when stations_data is not a list (line 199)."""
+ from py_gasbuddy.parsers import parse_ev_stations
+
+ assert parse_ev_stations("not-a-list") == []
+ assert parse_ev_stations(None) == []