test(python-sdk): add regression tests for template upload Content-Length

packages/python-sdk/tests/async/template_async/test_upload_file.py

@@ -0,0 +1,71 @@
+import threading
+from http.server import BaseHTTPRequestHandler, HTTPServer
+
+from e2b.api.client.client import AuthenticatedClient
+from e2b.template_async.build_api import upload_file
+
+
+# Regression test for e2b-dev/e2b#1243 — upload_file must set Content-Length
+# and must not fall back to Transfer-Encoding: chunked. S3 presigned PUT URLs
+# reject chunked encoding with 501 NotImplemented. httpx sets Content-Length
+# automatically when we pass bytes (tar_buffer.getvalue()); this test guards
+# against someone swapping the bytes for a generator/stream later.
+#
+# The mock server runs in a daemon thread and doesn't need to be async — the
+# httpx.AsyncClient connects to it via asyncio sockets without blocking the
+# event loop.
+
+
+def _make_server():
+    state = {"headers": None, "body_length": 0}
+
+    class Handler(BaseHTTPRequestHandler):
+        def do_PUT(self):
+            state["headers"] = dict(self.headers)
+            length = int(self.headers.get("Content-Length", 0))
+            body = self.rfile.read(length) if length else b""
+            state["body_length"] = len(body)
+            self.send_response(200)
+            self.end_headers()
+
+        def log_message(self, *args, **kwargs):
+            return
+
+    server = HTTPServer(("127.0.0.1", 0), Handler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    return server, thread, state
+
+
+async def test_upload_file_sets_content_length_and_no_chunked_encoding(tmp_path):
+    (tmp_path / "hello.txt").write_text("hello world")
+
+    server, thread, state = _make_server()
+    host, port = server.server_address
+    url = f"http://{host}:{port}/upload"
+
+    try:
+        client = AuthenticatedClient(base_url="http://test", token="test")
+        await upload_file(
+            api_client=client,
+            file_name="*.txt",
+            context_path=str(tmp_path),
+            url=url,
+            ignore_patterns=[],
+            resolve_symlinks=False,
+            stack_trace=None,
+        )
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=5)
+
+    assert state["headers"] is not None
+    content_length = state["headers"].get("Content-Length")
+    assert content_length is not None
+    assert int(content_length) > 0
+    assert int(content_length) == state["body_length"]
+
+    transfer_encoding = state["headers"].get("Transfer-Encoding")
+    if transfer_encoding is not None:
+        assert "chunked" not in transfer_encoding.lower()

packages/python-sdk/tests/sync/template_sync/test_upload_file.py

@@ -0,0 +1,67 @@
+import threading
+from http.server import BaseHTTPRequestHandler, HTTPServer
+
+from e2b.api.client.client import AuthenticatedClient
+from e2b.template_sync.build_api import upload_file
+
+
+# Regression test for e2b-dev/e2b#1243 — upload_file must set Content-Length
+# and must not fall back to Transfer-Encoding: chunked. S3 presigned PUT URLs
+# reject chunked encoding with 501 NotImplemented. httpx sets Content-Length
+# automatically when we pass bytes (tar_buffer.getvalue()); this test guards
+# against someone swapping the bytes for a generator/stream later.
+
+
+def _make_server():
+    state = {"headers": None, "body_length": 0}
+
+    class Handler(BaseHTTPRequestHandler):
+        def do_PUT(self):
+            state["headers"] = dict(self.headers)
+            length = int(self.headers.get("Content-Length", 0))
+            body = self.rfile.read(length) if length else b""
+            state["body_length"] = len(body)
+            self.send_response(200)
+            self.end_headers()
+
+        def log_message(self, *args, **kwargs):
+            return
+
+    server = HTTPServer(("127.0.0.1", 0), Handler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    return server, thread, state
+
+
+def test_upload_file_sets_content_length_and_no_chunked_encoding(tmp_path):
+    (tmp_path / "hello.txt").write_text("hello world")
+
+    server, thread, state = _make_server()
+    host, port = server.server_address
+    url = f"http://{host}:{port}/upload"
+
+    try:
+        client = AuthenticatedClient(base_url="http://test", token="test")
+        upload_file(
+            api_client=client,
+            file_name="*.txt",
+            context_path=str(tmp_path),
+            url=url,
+            ignore_patterns=[],
+            resolve_symlinks=False,
+            stack_trace=None,
+        )
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=5)
+
+    assert state["headers"] is not None
+    content_length = state["headers"].get("Content-Length")
+    assert content_length is not None
+    assert int(content_length) > 0
+    assert int(content_length) == state["body_length"]
+
+    transfer_encoding = state["headers"].get("Transfer-Encoding")
+    if transfer_encoding is not None:
+        assert "chunked" not in transfer_encoding.lower()