failed WRONG_FIX UNSUBMITTED wrong_fix_unsubmitted(None) · None tool calls · 0 s · encode/httpx
(not found in data/tasks.jsonl)
| # | Tool | Arguments | Result |
|---|---|---|---|
| No trace captured. | |||
--- a/httpx/_models.py
+++ b/httpx/_models.py
@@ -1,8 +1,10 @@
from __future__ import annotations
+import codecs
import datetime
import email.message
import json as jsonlib
+import re
import typing
import urllib.request
from collections.abc import Mapping
@@ -44,15 +46,23 @@ from ._types import (
SyncByteStream,
)
from ._urls import URL
-from ._utils import (
- is_known_encoding,
- obfuscate_sensitive_headers,
- parse_content_type_charset,
- parse_header_links,
-)
+from ._utils import to_bytes_or_str, to_str
__all__ = ["Cookies", "Headers", "Request", "Response"]
+SENSITIVE_HEADERS = {"authorization", "proxy-authorization"}
+
+
+def _is_known_encoding(encoding: str) -> bool:
+ """
+ Return `True` if `encoding` is a known codec.
+ """
+ try:
+ codecs.lookup(encoding)
+ except LookupError:
+ return False
+ return True
+
def _normalize_header_key(key: str | bytes, encoding: str | None = None) -> bytes:
"""
@@ -72,6 +82,60 @@ def _normalize_header_value(value: str | bytes, encoding: str | None = None) ->
return value.encode(encoding or "ascii")
+def _parse_content_type_charset(content_type: str) -> str | None:
+ # We used to use `cgi.parse_header()` here, but `cgi` became a dead battery.
+ # See: https://peps.python.org/pep-0594/#cgi
+ msg = email.message.Message()
+ msg["content-type"] = content_type
+ return msg.get_content_charset(failobj=None)
+
+
+def _parse_header_links(value: str) -> list[dict[str, str]]:
+ """
+ Returns a list of parsed link headers, for more info see:
+ https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Link
+ The generic syntax of those is:
+ Link: < uri-reference >; param1=value1; param2="value2"
+ So for instance:
+ Link; '<http:/.../front.jpeg>; type="image/jpeg",<http://.../back.jpeg>;'
+ would return
+ [
+ {"url": "http:/.../front.jpeg", "type": "image/jpeg"},
+ {"url": "http://.../back.jpeg"},
+ ]
+ :param value: HTTP Link entity-header field
+ :return: list of parsed link headers
+ """
+ links: list[dict[str, str]] = []
+ replace_chars = " '\""
+ value = value.strip(replace_chars)
+ if not value:
+ return links
+ for val in re.split(", *<", value):
+ try:
+ url, params = val.split(";", 1)
+ except ValueError:
+ url, params = val, ""
+ link = {"url": url.strip("<> '\"")}
+ for param in params.split(";"):
+ try:
+ key, value = param.split("=")
+ except ValueError:
+ break
+ link[key.strip(replace_chars)] = value.strip(replace_chars)
+ links.append(link)
+ return links
+
+
+def _obfuscate_sensitive_headers(
+ items: typing.Iterable[tuple[typing.AnyStr, typing.AnyStr]],
+) -> typing.Iterator[tuple[typing.AnyStr, typing.AnyStr]]:
+ for k, v in items:
+ if to_str(k.lower()) in SENSITIVE_HEADERS:
+ v = to_bytes_or_str("[secure]", match_type_of=v)
+ yield k, v
+
+
class Headers(typing.MutableMapping[str, str]):
"""
HTTP headers, as a case-insensitive multi-dict.
@@ -306,7 +370,7 @@ class Headers(typing.MutableMapping[str, str]):
if self.encoding != "ascii":
encoding_str = f", encoding={self.encoding!r}"
- as_list = list(obfuscate_sensitive_headers(self.multi_items()))
+ as_list = list(_obfuscate_sensitive_headers(self.multi_items()))
as_dict = dict(as_list)
no_duplicate_keys = len(as_dict) == len(as_list)
@@ -599,7 +663,7 @@ class Response:
"""
if not hasattr(self, "_encoding"):
encoding = self.charset_encoding
- if encoding is None or not is_known_encoding(encoding):
+ if encoding is None or not _is_known_encoding(encoding):
if isinstance(self.default_encoding, str):
encoding = self.default_encoding
elif hasattr(self, "_content"):
@@ -630,7 +694,7 @@ class Response:
if content_type is None:
return None
- return parse_content_type_charset(content_type)
+ return _parse_content_type_charset(content_type)
def _get_content_decoder(self) -> ContentDecoder:
"""
@@ -785,7 +849,7 @@ class Response:
return {
(link.get("rel") or link.get("url")): link
- for link in parse_header_links(header)
+ for link in _parse_header_links(header)
}
@property
--- a/httpx/_utils.py
+++ b/httpx/_utils.py
@@ -1,7 +1,5 @@
from __future__ import annotations
-import codecs
-import email.message
import ipaddress
import os
import re
@@ -29,74 +27,6 @@ def primitive_value_to_str(value: PrimitiveData) -> str:
return str(value)
-def is_known_encoding(encoding: str) -> bool:
- """
- Return `True` if `encoding` is a known codec.
- """
- try:
- codecs.lookup(encoding)
- except LookupError:
- return False
- return True
-
-
-def parse_header_links(value: str) -> list[dict[str, str]]:
- """
- Returns a list of parsed link headers, for more info see:
- https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Link
- The generic syntax of those is:
- Link: < uri-reference >; param1=value1; param2="value2"
- So for instance:
- Link; '<http:/.../front.jpeg>; type="image/jpeg",<http://.../back.jpeg>;'
- would return
- [
- {"url": "http:/.../front.jpeg", "type": "image/jpeg"},
- {"url": "http://.../back.jpeg"},
- ]
- :param value: HTTP Link entity-header field
- :return: list of parsed link headers
- """
- links: list[dict[str, str]] = []
- replace_chars = " '\""
- value = value.strip(replace_chars)
- if not value:
- return links
- for val in re.split(", *<", value):
- try:
- url, params = val.split(";", 1)
- except ValueError:
- url, params = val, ""
- link = {"url": url.strip("<> '\"")}
- for param in params.split(";"):
- try:
- key, value = param.split("=")
- except ValueError:
- break
- link[key.strip(replace_chars)] = value.strip(replace_chars)
- links.append(link)
- return links
-
-
-def parse_content_type_charset(content_type: str) -> str | None:
- # We used to use `cgi.parse_header()` here, but `cgi` became a dead battery.
- # See: https://peps.python.org/pep-0594/#cgi
- msg = email.message.Message()
- msg["content-type"] = content_type
- return msg.get_content_charset(failobj=None)
-
-
-SENSITIVE_HEADERS = {"authorization", "proxy-authorization"}
-
-
-def obfuscate_sensitive_headers(
- items: typing.Iterable[tuple[typing.AnyStr, typing.AnyStr]],
-) -> typing.Iterator[tuple[typing.AnyStr, typing.AnyStr]]:
- for k, v in items:
- if to_str(k.lower()) in SENSITIVE_HEADERS:
- v = to_bytes_or_str("[secure]", match_type_of=v)
- yield k, v
-
-
def port_or_default(url: URL) -> int | None:
if url.port is not None:
return url.port
--- a/scripts/lint
+++ b/scripts/lint
@@ -8,5 +8,5 @@ export SOURCE_FILES="httpx tests"
set -x
-${PREFIX}ruff --fix $SOURCE_FILES
+${PREFIX}ruff check --fix $SOURCE_FILES
${PREFIX}ruff format $SOURCE_FILES
........................................F
=================================== FAILURES ===================================
_____________________________ test_logging_request _____________________________
server = <tests.conftest.TestServer object at 0x1037e7770>
caplog = <_pytest.logging.LogCaptureFixture object at 0x1045056a0>
def test_logging_request(server, caplog):
caplog.set_level(logging.INFO)
> with httpx.Client() as client:
tests/test_utils.py:58:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
httpx/_client.py:661: in __init__
self._transport = self._init_transport(
httpx/_client.py:709: in _init_transport
return HTTPTransport(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <httpx.HTTPTransport object at 0x104505fd0>, ssl_context = None
http1 = True, http2 = False
limits = Limits(max_connections=100, max_keepalive_connections=20, keepalive_expiry=5.0)
proxy = None, uds = None, local_address = None, retries = 0
socket_options = None, verify = None, cert = None
def __init__(
self,
ssl_context: ssl.SSLContext | None = None,
http1: bool = True,
http2: bool = False,
limits: Limits = DEFAULT_LIMITS,
proxy: ProxyTypes | None = None,
uds: str | None = None,
local_address: str | None = None,
retries: int = 0,
socket_options: typing.Iterable[SOCKET_OPTION] | None = None,
# Deprecated...
verify: typing.Any = None,
cert: typing.Any = None,
) -> None:
> import httpcore
E ModuleNotFoundError: No module named 'httpcore'
httpx/_transports/default.py:151: ModuleNotFoundError
---------------------------- Captured stderr setup -----------------------------
INFO: Started server process [41180]
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
--------------------------- Captured stderr teardown ---------------------------
INFO: Shutting down
INFO: Finished server process [41180]
!!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!
1 failed, 40 passed in 0.28s