Set content length for empty stream uploads

src/requests/models.py

@@ -70,6 +70,7 @@
 from .status_codes import codes
 from .structures import CaseInsensitiveDict
 from .utils import (
+    _super_len,  # type: ignore[reportPrivateUsage]
     check_header_validity,
     get_auth_from_url,
     guess_filename,
@@ -600,9 +601,10 @@ def prepare_body(
         is_iterable = isinstance(data, Iterable) or hasattr(data, "__iter__")
         if is_iterable and not isinstance(data, (str, bytes, list, tuple, Mapping)):
             try:
-                length = super_len(data)
+                length, is_known_length = _super_len(data)
             except (TypeError, AttributeError, UnsupportedOperation):
                 length = None
+                is_known_length = False
 
             body = data
 
@@ -622,9 +624,9 @@ def prepare_body(
                     "Streamed bodies and files are mutually exclusive."
                 )
 
-            if length:
+            if length or is_known_length:
                 self.headers["Content-Length"] = builtin_str(length)
-            else:
+            elif "Content-Length" not in self.headers:
                 self.headers["Transfer-Encoding"] = "chunked"
         else:
             # After is_stream filtering, remaining data is raw (not streamed)

src/requests/utils.py

@@ -157,9 +157,10 @@ def dict_to_sequence(
     return d
 
 
-def super_len(o: Any) -> int:
+def _super_len(o: Any) -> tuple[int, bool]:
     total_length = None
     current_position = 0
+    is_known_length = False
 
     if not is_urllib3_1 and isinstance(o, str):
         # urllib3 2.x+ treats all strings as utf-8 instead
@@ -168,9 +169,11 @@ def super_len(o: Any) -> int:
 
     if hasattr(o, "__len__"):
         total_length = len(o)
+        is_known_length = True
 
     elif hasattr(o, "len"):
         total_length = o.len
+        is_known_length = True
 
     elif hasattr(o, "fileno"):
         try:
@@ -182,6 +185,7 @@ def super_len(o: Any) -> int:
             pass
         else:
             total_length = os.fstat(fileno).st_size
+            is_known_length = True
 
             # Having used fstat to determine the file length, we need to
             # confirm that this file was opened up in binary mode.
@@ -208,6 +212,7 @@ def super_len(o: Any) -> int:
             # let requests chunk it instead.
             if total_length is not None:
                 current_position = total_length
+                is_known_length = False
         else:
             if hasattr(o, "seek") and total_length is None:
                 # StringIO and BytesIO have seek but no usable fileno
@@ -219,13 +224,18 @@ def super_len(o: Any) -> int:
                     # seek back to current position to support
                     # partially read file-like objects
                     o.seek(current_position or 0)
+                    is_known_length = True
                 except OSError:
-                    total_length = 0
+                    pass
 
     if total_length is None:
         total_length = 0
 
-    return max(0, total_length - current_position)
+    return max(0, total_length - current_position), is_known_length
+
+
+def super_len(o: Any) -> int:
+    return _super_len(o)[0]
 
 
 def get_netrc_auth(

tests/test_requests.py

@@ -2222,17 +2222,39 @@ def test_response_without_release_conn(self):
         resp.close()
         assert resp.raw.closed
 
-    def test_empty_stream_with_auth_does_not_set_content_length_header(self, httpbin):
-        """Ensure that a byte stream with size 0 will not set both a Content-Length
-        and Transfer-Encoding header.
-        """
+    def test_empty_stream_with_auth_sets_zero_content_length_header(self, httpbin):
+        """Ensure that a byte stream with size 0 sends Content-Length: 0."""
         auth = ("user", "pass")
         url = httpbin("post")
         file_obj = io.BytesIO(b"")
         r = requests.Request("POST", url, auth=auth, data=file_obj)
         prepared_request = r.prepare()
-        assert "Transfer-Encoding" in prepared_request.headers
-        assert "Content-Length" not in prepared_request.headers
+        assert prepared_request.headers["Content-Length"] == "0"
+        assert "Transfer-Encoding" not in prepared_request.headers
+
+    def test_empty_file_stream_sets_zero_content_length_header(self, tmp_path, httpbin):
+        """Ensure that empty file uploads send Content-Length: 0."""
+        url = httpbin("put")
+        empty_file = tmp_path / "empty.txt"
+        empty_file.write_bytes(b"")
+
+        with empty_file.open("rb") as file_obj:
+            r = requests.Request("PUT", url, data=file_obj)
+            prepared_request = r.prepare()
+
+        assert prepared_request.headers["Content-Length"] == "0"
+        assert "Transfer-Encoding" not in prepared_request.headers
+
+    def test_explicit_zero_content_length_stream_does_not_set_chunked_header(
+        self, httpbin
+    ):
+        """Ensure explicit Content-Length: 0 is not combined with chunked."""
+        url = httpbin("put")
+        file_obj = io.BytesIO(b"")
+        r = requests.Request("PUT", url, data=file_obj, headers={"Content-Length": "0"})
+        prepared_request = r.prepare()
+        assert prepared_request.headers["Content-Length"] == "0"
+        assert "Transfer-Encoding" not in prepared_request.headers
 
     def test_stream_with_auth_does_not_set_transfer_encoding_header(self, httpbin):
         """Ensure that a byte stream with size > 0 will not set both a Content-Length
@@ -2257,6 +2279,21 @@ def test_chunked_upload_does_not_set_content_length_header(self, httpbin):
         assert "Transfer-Encoding" in prepared_request.headers
         assert "Content-Length" not in prepared_request.headers
 
+    def test_unknown_length_file_stream_sets_transfer_encoding_header(self, httpbin):
+        """Ensure that file streams with unknown size use chunked uploads."""
+        read_fd, write_fd = os.pipe()
+        url = httpbin("put")
+
+        try:
+            with os.fdopen(read_fd, "rb") as file_obj:
+                r = requests.Request("PUT", url, data=file_obj)
+                prepared_request = r.prepare()
+        finally:
+            os.close(write_fd)
+
+        assert "Transfer-Encoding" in prepared_request.headers
+        assert "Content-Length" not in prepared_request.headers
+
     def test_custom_redirect_mixin(self, httpbin):
         """Tests a custom mixin to overwrite ``get_redirect_target``.