ronantakizawa/codereview-bench huggingface dataset:
Repository: encode/httpx
File: tests/models/test_queryparams.py (Python)

Code before:
    assert str(q) == "a="

    q = httpx.QueryParams({"a": 1.23})
    assert str(q) == "a=1.23"

    q = httpx.QueryParams({"a": 123})
    assert str(q) == "a=123"

    q = httpx.QueryParams({"a": [1, 2]})
    assert str(q) == "a=1&a=2"


def test_empty_query_params():
    q = httpx.QueryParams({"a": ""})
    assert str(q) == "a="

    q = httpx.QueryParams("a=")
    assert str(q) == "a="

    q = httpx.QueryParams("a")
    assert str(q) == "a="


def test_queryparam_update_is_hard_deprecated():
    q = httpx.QueryParams("a=123")
    with pytest.raises(RuntimeError):
        q.update({"a": "456"})


def test_queryparam_setter_is_hard_deprecated():
    q = httpx.QueryParams("a=123")
    with pytest.raises(RuntimeError):
        q["a"] = "456"


def test_queryparam_set():
    q = httpx.QueryParams("a=123")
    q = q.set("a", "456")
    assert q == httpx.QueryParams("a=456")


def test_queryparam_add():
    q = httpx.QueryParams("a=123")
    q = q.add("a", "456")
    assert q == httpx.QueryParams("a=123&a=456")


def test_queryparam_remove():
    q = httpx.QueryParams("a=123")
    q = q.remove("a")
    assert q == httpx.QueryParams("")

Review comment (security):
You don't need the `r` if you aren't escaping the periods — this is doing a `re.search` so technically the `.` will be a wildcard but 🤷‍♀️ it's not likely to cause you any problems. You could use `re.escape` if you really wanted to be correct.

Code after:
    assert str(q) == "a="

    q = httpx.QueryParams({"a": 1.23})
    assert str(q) == "a=1.23"

    q = httpx.QueryParams({"a": 123})
    assert str(q) == "a=123"

    q = httpx.QueryParams({"a": [1, 2]})
    assert str(q) == "a=1&a=2"


def test_empty_query_params():
    q = httpx.QueryParams({"a": ""})
    assert str(q) == "a="

    q = httpx.QueryParams("a=")
    assert str(q) == "a="

    q = httpx.QueryParams("a")
    assert str(q) == "a="


def test_invalid_query_params():
    with pytest.raises(
        TypeError, match=r"Expected str, int, float, bool, or None. Got 'bytes'."
    ):
        httpx.QueryParams({"a": b"bytes"})


def test_queryparam_update_is_hard_deprecated():
    q = httpx.QueryParams("a=123")
    with pytest.raises(RuntimeError):
        q.update({"a": "456"})


def test_queryparam_setter_is_hard_deprecated():
    q = httpx.QueryParams("a=123")
    with pytest.raises(RuntimeError):
        q["a"] = "456"


def test_queryparam_set():
    q = httpx.QueryParams("a=123")
    q = q.set("a", "456")
    assert q == httpx.QueryParams("a=456")


def test_queryparam_add():
    q = httpx.QueryParams("a=123")
    q = q.add("a", "456")
Repository: encode/httpx
File: tests/test_asgi.py (Python)

Code before:
import json

import pytest

import httpx
from httpx._transports.asgi import ASGITransport


async def hello_world(scope, receive, send):
    status = 200
    output = b"Hello, World!"
    headers = [(b"content-type", "text/plain"), (b"content-length", str(len(output)))]

    await send({"type": "http.response.start", "status": status, "headers": headers})
    await send({"type": "http.response.body", "body": output})


async def echo_path(scope, receive, send):
    status = 200
    output = json.dumps({"path": scope["path"]}).encode("utf-8")
    headers = [(b"content-type", "text/plain"), (b"content-length", str(len(output)))]

    await send({"type": "http.response.start", "status": status, "headers": headers})
    await send({"type": "http.response.body", "body": output})


async def echo_raw_path(scope, receive, send):
    status = 200
    output = json.dumps({"raw_path": scope["raw_path"].decode("ascii")}).encode("utf-8")
    headers = [(b"content-type", "text/plain"), (b"content-length", str(len(output)))]


Review comment (suggestion):
We don't need the import from private API space here.
We can use `httpx.ASGITransport`, same as the other tests.

Code after:
import json

import pytest

import httpx
from httpx import ASGITransport


async def hello_world(scope, receive, send):
    status = 200
    output = b"Hello, World!"
    headers = [(b"content-type", "text/plain"), (b"content-length", str(len(output)))]

    await send({"type": "http.response.start", "status": status, "headers": headers})
    await send({"type": "http.response.body", "body": output})


async def echo_path(scope, receive, send):
    status = 200
    output = json.dumps({"path": scope["path"]}).encode("utf-8")
    headers = [(b"content-type", "text/plain"), (b"content-length", str(len(output)))]

    await send({"type": "http.response.start", "status": status, "headers": headers})
    await send({"type": "http.response.body", "body": output})


async def echo_raw_path(scope, receive, send):
    status = 200
    output = json.dumps({"raw_path": scope["raw_path"].decode("ascii")}).encode("utf-8")
    headers = [(b"content-type", "text/plain"), (b"content-length", str(len(output)))]
Repository: encode/httpx
File: tests/test_asgi.py (Python)

Code before:
        )
        more_body = True
        while more_body:
            message = await receive()
            more_body = message.get("more_body", False)

        await send({"type": "http.response.body", "body": b"", "more_body": False})

        # The ASGI spec says of the Disconnect message:
        # "Sent to the application when a HTTP connection is closed or if receive is
        # called after a response has been sent."
        # So if receive() is called again, the disconnect message should be received
        message = await receive()
        disconnect = message.get("type") == "http.disconnect"

    async with httpx.AsyncClient(app=read_body) as client:
        response = await client.post("http://www.example.org/", content=b"example")

    assert response.status_code == 200
    assert disconnect


@pytest.mark.anyio
async def test_asgi_exc_no_raise():
    transport = ASGITransport(app=raise_exc, raise_app_exceptions=False)
    async with httpx.AsyncClient(app=raise_exc, transport=transport) as client:
        response = await client.get("http://www.example.org/")

        assert response.status_code == 500


Review comment (suggestion):
Since we're specifying `transport=...` the `app=...` parameter is redundant.

```suggestion
    async with httpx.AsyncClient(transport=transport) as client:
```

Code after:
        )
        more_body = True
        while more_body:
            message = await receive()
            more_body = message.get("more_body", False)

        await send({"type": "http.response.body", "body": b"", "more_body": False})

        # The ASGI spec says of the Disconnect message:
        # "Sent to the application when a HTTP connection is closed or if receive is
        # called after a response has been sent."
        # So if receive() is called again, the disconnect message should be received
        message = await receive()
        disconnect = message.get("type") == "http.disconnect"

    async with httpx.AsyncClient(app=read_body) as client:
        response = await client.post("http://www.example.org/", content=b"example")

    assert response.status_code == 200
    assert disconnect


@pytest.mark.anyio
async def test_asgi_exc_no_raise():
    transport = ASGITransport(app=raise_exc, raise_app_exceptions=False)
    async with httpx.AsyncClient(transport=transport) as client:
        response = await client.get("http://www.example.org/")

        assert response.status_code == 500
Repository: encode/httpx
File: httpx/_transports/default.py (Python)

Code before:
* uds: str
* local_address: str
* retries: int

Example usages...

# Disable HTTP/2 on a single specific domain.
mounts = {
    "all://": httpx.HTTPTransport(http2=True),
    "all://*example.org": httpx.HTTPTransport()
}

# Using advanced httpcore configuration, with connection retries.
transport = httpx.HTTPTransport(retries=1)
client = httpx.Client(transport=transport)

# Using advanced httpcore configuration, with unix domain sockets.
transport = httpx.HTTPTransport(uds="socket.uds")
client = httpx.Client(transport=transport)
"""
import contextlib
import typing
from types import TracebackType

import httpcore
from httpcore.backends.base import SOCKET_OPTION

from .._config import DEFAULT_LIMITS, Limits, Proxy, create_ssl_context
from .._exceptions import (
    ConnectError,
    ConnectTimeout,
    LocalProtocolError,
    NetworkError,
    PoolTimeout,
    ProtocolError,
    ProxyError,
    ReadError,
    ReadTimeout,
    RemoteProtocolError,
    TimeoutException,
    UnsupportedProtocol,
    WriteError,
    WriteTimeout,
)
from .._models import Request, Response
from .._types import AsyncByteStream, CertTypes, SyncByteStream, VerifyTypes
from .base import AsyncBaseTransport, BaseTransport

T = typing.TypeVar("T", bound="HTTPTransport")
A = typing.TypeVar("A", bound="AsyncHTTPTransport")


Review comment (suggestion):
Let's not import that type definition, it's not documented public API.

It's be okay to instead define the socket option types here.

```python
SOCKET_OPTION = typing.Union[
    typing.Tuple[int, int, int],
    typing.Tuple[int, int, typing.Union[bytes, bytearray]],
    typing.Tuple[int, int, None, int],
]
```

Code after:
* uds: str
* local_address: str
* retries: int

Example usages...

# Disable HTTP/2 on a single specific domain.
mounts = {
    "all://": httpx.HTTPTransport(http2=True),
    "all://*example.org": httpx.HTTPTransport()
}

# Using advanced httpcore configuration, with connection retries.
transport = httpx.HTTPTransport(retries=1)
client = httpx.Client(transport=transport)

# Using advanced httpcore configuration, with unix domain sockets.
transport = httpx.HTTPTransport(uds="socket.uds")
client = httpx.Client(transport=transport)
"""
import contextlib
import typing
from types import TracebackType

import httpcore

from .._config import DEFAULT_LIMITS, Limits, Proxy, create_ssl_context
from .._exceptions import (
    ConnectError,
    ConnectTimeout,
    LocalProtocolError,
    NetworkError,
    PoolTimeout,
    ProtocolError,
    ProxyError,
    ReadError,
    ReadTimeout,
    RemoteProtocolError,
    TimeoutException,
    UnsupportedProtocol,
    WriteError,
    WriteTimeout,
)
from .._models import Request, Response
from .._types import AsyncByteStream, CertTypes, SyncByteStream, VerifyTypes
from .base import AsyncBaseTransport, BaseTransport

T = typing.TypeVar("T", bound="HTTPTransport")
A = typing.TypeVar("A", bound="AsyncHTTPTransport")

SOCKET_OPTION = typing.Union[
Repository: encode/httpx
File: tests/models/test_responses.py (Python)

Code before:
    assert response.encoding == "latin-1"


def test_response_force_encoding():
    response = httpx.Response(
        200,
        content="Snowman: ☃".encode("utf-8"),
    )
    response.encoding = "iso-8859-1"
    assert response.status_code == 200
    assert response.reason_phrase == "OK"
    assert response.text == "Snowman: â\x98\x83"
    assert response.encoding == "iso-8859-1"


def test_response_force_encoding_after_text_accessed():
    response = httpx.Response(
        200,
        content=b"Hello, world!",
    )
    assert response.status_code == 200
    assert response.reason_phrase == "OK"
    assert response.text == "Hello, world!"
    assert response.encoding == "utf-8"

    response.encoding = "UTF8"

    with pytest.raises(ValueError):
        response.encoding = "iso-8859-1"


def test_read():
    response = httpx.Response(
        200,
        content=b"Hello, world!",
    )

    assert response.status_code == 200
    assert response.text == "Hello, world!"
    assert response.encoding == "utf-8"
    assert response.is_closed

    content = response.read()

    assert content == b"Hello, world!"
    assert response.content == b"Hello, world!"
    assert response.is_closed


def test_empty_read():
    response = httpx.Response(200)

Review comment (performance):
I'd expect the `ValueError` to be raised here. I don't think we need the conditional "don't raise if the encoding is being set but the resulting codec won't change".

Simplicity over complexity where possible.

Code after:
    assert response.encoding == "latin-1"


def test_response_force_encoding():
    response = httpx.Response(
        200,
        content="Snowman: ☃".encode("utf-8"),
    )
    response.encoding = "iso-8859-1"
    assert response.status_code == 200
    assert response.reason_phrase == "OK"
    assert response.text == "Snowman: â\x98\x83"
    assert response.encoding == "iso-8859-1"


def test_response_force_encoding_after_text_accessed():
    response = httpx.Response(
        200,
        content=b"Hello, world!",
    )
    assert response.status_code == 200
    assert response.reason_phrase == "OK"
    assert response.text == "Hello, world!"
    assert response.encoding == "utf-8"

    with pytest.raises(ValueError):
        response.encoding = "UTF8"

    with pytest.raises(ValueError):
        response.encoding = "iso-8859-1"


def test_read():
    response = httpx.Response(
        200,
        content=b"Hello, world!",
    )

    assert response.status_code == 200
    assert response.text == "Hello, world!"
    assert response.encoding == "utf-8"
    assert response.is_closed

    content = response.read()

    assert content == b"Hello, world!"
    assert response.content == b"Hello, world!"
    assert response.is_closed


def test_empty_read():
Repository: encode/httpx
File: httpx/_transports/default.py (Python)

Code before:
"""
import contextlib
import typing
from types import TracebackType

import httpcore

from httpx import URL

from .._config import DEFAULT_LIMITS, Limits, Proxy, create_ssl_context
from .._exceptions import (
    ConnectError,
    ConnectTimeout,
    LocalProtocolError,
    NetworkError,
    PoolTimeout,
    ProtocolError,
    ProxyError,
    ReadError,
    ReadTimeout,
    RemoteProtocolError,
    TimeoutException,
    UnsupportedProtocol,
    WriteError,
    WriteTimeout,
)
from .._models import Request, Response
from .._types import AsyncByteStream, CertTypes, ProxyTypes, SyncByteStream, VerifyTypes
from .base import AsyncBaseTransport, BaseTransport

T = typing.TypeVar("T", bound="HTTPTransport")
A = typing.TypeVar("A", bound="AsyncHTTPTransport")

SOCKET_OPTION = typing.Union[
    typing.Tuple[int, int, int],
    typing.Tuple[int, int, typing.Union[bytes, bytearray]],
    typing.Tuple[int, int, None, int],
]


@contextlib.contextmanager
def map_httpcore_exceptions() -> typing.Iterator[None]:
    try:
        yield
    except Exception as exc:  # noqa: PIE-786
        mapped_exc = None

        for from_exc, to_exc in HTTPCORE_EXC_MAP.items():
            if not isinstance(exc, from_exc):
                continue
            # We want to map to the most specific exception we can find.

Review comment (suggestion):
```suggestion
from .._types import AsyncByteStream, CertTypes, ProxyTypes, SyncByteStream, VerifyTypes
from .._urls import URL
```

Code after:
"""
import contextlib
import typing
from types import TracebackType

import httpcore

from .._config import DEFAULT_LIMITS, Limits, Proxy, create_ssl_context
from .._exceptions import (
    ConnectError,
    ConnectTimeout,
    LocalProtocolError,
    NetworkError,
    PoolTimeout,
    ProtocolError,
    ProxyError,
    ReadError,
    ReadTimeout,
    RemoteProtocolError,
    TimeoutException,
    UnsupportedProtocol,
    WriteError,
    WriteTimeout,
)
from .._models import Request, Response
from .._types import AsyncByteStream, CertTypes, ProxyTypes, SyncByteStream, VerifyTypes
from .._urls import URL
from .base import AsyncBaseTransport, BaseTransport

T = typing.TypeVar("T", bound="HTTPTransport")
A = typing.TypeVar("A", bound="AsyncHTTPTransport")

SOCKET_OPTION = typing.Union[
    typing.Tuple[int, int, int],
    typing.Tuple[int, int, typing.Union[bytes, bytearray]],
    typing.Tuple[int, int, None, int],
]


@contextlib.contextmanager
def map_httpcore_exceptions() -> typing.Iterator[None]:
    try:
        yield
    except Exception as exc:
        mapped_exc = None

        for from_exc, to_exc in HTTPCORE_EXC_MAP.items():
            if not isinstance(exc, from_exc):
                continue
            # We want to map to the most specific exception we can find.
            # Eg if `exc` is an `httpcore.ReadTimeout`, we want to map to
Repository: encode/httpx
File: tests/test_decoders.py (Python)

Code before:

    # Issue #1033
    response = httpx.Response(
        200, content=[b"", b"12345\r", b"foo ", b"bar ", b"baz\r"]
    )
    assert list(response.iter_lines()) == ["12345", "foo bar baz"]


def test_line_decoder_crnl():
    response = httpx.Response(200, content=[b"", b"a\r\n\r\nb\r\nc"])
    assert list(response.iter_lines()) == ["a", "", "b", "c"]

    response = httpx.Response(200, content=[b"", b"a\r\n\r\nb\r\nc\r\n"])
    assert list(response.iter_lines()) == ["a", "", "b", "c"]

    response = httpx.Response(200, content=[b"", b"a\r", b"\n\r\nb\r\nc"])
    assert list(response.iter_lines()) == ["a", "", "b", "c"]

    # Issue #1033
    response = httpx.Response(200, content=[b"", b"12345\r\n", b"foo bar baz\r\n"])
    assert list(response.iter_lines()) == ["12345", "foo bar baz"]


@pytest.mark.parametrize(["text", "expected"], [("", [])])
def test_line_decoding_edge_cases(text: str, expected: typing.List[str]) -> None:
    assert httpx._decoders.LineDecoder().decode(text) == expected


def test_invalid_content_encoding_header():
    headers = [(b"Content-Encoding", b"invalid-header")]
    body = b"test 123"

    response = httpx.Response(
        200,
        headers=headers,
        content=body,
    )
    assert response.content == body


Review comment (suggestion):
This was the only way I could figure out how to cover an empty input to `LineDecoder.decode`.

Code after:

    # Issue #1033
    response = httpx.Response(
        200, content=[b"", b"12345\r", b"foo ", b"bar ", b"baz\r"]
    )
    assert list(response.iter_lines()) == ["12345", "foo bar baz"]


def test_line_decoder_crnl():
    response = httpx.Response(200, content=[b"", b"a\r\n\r\nb\r\nc"])
    assert list(response.iter_lines()) == ["a", "", "b", "c"]

    response = httpx.Response(200, content=[b"", b"a\r\n\r\nb\r\nc\r\n"])
    assert list(response.iter_lines()) == ["a", "", "b", "c"]

    response = httpx.Response(200, content=[b"", b"a\r", b"\n\r\nb\r\nc"])
    assert list(response.iter_lines()) == ["a", "", "b", "c"]

    # Issue #1033
    response = httpx.Response(200, content=[b"", b"12345\r\n", b"foo bar baz\r\n"])
    assert list(response.iter_lines()) == ["12345", "foo bar baz"]


def test_invalid_content_encoding_header():
    headers = [(b"Content-Encoding", b"invalid-header")]
    body = b"test 123"

    response = httpx.Response(
        200,
        headers=headers,
        content=body,
    )
    assert response.content == body
Repository: encode/httpx
File: httpx/_config.py (Python)

Code before:
from ._models import Headers
from ._types import HeaderTypes, TimeoutTypes
from ._urls import URL

__all__ = ["Limits", "Proxy", "SSLContext", "Timeout"]

DEFAULT_CIPHERS = ":".join(
    [
        "ECDHE+AESGCM",
        "ECDHE+CHACHA20",
        "DHE+AESGCM",
        "DHE+CHACHA20",
        "ECDH+AESGCM",
        "DH+AESGCM",
        "ECDH+AES",
        "DH+AES",
        "RSA+AESGCM",
        "RSA+AES",
        "!aNULL",
        "!eNULL",
        "!MD5",
        "!DSS",
    ]
)


logger = logging.getLogger("httpx")


class UnsetType:
    pass  # pragma: no cover


UNSET = UnsetType()


class SSLContext(ssl.SSLContext):
    def __init__(
        self,
        verify: bool = True,
    ) -> None:
        super().__init__()
        self._verify = verify

        # Our SSL setup here is similar to the stdlib `ssl.create_default_context()`
        # implementation, except with `certifi` used for certificate verification.
        if not verify:
            self.check_hostname = False
            self.verify_mode = ssl.CERT_NONE
            return


Review comment (refactor):
when `verify=False`, we actually don't load certifi. let's move this comment to down.

Code after:
from ._types import HeaderTypes, TimeoutTypes
from ._urls import URL

__all__ = ["Limits", "Proxy", "SSLContext", "Timeout"]


class UnsetType:
    pass  # pragma: no cover


UNSET = UnsetType()


class SSLContext(ssl.SSLContext):
    def __init__(
        self,
        verify: bool = True,
    ) -> None:
        # ssl.SSLContext sets OP_NO_SSLv2, OP_NO_SSLv3, OP_NO_COMPRESSION,
        # OP_CIPHER_SERVER_PREFERENCE, OP_SINGLE_DH_USE and OP_SINGLE_ECDH_USE
        # by default. (from `ssl.create_default_context`)
        super().__init__()
        self._verify = verify

        # Our SSL setup here is similar to the stdlib `ssl.create_default_context()`
        # implementation, except with `certifi` used for certificate verification.
        if not verify:
            self.check_hostname = False
            self.verify_mode = ssl.CERT_NONE
            return

        self.verify_mode = ssl.CERT_REQUIRED
        self.check_hostname = True

        # Use stricter verify flags where possible.
        if hasattr(ssl, "VERIFY_X509_PARTIAL_CHAIN"):  # pragma: nocover
            self.verify_flags |= ssl.VERIFY_X509_PARTIAL_CHAIN
        if hasattr(ssl, "VERIFY_X509_STRICT"):  # pragma: nocover
            self.verify_flags |= ssl.VERIFY_X509_STRICT

        # Default to `certifi` for certificiate verification.
        self.load_verify_locations(cafile=certifi.where())

        # OpenSSL keylog file support.
        if hasattr(self, "keylog_filename"):
            keylogfile = os.environ.get("SSLKEYLOGFILE")
            if keylogfile and not sys.flags.ignore_environment:
                self.keylog_filename = keylogfile

    def __repr__(self) -> str:
        class_name = self.__class__.__name__
Repository: encode/httpx
File: tests/test_content.py (Python)

Code before:
import io
import typing

import pytest

import httpx

method = "POST"
url = "https://www.example.com"


@pytest.mark.anyio
async def test_empty_content():
    request = httpx.Request(method, url)
    assert isinstance(request.stream, httpx.SyncByteStream)
    assert isinstance(request.stream, httpx.AsyncByteStream)

    sync_content = b"".join(list(request.stream))
    async_content = b"".join([part async for part in request.stream])

    assert request.headers == {"Host": "www.example.com", "Content-Length": "0"}
    assert sync_content == b""
    assert async_content == b""


@pytest.mark.anyio
async def test_bytes_content():
    request = httpx.Request(method, url, content=b"Hello, world!")
    assert isinstance(request.stream, typing.Iterable)
    assert isinstance(request.stream, typing.AsyncIterable)

    sync_content = b"".join(list(request.stream))

Review comment (refactor):
I've just noticed this accidentally snuck in a private import.

@BERRADA-Omar would you be up for refactoring these tests?

We can test this against public API using `httpx.Request()`. Eg...

```python
data = {...}
req = httpx.Request("POST", "https://www.example.com/", json=data)
assert req.content == ...
assert req.headers = ...
```

Code after:
import io
import typing

import pytest

import httpx
from httpx._content import encode_json

method = "POST"
url = "https://www.example.com"


@pytest.mark.anyio
async def test_empty_content():
    request = httpx.Request(method, url)
    assert isinstance(request.stream, httpx.SyncByteStream)
    assert isinstance(request.stream, httpx.AsyncByteStream)

    sync_content = b"".join(list(request.stream))
    async_content = b"".join([part async for part in request.stream])

    assert request.headers == {"Host": "www.example.com", "Content-Length": "0"}
    assert sync_content == b""
    assert async_content == b""


@pytest.mark.anyio
async def test_bytes_content():
    request = httpx.Request(method, url, content=b"Hello, world!")
    assert isinstance(request.stream, typing.Iterable)
    assert isinstance(request.stream, typing.AsyncIterable)