| Server IP : 185.124.137.213 / Your IP : 216.73.217.6 Web Server : LiteSpeed System : Linux id-dci-web1986.main-hosting.eu 5.14.0-611.26.1.el9_7.x86_64 #1 SMP PREEMPT_DYNAMIC Thu Jan 29 05:24:47 EST 2026 x86_64 User : u686484674 ( 686484674) PHP Version : 8.0.30 Disable Function : system, exec, shell_exec, passthru, mysql_list_dbs, ini_alter, dl, symlink, link, chgrp, leak, popen, apache_child_terminate, virtual, mb_send_mail MySQL : OFF | cURL : ON | WGET : ON | Perl : OFF | Python : OFF | Sudo : OFF | Pkexec : OFF Directory : /proc/self/root/opt/gsutil/third_party/urllib3/test/with_dummyserver/ |
Upload File : |
from __future__ import annotations
import socket
import pytest
from dummyserver.testcase import (
ConnectionMarker,
SocketDummyServerTestCase,
consume_socket,
)
from urllib3 import HTTPConnectionPool
from urllib3.util import SKIP_HEADER
from urllib3.util.retry import Retry
class TestChunkedTransfer(SocketDummyServerTestCase):
def start_chunked_handler(self) -> None:
self.buffer = b""
def socket_handler(listener: socket.socket) -> None:
sock = listener.accept()[0]
while not self.buffer.endswith(b"\r\n0\r\n\r\n"):
self.buffer += sock.recv(65536)
sock.send(
b"HTTP/1.1 200 OK\r\n"
b"Content-type: text/plain\r\n"
b"Content-Length: 0\r\n"
b"\r\n"
)
sock.close()
self._start_server(socket_handler)
@pytest.mark.parametrize(
"chunks",
[
["foo", "bar", "", "bazzzzzzzzzzzzzzzzzzzzzz"],
[b"foo", b"bar", b"", b"bazzzzzzzzzzzzzzzzzzzzzz"],
],
)
def test_chunks(self, chunks: list[bytes | str]) -> None:
self.start_chunked_handler()
with HTTPConnectionPool(self.host, self.port, retries=False) as pool:
pool.urlopen("GET", "/", body=chunks, headers=dict(DNT="1"), chunked=True) # type: ignore[arg-type]
assert b"Transfer-Encoding" in self.buffer
body = self.buffer.split(b"\r\n\r\n", 1)[1]
lines = body.split(b"\r\n")
# Empty chunks should have been skipped, as this could not be distinguished
# from terminating the transmission
for i, chunk in enumerate(
[c.decode() if isinstance(c, bytes) else c for c in chunks if c]
):
assert lines[i * 2] == hex(len(chunk))[2:].encode("utf-8")
assert lines[i * 2 + 1] == chunk.encode("utf-8")
def _test_body(self, data: bytes | str | None) -> None:
self.start_chunked_handler()
with HTTPConnectionPool(self.host, self.port, retries=False) as pool:
pool.urlopen("GET", "/", data, chunked=True)
header, body = self.buffer.split(b"\r\n\r\n", 1)
assert b"Transfer-Encoding: chunked" in header.split(b"\r\n")
if data:
bdata = data if isinstance(data, bytes) else data.encode("utf-8")
assert b"\r\n" + bdata + b"\r\n" in body
assert body.endswith(b"\r\n0\r\n\r\n")
len_str = body.split(b"\r\n", 1)[0]
stated_len = int(len_str, 16)
assert stated_len == len(bdata)
else:
assert body == b"0\r\n\r\n"
def test_bytestring_body(self) -> None:
self._test_body(b"thisshouldbeonechunk\r\nasdf")
def test_unicode_body(self) -> None:
self._test_body("thisshouldbeonechunk\r\näöüß")
def test_empty_body(self) -> None:
self._test_body(None)
def test_empty_string_body(self) -> None:
self._test_body("")
def test_empty_iterable_body(self) -> None:
self._test_body(None)
def _get_header_lines(self, prefix: bytes) -> list[bytes]:
header_block = self.buffer.split(b"\r\n\r\n", 1)[0].lower()
header_lines = header_block.split(b"\r\n")[1:]
return [x for x in header_lines if x.startswith(prefix)]
def test_removes_duplicate_host_header(self) -> None:
self.start_chunked_handler()
chunks = [b"foo", b"bar", b"", b"bazzzzzzzzzzzzzzzzzzzzzz"]
with HTTPConnectionPool(self.host, self.port, retries=False) as pool:
pool.urlopen(
"GET", "/", body=chunks, headers={"Host": "test.org"}, chunked=True
)
host_headers = self._get_header_lines(b"host")
assert len(host_headers) == 1
def test_provides_default_host_header(self) -> None:
self.start_chunked_handler()
chunks = [b"foo", b"bar", b"", b"bazzzzzzzzzzzzzzzzzzzzzz"]
with HTTPConnectionPool(self.host, self.port, retries=False) as pool:
pool.urlopen("GET", "/", body=chunks, chunked=True)
host_headers = self._get_header_lines(b"host")
assert len(host_headers) == 1
def test_provides_default_user_agent_header(self) -> None:
self.start_chunked_handler()
chunks = [b"foo", b"bar", b"", b"bazzzzzzzzzzzzzzzzzzzzzz"]
with HTTPConnectionPool(self.host, self.port, retries=False) as pool:
pool.urlopen("GET", "/", body=chunks, chunked=True)
ua_headers = self._get_header_lines(b"user-agent")
assert len(ua_headers) == 1
def test_preserve_user_agent_header(self) -> None:
self.start_chunked_handler()
chunks = [b"foo", b"bar", b"", b"bazzzzzzzzzzzzzzzzzzzzzz"]
with HTTPConnectionPool(self.host, self.port, retries=False) as pool:
pool.urlopen(
"GET",
"/",
body=chunks,
headers={"user-Agent": "test-agent"},
chunked=True,
)
ua_headers = self._get_header_lines(b"user-agent")
# Validate that there is only one User-Agent header.
assert len(ua_headers) == 1
# Validate that the existing User-Agent header is the one that was
# provided.
assert ua_headers[0] == b"user-agent: test-agent"
def test_remove_user_agent_header(self) -> None:
self.start_chunked_handler()
chunks = [b"foo", b"bar", b"", b"bazzzzzzzzzzzzzzzzzzzzzz"]
with HTTPConnectionPool(self.host, self.port, retries=False) as pool:
pool.urlopen(
"GET",
"/",
body=chunks,
headers={"User-Agent": SKIP_HEADER},
chunked=True,
)
ua_headers = self._get_header_lines(b"user-agent")
assert len(ua_headers) == 0
def test_provides_default_transfer_encoding_header(self) -> None:
self.start_chunked_handler()
chunks = [b"foo", b"bar", b"", b"bazzzzzzzzzzzzzzzzzzzzzz"]
with HTTPConnectionPool(self.host, self.port, retries=False) as pool:
pool.urlopen("GET", "/", body=chunks, chunked=True)
te_headers = self._get_header_lines(b"transfer-encoding")
assert len(te_headers) == 1
def test_preserve_transfer_encoding_header(self) -> None:
self.start_chunked_handler()
chunks = [b"foo", b"bar", b"", b"bazzzzzzzzzzzzzzzzzzzzzz"]
with HTTPConnectionPool(self.host, self.port, retries=False) as pool:
pool.urlopen(
"GET",
"/",
body=chunks,
headers={"transfer-Encoding": "test-transfer-encoding"},
chunked=True,
)
te_headers = self._get_header_lines(b"transfer-encoding")
# Validate that there is only one Transfer-Encoding header.
assert len(te_headers) == 1
# Validate that the existing Transfer-Encoding header is the one that
# was provided.
assert te_headers[0] == b"transfer-encoding: test-transfer-encoding"
def test_preserve_chunked_on_retry_after(self) -> None:
self.chunked_requests = 0
self.socks: list[socket.socket] = []
def socket_handler(listener: socket.socket) -> None:
for _ in range(2):
sock = listener.accept()[0]
self.socks.append(sock)
request = consume_socket(sock)
if b"Transfer-Encoding: chunked" in request.split(b"\r\n"):
self.chunked_requests += 1
sock.send(
b"HTTP/1.1 429 Too Many Requests\r\n"
b"Content-Type: text/plain\r\n"
b"Retry-After: 1\r\n"
b"Content-Length: 0\r\n"
b"Connection: close\r\n"
b"\r\n"
)
self._start_server(socket_handler)
with HTTPConnectionPool(self.host, self.port) as pool:
retries = Retry(total=1)
pool.urlopen("GET", "/", chunked=True, retries=retries)
for sock in self.socks:
sock.close()
assert self.chunked_requests == 2
def test_preserve_chunked_on_redirect(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
self.chunked_requests = 0
def socket_handler(listener: socket.socket) -> None:
for i in range(2):
sock = listener.accept()[0]
request = ConnectionMarker.consume_request(sock)
if b"Transfer-Encoding: chunked" in request.split(b"\r\n"):
self.chunked_requests += 1
if i == 0:
sock.sendall(
b"HTTP/1.1 301 Moved Permanently\r\n"
b"Location: /redirect\r\n\r\n"
)
else:
sock.sendall(b"HTTP/1.1 200 OK\r\n\r\n")
sock.close()
self._start_server(socket_handler)
with ConnectionMarker.mark(monkeypatch):
with HTTPConnectionPool(self.host, self.port) as pool:
retries = Retry(redirect=1)
pool.urlopen(
"GET", "/", chunked=True, preload_content=False, retries=retries
)
assert self.chunked_requests == 2
def test_preserve_chunked_on_broken_connection(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
self.chunked_requests = 0
def socket_handler(listener: socket.socket) -> None:
for i in range(2):
sock = listener.accept()[0]
request = ConnectionMarker.consume_request(sock)
if b"Transfer-Encoding: chunked" in request.split(b"\r\n"):
self.chunked_requests += 1
if i == 0:
# Bad HTTP version will trigger a connection close
sock.sendall(b"HTTP/0.5 200 OK\r\n\r\n")
else:
sock.sendall(b"HTTP/1.1 200 OK\r\n\r\n")
sock.close()
self._start_server(socket_handler)
with ConnectionMarker.mark(monkeypatch):
with HTTPConnectionPool(self.host, self.port) as pool:
retries = Retry(read=1)
pool.urlopen(
"GET", "/", chunked=True, preload_content=False, retries=retries
)
assert self.chunked_requests == 2