diff --git a/clients/python/README.md b/clients/python/README.md index f7a6bb41..924eb82a 100644 --- a/clients/python/README.md +++ b/clients/python/README.md @@ -102,6 +102,12 @@ session.put(video_data, compress="none") session.put(zstd_data, precompressed="zstd") ``` +### Resumable Uploads + +The client automatically uses resumable uploads when eligible and may create +temporary files for spooling. See [`Session.put()`](src/objectstore_client/client.py) +for eligibility, configuration, and upload behavior. + ### Custom Metadata Arbitrary key-value pairs can be attached to objects and retrieved on download. diff --git a/clients/python/src/objectstore_client/_resumable.py b/clients/python/src/objectstore_client/_resumable.py index 8c467ade..56aa0bd1 100644 --- a/clients/python/src/objectstore_client/_resumable.py +++ b/clients/python/src/objectstore_client/_resumable.py @@ -1,12 +1,21 @@ from __future__ import annotations +import random +import time +from collections.abc import Iterator +from contextlib import contextmanager, nullcontext from dataclasses import dataclass +from io import SEEK_END, BytesIO +from os import PathLike +from tempfile import TemporaryFile from typing import IO, TYPE_CHECKING from urllib.parse import urlencode import urllib3 +import zstandard from objectstore_client.errors import RequestError, raise_for_status +from objectstore_client.metadata import Compression, ExpirationPolicy from objectstore_client.metrics import measure_storage_operation from objectstore_client.tracing import storage_span @@ -34,6 +43,13 @@ class ChunkTooSmall(ValueError): """A non-final chunk is shorter than the upload granularity.""" +class NoUploadProgress(Exception): + """A write completed without advancing the server's upload offset.""" + + def __init__(self) -> None: + super().__init__("Upload made no progress") + + @dataclass(frozen=True) class UploadIncomplete: """The upload expects more bytes, starting at the given offset.""" @@ -156,6 +172,12 @@ def put( headers = session._make_headers() headers["Upload-Offset"] = str(offset) headers["Content-Length"] = str(length) + # Disable pool retries that replay the body. The resumable upload loop + # probes the persisted offset before resending data; connection retries + # retain the pool's policy. + retries = urllib3.Retry.from_int(session._pool.retries).new( + read=0, status=0, other=0, raise_on_status=False + ) with ( storage_span( "resumable.put", @@ -174,6 +196,7 @@ def put( f"{session._make_url(self.key)}?{query}", headers=headers, body=body, + retries=retries, preload_content=True, decode_content=True, ) @@ -207,3 +230,156 @@ def cancel(self) -> None: else RequestError ) raise_for_status(response, error_type=error_type) + + +@contextmanager +def prepare_body( + contents: bytes | IO[bytes], + source_size: int, + compress: bool, + spool_dir: str | PathLike[str] | None, +) -> Iterator[tuple[IO[bytes], int, int] | None]: + """Yield the body, source size, and compressed size (equal if uncompressed). + + Reuse sized, seekable input when no compression is needed. Otherwise spool + the compressed bytes without closing the caller's stream. + Yield None if temporary-file creation fails, before consuming any input. + """ + with ( + BytesIO(contents) if isinstance(contents, bytes) else nullcontext(contents) + ) as source: + if not compress: + yield source, source_size, source_size + return + + try: + temporary = TemporaryFile(dir=spool_dir) + except OSError: + yield None + return + + with temporary as body: + source_size, compressed_size = zstandard.ZstdCompressor().copy_stream( + source, body + ) + body.seek(0) + yield body, source_size, compressed_size + + +def get_size(source: bytes | IO[bytes]) -> int | None: + """Return the buffer length or remaining stream size, or None if unsupported. + + Streams are measured by seeking to the end, which may read data in transforming + streams. Failure to restore the original position propagates so callers cannot + upload from an unexpected position. + """ + if isinstance(source, bytes): + return len(source) + + try: + if not source.seekable(): + return None + start = source.tell() + except (AttributeError, OSError): + return None + + try: + source.seek(0, SEEK_END) + return max(0, source.tell() - start) + except OSError: + return None + finally: + source.seek(start) + + +def is_transient(error: Exception) -> bool: + if isinstance(error, urllib3.exceptions.MaxRetryError): + # Exhausted status retries carry ResponseError rather than the response. + return isinstance(error.reason, urllib3.exceptions.ResponseError) or ( + isinstance(error.reason, Exception) and is_transient(error.reason) + ) + if isinstance(error, RequestError): + return error.status in (408, 429, 502, 503, 504) + return isinstance( + error, + ( + NoUploadProgress, + urllib3.exceptions.ReadTimeoutError, + urllib3.exceptions.ProtocolError, + ), + ) + + +def upload( + session: Session, + body: IO[bytes], + compressed_size: int, + key: str | None = None, + compression: Compression | None = None, + content_type: str | None = None, + metadata: dict[str, str] | None = None, + expiration_policy: ExpirationPolicy | None = None, + origin: str | None = None, + filename: str | None = None, +) -> str | None: + start = body.tell() + try: + handle = session._create_upload( + compressed_size, + key=key, + compression=compression, + content_type=content_type, + metadata=metadata, + expiration_policy=expiration_policy, + origin=origin, + filename=filename, + ) + except Exception: + handle = None + if handle is None: + return None + + try: + offset = 0 + retries = 0 + probe = False + while True: + try: + if probe: + result = handle.progress() + else: + body.seek(start + offset) + try: + result = handle.put(offset, (body, compressed_size - offset)) + except UploadOffsetMismatch as error: + result = UploadIncomplete(error.offset) + if ( + isinstance(result, UploadIncomplete) + and result.offset == offset + and not probe + ): + raise NoUploadProgress() + except Exception as error: + if not is_transient(error) or retries >= 2: + raise + time.sleep(2 * 2**retries + random.uniform(0, 1)) + retries += 1 + probe = True + continue + + if isinstance(result, UploadComplete): + return handle.key + if not offset <= result.offset <= compressed_size: + raise ValueError("Invalid upload offset") + if result.offset > offset: + retries = 0 + offset = result.offset + probe = False + except Exception as error: + try: + handle.cancel() + except Exception: + pass + status = error.status if isinstance(error, RequestError) else None + response = error.response if isinstance(error, RequestError) else None + raise RequestError("upload failed", status, response) from error diff --git a/clients/python/src/objectstore_client/client.py b/clients/python/src/objectstore_client/client.py index 7ae00d65..d4f82b3f 100644 --- a/clients/python/src/objectstore_client/client.py +++ b/clients/python/src/objectstore_client/client.py @@ -3,10 +3,12 @@ import math import warnings from collections.abc import Mapping, Sequence +from contextlib import nullcontext from dataclasses import asdict, dataclass from datetime import UTC, datetime, timedelta from importlib.metadata import version from io import BytesIO +from os import PathLike from typing import IO, Any, Literal, NamedTuple, cast from urllib.parse import urlparse @@ -78,10 +80,12 @@ def __init__( name: str, compression: Compression = "zstd", expiration_policy: ExpirationPolicy | None = None, + resumable_threshold_bytes: int | Literal["none"] = 256 * 1024 * 1024, ): """Create a usecase. - Raises ``ValueError`` if the name violates the naming rules on ``Usecase``. + Raises ``ValueError`` if the name violates the naming rules on ``Usecase``, + or if the ``resumable_threshold_bytes`` is invalid. """ if not 3 <= len(name) <= 64: raise ValueError("usecase name must be 3–64 ASCII characters long") @@ -90,6 +94,14 @@ def __init__( "usecase name must contain only ASCII letters, digits, '_' or '-'" ) self._name = name + + if resumable_threshold_bytes != "none" and ( + type(resumable_threshold_bytes) is not int or resumable_threshold_bytes < 0 + ): + raise ValueError( + 'resumable_threshold_bytes must be a non-negative integer or "none"' + ) + self._resumable_threshold_bytes = resumable_threshold_bytes self._compression = compression self._expiration_policy = expiration_policy @@ -165,6 +177,8 @@ class Client: as-is for every request. Use a ``SecretKey`` for internal services that have access to the signing key, and a string for external services that receive a token from another source. + spool_dir: Directory that temporary files are created in when preparing + resumable uploads. ``None`` uses Python's default temporary directory. """ def __init__( @@ -176,6 +190,7 @@ def __init__( timeout_ms: float | None = None, connection_kwargs: Mapping[str, Any] | None = None, token: TokenProvider | None = None, + spool_dir: str | PathLike[str] | None = None, ): if propagate_traces: warnings.warn( @@ -214,6 +229,7 @@ def __init__( self._metrics_backend = metrics_backend or NoOpMetricsBackend() self._propagate_traces = propagate_traces self._token = token + self._spool_dir = spool_dir def session(self, usecase: Usecase, **scopes: str | int | bool) -> Session: """ @@ -249,6 +265,7 @@ def session(self, usecase: Usecase, **scopes: str | int | bool) -> Session: usecase, Scope(**scopes), self._token, + self._spool_dir, ) @@ -268,6 +285,7 @@ def __init__( usecase: Usecase, scope: Scope, token: TokenProvider | None = None, + spool_dir: str | PathLike[str] | None = None, ): self._pool = pool self._base_path = base_path @@ -276,6 +294,7 @@ def __init__( self._usecase = usecase self._scope = scope self._token = token + self._spool_dir = spool_dir def mint_token( self, @@ -412,6 +431,7 @@ def put( expiration_policy: ExpirationPolicy | None = None, origin: str | None = None, filename: str | None = None, + resumable_threshold_bytes: int | Literal["none"] | None = None, ) -> str: """ Uploads the given `contents` to blob storage. @@ -419,7 +439,7 @@ def put( If no `key` is provided, one will be automatically generated and returned from this function. - The client will select the configured `default_compression` if none is given + The client will select the Usecase's configured compression if none is given explicitly. This can be overridden by explicitly giving a `compress` argument. Providing `"none"` as the argument will instruct the client to not apply any compression to this upload, which is useful for uncompressible formats. @@ -432,6 +452,28 @@ def put( to attempt to guess a `content_type` based on magic bytes. `compression` is deprecated in favor of `compress`. + + .. rubric:: Resumable Uploads + + This method uses Objectstore's Resumable Uploads API when the size of + ``contents`` is at least ``resumable_threshold_bytes``. + Pass ``resumable_threshold_bytes="none"`` to disable Resumable Uploads. + When not overridden, this threshold defaults to the one set on the Usecase. + + Transient failures and writes that make no progress are retried up to + twice. The retry count resets whenever the server reports a higher offset. + + For seekable binary streams, the remaining size is measured by seeking to + the end and restoring the original position. Prefer streams with efficient + seeking: transforming streams may read their entire input to find the end. + Non-seekable streams and streams whose size cannot be determined use a + regular upload without spooling. + + Resumable uploads reuse the input unless compression is needed, in which + case the compressed payload is spooled in a temporary file + (``tempfile.TemporaryFile``) on disk before sending. + Set ``Client(spool_dir=)`` if you wish to customize the directory where + the temporary files are created in. """ if compression is not None: warnings.warn( @@ -450,16 +492,22 @@ def put( if precompressed and precompressed != "zstd": raise ValueError(f"Invalid compression: {precompressed}") - body = BytesIO(contents) if isinstance(contents, bytes) else contents - original_body: IO[bytes] = body + if resumable_threshold_bytes is None: + resumable_threshold_bytes = self._usecase._resumable_threshold_bytes + if resumable_threshold_bytes != "none" and ( + type(resumable_threshold_bytes) is not int or resumable_threshold_bytes < 0 + ): + raise ValueError( + 'resumable_threshold_bytes must be a non-negative integer or "none"' + ) + + source_size = None + if resumable_threshold_bytes != "none": + source_size = _resumable.get_size(contents) encoding = precompressed or compress or self._usecase._compression compress_with = encoding if precompressed is None else "none" - if compress_with == "zstd": - cctx = zstandard.ZstdCompressor() - body = cctx.stream_reader(original_body) - body = cast(IO[bytes], utils._ZstdCompressionReaderWrapper(body)) headers = self._metadata_headers( compression=encoding, @@ -479,37 +527,56 @@ def put( self._metrics_backend, "put", self._usecase.name ) as metrics, ): - retries = None # by default use the pool's value, set by the Client - if compress_with != "none": - # For on-the-fly compression, don't attempt read retries, - # as the stream cannot be rewound after data has been consumed. - pool_retries = self._pool.retries - if isinstance(pool_retries, urllib3.Retry): - retries = pool_retries.new(read=0) - elif isinstance(pool_retries, int): - retries = urllib3.Retry(pool_retries, read=0) - - response = self._pool.request( - "POST" if not key else "PUT", - self._make_url(key), - body=body, - headers=headers, - preload_content=True, - decode_content=True, - retries=retries, - ) - raise_for_status(response) - res = response.json() + result_key: str | None + if ( + resumable_threshold_bytes != "none" + and source_size is not None + and source_size >= resumable_threshold_bytes + ): + with _resumable.prepare_body( + contents, + source_size, + compress_with != "none", + self._spool_dir, + ) as prepared: + if prepared is None: + # Temporary-file creation failed before consuming the input. + result_key, source_size, compressed_size = self._put_direct( + contents, key, headers, compress=compress_with != "none" + ) + else: + body, source_size, compressed_size = prepared + result_key = _resumable.upload( + self, + body, + compressed_size, + key=key, + compression=encoding, + content_type=content_type, + metadata=metadata, + expiration_policy=expiration_policy, + origin=origin, + filename=filename, + ) + if result_key is None: + # Session creation failed or resumables are unsupported. + # Send the prepared body without compressing it again. + headers["Content-Length"] = str(compressed_size) + result_key, _, _ = self._put_direct( + body, key, headers, compress=False + ) + else: + result_key, source_size, compressed_size = self._put_direct( + contents, key, headers, compress=compress_with != "none" + ) - # Must do this after streaming `body` as that's what is responsible - # for advancing the seek position in both streams - if precompressed is None: - metrics.record_uncompressed_size(original_body.tell()) - if encoding != "none": - metrics.record_compressed_size(body.tell(), encoding) + if precompressed is None and source_size is not None: + metrics.record_uncompressed_size(source_size) + if encoding != "none" and compressed_size is not None: + metrics.record_compressed_size(compressed_size, encoding) # Set after the response, since the key may be server-generated. - span.set_attribute("objectstore.key", res["key"]) + span.set_attribute("objectstore.key", result_key) span.set_attribute("objectstore.compression", encoding) if metrics.uncompressed_size is not None: span.set_attribute( @@ -519,7 +586,62 @@ def put( span.set_attribute( "objectstore.compressed_size", metrics.compressed_size ) - return res["key"] + return result_key + + def _put_direct( + self, + contents: bytes | IO[bytes], + key: str | None, + headers: dict[str, str], + compress: bool, + ) -> tuple[str, int | None, int | None]: + """Return the key and final source/body positions when available.""" + with ( + BytesIO(contents) if isinstance(contents, bytes) else nullcontext(contents) + ) as original_body: + body = original_body + retries = None # by default use the pool's value, set by the Client + try: + if compress: + cctx = zstandard.ZstdCompressor() + body = cctx.stream_reader(original_body, closefd=False) + body = cast(IO[bytes], utils._ZstdCompressionReaderWrapper(body)) + + if compress: + # For on-the-fly compression, don't attempt read retries, + # as the stream cannot be rewound after data has been consumed. + pool_retries = self._pool.retries + if isinstance(pool_retries, urllib3.Retry): + retries = pool_retries.new(read=0) + elif isinstance(pool_retries, int): + retries = urllib3.Retry(pool_retries, read=0) + + response = self._pool.request( + "POST" if not key else "PUT", + self._make_url(key), + body=body, + headers=headers, + preload_content=True, + decode_content=True, + retries=retries, + ) + raise_for_status(response) + res = response.json() + + # Read positions after streaming, without failing an upload if + # the source does not support tell(). + try: + source_size = original_body.tell() + except (AttributeError, OSError, ValueError): + source_size = None + try: + compressed_size = body.tell() + except (AttributeError, OSError, ValueError): + compressed_size = None + return res["key"], source_size, compressed_size + finally: + if body is not original_body: + body.close() def get( self, diff --git a/clients/python/src/objectstore_client/errors.py b/clients/python/src/objectstore_client/errors.py index 2f94eff6..2f623cbd 100644 --- a/clients/python/src/objectstore_client/errors.py +++ b/clients/python/src/objectstore_client/errors.py @@ -6,7 +6,7 @@ class RequestError(Exception): """Exception raised if an API call to Objectstore fails.""" - def __init__(self, message: str, status: int, response: str): + def __init__(self, message: str, status: int | None, response: str | None): super().__init__(message) self.status = status self.response = response diff --git a/clients/python/tests/test_e2e.py b/clients/python/tests/test_e2e.py index 8529d57c..1f08b373 100644 --- a/clients/python/tests/test_e2e.py +++ b/clients/python/tests/test_e2e.py @@ -12,6 +12,7 @@ from datetime import timedelta from io import BytesIO from pathlib import Path +from unittest.mock import Mock, patch import pytest import urllib3 @@ -24,6 +25,7 @@ Usecase, ) from objectstore_client._resumable import ( + ResumableUpload, ResumableUploadUnavailable, UploadComplete, UploadIncomplete, @@ -47,6 +49,9 @@ class UnrewindableStream(BytesIO): """Read-only stream that cannot report or restore position.""" + def seekable(self) -> bool: + return False + def seek(self, offset: int, whence: int = 0) -> int: raise OSError("stream is not seekable") @@ -1153,3 +1158,94 @@ def test_put_stores_under_literal_key(server_url: str) -> None: status, body = _fetch(url) assert status == 200 assert body == payload + + +@pytest.mark.parametrize("source_kind", ["bytes", "buffer", "stream"]) +@pytest.mark.parametrize("compression", ["none", "zstd", "precompressed"]) +@pytest.mark.parametrize("threshold", [1, 2048]) +def test_automatic_resumable_upload( + server_url: str, source_kind: str, compression: str, threshold: int +) -> None: + session = Client(server_url, token=TestSecretKey.get()).session( + Usecase( + "test-usecase", + compression="zstd" if compression == "zstd" else "none", + expiration_policy=TimeToLive(timedelta(days=1)), + resumable_threshold_bytes=threshold, + ), + org=42, + ) + contents = b"file contents\n" * 100 + precompressed = compression == "precompressed" + encoded = ( + zstandard.ZstdCompressor().compress(contents) if precompressed else contents + ) + with ( + BytesIO() + if source_kind == "buffer" + else tempfile.NamedTemporaryFile() as source, + patch.object( + ResumableUpload, "put", autospec=True, side_effect=ResumableUpload.put + ) as put, + ): + source.write(b"skip this prefix" + encoded) + source.seek(len(b"skip this prefix")) + key = session.put( + source if source_kind in ("buffer", "stream") else encoded, + precompressed="zstd" if precompressed else None, + content_type="text/plain", + metadata={"source": "file"}, + origin="203.0.113.42", + filename="example.txt", + ) + assert put.call_count == int(threshold == 1) + assert not source.closed + stored = session.head(key) + assert stored is not None + assert stored.compression == ("zstd" if compression != "none" else None) + assert stored.content_type == "text/plain" + assert stored.filename == "example.txt" + assert stored.origin == "203.0.113.42" + assert stored.custom == {"source": "file"} + retrieved = session.get(key) + assert retrieved is not None + assert retrieved.payload.read() == contents + if precompressed: + raw = session.get(key, decompress=False) + assert raw is not None + assert raw.payload.read() == encoded + + +@pytest.mark.parametrize( + "error", [None, RequestError("creation rejected", 403, "forbidden")] +) +def test_resumable_creation_fallback( + server_url: str, monkeypatch: pytest.MonkeyPatch, error: RequestError | None +) -> None: + session = Client(server_url, token=TestSecretKey.get()).session( + Usecase("test-usecase", resumable_threshold_bytes=1) + ) + create = Mock(return_value=None, side_effect=error) + monkeypatch.setattr(session, "_create_upload", create) + source = BytesIO(b"prefixpayload") + source.seek(len(b"prefix")) + key = session.put(source, compress="none") + create.assert_called_once() + assert not source.closed + retrieved = session.get(key) + assert retrieved is not None + assert retrieved.payload.read() == b"payload" + + +@pytest.mark.parametrize("compress", [False, True]) +def test_nonseekable_upload(server_url: str, compress: bool) -> None: + session = Client(server_url, token=TestSecretKey.get()).session( + Usecase("test-usecase", resumable_threshold_bytes=1) + ) + with UnrewindableStream(b"payload") as source: + key = session.put(source, compress="zstd" if compress else "none") + assert not source.closed + + retrieved = session.get(key) + assert retrieved is not None + assert retrieved.payload.read() == b"payload" diff --git a/clients/python/tests/test_upload.py b/clients/python/tests/test_upload.py new file mode 100644 index 00000000..39e9c33b --- /dev/null +++ b/clients/python/tests/test_upload.py @@ -0,0 +1,124 @@ +from io import BytesIO +from typing import Any +from unittest.mock import Mock + +import pytest +import urllib3 +from objectstore_client import Client, RequestError, Session, Usecase +from objectstore_client._resumable import NoUploadProgress + + +@pytest.fixture +def session(monkeypatch: pytest.MonkeyPatch) -> Session: + monkeypatch.setattr("objectstore_client._resumable.time.sleep", Mock()) + return Client("http://localhost:8888").session( + Usecase("test", compression="none", resumable_threshold_bytes=4) + ) + + +@pytest.mark.parametrize("complete", [False, True]) +@pytest.mark.parametrize("offsets,delays", [([3], [2]), ([0, 3, 3, 6], [2, 4, 2, 4])]) +def test_failure_recovery( + session: Session, + monkeypatch: pytest.MonkeyPatch, + complete: bool, + offsets: list[int], + delays: list[int], +) -> None: + failure = urllib3.exceptions.ReadTimeoutError(session._pool, "/", "lost response") + responses: list[urllib3.HTTPResponse | Exception] = [ + urllib3.HTTPResponse(status=201, body=b'{"key":"key","session":"token"}') + ] + for index, offset in enumerate(offsets): + responses.extend( + [ + failure, + urllib3.HTTPResponse( + status=201 if complete and index == len(offsets) - 1 else 204, + headers={"Upload-Offset": str(offset)}, + ), + ] + ) + responses.append(urllib3.HTTPResponse(status=201)) + outcomes = Mock(side_effect=responses) + sleep = Mock() + monkeypatch.setattr("objectstore_client._resumable.time.sleep", sleep) + sent = [] + + def request(*args: Any, **kwargs: Any) -> urllib3.HTTPResponse: + if body := kwargs.get("body"): + sent.append((kwargs["headers"]["Upload-Offset"], body.read())) + return outcomes() + + monkeypatch.setattr(session._pool, "_make_request", request) + source = BytesIO(b"prefixabcdefgh") + source.seek(len(b"prefix")) + assert session.put(source) == "key" + assert not source.closed + write_offsets = [0, *offsets[:-1]] if complete else [0, *offsets] + assert sent == [(str(offset), b"abcdefgh"[offset:]) for offset in write_offsets] + assert sleep.call_count == len(delays) + for call, delay in zip(sleep.call_args_list, delays, strict=True): + assert delay <= call.args[0] <= delay + 1 + + +@pytest.mark.parametrize("failure_kind", ["timeout", "status", "no_progress"]) +def test_retry_exhaustion( + session: Session, + monkeypatch: pytest.MonkeyPatch, + failure_kind: str, +) -> None: + failure = urllib3.exceptions.ReadTimeoutError(session._pool, "/", "lost response") + sleep = Mock() + monkeypatch.setattr("objectstore_client._resumable.time.sleep", sleep) + + def request(*args: Any, **kwargs: Any) -> urllib3.HTTPResponse: + headers = kwargs["headers"] + if args[1] == "DELETE": + return urllib3.HTTPResponse(status=204) + if "Upload-Length" in headers: + return urllib3.HTTPResponse( + status=201, body=b'{"key":"key","session":"token"}' + ) + if headers["Upload-Offset"] == "*": + return urllib3.HTTPResponse(status=204, headers={"Upload-Offset": "0"}) + if failure_kind == "status": + return urllib3.HTTPResponse(status=503, body=b"unavailable") + if failure_kind == "no_progress": + return urllib3.HTTPResponse(status=204, headers={"Upload-Offset": "0"}) + raise failure + + make_request = Mock(side_effect=request) + monkeypatch.setattr(session._pool, "_make_request", make_request) + with pytest.raises(RequestError, match="^upload failed$") as raised: + session.put(b"payload") + if failure_kind == "status": + assert isinstance(raised.value.__cause__, RequestError) + assert raised.value.__cause__.status == 503 + assert raised.value.response == "unavailable" + elif failure_kind == "timeout": + assert isinstance(raised.value.__cause__, urllib3.exceptions.MaxRetryError) + assert raised.value.__cause__.reason is failure + assert raised.value.response is None + else: + assert isinstance(raised.value.__cause__, NoUploadProgress) + assert str(raised.value.__cause__) == "Upload made no progress" + # Creation, two progress queries, three writes, and cancellation. + assert make_request.call_count == 1 + 2 + 3 + 1 + assert sleep.call_count == 2 + for call, delay in zip(sleep.call_args_list, [2, 4], strict=True): + assert delay <= call.args[0] <= delay + 1 + + +def test_sizing_must_restore_position( + session: Session, monkeypatch: pytest.MonkeyPatch +) -> None: + source = Mock(wraps=BytesIO(b"payload")) + source.seek.side_effect = [None, OSError("cannot restore position")] + request = Mock() + monkeypatch.setattr(session._pool, "request", request) + + with pytest.raises(OSError, match="cannot restore position"): + session.put(source) + + request.assert_not_called()