From 76549e4fb21d7ef1a991b72c880a0d3e231c1888 Mon Sep 17 00:00:00 2001 From: Waylon Date: Tue, 20 Oct 2026 23:04:14 +0800 Subject: [PATCH 1/2] feat: multipart concurrent upload to ftrack.server Large files written through the server accessor are split into parts and uploaded concurrently (ThreadPoolExecutor, one requests.Session per worker, default 6 workers via FTRACK_API_UPLOAD_MAX_CONCURRENCY). Session.get_upload_metadata accepts parts; adds Session.complete_multipart_upload. Small files keep the single PUT. --- AGENTS.md | 1 + source/ftrack_api/_uploader.py | 234 +++++++++++++++++++++++++++ source/ftrack_api/_uploader.pyi | 21 +++ source/ftrack_api/accessor/server.py | 76 ++++++--- source/ftrack_api/session.py | 35 +++- source/ftrack_api/session.pyi | 8 +- test/unit/accessor/test_server.py | 107 ++++++++++++ test/unit/test_uploader.py | 221 +++++++++++++++++++++++++ 8 files changed, 683 insertions(+), 20 deletions(-) create mode 100644 source/ftrack_api/_uploader.py create mode 100644 source/ftrack_api/_uploader.pyi create mode 100644 test/unit/test_uploader.py diff --git a/AGENTS.md b/AGENTS.md index 2df4b3df..1b960256 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -83,6 +83,7 @@ source/ftrack_api/ ├── exception.py # Custom exceptions ├── plugin.py # Plugin system ├── operation.py # Operation tracking +├── _uploader.py # Multi-part concurrent upload to ftrack.server └── _centralized_storage_scenario.py ``` diff --git a/source/ftrack_api/_uploader.py b/source/ftrack_api/_uploader.py new file mode 100644 index 00000000..e63db0a9 --- /dev/null +++ b/source/ftrack_api/_uploader.py @@ -0,0 +1,234 @@ +# :coding: utf-8 +# :copyright: Copyright (c) 2026 ftrack + +"""Multi-part upload of file data to the ftrack.server location. + +Part sizing and retry back-off follow the ftrack JavaScript API client +(``source/util/get_chunk_size.ts`` and ``source/util/back_off.ts``). +""" + +import concurrent.futures +import logging +import os +import random +import threading + +import requests + +logger = logging.getLogger(__name__) + +MEGABYTE = 1024 * 1024 +GIGABYTE = 1024 * MEGABYTE + +#: Maximum number of parts the server accepts for one upload. +MAX_PARTS = 10000 + +#: Maximum number of parts uploaded concurrently. +MAX_CONCURRENCY = int(os.getenv("FTRACK_API_UPLOAD_MAX_CONCURRENCY", 0)) or 6 + +#: Maximum number of attempts per part, including the first one. +MAX_ATTEMPTS = 16 + +#: Delay before the first retry is twice this value, in seconds. +INITIAL_DELAY = 0.5 + +#: Maximum delay between attempts, in seconds. +MAX_DELAY = 64 + +#: Random delay of up to this many seconds added to each retry. +MAX_JITTER = 0.5 + + +def get_chunk_size(file_size): + """Return part size in bytes for a multi-part upload of *file_size* bytes. + + Raise :exc:`ValueError` if *file_size* exceeds what can be uploaded within + :data:`MAX_PARTS` parts. + """ + if file_size > 128 * MEGABYTE * MAX_PARTS: + raise ValueError("File is larger than the maximum allowed (1,280 GB).") + + if file_size > 64 * GIGABYTE: + return 128 * MEGABYTE + if file_size > 8 * GIGABYTE: + return 64 * MEGABYTE + if file_size > 1 * GIGABYTE: + return 32 * MEGABYTE + if file_size > 128 * MEGABYTE: + return 16 * MEGABYTE + + return 8 * MEGABYTE + + +def get_part_count(file_size): + """Return number of parts to upload *file_size* bytes in. + + Return None when the data should be uploaded with a single PUT request. + """ + chunk_size = get_chunk_size(file_size) + part_count = -(-file_size // chunk_size) + if part_count <= 2: + return None + + return part_count + + +def upload_parts(file_object, file_size, urls, timeout): + """Upload *file_object* to signed part *urls* and return uploaded parts. + + *file_object* must be seekable and hold *file_size* bytes. It is shared by + all worker threads and must not be used by the caller until this returns. + + *urls* is the list of ``{"part_number": int, "signed_url": str}`` + returned by :meth:`ftrack_api.session.Session.get_upload_metadata`. + + *timeout* is passed to each PUT request. + + Return a list of ``{"part_number": int, "e_tag": str}`` sorted by part + number, suitable for + :meth:`ftrack_api.session.Session.complete_multipart_upload`. + + If any part fails after retries, parts not yet started are cancelled and + the error of that part is raised. + """ + upload = _MultipartUpload(file_object, get_chunk_size(file_size), timeout) + return upload.run(urls) + + +def _is_retryable(error): + """Return whether a failed part PUT raising *error* should be retried.""" + # Certificate failures are configuration errors that retries cannot fix. + if isinstance(error, requests.exceptions.SSLError): + return False + + if isinstance( + error, (requests.exceptions.ConnectionError, requests.exceptions.Timeout) + ): + return True + + if isinstance(error, requests.exceptions.HTTPError): + response = error.response + return response is not None and ( + response.status_code == 429 or response.status_code >= 500 + ) + + return False + + +class _MultipartUpload(object): + """Concurrent upload of parts read from one shared file object.""" + + def __init__(self, file_object, chunk_size, timeout): + """Initialise upload of *file_object* in *chunk_size* parts.""" + self._file = file_object + self._file_lock = threading.Lock() + self._chunk_size = chunk_size + self._timeout = timeout + self._aborted = threading.Event() + self._error = None + self._error_lock = threading.Lock() + self._local = threading.local() + self._http_sessions = [] + self._http_sessions_lock = threading.Lock() + + def run(self, urls): + """Upload all part *urls* and return uploaded parts.""" + executor = concurrent.futures.ThreadPoolExecutor( + max_workers=min(MAX_CONCURRENCY, len(urls)), + thread_name_prefix="ftrack-api-upload", + ) + futures = [] + try: + for url in urls: + futures.append(executor.submit(self._upload_part, url)) + + concurrent.futures.wait( + futures, return_when=concurrent.futures.FIRST_EXCEPTION + ) + finally: + # Stops retries and queued parts on failure or interruption; no-op + # once every part has completed. + self._aborted.set() + for future in futures: + future.cancel() + + executor.shutdown(wait=True) + for http_session in self._http_sessions: + http_session.close() + + if self._error is not None: + raise self._error + + parts = [future.result() for future in futures] + parts.sort(key=lambda part: part["part_number"]) + return parts + + def _upload_part(self, url): + """Upload part described by *url*; abort the upload if it fails.""" + try: + return self._send_part(url) + except Exception as error: + # Record before aborting so the first failure is the one raised, + # not the abort it causes in other parts. + with self._error_lock: + if self._error is None: + self._error = error + self._aborted.set() + raise + + def _send_part(self, url): + """Upload part described by *url* and return its number and ETag.""" + part_number = url["part_number"] + if self._aborted.is_set(): + raise RuntimeError("Upload aborted before part {0}.".format(part_number)) + + with self._file_lock: + self._file.seek((part_number - 1) * self._chunk_size) + data = self._file.read(self._chunk_size) + + http_session = self._get_http_session() + attempt = 0 + delay = INITIAL_DELAY + while True: + try: + e_tag = self._put(http_session, url["signed_url"], part_number, data) + return {"part_number": part_number, "e_tag": e_tag} + except Exception as error: + attempt += 1 + if attempt >= MAX_ATTEMPTS or not _is_retryable(error): + raise + + delay = min(MAX_DELAY, delay * 2) + wait = delay + random.uniform(0, MAX_JITTER) + logger.warning( + "Failed to upload part %d (%s), retrying in %.1f seconds.", + part_number, + error, + wait, + ) + if self._aborted.wait(wait): + raise + + def _put(self, http_session, signed_url, part_number, data): + """PUT *data* to *signed_url* and return the ETag without quotes.""" + response = http_session.put(signed_url, data=data, timeout=self._timeout) + response.raise_for_status() + + e_tag = response.headers.get("ETag") + if not e_tag: + raise RuntimeError( + "Upload of part {0} returned no ETag header.".format(part_number) + ) + + return e_tag.replace('"', "") + + def _get_http_session(self): + """Return the requests session of the calling worker thread.""" + http_session = getattr(self._local, "http_session", None) + if http_session is None: + http_session = requests.Session() + self._local.http_session = http_session + with self._http_sessions_lock: + self._http_sessions.append(http_session) + + return http_session diff --git a/source/ftrack_api/_uploader.pyi b/source/ftrack_api/_uploader.pyi new file mode 100644 index 00000000..66b91c93 --- /dev/null +++ b/source/ftrack_api/_uploader.pyi @@ -0,0 +1,21 @@ +from _typeshed import Incomplete + +logger: Incomplete +MEGABYTE: int +GIGABYTE: int +MAX_PARTS: int +MAX_CONCURRENCY: int +MAX_ATTEMPTS: int +INITIAL_DELAY: float +MAX_DELAY: int +MAX_JITTER: float + +def get_chunk_size(file_size: int) -> int: ... +def get_part_count(file_size: int) -> int | None: ... +def upload_parts( + file_object, file_size: int, urls, timeout +) -> list[dict[str, Incomplete]]: ... + +class _MultipartUpload: + def __init__(self, file_object, chunk_size: int, timeout) -> None: ... + def run(self, urls) -> list[dict[str, Incomplete]]: ... diff --git a/source/ftrack_api/accessor/server.py b/source/ftrack_api/accessor/server.py index 0c9e5f78..0e7cac87 100644 --- a/source/ftrack_api/accessor/server.py +++ b/source/ftrack_api/accessor/server.py @@ -11,6 +11,7 @@ from ..data import String import ftrack_api.exception import ftrack_api.symbol +import ftrack_api._uploader class ServerFile(String): @@ -94,32 +95,71 @@ def _write(self): if component["file_type"]: name = "{0}.{1}".format(name, component["file_type"].lstrip(".")) + file_size = self._get_size() try: - metadata = self._session.get_upload_metadata( - component_id=self.resource_identifier, - file_name=name, - file_size=self._get_size(), - checksum=self._compute_checksum(), + parts = ftrack_api._uploader.get_part_count(file_size) + except ValueError as error: + raise ftrack_api.exception.AccessorOperationFailedError( + "Failed to put file to server: {0}".format(error) ) + + try: + metadata = None + if parts is not None: + # Content-MD5 covers a single PUT body, so it is not sent for + # multi-part uploads. + try: + metadata = self._session.get_upload_metadata( + component_id=self.resource_identifier, + file_name=name, + file_size=file_size, + parts=parts, + ) + except ftrack_api.exception.ServerError as error: + # Raised when multi-part uploads are enabled on the server + # but its storage is not S3 compatible. + self._session.logger.debug( + "Multi-part upload unavailable, using single PUT: " + "{0}".format(error) + ) + else: + if "urls" not in metadata: + # Multi-part uploads are disabled on the server. + metadata = None + + if metadata is None: + metadata = self._session.get_upload_metadata( + component_id=self.resource_identifier, + file_name=name, + file_size=file_size, + checksum=self._compute_checksum(), + ) except Exception as error: raise ftrack_api.exception.AccessorOperationFailedError( "Failed to get put metadata: {0}.".format(error) ) - # Ensure at beginning of file before put. - self.seek(0) - - # Put the file based on the metadata. - response = requests.put( - metadata["url"], - data=self.wrapped_file, - headers=metadata["headers"], - timeout=self._timeout, - ) - try: - response.raise_for_status() - except requests.exceptions.HTTPError as error: + if "urls" in metadata: + uploaded_parts = ftrack_api._uploader.upload_parts( + self.wrapped_file, file_size, metadata["urls"], self._timeout + ) + self._session.complete_multipart_upload( + component_id=self.resource_identifier, + upload_id=metadata["upload_id"], + parts=uploaded_parts, + ) + else: + # Ensure at beginning of file before put. + self.seek(0) + response = requests.put( + metadata["url"], + data=self.wrapped_file, + headers=metadata["headers"], + timeout=self._timeout, + ) + response.raise_for_status() + except Exception as error: raise ftrack_api.exception.AccessorOperationFailedError( "Failed to put file to server: {0}.".format(error) ) diff --git a/source/ftrack_api/session.py b/source/ftrack_api/session.py index fdc79d08..51c892b4 100644 --- a/source/ftrack_api/session.py +++ b/source/ftrack_api/session.py @@ -2249,7 +2249,9 @@ def encode_media(self, media, version_id=None, keep_original="auto"): return self.get("Job", result[0]["job_id"]) - def get_upload_metadata(self, component_id, file_name, file_size, checksum=None): + def get_upload_metadata( + self, component_id, file_name, file_size, checksum=None, parts=None + ): """Return URL and headers used to upload data for *component_id*. *file_name* and *file_size* should match the components details. @@ -2261,6 +2263,19 @@ def get_upload_metadata(self, component_id, file_name, file_size, checksum=None) the base64-encoded 128-bit MD5 digest of the message (without the headers) according to RFC 1864. This can be used as a message integrity check to verify that the data is the same data that was originally sent. + + Specify *parts* to request a multi-part upload in that many parts. If + multi-part uploads are enabled on the server the result contains + ``upload_id`` and ``urls``, a list of + ``{"part_number": int, "signed_url": str}`` to PUT each part to, + instead of ``url`` and ``headers``. Finish the upload with + :meth:`complete_multipart_upload`. If they are disabled *parts* is + ignored and the result is for a single PUT. If they are enabled but the + server storage is not S3 compatible + :exc:`ftrack_api.exception.ServerError` is raised. + + *checksum* only applies to a single PUT and should not be combined with + *parts*. """ operation = { "action": "get_upload_metadata", @@ -2269,6 +2284,8 @@ def get_upload_metadata(self, component_id, file_name, file_size, checksum=None) "file_size": file_size, "checksum": checksum, } + if parts is not None: + operation["parts"] = parts try: result = self.call([operation]) @@ -2286,6 +2303,22 @@ def get_upload_metadata(self, component_id, file_name, file_size, checksum=None) return result[0] + def complete_multipart_upload(self, component_id, upload_id, parts): + """Assemble uploaded *parts* of multi-part upload *upload_id*. + + *upload_id* is returned by :meth:`get_upload_metadata` when called with + *parts*. *parts* is a list of ``{"part_number": int, "e_tag": str}`` + for every uploaded part, ordered by part number. + """ + operation = { + "action": "complete_multipart_upload", + "component_id": component_id, + "upload_id": upload_id, + "parts": parts, + } + + return self.call([operation])[0] + def send_user_invite(self, user): """Send a invitation to the provided *user*. diff --git a/source/ftrack_api/session.pyi b/source/ftrack_api/session.pyi index 24b78a78..5d2bc53e 100644 --- a/source/ftrack_api/session.pyi +++ b/source/ftrack_api/session.pyi @@ -106,8 +106,14 @@ class Session: self, media, version_id: Incomplete | None = None, keep_original: str = "auto" ): ... def get_upload_metadata( - self, component_id, file_name, file_size, checksum: Incomplete | None = None + self, + component_id, + file_name, + file_size, + checksum: Incomplete | None = None, + parts: int | None = None, ): ... + def complete_multipart_upload(self, component_id, upload_id, parts): ... def send_user_invite(self, user) -> None: ... def send_user_invites(self, users) -> None: ... def send_review_session_invite(self, invitee) -> None: ... diff --git a/test/unit/accessor/test_server.py b/test/unit/accessor/test_server.py index b1789428..b8af2924 100644 --- a/test/unit/accessor/test_server.py +++ b/test/unit/accessor/test_server.py @@ -1,14 +1,17 @@ # :coding: utf-8 # :copyright: Copyright (c) 2015 ftrack +import os import uuid import pytest +import requests import ftrack_api import ftrack_api.exception import ftrack_api.accessor.server import ftrack_api.data +import ftrack_api._uploader @pytest.fixture(scope="module") @@ -16,6 +19,12 @@ def random_binary_data(): return uuid.uuid1().hex.encode() +@pytest.fixture(scope="module") +def large_binary_data(): + """Return data large enough to be uploaded in three parts.""" + return os.urandom(2 * 8 * ftrack_api._uploader.MEGABYTE + 1) + + def test_read_and_write(new_component, random_binary_data, session): """Read and write data from server accessor.""" accessor = ftrack_api.accessor.server._ServerAccessor(session) @@ -93,3 +102,101 @@ def test_remove_timeout(new_component, random_binary_data, session, monkeypatch) ftrack_api.exception.AccessorOperationFailedError, match="timed out" ): accessor.remove(new_component["id"]) + + +def test_read_and_write_multipart(new_component, large_binary_data, session, mocker): + """Read and write data uploaded in multiple parts.""" + mocker.patch.object( + session, + "complete_multipart_upload", + wraps=session.complete_multipart_upload, + ) + + accessor = ftrack_api.accessor.server._ServerAccessor(session) + http_file = accessor.open(new_component["id"], mode="wb") + http_file.write(large_binary_data) + http_file.close() + + assert session.complete_multipart_upload.call_count == 1 + + data = accessor.open(new_component["id"], "r") + assert data.read() == large_binary_data, "Read data is the same as written." + data.close() + + +@pytest.mark.parametrize("server_response", ["error", "single_url"]) +def test_write_multipart_unavailable( + new_component, large_binary_data, session, mocker, server_response +): + """Fall back to a single PUT when server cannot upload in parts.""" + get_upload_metadata = session.get_upload_metadata + + def without_multipart( + component_id, file_name, file_size, checksum=None, parts=None + ): + if parts is not None: + if server_response == "error": + # Raised when multi-part uploads are enabled on the server but + # its storage is not S3 compatible. + raise ftrack_api.exception.ServerError( + "Multipart upload is not supported." + ) + # Returned by servers with multi-part uploads disabled. + return get_upload_metadata(component_id, file_name, file_size) + + return get_upload_metadata(component_id, file_name, file_size, checksum) + + mocker.patch.object(session, "get_upload_metadata", side_effect=without_multipart) + mocker.patch.object(session, "complete_multipart_upload") + + accessor = ftrack_api.accessor.server._ServerAccessor(session) + http_file = accessor.open(new_component["id"], mode="wb") + http_file.write(large_binary_data) + http_file.close() + + assert session.complete_multipart_upload.call_count == 0 + assert session.get_upload_metadata.call_args.kwargs["checksum"] + + data = accessor.open(new_component["id"], "r") + assert data.read() == large_binary_data, "Read data is the same as written." + data.close() + + +def test_write_multipart_part_failure( + new_component, large_binary_data, session, mocker +): + """Fail without completing the upload when a part cannot be uploaded.""" + mocker.patch.object( + ftrack_api._uploader, + "upload_parts", + side_effect=requests.exceptions.HTTPError("403 Client Error"), + ) + mocker.patch.object(session, "complete_multipart_upload") + + accessor = ftrack_api.accessor.server._ServerAccessor(session) + http_file = accessor.open(new_component["id"], mode="wb") + http_file.write(large_binary_data) + + with pytest.raises(ftrack_api.exception.AccessorOperationFailedError, match="403"): + http_file.close() + + assert session.complete_multipart_upload.call_count == 0 + + +def test_write_too_large_file(new_component, random_binary_data, session, mocker): + """Fail before requesting upload metadata when the file has too many parts.""" + # Every non-empty file exceeds the maximum size when no parts are allowed. + mocker.patch.object(ftrack_api._uploader, "MAX_PARTS", 0) + mocker.patch.object(session, "get_upload_metadata") + + accessor = ftrack_api.accessor.server._ServerAccessor(session) + http_file = accessor.open(new_component["id"], mode="wb") + http_file.write(random_binary_data) + + with pytest.raises( + ftrack_api.exception.AccessorOperationFailedError, + match="Failed to put file to server: File is larger than the maximum", + ): + http_file.close() + + assert session.get_upload_metadata.call_count == 0 diff --git a/test/unit/test_uploader.py b/test/unit/test_uploader.py new file mode 100644 index 00000000..08128b75 --- /dev/null +++ b/test/unit/test_uploader.py @@ -0,0 +1,221 @@ +# :coding: utf-8 +# :copyright: Copyright (c) 2026 ftrack + +import io +import threading + +import pytest +import requests + +import ftrack_api._uploader as uploader + +MB = uploader.MEGABYTE +GB = uploader.GIGABYTE + + +@pytest.mark.parametrize( + "file_size, expected", + [ + (128 * MB, 8 * MB), + (128 * MB + 1, 16 * MB), + (1 * GB + 1, 32 * MB), + (8 * GB + 1, 64 * MB), + (64 * GB + 1, 128 * MB), + (128 * MB * uploader.MAX_PARTS, 128 * MB), + ], +) +def test_chunk_size_boundaries(file_size, expected): + """Part size steps up just above each threshold.""" + assert uploader.get_chunk_size(file_size) == expected + + +def test_chunk_size_rejects_too_large_file(): + with pytest.raises(ValueError): + uploader.get_chunk_size(128 * MB * uploader.MAX_PARTS + 1) + + +@pytest.mark.parametrize( + "file_size, expected", + [(0, None), (16 * MB, None), (16 * MB + 1, 3), (128 * MB, 16)], +) +def test_part_count_single_put_up_to_two_parts(file_size, expected): + """Up to two parts is uploaded with a single PUT.""" + assert uploader.get_part_count(file_size) == expected + + +class _Response(object): + def __init__(self, status_code=200, e_tag=None): + self.status_code = status_code + self.headers = {} if e_tag is None else {"ETag": e_tag} + + def raise_for_status(self): + if self.status_code >= 400: + raise requests.exceptions.HTTPError( + "{0} Error".format(self.status_code), response=self + ) + + +@pytest.fixture() +def fake_put(monkeypatch): + """Route part PUTs to a handler; return the recorded calls.""" + monkeypatch.setattr(uploader, "INITIAL_DELAY", 0) + monkeypatch.setattr(uploader, "MAX_JITTER", 0) + + state = {"handler": None, "calls": [], "lock": threading.Lock()} + + def put(session, url, data=None, timeout=None): + with state["lock"]: + state["calls"].append((url, data, timeout)) + attempt = sum(1 for call in state["calls"] if call[0] == url) + return state["handler"](url, data, attempt) + + monkeypatch.setattr(requests.Session, "put", put) + return state + + +def _urls(count, reverse=False): + numbers = range(count, 0, -1) if reverse else range(1, count + 1) + return [ + {"part_number": number, "signed_url": "part-{0}".format(number)} + for number in numbers + ] + + +def _upload(data, urls, chunk_size): + upload = uploader._MultipartUpload(io.BytesIO(data), chunk_size, timeout=7) + return upload.run(urls) + + +def test_upload_reads_part_offsets_and_sorts_by_part_number(fake_put): + """Each part gets its own slice; result is ordered with ETags unquoted.""" + data = bytes(bytearray(range(10))) + fake_put["handler"] = lambda url, body, attempt: _Response( + e_tag='"etag-{0}"'.format(url) + ) + + parts = _upload(data, _urls(4, reverse=True), chunk_size=3) + + assert parts == [ + {"part_number": number, "e_tag": "etag-part-{0}".format(number)} + for number in range(1, 5) + ] + bodies = {url: body for url, body, _ in fake_put["calls"]} + assert bodies == { + "part-1": data[0:3], + "part-2": data[3:6], + "part-3": data[6:9], + "part-4": data[9:10], + } + assert {timeout for _, _, timeout in fake_put["calls"]} == {7} + + +@pytest.mark.parametrize( + "failure", + [ + lambda: _Response(503), + lambda: _Response(429), + lambda: (_ for _ in ()).throw(requests.exceptions.ConnectionError("down")), + lambda: (_ for _ in ()).throw(requests.exceptions.ReadTimeout("slow")), + ], + ids=["5xx", "429", "connection-error", "timeout"], +) +def test_transient_failure_is_retried(fake_put, failure): + def handler(url, body, attempt): + if url == "part-2" and attempt < 3: + return failure() + return _Response(e_tag="e") + + fake_put["handler"] = handler + + parts = _upload(b"x" * 9, _urls(3), chunk_size=3) + + assert [part["part_number"] for part in parts] == [1, 2, 3] + assert sum(1 for url, _, _ in fake_put["calls"] if url == "part-2") == 3 + + +def test_client_error_is_not_retried(fake_put): + def handler(url, body, attempt): + if url == "part-2": + return _Response(403) + return _Response(e_tag="e") + + fake_put["handler"] = handler + + with pytest.raises(requests.exceptions.HTTPError, match="403"): + _upload(b"x" * 9, _urls(3), chunk_size=3) + + assert sum(1 for url, _, _ in fake_put["calls"] if url == "part-2") == 1 + + +def test_certificate_error_is_not_retried(fake_put): + """SSLError subclasses ConnectionError but retrying cannot fix it.""" + + def handler(url, body, attempt): + if url == "part-2": + raise requests.exceptions.SSLError("certificate verify failed") + return _Response(e_tag="e") + + fake_put["handler"] = handler + + with pytest.raises(requests.exceptions.SSLError): + _upload(b"x" * 9, _urls(3), chunk_size=3) + + assert sum(1 for url, _, _ in fake_put["calls"] if url == "part-2") == 1 + + +def test_retries_stop_after_max_attempts(fake_put, monkeypatch): + monkeypatch.setattr(uploader, "MAX_ATTEMPTS", 4) + fake_put["handler"] = lambda url, body, attempt: _Response(500) + + with pytest.raises(requests.exceptions.HTTPError, match="500"): + _upload(b"x" * 3, _urls(1), chunk_size=3) + + assert len(fake_put["calls"]) == 4 + + +def test_missing_etag_fails_upload(fake_put): + fake_put["handler"] = lambda url, body, attempt: _Response(e_tag=None) + + with pytest.raises(RuntimeError, match="ETag"): + _upload(b"x" * 3, _urls(1), chunk_size=3) + + +def test_failed_part_cancels_parts_not_started(fake_put, monkeypatch): + """With one worker, parts queued behind a failed part are never sent.""" + monkeypatch.setattr(uploader, "MAX_CONCURRENCY", 1) + + def handler(url, body, attempt): + if url == "part-2": + return _Response(400) + return _Response(e_tag="e") + + fake_put["handler"] = handler + + with pytest.raises(requests.exceptions.HTTPError, match="400"): + _upload(b"x" * 30, _urls(10), chunk_size=3) + + assert [url for url, _, _ in fake_put["calls"]] == ["part-1", "part-2"] + + +def test_failed_part_stops_retries_of_other_parts(fake_put, monkeypatch): + """A part waiting to retry gives up once another part has failed.""" + # Long enough that the retry only ends early through the abort. + monkeypatch.setattr(uploader, "INITIAL_DELAY", 30) + started = threading.Event() + failed = threading.Event() + + def handler(url, body, attempt): + if url == "part-1": + started.wait(5) + failed.set() + return _Response(400) + started.set() + failed.wait(5) + return _Response(503) + + fake_put["handler"] = handler + + with pytest.raises(requests.exceptions.HTTPError, match="400"): + _upload(b"x" * 6, _urls(2), chunk_size=3) + + assert sum(1 for url, _, _ in fake_put["calls"] if url == "part-2") == 1 From a669434dc74b1335a8ad499381f61ad88b444b57 Mon Sep 17 00:00:00 2001 From: Waylon Date: Wed, 21 Oct 2026 08:04:43 +0800 Subject: [PATCH 2/2] test: stub multipart metadata in part failure test Document that test_read_and_write_multipart requires a server with multi-part uploads enabled on S3 compatible storage. --- test/unit/accessor/test_server.py | 20 +++++++++++++++++++- 1 file changed, 19 insertions(+), 1 deletion(-) diff --git a/test/unit/accessor/test_server.py b/test/unit/accessor/test_server.py index b8af2924..aee72faf 100644 --- a/test/unit/accessor/test_server.py +++ b/test/unit/accessor/test_server.py @@ -105,7 +105,12 @@ def test_remove_timeout(new_component, random_binary_data, session, monkeypatch) def test_read_and_write_multipart(new_component, large_binary_data, session, mocker): - """Read and write data uploaded in multiple parts.""" + """Read and write data uploaded in multiple parts. + + Requires a server with multi-part uploads enabled on S3 compatible storage, + otherwise the data is uploaded with a single PUT and the test fails. + + """ mocker.patch.object( session, "complete_multipart_upload", @@ -166,6 +171,19 @@ def test_write_multipart_part_failure( new_component, large_binary_data, session, mocker ): """Fail without completing the upload when a part cannot be uploaded.""" + # Multi-part metadata is stubbed so the test does not depend on server + # support. No request is made to these URLs as upload_parts is patched. + mocker.patch.object( + session, + "get_upload_metadata", + return_value={ + "urls": [ + {"part_number": number, "signed_url": "https://example.com"} + for number in range(1, 4) + ], + "upload_id": "upload-id", + }, + ) mocker.patch.object( ftrack_api._uploader, "upload_parts",