Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 41 additions & 0 deletions tests/test_transmitter.py
Original file line number Diff line number Diff line change
Expand Up @@ -182,3 +182,44 @@ def test_send_301_redirect_returns_error_no_retry(self):
resp = t.send({"batch_id": "b-1", "events": []})

assert resp.status == "error"


@pytest.mark.parametrize("status_code", [200, 201, 203, 206])
def test_send_unexpected_status_raises_transmit_error(status_code):
# Regression: any status not explicitly handled must raise TransmitError rather
# than falling through the elif chain and returning None. A None return would
# cause AttributeError in the consumer (response.status), silently killing the
# background thread and freezing the queue.
t = Transmitter(api_key="test-key", host="https://app.wildedge.dev")
with patch.object(
t._opener, "open", return_value=_make_response(status_code, {"status": "ok"})
):
with pytest.raises(TransmitError, match=f"Unexpected HTTP {status_code}"):
t.send({"batch_id": "b-1", "events": []})


def test_unexpected_status_keeps_events_in_consumer_queue():
# End-to-end regression: TransmitError from an unexpected status must cause the
# consumer to retain events for retry, not lose them.
from wildedge.consumer import Consumer
from wildedge.device import DeviceInfo
from wildedge.queue import EventQueue

queue = EventQueue(max_size=100)
queue.add({"event_id": "e1", "event_type": "inference", "model_id": "m"})

mock_transmitter = MagicMock(spec=Transmitter)
mock_transmitter.send.side_effect = TransmitError("Unexpected HTTP 200")

consumer = Consumer(
queue=queue,
transmitter=mock_transmitter,
device=DeviceInfo(device_id="d", device_type="linux"),
get_models=lambda: {},
session_id="sess-1",
)
result = consumer.drain_once()
consumer.stop()

assert result is False
assert queue.length() == 1
25 changes: 12 additions & 13 deletions wildedge/transmitter.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ class IngestResponse:


class TransmitError(Exception):
"""Raised for retryable errors (429 / 5xx / network)."""
"""Raised for retryable errors (429 / 5xx / network / unexpected status)."""


class Transmitter:
Expand Down Expand Up @@ -78,8 +78,7 @@ def send(self, batch: dict) -> IngestResponse:
server_time=data.get("server_time"),
rejected=data.get("rejected"),
)

if status_code == 400:
elif status_code == 400:
logger.warning(
"wildedge: batch rejected (400) - discarding: %s",
raw[: constants.ERROR_MSG_MAX_LEN],
Expand All @@ -90,17 +89,15 @@ def send(self, batch: dict) -> IngestResponse:
events_accepted=0,
events_rejected=len(batch.get("events", [])),
)

if status_code == 401:
elif status_code == 401:
logger.error("wildedge: authentication failed (401) - check your API key")
return IngestResponse(
status="unauthorized",
batch_id=batch.get("batch_id", ""),
events_accepted=0,
events_rejected=len(batch.get("events", [])),
)

if 300 <= status_code < 400:
elif 300 <= status_code < 400:
# Redirects should never occur (we disable redirect following).
# Treat as a permanent config error so we don't loop.
logger.error(
Expand All @@ -114,8 +111,7 @@ def send(self, batch: dict) -> IngestResponse:
events_accepted=0,
events_rejected=len(batch.get("events", [])),
)

if status_code == 404:
elif status_code == 404:
logger.error(
"wildedge: endpoint not found (404) at %s; check WILDEDGE_DSN", url
)
Expand All @@ -125,13 +121,11 @@ def send(self, batch: dict) -> IngestResponse:
events_accepted=0,
events_rejected=len(batch.get("events", [])),
)

if status_code == 429 or status_code >= 500:
elif status_code == 429 or status_code >= 500:
raise TransmitError(
f"HTTP {status_code}: {raw[: constants.ERROR_MSG_MAX_LEN]!r}"
)

if 400 <= status_code < 500:
elif 400 <= status_code < 500:
# Other 4xx (e.g. 422 Unprocessable) are permanent client errors; discard.
logger.warning(
"wildedge: batch rejected (%d) - discarding: %s",
Expand All @@ -144,6 +138,11 @@ def send(self, batch: dict) -> IngestResponse:
events_accepted=0,
events_rejected=len(batch.get("events", [])),
)
else:
# Guard against unexpected status codes that would otherwise return None.
raise TransmitError(
f"Unexpected HTTP {status_code}: {raw[: constants.ERROR_MSG_MAX_LEN]!r}"
)

def close(self) -> None:
pass
Loading