fix(storage): close the upload handle we opened, and only that one

src/storage/src/storage3/_async/file_api.py

@@ -89,10 +89,6 @@ async def _request(
                     message, "InternalError", exc.response.status_code
                 ) from err
 
-        # close the resource before returning the response
-        if files and "file" in files and isinstance(files["file"][1], BufferedReader):
-            files["file"][1].close()
-
         return response
 
     async def create_signed_upload_url(
@@ -185,25 +181,27 @@ async def upload_to_signed_url(
             or isinstance(file, bytes)
             or isinstance(file, FileIO)
         ):
-            # bytes or byte-stream-like object received
+            # bytes or byte-stream-like object received -- the caller owns it
+            opened = None
             _file = {"file": (filename, file, content_type)}
         else:
-            # str or pathlib.path received
-            _file = {
-                "file": (
-                    filename,
-                    open(file, "rb"),
-                    content_type,
-                )
-            }
-        response = await self._request(
-            "PUT",
-            final_url,
-            files=_file,
-            headers=headers,
-            data=_data,
-            query_params=query_params,
-        )
+            # str or pathlib.path received -- we own the handle, so we close it
+            opened = open(file, "rb")
+            _file = {"file": (filename, opened, content_type)}
+
+        try:
+            response = await self._request(
+                "PUT",
+                final_url,
+                files=_file,
+                headers=headers,
+                data=_data,
+                query_params=query_params,
+            )
+        finally:
+            if opened is not None:
+                opened.close()
+
         data: UploadData = response.json()
 
         return UploadResponse(path=path, Key=data["Key"])
@@ -562,21 +560,25 @@ async def _upload_or_update(
             or isinstance(file, bytes)
             or isinstance(file, FileIO)
         ):
-            # bytes or byte-stream-like object received
+            # bytes or byte-stream-like object received -- the caller owns it
+            opened = None
             files = {"file": (filename, file, content_type)}
         else:
-            # str or pathlib.path received
-            files = {
-                "file": (
-                    filename,
-                    open(file, "rb"),
-                    content_type,
-                )
-            }
+            # str or pathlib.path received -- we own the handle, so we close it
+            opened = open(file, "rb")
+            files = {"file": (filename, opened, content_type)}
 
-        response = await self._request(
-            method, ["object", self.id, *path], files=files, headers=headers, data=_data
-        )
+        try:
+            response = await self._request(
+                method,
+                ["object", self.id, *path],
+                files=files,
+                headers=headers,
+                data=_data,
+            )
+        finally:
+            if opened is not None:
+                opened.close()
 
         data: UploadData = response.json()
 

src/storage/src/storage3/_sync/file_api.py

@@ -89,10 +89,6 @@ def _request(
                     message, "InternalError", exc.response.status_code
                 ) from err
 
-        # close the resource before returning the response
-        if files and "file" in files and isinstance(files["file"][1], BufferedReader):
-            files["file"][1].close()
-
         return response
 
     def create_signed_upload_url(
@@ -185,25 +181,27 @@ def upload_to_signed_url(
             or isinstance(file, bytes)
             or isinstance(file, FileIO)
         ):
-            # bytes or byte-stream-like object received
+            # bytes or byte-stream-like object received -- the caller owns it
+            opened = None
             _file = {"file": (filename, file, content_type)}
         else:
-            # str or pathlib.path received
-            _file = {
-                "file": (
-                    filename,
-                    open(file, "rb"),
-                    content_type,
-                )
-            }
-        response = self._request(
-            "PUT",
-            final_url,
-            files=_file,
-            headers=headers,
-            data=_data,
-            query_params=query_params,
-        )
+            # str or pathlib.path received -- we own the handle, so we close it
+            opened = open(file, "rb")
+            _file = {"file": (filename, opened, content_type)}
+
+        try:
+            response = self._request(
+                "PUT",
+                final_url,
+                files=_file,
+                headers=headers,
+                data=_data,
+                query_params=query_params,
+            )
+        finally:
+            if opened is not None:
+                opened.close()
+
         data: UploadData = response.json()
 
         return UploadResponse(path=path, Key=data["Key"])
@@ -560,21 +558,25 @@ def _upload_or_update(
             or isinstance(file, bytes)
             or isinstance(file, FileIO)
         ):
-            # bytes or byte-stream-like object received
+            # bytes or byte-stream-like object received -- the caller owns it
+            opened = None
             files = {"file": (filename, file, content_type)}
         else:
-            # str or pathlib.path received
-            files = {
-                "file": (
-                    filename,
-                    open(file, "rb"),
-                    content_type,
-                )
-            }
+            # str or pathlib.path received -- we own the handle, so we close it
+            opened = open(file, "rb")
+            files = {"file": (filename, opened, content_type)}
 
-        response = self._request(
-            method, ["object", self.id, *path], files=files, headers=headers, data=_data
-        )
+        try:
+            response = self._request(
+                method,
+                ["object", self.id, *path],
+                files=files,
+                headers=headers,
+                data=_data,
+            )
+        finally:
+            if opened is not None:
+                opened.close()
 
         data: UploadData = response.json()
 

src/storage/tests/_async/test_file_api.py

@@ -0,0 +1,95 @@
+from pathlib import Path
+from typing import Any
+from unittest.mock import Mock
+
+import pytest
+from httpx import Headers, HTTPStatusError, Request, Response
+from storage3.exceptions import StorageApiError
+from yarl import URL
+
+from .. import AsyncBucketProxy
+
+
+def _error_response() -> Mock:
+    response = Mock(spec=Response)
+    response.status_code = 409
+    response.json.return_value = {
+        "message": "The resource already exists",
+        "error": "Duplicate",
+        "statusCode": 409,
+    }
+    response.raise_for_status = Mock(
+        side_effect=HTTPStatusError(
+            "Conflict", request=Mock(spec=Request), response=response
+        )
+    )
+    return response
+
+
+def _success_response() -> Mock:
+    response = Mock(spec=Response)
+    response.status_code = 200
+    response.json.return_value = {"Key": "bucket/upload.txt"}
+    response.raise_for_status = Mock()
+    return response
+
+
+def _proxy(response: Mock, captured: dict) -> AsyncBucketProxy:
+    """A proxy whose transport records the `files` mapping storage3 built for the request."""
+
+    async def request(*args: Any, **kwargs: Any) -> Mock:
+        captured["files"] = kwargs["files"]
+        return response
+
+    client = Mock(headers=Headers())
+    client.request = request
+    return AsyncBucketProxy("bucket", URL("http://example.com"), Headers(), client)
+
+
+@pytest.fixture
+def source(tmp_path: Path) -> str:
+    path = tmp_path / "upload.txt"
+    path.write_bytes(b"payload")
+    return str(path)
+
+
+async def test_upload_closes_file_handle_on_error(source: str) -> None:
+    """A handle storage3 opened itself must be closed when the upload fails."""
+    captured: dict = {}
+
+    with pytest.raises(StorageApiError):
+        await _proxy(_error_response(), captured).upload("upload.txt", source)
+
+    assert captured["files"]["file"][1].closed
+
+
+async def test_upload_closes_file_handle_on_success(source: str) -> None:
+    """Regression: the success path must keep closing the handle it opened."""
+    captured: dict = {}
+
+    await _proxy(_success_response(), captured).upload("upload.txt", source)
+
+    assert captured["files"]["file"][1].closed
+
+
+async def test_update_closes_file_handle_on_error(source: str) -> None:
+    """update() shares _upload_or_update, so it leaks the same way."""
+    captured: dict = {}
+
+    with pytest.raises(StorageApiError):
+        await _proxy(_error_response(), captured).update("upload.txt", source)
+
+    assert captured["files"]["file"][1].closed
+
+
+async def test_caller_supplied_handle_is_left_open(source: str) -> None:
+    """storage3 must not close a stream the caller owns, on either path."""
+    captured: dict = {}
+
+    with open(source, "rb") as handle:
+        with pytest.raises(StorageApiError):
+            await _proxy(_error_response(), captured).upload("upload.txt", handle)
+        assert not handle.closed
+
+        await _proxy(_success_response(), captured).upload("upload.txt", handle)
+        assert not handle.closed

src/storage/tests/_sync/test_file_api.py

@@ -0,0 +1,95 @@
+from pathlib import Path
+from typing import Any
+from unittest.mock import Mock
+
+import pytest
+from httpx import Headers, HTTPStatusError, Request, Response
+from storage3.exceptions import StorageApiError
+from yarl import URL
+
+from .. import SyncBucketProxy
+
+
+def _error_response() -> Mock:
+    response = Mock(spec=Response)
+    response.status_code = 409
+    response.json.return_value = {
+        "message": "The resource already exists",
+        "error": "Duplicate",
+        "statusCode": 409,
+    }
+    response.raise_for_status = Mock(
+        side_effect=HTTPStatusError(
+            "Conflict", request=Mock(spec=Request), response=response
+        )
+    )
+    return response
+
+
+def _success_response() -> Mock:
+    response = Mock(spec=Response)
+    response.status_code = 200
+    response.json.return_value = {"Key": "bucket/upload.txt"}
+    response.raise_for_status = Mock()
+    return response
+
+
+def _proxy(response: Mock, captured: dict) -> SyncBucketProxy:
+    """A proxy whose transport records the `files` mapping storage3 built for the request."""
+
+    def request(*args: Any, **kwargs: Any) -> Mock:
+        captured["files"] = kwargs["files"]
+        return response
+
+    client = Mock(headers=Headers())
+    client.request = request
+    return SyncBucketProxy("bucket", URL("http://example.com"), Headers(), client)
+
+
+@pytest.fixture
+def source(tmp_path: Path) -> str:
+    path = tmp_path / "upload.txt"
+    path.write_bytes(b"payload")
+    return str(path)
+
+
+def test_upload_closes_file_handle_on_error(source: str) -> None:
+    """A handle storage3 opened itself must be closed when the upload fails."""
+    captured: dict = {}
+
+    with pytest.raises(StorageApiError):
+        _proxy(_error_response(), captured).upload("upload.txt", source)
+
+    assert captured["files"]["file"][1].closed
+
+
+def test_upload_closes_file_handle_on_success(source: str) -> None:
+    """Regression: the success path must keep closing the handle it opened."""
+    captured: dict = {}
+
+    _proxy(_success_response(), captured).upload("upload.txt", source)
+
+    assert captured["files"]["file"][1].closed
+
+
+def test_update_closes_file_handle_on_error(source: str) -> None:
+    """update() shares _upload_or_update, so it leaks the same way."""
+    captured: dict = {}
+
+    with pytest.raises(StorageApiError):
+        _proxy(_error_response(), captured).update("upload.txt", source)
+
+    assert captured["files"]["file"][1].closed
+
+
+def test_caller_supplied_handle_is_left_open(source: str) -> None:
+    """storage3 must not close a stream the caller owns, on either path."""
+    captured: dict = {}
+
+    with open(source, "rb") as handle:
+        with pytest.raises(StorageApiError):
+            _proxy(_error_response(), captured).upload("upload.txt", handle)
+        assert not handle.closed
+
+        _proxy(_success_response(), captured).upload("upload.txt", handle)
+        assert not handle.closed