From 034815146af072d92c64a2f3bf39ef88c60e1eb7 Mon Sep 17 00:00:00 2001 From: jonathan343 Date: Sun, 18 Oct 2026 17:11:21 -0400 Subject: [PATCH 1/5] Compute Content-Length for seekable streaming payloads The HTTP serializers only set Content-Length for sized payloads, so files and BytesIO bodies were sent without one and aiohttp fell back to Transfer-Encoding: chunked, which S3 rejects. Synchronously seekable streams are now measured from their current position to the end without consuming them. Streams whose length can't be determined, including async streams, still get no Content-Length. A Content-Length the caller already set, such as S3's modeled ContentLength member, now takes precedence. The serializers used to append their own length too, sending and signing "3, 3". This also fixes @requiresLength request payloads. Their length could never be computed, because the request serializer measured the async wrapper rather than the caller's stream. An unset payload of such an operation is sent with a length of 0. The length is computed at serialization, before signing, as the Go v2, Java v2 and Rust SDKs do, so steps that need it before signing, such as aws-chunked encoding, have it. --- ...gfix-dbdfcc08b0ac4033a2f69774df0decdd.json | 4 + .../src/smithy_http/serializers.py | 145 +++++---- .../tests/unit/test_serializers.py | 292 +++++++++++++++++- 3 files changed, 384 insertions(+), 57 deletions(-) create mode 100644 packages/smithy-http/.changes/next-release/smithy-http-bugfix-dbdfcc08b0ac4033a2f69774df0decdd.json diff --git a/packages/smithy-http/.changes/next-release/smithy-http-bugfix-dbdfcc08b0ac4033a2f69774df0decdd.json b/packages/smithy-http/.changes/next-release/smithy-http-bugfix-dbdfcc08b0ac4033a2f69774df0decdd.json new file mode 100644 index 000000000..0a770cb71 --- /dev/null +++ b/packages/smithy-http/.changes/next-release/smithy-http-bugfix-dbdfcc08b0ac4033a2f69774df0decdd.json @@ -0,0 +1,4 @@ +{ + "type": "bugfix", + "description": "Fixed HTTP serializers sending seekable streaming payloads, such as files, without a Content-Length and sending a modeled Content-Length twice." +} diff --git a/packages/smithy-http/src/smithy_http/serializers.py b/packages/smithy-http/src/smithy_http/serializers.py index 471cb154b..182ae3562 100644 --- a/packages/smithy-http/src/smithy_http/serializers.py +++ b/packages/smithy-http/src/smithy_http/serializers.py @@ -1,11 +1,11 @@ # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. # SPDX-License-Identifier: Apache-2.0 from base64 import b64encode -from collections.abc import Callable, Iterator, Sized +from collections.abc import Callable, Iterator from contextlib import AbstractContextManager, contextmanager from datetime import datetime from decimal import Decimal -from inspect import iscoroutinefunction +from inspect import iscoroutine from io import BytesIO from types import TracebackType from typing import TYPE_CHECKING @@ -34,7 +34,7 @@ from smithy_core.types import PathPattern, TimestampFormat from smithy_core.utils import serialize_float -from . import Field, tuples_to_fields +from . import Field, Fields, tuples_to_fields from .aio import HTTPRequest as _HTTPRequest from .aio import HTTPResponse as _HTTPResponse from .aio.interfaces import HTTPRequest, HTTPResponse @@ -110,26 +110,75 @@ def begin_struct(self, schema: Schema) -> Iterator[ShapeSerializer]: self.result = binding_serializer.build_request() -def _compute_content_length( - payload: object, -) -> int | None: - if (tell := getattr(payload, "tell", None)) is not None and not iscoroutinefunction( - tell - ): - start: int = tell() - if (end := _seek(payload, 0, 2)) is not None: - content_length: int = end - start - _seek(payload, start, 0) - return content_length - return None +def _set_content_length( + fields: Fields, + content_length: int | None, + raw_payload: "AsyncStreamingBlob | None", + *, + required: bool, +) -> None: + # A length the caller already set, such as with a modeled Content-Length + # member, takes precedence and isn't computed again. + if "content-length" in fields: + return + + # A body sent with Transfer-Encoding can't also have a Content-Length. + if "transfer-encoding" not in fields: + if raw_payload is not None: + content_length = _compute_content_length(raw_payload) + if content_length is not None: + fields.set_field(Field(name="content-length", values=[str(content_length)])) + return + if required: + raise SerializationError( + "This operation requires the content length of the input stream, but " + "it was not provided and was unable to be computed. Provide a " + "seekable stream or set the content length explicitly." + ) -def _seek(payload: object, pos: int, whence: int = 0) -> int | None: - if (seek := getattr(payload, "seek", None)) is not None and not iscoroutinefunction( - seek - ): - return seek(pos, whence) - return None + +def _compute_content_length(payload: "AsyncStreamingBlob") -> int | None: + """Determine the length of a raw payload without consuming it. + + ``bytes`` and ``bytearray`` report their length, and synchronously seekable + streams are measured from their current position to the end. The length of any + other stream, including async streams, is unknown. + """ + if isinstance(payload, bytes | bytearray): + return len(payload) + + tell = getattr(payload, "tell", None) + seek = getattr(payload, "seek", None) + if tell is None or seek is None: + return None + + try: + if (seekable := getattr(payload, "seekable", None)) is not None: + if _is_async(is_seekable := seekable()) or not is_seekable: + return None + if _is_async(start := tell()) or _is_async(seek(0, 2)): + return None + except (OSError, ValueError, NotImplementedError): + # Pipes raise io.UnsupportedOperation, closed files raise ValueError, and + # AsyncBytesReader raises NotImplementedError if its source can't tell. + return None + + # The stream is now at its end, so errors are raised instead of sending it on + # with an empty body. + end: int = tell() + seek(start, 0) + # A stream positioned past its end has nothing left to send. + return max(end - start, 0) + + +def _is_async(result: object) -> bool: + # Wrappers of async methods return a coroutine without being declared with + # ``async def``. Closing it avoids a "never awaited" warning. + if iscoroutine(result): + result.close() + return True + return False class HTTPBindingSerializer(InterceptingSerializer): @@ -171,7 +220,6 @@ def __init__( self._raw_payload_serializer: RawPayloadSerializer | None = None self._payload: AsyncBytesReader | AsyncBytesProvider | None = None self._content_type: str | None = payload_codec.media_type - self._content_length: int | None = None self._content_length_required = False self._writes_document_body = False self._result: HTTPRequest | None = None @@ -253,22 +301,22 @@ def build_request(self) -> HTTPRequest: self._body_context = None payload = self._payload + content_length: int | None = None + raw_payload: AsyncStreamingBlob | None = None if self._raw_payload_serializer is not None: raw_payload = self._raw_payload_serializer.payload - if isinstance(raw_payload, Sized): - self._content_length = len(raw_payload) - payload = AsyncBytesReader(raw_payload or b"") + if raw_payload is None and self._content_length_required: + # An unset payload is sent as an empty body, whose length is 0. + raw_payload = b"" elif self._sync_payload is not None: if ( self._binding_metadata.request.payload is not None or self._writes_document_body ): - self._content_length = self._sync_payload.tell() + content_length = self._sync_payload.tell() self._sync_payload.seek(0) payload = AsyncBytesReader(self._sync_payload) - assert payload is not None # noqa: S101 - headers = self.header_serializer.headers if self._content_type is not None and ( not self._binding_metadata.request.has_content_type_header @@ -276,18 +324,14 @@ def build_request(self) -> HTTPRequest: ): headers.append(("content-type", self._content_type)) - if self._content_length is not None: - headers.append(("content-length", str(self._content_length))) - fields = tuples_to_fields(headers) - if self._content_length_required and "content-length" not in fields: - content_length = _compute_content_length(payload) - if content_length is None: - raise SerializationError( - "This operation requires the content length of the input " - "stream, but it was not provided and was unable to be computed." - ) - fields.set_field(Field(name="content-length", values=[str(content_length)])) + _set_content_length( + fields, content_length, raw_payload, required=self._content_length_required + ) + if self._raw_payload_serializer is not None: + payload = AsyncBytesReader(raw_payload or b"") + + assert payload is not None # noqa: S101 self._result = _HTTPRequest( method=self._operation_metadata.method, @@ -388,8 +432,8 @@ def begin_struct(self, schema: Schema) -> Iterator[ShapeSerializer]: content_type: str | None = self._payload_codec.media_type content_length: int | None = None - content_length_source: object | None = None content_length_required = False + raw_payload: AsyncStreamingBlob | None = None binding_metadata = schema.get_extension(HTTP_BINDING_SCHEMA_EXTENSION) response_metadata = binding_metadata.response @@ -410,9 +454,8 @@ def begin_struct(self, schema: Schema) -> Iterator[ShapeSerializer]: ) yield binding_serializer raw_payload = payload_serializer.payload - if isinstance(raw_payload, Sized): - content_length = len(raw_payload) - content_length_source = raw_payload + if raw_payload is None and content_length_required: + raw_payload = b"" payload = AsyncBytesReader(raw_payload or b"") else: if payload_metadata.media_type is not None: @@ -454,20 +497,10 @@ def begin_struct(self, schema: Schema) -> Iterator[ShapeSerializer]: ): headers.append(("content-type", content_type)) - if content_length is not None: - headers.append(("content-length", str(content_length))) - fields = tuples_to_fields(headers) - if content_length_required and "content-length" not in fields: - content_length = _compute_content_length( - content_length_source if content_length_source is not None else payload - ) - if content_length is None: - raise SerializationError( - "This operation requires the content length of the input " - "stream, but it was not provided and was unable to be computed." - ) - fields.set_field(Field(name="content-length", values=[str(content_length)])) + _set_content_length( + fields, content_length, raw_payload, required=content_length_required + ) status = binding_serializer.response_code_serializer.response_code if status is None: diff --git a/packages/smithy-http/tests/unit/test_serializers.py b/packages/smithy-http/tests/unit/test_serializers.py index 9708ee2b7..93963a613 100644 --- a/packages/smithy-http/tests/unit/test_serializers.py +++ b/packages/smithy-http/tests/unit/test_serializers.py @@ -1,11 +1,13 @@ # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. # SPDX-License-Identifier: Apache-2.0 import datetime +from collections.abc import AsyncIterator, Callable from dataclasses import dataclass, field from datetime import UTC from decimal import Decimal from inspect import iscoroutinefunction from io import BytesIO +from pathlib import Path from typing import Any, ClassVar, Protocol, Self import pytest @@ -13,16 +15,18 @@ from smithy_core.aio.interfaces import StreamingBlob from smithy_core.aio.types import AsyncBytesReader from smithy_core.deserializers import DeserializeableShape, ShapeDeserializer +from smithy_core.exceptions import SerializationError from smithy_core.prelude import ( BIG_DECIMAL, BLOB, BOOLEAN, FLOAT, INTEGER, + LONG, STRING, TIMESTAMP, ) -from smithy_core.schemas import Schema +from smithy_core.schemas import MemberSchema, Schema from smithy_core.serializers import ( SerializeableStruct, ShapeSerializer, @@ -47,6 +51,7 @@ from smithy_http import Fields, tuples_to_fields from smithy_http.aio import HTTPResponse as _HTTPResponse from smithy_http.deserializers import HTTPResponseDeserializer +from smithy_http.interfaces import Fields as FieldsInterface from smithy_http.serializers import ( HTTPBindingSerializer, HTTPRequestSerializer, @@ -2309,3 +2314,288 @@ async def test_serialize_response_event_stream_creates_writeable_body() -> None: assert "header" in actual.fields assert actual.fields["header"].as_string() == "foo" + + +def _streaming_payload_schema(*, requires_length: bool = False) -> Schema: + payload: MemberSchema = { + "target": BLOB, + "traits": [HTTPPayloadTrait(), StreamingTrait()], + } + if requires_length: + payload["traits"].append(RequiresLengthTrait()) + return Schema.collection( + id=ShapeID( + "com.smithy#RequiredLengthStreamingPayload" + if requires_length + else "com.smithy#StreamingPayload" + ), + members={ + "payload": payload, + "contentLength": { + "target": LONG, + "traits": [HTTPHeaderTrait("Content-Length")], + }, + "transferEncoding": { + "target": STRING, + "traits": [HTTPHeaderTrait("Transfer-Encoding")], + }, + }, + ) + + +STREAMING_PAYLOAD = _streaming_payload_schema() +REQUIRED_LENGTH_STREAMING_PAYLOAD = _streaming_payload_schema(requires_length=True) +STREAMING_DATA = b"foobar" + + +class SeekableFalseBytesReader(NonSeekableBytesReader): + """A stream with seek and tell that reports it can't seek, like a pipe.""" + + def seekable(self) -> bool: + return False + + def seek(self, offset: int, whence: int = 0, /) -> int: + raise OSError("unseekable") + + def tell(self) -> int: + raise OSError("unseekable") + + +class CoroutineTellBytesReader(NonSeekableBytesReader): + """A stream whose tell returns a coroutine without being declared async.""" + + def seek(self, offset: int, whence: int = 0, /) -> int: + return self._wrapped.seek(offset, whence) + + def tell(self) -> Any: + return self._tell() + + async def _tell(self) -> int: + return self._wrapped.tell() + + +class CoroutineSeekBytesReader(NonSeekableBytesReader): + """A stream whose seek returns a coroutine without being declared async.""" + + def seek(self, offset: int, whence: int = 0, /) -> Any: + return self._seek(offset, whence) + + def tell(self) -> int: + return self._wrapped.tell() + + async def _seek(self, offset: int, whence: int) -> int: + return self._wrapped.seek(offset, whence) + + +class SeekOnlyBytesReader(NonSeekableBytesReader): + """A stream that can seek but can't report its position.""" + + def seek(self, offset: int, whence: int = 0, /) -> int: + return self._wrapped.seek(offset, whence) + + +class FailsAfterSeekingBytesReader(NonSeekableBytesReader): + """A stream that can seek to its end but fails to report the position there.""" + + def seek(self, offset: int, whence: int = 0, /) -> int: + return self._wrapped.seek(offset, whence) + + def tell(self) -> int: + if self._wrapped.tell() != 0: + raise OSError("tell failed") + return 0 + + +async def _async_chunks(data: bytes) -> AsyncIterator[bytes]: + yield data + + +def _serialize_streaming( + kind: str, + payload: StreamingBlob | None, + *, + content_length: int | None = None, + transfer_encoding: str | None = None, + schema: Schema = STREAMING_PAYLOAD, +) -> FieldsInterface: + """Serialize a request the way generated clients do, or a response, and return + its fields. A payload of None leaves the payload member unset.""" + + def write(serializer: ShapeSerializer) -> None: + if payload is not None: + serializer.write_data_stream(schema.members["payload"], payload) + if content_length is not None: + serializer.write_long(schema.members["contentLength"], content_length) + if transfer_encoding is not None: + serializer.write_string( + schema.members["transferEncoding"], transfer_encoding + ) + + http_trait = HTTPTrait({"method": "PUT", "code": 200, "uri": "/"}) + if kind == "request": + serializer = HTTPBindingSerializer( + payload_codec=JSONCodec(), schema=schema, http_trait=http_trait + ) + write(serializer) + return serializer.build_request().fields + + response_serializer = HTTPResponseSerializer( + payload_codec=JSONCodec(), http_trait=http_trait + ) + with response_serializer.begin_struct(schema) as struct_serializer: + write(struct_serializer) + assert response_serializer.result is not None + return response_serializer.result.fields + + +@pytest.mark.parametrize("kind", ["request", "response"]) +@pytest.mark.parametrize("offset", [0, 2]) +async def test_streaming_payload_sets_content_length(kind: str, offset: int) -> None: + payload = BytesIO(STREAMING_DATA) + payload.seek(offset) + fields = _serialize_streaming(kind, payload) + + # The length counts what's left from the current position, which is kept. + assert fields["content-length"].values == [str(len(STREAMING_DATA) - offset)] + assert payload.tell() == offset + + +@pytest.mark.parametrize("kind", ["request", "response"]) +async def test_streaming_payload_past_its_end_has_zero_content_length( + kind: str, +) -> None: + payload = BytesIO(STREAMING_DATA) + payload.seek(len(STREAMING_DATA) + 4) + fields = _serialize_streaming(kind, payload) + assert fields["content-length"].values == ["0"] + + +@pytest.mark.parametrize("kind", ["request", "response"]) +async def test_streaming_payload_that_fails_after_seeking_raises(kind: str) -> None: + # Swallowing the error would send the stream from its end, with an empty body. + with pytest.raises(OSError, match="tell failed"): + _serialize_streaming(kind, FailsAfterSeekingBytesReader(STREAMING_DATA)) + + +@pytest.mark.parametrize("kind", ["request", "response"]) +async def test_streaming_file_payload_sets_content_length( + kind: str, tmp_path: Path +) -> None: + path = tmp_path / "payload" + path.write_bytes(STREAMING_DATA) + with path.open("rb") as payload: + fields = _serialize_streaming(kind, payload) + assert fields["content-length"].values == [str(len(STREAMING_DATA))] + assert payload.tell() == 0 + + +@pytest.mark.parametrize("kind", ["request", "response"]) +@pytest.mark.parametrize( + "factory", + [ + lambda: NonSeekableBytesReader(STREAMING_DATA), + lambda: SeekableFalseBytesReader(STREAMING_DATA), + lambda: AsyncBytesReader(BytesIO(STREAMING_DATA)), + lambda: _async_chunks(STREAMING_DATA), + lambda: CoroutineTellBytesReader(STREAMING_DATA), + lambda: CoroutineSeekBytesReader(STREAMING_DATA), + lambda: AsyncBytesReader(SeekOnlyBytesReader(STREAMING_DATA)), + ], + ids=[ + "no-seek", + "seekable-false", + "async-seek", + "async-iterable", + "coroutine-tell", + "coroutine-seek", + "async-reader-without-tell", + ], +) +async def test_streaming_payload_of_unknown_length_omits_content_length( + kind: str, factory: Callable[[], StreamingBlob] +) -> None: + fields = _serialize_streaming(kind, factory()) + assert "content-length" not in fields + + +@pytest.mark.parametrize("kind", ["request", "response"]) +@pytest.mark.parametrize( + "factory", + [lambda: STREAMING_DATA, lambda: NonSeekableBytesReader(STREAMING_DATA)], + ids=["bytes", "no-seek"], +) +async def test_modeled_content_length_is_not_duplicated( + kind: str, factory: Callable[[], StreamingBlob] +) -> None: + fields = _serialize_streaming(kind, factory(), content_length=6) + assert fields["content-length"].values == ["6"] + + +async def test_requires_length_computes_content_length() -> None: + fields = _serialize_streaming( + "request", BytesIO(STREAMING_DATA), schema=REQUIRED_LENGTH_STREAMING_PAYLOAD + ) + assert fields["content-length"].values == [str(len(STREAMING_DATA))] + + +@pytest.mark.parametrize("kind", ["request", "response"]) +@pytest.mark.parametrize( + "factory", + [ + lambda: NonSeekableBytesReader(STREAMING_DATA), + lambda: _async_chunks(STREAMING_DATA), + ], + ids=["no-seek", "async-iterable"], +) +async def test_requires_length_raises_for_stream_of_unknown_length( + kind: str, factory: Callable[[], StreamingBlob] +) -> None: + with pytest.raises(SerializationError): + _serialize_streaming(kind, factory(), schema=REQUIRED_LENGTH_STREAMING_PAYLOAD) + + +async def test_requires_length_uses_modeled_content_length() -> None: + fields = _serialize_streaming( + "request", + NonSeekableBytesReader(STREAMING_DATA), + content_length=6, + schema=REQUIRED_LENGTH_STREAMING_PAYLOAD, + ) + assert fields["content-length"].values == ["6"] + + +@pytest.mark.parametrize("kind", ["request", "response"]) +async def test_unset_streaming_payload_that_requires_length_has_zero_length( + kind: str, +) -> None: + fields = _serialize_streaming(kind, None, schema=REQUIRED_LENGTH_STREAMING_PAYLOAD) + assert fields["content-length"].values == ["0"] + + +@pytest.mark.parametrize("kind", ["request", "response"]) +async def test_unset_streaming_payload_omits_content_length(kind: str) -> None: + # Requests without content, such as GETs, shouldn't send a Content-Length. + fields = _serialize_streaming(kind, None) + assert "content-length" not in fields + + +@pytest.mark.parametrize("kind", ["request", "response"]) +async def test_streaming_payload_with_transfer_encoding_omits_content_length( + kind: str, +) -> None: + fields = _serialize_streaming( + kind, BytesIO(STREAMING_DATA), transfer_encoding="chunked" + ) + assert "content-length" not in fields + assert fields["transfer-encoding"].values == ["chunked"] + + +@pytest.mark.parametrize("kind", ["request", "response"]) +async def test_requires_length_with_transfer_encoding_raises(kind: str) -> None: + with pytest.raises(SerializationError): + _serialize_streaming( + kind, + BytesIO(STREAMING_DATA), + transfer_encoding="chunked", + schema=REQUIRED_LENGTH_STREAMING_PAYLOAD, + ) From f83729672c52bc5401f3ca2b31ab278dd1b87bc6 Mon Sep 17 00:00:00 2001 From: jonathan343 Date: Sun, 18 Oct 2026 17:11:22 -0400 Subject: [PATCH 2/5] Retry seekable streaming payloads from where they started Operations with a streaming input were never retried, so a throttled or failed upload from a file raised instead of being resent. A data stream that can be rewound is now retried, and each retry rewinds it to the position it had before the first attempt rather than to 0, so a caller that skipped the first bytes of a file doesn't have them sent. Event streams, and streams that can't be rewound or fail to seek, are still sent once. --- ...ment-ce57b5cfed2740908d9e5222d3f58f47.json | 4 + .../smithy-core/src/smithy_core/aio/client.py | 34 +++++- .../smithy-core/tests/unit/aio/test_client.py | 100 +++++++++++++++++- .../smithy-core/tests/unit/test_retries.py | 18 +++- 4 files changed, 149 insertions(+), 7 deletions(-) create mode 100644 packages/smithy-core/.changes/next-release/smithy-core-enhancement-ce57b5cfed2740908d9e5222d3f58f47.json diff --git a/packages/smithy-core/.changes/next-release/smithy-core-enhancement-ce57b5cfed2740908d9e5222d3f58f47.json b/packages/smithy-core/.changes/next-release/smithy-core-enhancement-ce57b5cfed2740908d9e5222d3f58f47.json new file mode 100644 index 000000000..f5c7be6c4 --- /dev/null +++ b/packages/smithy-core/.changes/next-release/smithy-core-enhancement-ce57b5cfed2740908d9e5222d3f58f47.json @@ -0,0 +1,4 @@ +{ + "type": "enhancement", + "description": "Added retries for requests with a seekable streaming payload, which are resent from the position the stream was at when the operation was called." +} \ No newline at end of file diff --git a/packages/smithy-core/src/smithy_core/aio/client.py b/packages/smithy-core/src/smithy_core/aio/client.py index ec43db2be..034c9028e 100644 --- a/packages/smithy-core/src/smithy_core/aio/client.py +++ b/packages/smithy-core/src/smithy_core/aio/client.py @@ -92,10 +92,36 @@ class ClientCall[I: SerializeableShape, O: DeserializeableShape]: """The retry scope for the operation.""" def retryable(self) -> bool: - # TODO: check to see if the stream is seekable + """Whether the operation can be retried whatever its request body is. + + Operations with an input stream can't be, but a request with a data stream can + still be retried if the stream can be rewound. + """ return self.operation.input_stream_member is None +async def _retry_start( + call: ClientCall[Any, Any], + body: Any, + request_future: Future[Any] | None, +) -> int | None: + """Find where a request body is rewound to before each retry. + + :returns: The position to rewind to, or None if the request can't be retried. + """ + if call.retryable(): + return 0 + # Event streams are never retried. + if request_future is not None: + return None + # A data stream is resent from where it started, so it can be retried only if it + # can be rewound there. + try: + return await seek(body, 0, 1) + except (OSError, ValueError, NotImplementedError): + return None + + class RequestPipeline[TRequest: Request, TResponse: Response]: """Invokes client operations asynchronously.""" @@ -382,7 +408,9 @@ async def _retry[I: SerializeableShape, O: DeserializeableShape]( request_context: RequestContext[I, TRequest], request_future: Future[RequestContext[I, TRequest]] | None, ) -> OutputContext[I, O, TRequest | None, TResponse | None]: - if not call.retryable(): + body = request_context.transport_request.body + start = await _retry_start(call, body, request_future) + if start is None: return await self._send_with_rate_limiting( call, request_context, request_future ) @@ -444,7 +472,7 @@ async def _retry[I: SerializeableShape, O: DeserializeableShape]( retry_token.retry_delay, ) - await seek(request_context.transport_request.body, 0) + await seek(body, start) else: await retry_strategy.record_success(token=retry_token) return output_context diff --git a/packages/smithy-core/tests/unit/aio/test_client.py b/packages/smithy-core/tests/unit/aio/test_client.py index 8af452517..f627cc50b 100644 --- a/packages/smithy-core/tests/unit/aio/test_client.py +++ b/packages/smithy-core/tests/unit/aio/test_client.py @@ -1,7 +1,9 @@ # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. # SPDX-License-Identifier: Apache-2.0 +from collections.abc import AsyncIterator from dataclasses import replace +from io import BytesIO, UnsupportedOperation from typing import Any, cast from unittest.mock import AsyncMock, Mock, patch @@ -9,7 +11,8 @@ from smithy_core.aio.client import AsyncClient from smithy_core.aio.eventstream import DuplexEventStream, InputEventStream from smithy_core.aio.interfaces import ClientTransport -from smithy_core.aio.retries import AdaptiveRetryStrategy +from smithy_core.aio.retries import AdaptiveRetryStrategy, SimpleRetryStrategy +from smithy_core.aio.types import AsyncBytesReader from smithy_core.auth import AuthOption from smithy_core.exceptions import CallError, SmithyError, UnsupportedTransportError from smithy_core.interceptors import ( @@ -19,6 +22,7 @@ OutputContext, ) from smithy_core.response import EMPTY_RESPONSE_METADATA +from smithy_core.retries import ExponentialRetryBackoffStrategy from smithy_core.shapes import ShapeID from smithy_core.types import TypedProperties @@ -30,6 +34,7 @@ StubEventReceiver, StubInput, StubOutput, + StubRequest, UndeclaredTransport, client_call, pipeline_harness, @@ -319,3 +324,96 @@ async def test_aexit_closes_client() -> None: assert client.closed is True assert client.transport.close_calls == 1 + + +class BodyRecordingTransport(NonDuplexTransport): + def __init__(self) -> None: + super().__init__() + self.bodies: list[bytes] = [] + + async def send(self, request: Any) -> Any: + self.bodies.append(await AsyncBytesReader(request.body).read()) + return await super().send(request) + + +async def _chunks() -> AsyncIterator[bytes]: + yield b"foobar" + + +def _streaming_call_failing_once( + harness: Any, body: Any, *, max_attempts: int = 2 +) -> Any: + request = StubRequest() + request.body = body + harness.protocol.serialize_request = lambda **kwargs: request # type: ignore[method-assign] + + async def fail_first_attempt(**kwargs: Any) -> StubOutput: + if harness.protocol.deserialize_response_calls == 0: + harness.protocol.deserialize_response_calls += 1 + raise CallError("Service unavailable", is_retry_safe=True) + return StubOutput() + + harness.protocol.deserialize_response = fail_first_attempt # type: ignore[method-assign] + return replace( + client_call(), + retry_strategy=SimpleRetryStrategy( + backoff_strategy=ExponentialRetryBackoffStrategy(backoff_scale_value=0), + max_attempts=max_attempts, + ), + ) + + +async def test_seekable_data_stream_is_retried_from_its_start() -> None: + harness = pipeline_harness(BodyRecordingTransport()) + body = BytesIO(b"foobar") + body.seek(2) + call = _streaming_call_failing_once(harness, AsyncBytesReader(body)) + + await harness.pipeline(call) + + # The caller's position is where the body starts, so retries don't resend 'fo'. + assert cast(BodyRecordingTransport, harness.transport).bodies == [b"obar", b"obar"] + + +async def test_non_seekable_data_stream_is_not_retried() -> None: + harness = pipeline_harness(BodyRecordingTransport()) + call = _streaming_call_failing_once(harness, AsyncBytesReader(_chunks())) + + with pytest.raises(CallError, match="Service unavailable"): + await harness.pipeline(call) + + assert cast(BodyRecordingTransport, harness.transport).bodies == [b"foobar"] + + +class UnrewindableBody: + """A body that reports it's seekable but fails to seek.""" + + def seekable(self) -> bool: + return True + + def seek(self, offset: int, whence: int = 0) -> int: + raise UnsupportedOperation("seek") + + async def read(self, size: int = -1) -> bytes: + return b"foobar" + + +async def test_data_stream_that_fails_to_seek_is_sent_once() -> None: + harness = pipeline_harness(BodyRecordingTransport()) + call = _streaming_call_failing_once(harness, UnrewindableBody()) + + with pytest.raises(CallError, match="Service unavailable"): + await harness.pipeline(call) + + assert cast(BodyRecordingTransport, harness.transport).bodies == [b"foobar"] + + +async def test_input_event_stream_is_not_retried() -> None: + harness = pipeline_harness(NonDuplexTransport()) + call = _streaming_call_failing_once(harness, BytesIO(b"")) + + stream = await harness.pipeline.input_stream(call, StubEvent) + + with pytest.raises(CallError, match="Service unavailable"): + await stream.await_output() + assert harness.transport.send_calls == 1 diff --git a/packages/smithy-core/tests/unit/test_retries.py b/packages/smithy-core/tests/unit/test_retries.py index d81c8d687..0edccfbff 100644 --- a/packages/smithy-core/tests/unit/test_retries.py +++ b/packages/smithy-core/tests/unit/test_retries.py @@ -728,6 +728,12 @@ def _non_retryable_call(self, strategy: AdaptiveRetryStrategy) -> Any: call.retry_strategy = strategy return call + def _unseekable_request_context(self) -> Any: + # The request's input stream can't be rewound, so it can't be resent + context = MagicMock() + context.transport_request.body.seekable.return_value = False + return context + def _adaptive_strategy_with_mocked_limiter( self, ) -> tuple[AdaptiveRetryStrategy, ClientRateLimiter]: @@ -753,7 +759,9 @@ async def test_non_retryable_call_applies_rate_limiting_on_success(self): with patch.object(pipeline, "_handle_attempt", handle_attempt): await pipeline._retry( # type: ignore[reportPrivateUsage] - self._non_retryable_call(strategy), MagicMock(), None + self._non_retryable_call(strategy), + self._unseekable_request_context(), + None, ) handle_attempt.assert_awaited_once() @@ -775,7 +783,9 @@ async def test_non_retryable_call_reports_throttle_without_retrying(self): with patch.object(pipeline, "_handle_attempt", handle_attempt): output_context = await pipeline._retry( # type: ignore[reportPrivateUsage] - self._non_retryable_call(strategy), MagicMock(), None + self._non_retryable_call(strategy), + self._unseekable_request_context(), + None, ) # Sent once and returned as-is, even though the error is retry-safe @@ -798,7 +808,9 @@ async def test_failure_without_response_does_not_update_rate_limiter(self): with patch.object(pipeline, "_handle_attempt", handle_attempt): output_context = await pipeline._retry( # type: ignore[reportPrivateUsage] - self._non_retryable_call(strategy), MagicMock(), None + self._non_retryable_call(strategy), + self._unseekable_request_context(), + None, ) assert output_context.response is error From ab3a312e79d33c1f007df2636ea57682732f577d Mon Sep 17 00:00:00 2001 From: jonathan343 Date: Sun, 18 Oct 2026 18:29:50 -0400 Subject: [PATCH 3/5] Decide stream retries from the operation's schema Whether a request could be retried was split between retryable(), which reported every streaming operation as not retryable, and a check of the request future in the retry loop that told event streams apart. retryable() now returns False only for operations with an input event stream, and the retry loop checks only whether the body can be rewound. Event streams no longer depend on how the pipeline was invoked, and the rate-limiting tests no longer need a body that can't be rewound. If seeking a body back fails before a retry, the call now returns the attempt's error with a note, as when a send token times out, instead of failing with the seek error. --- .../smithy-core/src/smithy_core/aio/client.py | 37 ++++++++----------- .../tests/unit/aio/_pipeline_harness.py | 25 ++++++++++++- .../smithy-core/tests/unit/aio/test_client.py | 31 ++++++++++++++-- .../smithy-core/tests/unit/test_retries.py | 18 ++------- 4 files changed, 71 insertions(+), 40 deletions(-) diff --git a/packages/smithy-core/src/smithy_core/aio/client.py b/packages/smithy-core/src/smithy_core/aio/client.py index 034c9028e..f2e1eb7a2 100644 --- a/packages/smithy-core/src/smithy_core/aio/client.py +++ b/packages/smithy-core/src/smithy_core/aio/client.py @@ -30,7 +30,7 @@ from ..interfaces.auth import AuthOption, AuthSchemeResolver from ..schemas import APIOperation from ..serializers import SerializeableShape -from ..shapes import ShapeID +from ..shapes import ShapeID, ShapeType from ..types import PropertyKey from ..types import TypedProperties as _TypedProperties from .eventstream import DuplexEventStream, InputEventStream, OutputEventStream @@ -92,30 +92,22 @@ class ClientCall[I: SerializeableShape, O: DeserializeableShape]: """The retry scope for the operation.""" def retryable(self) -> bool: - """Whether the operation can be retried whatever its request body is. + """Whether the operation can be retried. - Operations with an input stream can't be, but a request with a data stream can - still be retried if the stream can be rewound. + Operations with an input event stream are never retried. A request for any + other operation is retried only if its body can be rewound to where it started. """ - return self.operation.input_stream_member is None + member = self.operation.input_stream_member + return member is None or member.shape_type is not ShapeType.UNION -async def _retry_start( - call: ClientCall[Any, Any], - body: Any, - request_future: Future[Any] | None, -) -> int | None: - """Find where a request body is rewound to before each retry. +async def _rewind_position(body: Any) -> int | None: + """Find where a request body is rewound to before it's resent. - :returns: The position to rewind to, or None if the request can't be retried. + :returns: The body's current position, or None if it can't be rewound. """ - if call.retryable(): + if body is None or isinstance(body, bytes | bytearray): return 0 - # Event streams are never retried. - if request_future is not None: - return None - # A data stream is resent from where it started, so it can be retried only if it - # can be rewound there. try: return await seek(body, 0, 1) except (OSError, ValueError, NotImplementedError): @@ -409,8 +401,7 @@ async def _retry[I: SerializeableShape, O: DeserializeableShape]( request_future: Future[RequestContext[I, TRequest]] | None, ) -> OutputContext[I, O, TRequest | None, TResponse | None]: body = request_context.transport_request.body - start = await _retry_start(call, body, request_future) - if start is None: + if not call.retryable() or (start := await _rewind_position(body)) is None: return await self._send_with_rate_limiting( call, request_context, request_future ) @@ -472,7 +463,11 @@ async def _retry[I: SerializeableShape, O: DeserializeableShape]( retry_token.retry_delay, ) - await seek(body, start) + try: + await seek(body, start) + except (OSError, ValueError, NotImplementedError) as e: + output_context.response.add_note(f"Retry not attempted: {e}") + return output_context else: await retry_strategy.record_success(token=retry_token) return output_context diff --git a/packages/smithy-core/tests/unit/aio/_pipeline_harness.py b/packages/smithy-core/tests/unit/aio/_pipeline_harness.py index c4420181a..1f3f18a23 100644 --- a/packages/smithy-core/tests/unit/aio/_pipeline_harness.py +++ b/packages/smithy-core/tests/unit/aio/_pipeline_harness.py @@ -12,6 +12,7 @@ from smithy_core.documents import TypeRegistry from smithy_core.endpoints import EndpointResolverParams from smithy_core.interceptors import InterceptorChain +from smithy_core.prelude import BLOB from smithy_core.response import EMPTY_RESPONSE_METADATA, ResponseMetadata from smithy_core.schemas import APIOperation, Schema from smithy_core.serializers import ShapeSerializer @@ -86,7 +87,29 @@ def deserialize(cls, deserializer: ShapeDeserializer) -> Self: effective_auth_schemes=[], ) -# ``ClientCall.retryable()`` is False for operations with a streaming input, so +_DATA_STREAM_INPUT_SCHEMA = Schema.collection( + id=ShapeID("com.example#DataStreamInput"), + members={"body": {"target": BLOB, "traits": [StreamingTrait()]}}, +) + +# An operation whose input is a data stream rather than an event stream. +DATA_STREAM_OPERATION = APIOperation( + input=StubInput, + output=StubOutput, + static_schema=( + ShapeType.OPERATION.value, + "com.example", + "DataStreamOperation", + 0, + _DATA_STREAM_INPUT_SCHEMA, + _UNARY_OUTPUT_SCHEMA, + (), + ), + error_registry=TypeRegistry({}), + effective_auth_schemes=[], +) + +# ``ClientCall.retryable()`` is False for operations with an input event stream, so # OPERATION above never enters the retry loop. This one has no input stream and so # exercises it. UNARY_OPERATION = APIOperation( diff --git a/packages/smithy-core/tests/unit/aio/test_client.py b/packages/smithy-core/tests/unit/aio/test_client.py index f627cc50b..04999de22 100644 --- a/packages/smithy-core/tests/unit/aio/test_client.py +++ b/packages/smithy-core/tests/unit/aio/test_client.py @@ -27,6 +27,8 @@ from smithy_core.types import TypedProperties from ._pipeline_harness import ( + DATA_STREAM_OPERATION, + OPERATION, DuplexTransport, NonDuplexTransport, StubEndpointResolver, @@ -341,7 +343,7 @@ async def _chunks() -> AsyncIterator[bytes]: def _streaming_call_failing_once( - harness: Any, body: Any, *, max_attempts: int = 2 + harness: Any, body: Any, *, operation: Any = DATA_STREAM_OPERATION ) -> Any: request = StubRequest() request.body = body @@ -356,9 +358,10 @@ async def fail_first_attempt(**kwargs: Any) -> StubOutput: harness.protocol.deserialize_response = fail_first_attempt # type: ignore[method-assign] return replace( client_call(), + operation=operation, retry_strategy=SimpleRetryStrategy( backoff_strategy=ExponentialRetryBackoffStrategy(backoff_scale_value=0), - max_attempts=max_attempts, + max_attempts=2, ), ) @@ -408,9 +411,31 @@ async def test_data_stream_that_fails_to_seek_is_sent_once() -> None: assert cast(BodyRecordingTransport, harness.transport).bodies == [b"foobar"] +class SeekBackFailsBody(UnrewindableBody): + """A body that reports its position but fails to seek back to it.""" + + def seek(self, offset: int, whence: int = 0) -> int: + if whence == 1: + return 0 + raise UnsupportedOperation("seek") + + +async def test_data_stream_that_fails_to_rewind_returns_the_attempts_error() -> None: + harness = pipeline_harness(BodyRecordingTransport()) + call = _streaming_call_failing_once(harness, SeekBackFailsBody()) + + with pytest.raises(CallError, match="Service unavailable") as exc_info: + await harness.pipeline(call) + + assert cast(BodyRecordingTransport, harness.transport).bodies == [b"foobar"] + assert any( + note.startswith("Retry not attempted") for note in exc_info.value.__notes__ + ) + + async def test_input_event_stream_is_not_retried() -> None: harness = pipeline_harness(NonDuplexTransport()) - call = _streaming_call_failing_once(harness, BytesIO(b"")) + call = _streaming_call_failing_once(harness, BytesIO(b""), operation=OPERATION) stream = await harness.pipeline.input_stream(call, StubEvent) diff --git a/packages/smithy-core/tests/unit/test_retries.py b/packages/smithy-core/tests/unit/test_retries.py index 0edccfbff..d81c8d687 100644 --- a/packages/smithy-core/tests/unit/test_retries.py +++ b/packages/smithy-core/tests/unit/test_retries.py @@ -728,12 +728,6 @@ def _non_retryable_call(self, strategy: AdaptiveRetryStrategy) -> Any: call.retry_strategy = strategy return call - def _unseekable_request_context(self) -> Any: - # The request's input stream can't be rewound, so it can't be resent - context = MagicMock() - context.transport_request.body.seekable.return_value = False - return context - def _adaptive_strategy_with_mocked_limiter( self, ) -> tuple[AdaptiveRetryStrategy, ClientRateLimiter]: @@ -759,9 +753,7 @@ async def test_non_retryable_call_applies_rate_limiting_on_success(self): with patch.object(pipeline, "_handle_attempt", handle_attempt): await pipeline._retry( # type: ignore[reportPrivateUsage] - self._non_retryable_call(strategy), - self._unseekable_request_context(), - None, + self._non_retryable_call(strategy), MagicMock(), None ) handle_attempt.assert_awaited_once() @@ -783,9 +775,7 @@ async def test_non_retryable_call_reports_throttle_without_retrying(self): with patch.object(pipeline, "_handle_attempt", handle_attempt): output_context = await pipeline._retry( # type: ignore[reportPrivateUsage] - self._non_retryable_call(strategy), - self._unseekable_request_context(), - None, + self._non_retryable_call(strategy), MagicMock(), None ) # Sent once and returned as-is, even though the error is retry-safe @@ -808,9 +798,7 @@ async def test_failure_without_response_does_not_update_rate_limiter(self): with patch.object(pipeline, "_handle_attempt", handle_attempt): output_context = await pipeline._retry( # type: ignore[reportPrivateUsage] - self._non_retryable_call(strategy), - self._unseekable_request_context(), - None, + self._non_retryable_call(strategy), MagicMock(), None ) assert output_context.response is error From b18d0da644bc8c7c4dd68ee316e93b1fae9fca39 Mon Sep 17 00:00:00 2001 From: jonathan343 Date: Sun, 18 Oct 2026 18:30:18 -0400 Subject: [PATCH 4/5] Trim overlapping Content-Length tests The file payload test went through the same code as the BytesIO one, and the @requiresLength modeled-length test repeated the duplicate-header test, so it's now a parameter of it. --- .../tests/unit/test_serializers.py | 34 +++++-------------- 1 file changed, 8 insertions(+), 26 deletions(-) diff --git a/packages/smithy-http/tests/unit/test_serializers.py b/packages/smithy-http/tests/unit/test_serializers.py index 93963a613..f8ba12503 100644 --- a/packages/smithy-http/tests/unit/test_serializers.py +++ b/packages/smithy-http/tests/unit/test_serializers.py @@ -7,7 +7,6 @@ from decimal import Decimal from inspect import iscoroutinefunction from io import BytesIO -from pathlib import Path from typing import Any, ClassVar, Protocol, Self import pytest @@ -2477,18 +2476,6 @@ async def test_streaming_payload_that_fails_after_seeking_raises(kind: str) -> N _serialize_streaming(kind, FailsAfterSeekingBytesReader(STREAMING_DATA)) -@pytest.mark.parametrize("kind", ["request", "response"]) -async def test_streaming_file_payload_sets_content_length( - kind: str, tmp_path: Path -) -> None: - path = tmp_path / "payload" - path.write_bytes(STREAMING_DATA) - with path.open("rb") as payload: - fields = _serialize_streaming(kind, payload) - assert fields["content-length"].values == [str(len(STREAMING_DATA))] - assert payload.tell() == 0 - - @pytest.mark.parametrize("kind", ["request", "response"]) @pytest.mark.parametrize( "factory", @@ -2519,15 +2506,20 @@ async def test_streaming_payload_of_unknown_length_omits_content_length( @pytest.mark.parametrize("kind", ["request", "response"]) +@pytest.mark.parametrize( + "schema", + [STREAMING_PAYLOAD, REQUIRED_LENGTH_STREAMING_PAYLOAD], + ids=["optional-length", "requires-length"], +) @pytest.mark.parametrize( "factory", [lambda: STREAMING_DATA, lambda: NonSeekableBytesReader(STREAMING_DATA)], ids=["bytes", "no-seek"], ) -async def test_modeled_content_length_is_not_duplicated( - kind: str, factory: Callable[[], StreamingBlob] +async def test_modeled_content_length_is_used_once( + kind: str, schema: Schema, factory: Callable[[], StreamingBlob] ) -> None: - fields = _serialize_streaming(kind, factory(), content_length=6) + fields = _serialize_streaming(kind, factory(), content_length=6, schema=schema) assert fields["content-length"].values == ["6"] @@ -2554,16 +2546,6 @@ async def test_requires_length_raises_for_stream_of_unknown_length( _serialize_streaming(kind, factory(), schema=REQUIRED_LENGTH_STREAMING_PAYLOAD) -async def test_requires_length_uses_modeled_content_length() -> None: - fields = _serialize_streaming( - "request", - NonSeekableBytesReader(STREAMING_DATA), - content_length=6, - schema=REQUIRED_LENGTH_STREAMING_PAYLOAD, - ) - assert fields["content-length"].values == ["6"] - - @pytest.mark.parametrize("kind", ["request", "response"]) async def test_unset_streaming_payload_that_requires_length_has_zero_length( kind: str, From b20ea46c4cab23463044b2ec9f76e69010976b2e Mon Sep 17 00:00:00 2001 From: jonathan343 Date: Sun, 18 Oct 2026 18:44:48 -0400 Subject: [PATCH 5/5] Drop tests for contrived streams The coroutine-returning tell and seek fixtures and the stream whose tell fails only at its end don't match any stream callers pass, and the async seek case already covers what _is_async is for. The two tests for data streams that can't be rewound are now one parametrized test. --- .../smithy-core/tests/unit/aio/test_client.py | 23 ++++----- .../tests/unit/test_serializers.py | 49 ------------------- 2 files changed, 10 insertions(+), 62 deletions(-) diff --git a/packages/smithy-core/tests/unit/aio/test_client.py b/packages/smithy-core/tests/unit/aio/test_client.py index 04999de22..215d7ff14 100644 --- a/packages/smithy-core/tests/unit/aio/test_client.py +++ b/packages/smithy-core/tests/unit/aio/test_client.py @@ -1,7 +1,7 @@ # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. # SPDX-License-Identifier: Apache-2.0 -from collections.abc import AsyncIterator +from collections.abc import AsyncIterator, Callable from dataclasses import replace from io import BytesIO, UnsupportedOperation from typing import Any, cast @@ -378,16 +378,6 @@ async def test_seekable_data_stream_is_retried_from_its_start() -> None: assert cast(BodyRecordingTransport, harness.transport).bodies == [b"obar", b"obar"] -async def test_non_seekable_data_stream_is_not_retried() -> None: - harness = pipeline_harness(BodyRecordingTransport()) - call = _streaming_call_failing_once(harness, AsyncBytesReader(_chunks())) - - with pytest.raises(CallError, match="Service unavailable"): - await harness.pipeline(call) - - assert cast(BodyRecordingTransport, harness.transport).bodies == [b"foobar"] - - class UnrewindableBody: """A body that reports it's seekable but fails to seek.""" @@ -401,9 +391,16 @@ async def read(self, size: int = -1) -> bytes: return b"foobar" -async def test_data_stream_that_fails_to_seek_is_sent_once() -> None: +@pytest.mark.parametrize( + "factory", + [lambda: AsyncBytesReader(_chunks()), UnrewindableBody], + ids=["not-seekable", "fails-to-seek"], +) +async def test_data_stream_that_cant_be_rewound_is_sent_once( + factory: Callable[[], Any], +) -> None: harness = pipeline_harness(BodyRecordingTransport()) - call = _streaming_call_failing_once(harness, UnrewindableBody()) + call = _streaming_call_failing_once(harness, factory()) with pytest.raises(CallError, match="Service unavailable"): await harness.pipeline(call) diff --git a/packages/smithy-http/tests/unit/test_serializers.py b/packages/smithy-http/tests/unit/test_serializers.py index f8ba12503..c9550fe58 100644 --- a/packages/smithy-http/tests/unit/test_serializers.py +++ b/packages/smithy-http/tests/unit/test_serializers.py @@ -2360,32 +2360,6 @@ def tell(self) -> int: raise OSError("unseekable") -class CoroutineTellBytesReader(NonSeekableBytesReader): - """A stream whose tell returns a coroutine without being declared async.""" - - def seek(self, offset: int, whence: int = 0, /) -> int: - return self._wrapped.seek(offset, whence) - - def tell(self) -> Any: - return self._tell() - - async def _tell(self) -> int: - return self._wrapped.tell() - - -class CoroutineSeekBytesReader(NonSeekableBytesReader): - """A stream whose seek returns a coroutine without being declared async.""" - - def seek(self, offset: int, whence: int = 0, /) -> Any: - return self._seek(offset, whence) - - def tell(self) -> int: - return self._wrapped.tell() - - async def _seek(self, offset: int, whence: int) -> int: - return self._wrapped.seek(offset, whence) - - class SeekOnlyBytesReader(NonSeekableBytesReader): """A stream that can seek but can't report its position.""" @@ -2393,18 +2367,6 @@ def seek(self, offset: int, whence: int = 0, /) -> int: return self._wrapped.seek(offset, whence) -class FailsAfterSeekingBytesReader(NonSeekableBytesReader): - """A stream that can seek to its end but fails to report the position there.""" - - def seek(self, offset: int, whence: int = 0, /) -> int: - return self._wrapped.seek(offset, whence) - - def tell(self) -> int: - if self._wrapped.tell() != 0: - raise OSError("tell failed") - return 0 - - async def _async_chunks(data: bytes) -> AsyncIterator[bytes]: yield data @@ -2469,13 +2431,6 @@ async def test_streaming_payload_past_its_end_has_zero_content_length( assert fields["content-length"].values == ["0"] -@pytest.mark.parametrize("kind", ["request", "response"]) -async def test_streaming_payload_that_fails_after_seeking_raises(kind: str) -> None: - # Swallowing the error would send the stream from its end, with an empty body. - with pytest.raises(OSError, match="tell failed"): - _serialize_streaming(kind, FailsAfterSeekingBytesReader(STREAMING_DATA)) - - @pytest.mark.parametrize("kind", ["request", "response"]) @pytest.mark.parametrize( "factory", @@ -2484,8 +2439,6 @@ async def test_streaming_payload_that_fails_after_seeking_raises(kind: str) -> N lambda: SeekableFalseBytesReader(STREAMING_DATA), lambda: AsyncBytesReader(BytesIO(STREAMING_DATA)), lambda: _async_chunks(STREAMING_DATA), - lambda: CoroutineTellBytesReader(STREAMING_DATA), - lambda: CoroutineSeekBytesReader(STREAMING_DATA), lambda: AsyncBytesReader(SeekOnlyBytesReader(STREAMING_DATA)), ], ids=[ @@ -2493,8 +2446,6 @@ async def test_streaming_payload_that_fails_after_seeking_raises(kind: str) -> N "seekable-false", "async-seek", "async-iterable", - "coroutine-tell", - "coroutine-seek", "async-reader-without-tell", ], )