fix(parser): pass non-JSON octet-stream payloads to the model in EventBridgeV2Envelope

aws_lambda_powertools/utilities/parser/envelopes/event_bridge_v2.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 import logging
 from typing import TYPE_CHECKING, Any
 
@@ -16,6 +17,20 @@ def _is_with_metadata(item: Any) -> bool:
     return isinstance(item, dict) and "Data" in item and "SystemMetadata" in item
 
 
+def _decode_octet_stream(data: bytes) -> Any:
+    """Decode an application/octet-stream payload.
+
+    JSON payloads are loaded so they can be parsed into a model. Anything else (binary or plain text)
+    is passed through as bytes, so a `bytes` or `str` model can receive it and any other model raises
+    a ValidationError instead of a UnicodeDecodeError or a JSON error.
+    """
+    try:
+        return json.loads(data)
+    except ValueError:  # JSONDecodeError and UnicodeDecodeError are both ValueError
+        logger.debug("application/octet-stream payload is not JSON, passing it to the model as bytes")
+        return data
+
+
 def _unwrap_put_events(data: Any) -> Any:
     if isinstance(data, dict) and "detail-type" in data and "detail" in data:
         return data["detail"]
@@ -54,9 +69,9 @@ def parse(self, data: dict[str, Any] | Any | None, model: type[T]) -> list[T | N
             else:
                 payload = item
             if isinstance(payload, bytes):
-                # application/octet-stream is delivered as bytes; let the JSON path of _parse handle it
-                payload = payload.decode()
-            payloads.append(_unwrap_put_events(payload))
+                payloads.append(_decode_octet_stream(payload))
+            else:
+                payloads.append(_unwrap_put_events(payload))
 
         logger.debug(f"Parsing EventBridge V2 payloads with {model}")
         return [self._parse(data=payload, model=model) for payload in payloads]

tests/unit/parser/_pydantic/test_eventbridge_v2.py

@@ -102,3 +102,39 @@ def test_envelope_parses_octet_stream_json_bytes():
 
     # THEN the bytes are decoded and parsed into the model, and the optional aws: fields may be absent
     assert orders == [Order(orderId="o-1", total=5)]
+
+
+def _octet_stream_event(payload: bytes) -> dict:
+    return {
+        "Data": base64.b64encode(payload).decode(),
+        "Metadata": {},
+        "SystemMetadata": {"ContentType": "application/octet-stream", "aws:DeliveryType": "LIVE"},
+    }
+
+
+@pytest.mark.parametrize(
+    "payload",
+    [
+        pytest.param(b"\xff\xd8\xff\xe0JFIF", id="binary"),
+        pytest.param(b"hello world", id="plain_text"),
+    ],
+)
+def test_envelope_parses_non_json_octet_stream_into_bytes_model(payload):
+    # GIVEN an application/octet-stream payload that is not JSON
+    event = _octet_stream_event(payload)
+
+    # WHEN parsing it with a bytes model
+    result = parse(event=event, model=bytes, envelope=envelopes.EventBridgeV2Envelope)
+
+    # THEN the raw bytes are returned, matching the data class `data` property
+    assert result == [payload]
+
+
+def test_envelope_raises_validation_error_for_binary_octet_stream_with_object_model():
+    # GIVEN a binary application/octet-stream payload that is not valid UTF-8
+    event = _octet_stream_event(b"\xff\xd8\xff\xe0JFIF")
+
+    # WHEN parsing it with an object model
+    # THEN a ValidationError is raised instead of a UnicodeDecodeError
+    with pytest.raises(ValidationError):
+        parse(event=event, model=Order, envelope=envelopes.EventBridgeV2Envelope)