diff --git a/CHANGELOG.md b/CHANGELOG.md index 43f0a9b8..e2cc4b95 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,31 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [2.6.5] - Unreleased + +- Improve header ergonomics: dictionary-style assignment and `Headers.update()` + accept strings or bytes, so `response.headers["X-Example"] = "value"` is + supported while headers remain stored internally as bytes. String names use + ASCII and string values use Latin-1; unrepresentable characters raise + `UnicodeEncodeError`. Assignment still replaces all case-insensitive matches. +- Add `Headers.get_text_values(name, encoding="latin-1")` to decode matching + values on demand as a tuple, preserving repeated header entries. + `get_first_text()` returns the first value as a string, or `None` when absent; + `get_single_text()` returns a string and requires exactly one matching entry, + raising `ValueError` otherwise. Both also accept an explicit encoding. + Existing byte-valued reads and low-level byte-based methods remain unchanged. +- Complete header dictionary symmetry: lookup, membership, and deletion accept + string or byte names; lookup always returns byte values. Add `Headers.add_text()` + to append without replacing matches, and `Headers.from_values()` to build a + byte-backed collection from mappings or iterables while preserving duplicates. +- Use the same ASCII-name/Latin-1-value policy for HTTP client `headers=` and + `default_headers=`, route header filters, and testing-client iterable headers. + Previously, HTTP client string values required ASCII. Testing-client dictionaries + retain their historical UTF-8 encoding for backward compatibility. + Pass explicitly encoded bytes for field-specific encodings. +- Preserve raw header values when sending HTTP/2 requests, without decoding them + as UTF-8, so Latin-1 values and explicitly encoded bytes reach the wire unchanged. + ## [2.6.4] - 2026-10-11 :libra: - Add a `vary` parameter to the `cache_control` decorator and to diff --git a/blacksheep/client/connection.py b/blacksheep/client/connection.py index 80f3b6d8..a6f9c4e0 100644 --- a/blacksheep/client/connection.py +++ b/blacksheep/client/connection.py @@ -362,7 +362,9 @@ async def connect(self) -> None: self._connected = True - def _convert_request_to_h2_headers(self, request: Request) -> list[tuple[str, str]]: + def _convert_request_to_h2_headers( + self, request: Request + ) -> list[tuple[bytes, bytes]]: """Convert a BlackSheep Request to HTTP/2 pseudo-headers and headers.""" # HTTP/2 pseudo-headers path = request.url.path or b"/" @@ -370,36 +372,36 @@ def _convert_request_to_h2_headers(self, request: Request) -> list[tuple[str, st path = path + b"?" + request.url.query headers = [ - (":method", request.method), - (":path", path.decode("utf-8")), - (":scheme", self._cached_scheme), - (":authority", self.host), + (b":method", request.method.encode("ascii")), + (b":path", path), + (b":scheme", self._cached_scheme.encode("ascii")), + (b":authority", self.host.encode("utf-8")), ] # Add Content-Type header from content if present if request.content and request.content.type: content_type = request.content.type - content_type_str = ( - content_type.decode("utf-8") + content_type_bytes = ( + content_type if isinstance(content_type, bytes) - else content_type + else content_type.encode("ascii") ) - headers.append(("content-type", content_type_str)) + headers.append((b"content-type", content_type_bytes)) # Add regular headers for name, value in request.headers: - name_str = ( - name.decode("utf-8").lower() + name_bytes = ( + name.lower() if isinstance(name, bytes) - else name.lower() + else name.encode("ascii").lower() ) - value_str = value.decode("utf-8") if isinstance(value, bytes) else value + value_bytes = value if isinstance(value, bytes) else value.encode("latin-1") # Skip headers that are represented as pseudo-headers in HTTP/2 - if name_str in ("host", "connection", "transfer-encoding"): + if name_bytes in (b"host", b"connection", b"transfer-encoding"): continue - headers.append((name_str, value_str)) + headers.append((name_bytes, value_bytes)) return headers @@ -457,7 +459,7 @@ async def send(self, request: Request) -> Response: # Check for Expect: 100-continue header expect_continue = any( - (h[0] == "expect" and h[1] == "100-continue") for h in h2_headers + (h[0] == b"expect" and h[1] == b"100-continue") for h in h2_headers ) # Send headers (without end_stream if expecting 100-continue with body) diff --git a/blacksheep/common/types.py b/blacksheep/common/types.py index 738d5534..9bcbc0e6 100644 --- a/blacksheep/common/types.py +++ b/blacksheep/common/types.py @@ -2,20 +2,21 @@ Common types annotations and functions. """ -from typing import AnyStr, Iterable +from collections.abc import Mapping +from typing import AnyStr, Iterable, TypeVar +from blacksheep.headers import Headers from blacksheep.url import URL KeyValuePair = tuple[AnyStr, AnyStr] -HeadersType = dict[AnyStr, AnyStr] | Iterable[KeyValuePair] +_HeaderName = TypeVar("_HeaderName", bound=str | bytes) +HeadersType = ( + Mapping[_HeaderName, str | bytes] | Iterable[tuple[str | bytes, str | bytes]] +) ParamsType = dict[AnyStr, AnyStr] | Iterable[KeyValuePair] URLType = str | bytes | URL -def _ensure_header_bytes(value: AnyStr) -> bytes: - return value if isinstance(value, bytes) else value.encode("ascii") - - def _ensure_param_str(value: AnyStr) -> str: return value if isinstance(value, str) else value.decode("ascii") @@ -25,15 +26,7 @@ def normalize_headers( ) -> list[tuple[bytes, bytes]] | None: if headers is None: return None - if isinstance(headers, dict): - return [ - (_ensure_header_bytes(key), _ensure_header_bytes(value)) # type: ignore - for key, value in headers.items() - ] - return [ - (_ensure_header_bytes(key), _ensure_header_bytes(value)) - for key, value in headers - ] + return Headers.from_values(headers).values def normalize_params(params: ParamsType | None) -> list[tuple[str, str]] | None: diff --git a/blacksheep/headers.py b/blacksheep/headers.py index 7cb5db07..075436a6 100644 --- a/blacksheep/headers.py +++ b/blacksheep/headers.py @@ -1,5 +1,23 @@ -from collections.abc import Mapping, MutableSequence -from typing import Generator +from collections.abc import Iterable, Mapping, MutableSequence +from typing import Generator, TypeVar + +_HeaderName = TypeVar("_HeaderName", bound=str | bytes) + + +def _encode_header_name(name: str | bytes) -> bytes: + if isinstance(name, str): + return name.encode("ascii") + if not isinstance(name, bytes): + raise TypeError("Header names must be str or bytes") + return name + + +def _encode_header_value(value: str | bytes) -> bytes: + if isinstance(value, str): + return value.encode("latin-1") + if not isinstance(value, bytes): + raise TypeError("Header values must be str or bytes") + return value class Header: @@ -26,6 +44,27 @@ def __init__(self, values: list[tuple[bytes, bytes]] | None = None): values = [] self.values = values + @classmethod + def from_values( + cls, + values: ( + Mapping[_HeaderName, str | bytes] + | Iterable[tuple[str | bytes, str | bytes]] + ), + ) -> "Headers": + """Create byte-backed headers from a mapping or iterable, preserving duplicates.""" + items = values.items() if isinstance(values, Mapping) else values + return cls( + [ + (_encode_header_name(name), _encode_header_value(value)) + for name, value in items + ] + ) + + def add_text(self, name: str | bytes, value: str | bytes) -> None: + """Append an entry without replacing matches, encoding strings at the boundary.""" + self.add(_encode_header_name(name), _encode_header_value(value)) + def get(self, name: bytes) -> tuple[bytes, ...]: results = [] name = name.lower() @@ -34,6 +73,25 @@ def get(self, name: bytes) -> tuple[bytes, ...]: results.append(header[1]) return tuple(results) + def get_text_values( + self, name: str | bytes, encoding: str = "latin-1" + ) -> tuple[str, ...]: + """Decode matching values on demand, preserving separate header entries.""" + return tuple( + value.decode(encoding) for value in self.get(_encode_header_name(name)) + ) + + def get_first_text( + self, name: str | bytes, encoding: str = "latin-1" + ) -> str | None: + """Decode the first matching value, or return None when absent.""" + value = self.get_first(_encode_header_name(name)) + return value.decode(encoding) if value is not None else None + + def get_single_text(self, name: str | bytes, encoding: str = "latin-1") -> str: + """Decode exactly one matching value; raise ValueError otherwise.""" + return self.get_single(_encode_header_name(name)).decode(encoding) + def get_tuples(self, name: bytes) -> list[tuple[bytes, bytes]]: results = [] name = name.lower() @@ -63,7 +121,7 @@ def merge(self, values: list[tuple[bytes, bytes]]): continue self.values.append(header) - def update(self, values: dict[bytes, bytes]): + def update(self, values: Mapping[_HeaderName, bytes | str]): for key, value in values.items(): self[key] = value @@ -139,11 +197,12 @@ def __iadd__( def __iter__(self) -> Generator[tuple[bytes, bytes], None, None]: yield from self.values - def __setitem__(self, key: bytes, value: bytes): - self.set(key, value) + def __setitem__(self, key: str | bytes, value: str | bytes): + """Replace matching entries, encoding text names as ASCII and values as Latin-1.""" + self.set(_encode_header_name(key), _encode_header_value(value)) - def __getitem__(self, item: bytes) -> tuple[bytes, ...]: - return self.get(item) + def __getitem__(self, item: str | bytes) -> tuple[bytes, ...]: + return self.get(_encode_header_name(item)) def keys(self) -> tuple[bytes, ...]: results = [] @@ -176,11 +235,11 @@ def contains(self, key: bytes) -> bool: return True return False - def __delitem__(self, key: bytes): - self.remove(key) + def __delitem__(self, key: str | bytes): + self.remove(_encode_header_name(key)) - def __contains__(self, key: bytes) -> bool: - return self.contains(key) + def __contains__(self, key: str | bytes) -> bool: + return self.contains(_encode_header_name(key)) def __repr__(self) -> str: return f"" diff --git a/blacksheep/headers.pyi b/blacksheep/headers.pyi index 885c805b..bf924da1 100644 --- a/blacksheep/headers.pyi +++ b/blacksheep/headers.pyi @@ -1,5 +1,7 @@ -from collections.abc import MutableSequence -from typing import Generator +from collections.abc import Iterable, Mapping, MutableSequence +from typing import Generator, TypeVar + +_HeaderName = TypeVar("_HeaderName", bound=str | bytes) class Header: def __init__(self, name: bytes, value: bytes): @@ -16,12 +18,28 @@ class Headers: def __init__(self, values: list[HeaderType] | None = None): self.values = values + @classmethod + def from_values( + cls, + values: Mapping[_HeaderName, str | bytes] + | Iterable[tuple[str | bytes, str | bytes]], + ) -> "Headers": ... + def add_text(self, name: str | bytes, value: str | bytes) -> None: ... def get(self, name: bytes) -> tuple[bytes]: ... + def get_text_values( + self, name: str | bytes, encoding: str = "latin-1" + ) -> tuple[str, ...]: ... + def get_first_text( + self, name: str | bytes, encoding: str = "latin-1" + ) -> str | None: ... + def get_single_text( + self, name: str | bytes, encoding: str = "latin-1" + ) -> str: ... def get_tuples(self, name: bytes) -> list[HeaderType]: ... def get_first(self, key: bytes) -> bytes | None: ... def get_single(self, key: bytes) -> bytes: ... def merge(self, values: list[HeaderType]) -> None: ... - def update(self, values: dict[bytes, bytes]) -> None: ... + def update(self, values: Mapping[_HeaderName, bytes | str]) -> None: ... def items(self) -> Generator[HeaderType, None, None]: ... def clone(self) -> "Headers": ... def add_many( @@ -37,13 +55,13 @@ class Headers: self, other: "Headers" | Header | HeaderType | MutableSequence ) -> "Headers": ... def __iter__(self) -> Generator[HeaderType, None, None]: ... - def __setitem__(self, key: bytes, value: bytes) -> None: ... - def __getitem__(self, item: bytes) -> tuple[bytes]: ... + def __setitem__(self, key: str | bytes, value: str | bytes) -> None: ... + def __getitem__(self, item: str | bytes) -> tuple[bytes, ...]: ... def keys(self) -> tuple[bytes, ...]: ... def add(self, name: bytes, value: bytes) -> None: ... def set(self, name: bytes, value: bytes) -> None: ... def remove(self, key: bytes) -> None: ... def contains(self, key: bytes) -> bool: ... - def __delitem__(self, key: bytes) -> None: ... - def __contains__(self, key: bytes) -> bool: ... + def __delitem__(self, key: str | bytes) -> None: ... + def __contains__(self, key: str | bytes) -> bool: ... def __repr__(self) -> str: ... diff --git a/blacksheep/headers.pyx b/blacksheep/headers.pyx index 8214daa7..d1bc0bc6 100644 --- a/blacksheep/headers.pyx +++ b/blacksheep/headers.pyx @@ -1,4 +1,20 @@ -from collections.abc import Mapping, MutableSequence +from collections.abc import Iterable, Mapping, MutableSequence +from typing import TypeVar + +_HeaderName = TypeVar("_HeaderName", bound=str | bytes) + +cdef inline bytes _encode_header_name(object name): + cdef bytes raw_name = name.encode("ascii") if isinstance(name, str) else name + if raw_name is None: + raise TypeError("Header names must be str or bytes") + return raw_name + + +cdef inline bytes _encode_header_value(object value): + cdef bytes raw_value = value.encode("latin-1") if isinstance(value, str) else value + if raw_value is None: + raise TypeError("Header values must be str or bytes") + return raw_value cdef class Header: @@ -27,6 +43,23 @@ cdef class Headers: values = [] self.values = values + @classmethod + def from_values( + cls, + values: Mapping[_HeaderName, str | bytes] + | Iterable[tuple[str | bytes, str | bytes]], + ): + """Create byte-backed headers from a mapping or iterable, preserving duplicates.""" + items = values.items() if isinstance(values, Mapping) else values + return cls([ + (_encode_header_name(name), _encode_header_value(value)) + for name, value in items + ]) + + def add_text(self, name: str | bytes, value: str | bytes): + """Append an entry without replacing matches, encoding strings at the boundary.""" + self.add(_encode_header_name(name), _encode_header_value(value)) + cpdef tuple get(self, bytes name): cdef list results = [] cdef tuple header @@ -36,6 +69,20 @@ cdef class Headers: results.append(header[1]) return tuple(results) + def get_text_values(self, name: str | bytes, str encoding = "latin-1"): + """Decode matching values on demand, preserving separate header entries.""" + cdef bytes raw_name = _encode_header_name(name) + return tuple(value.decode(encoding) for value in self.get(raw_name)) + + def get_first_text(self, name: str | bytes, str encoding = "latin-1"): + """Decode the first matching value, or return None when absent.""" + cdef bytes value = self.get_first(_encode_header_name(name)) + return value.decode(encoding) if value is not None else None + + def get_single_text(self, name: str | bytes, str encoding = "latin-1"): + """Decode exactly one matching value; raise ValueError otherwise.""" + return self.get_single(_encode_header_name(name)).decode(encoding) + cpdef list get_tuples(self, bytes name): cdef list results = [] cdef tuple header @@ -67,7 +114,7 @@ cdef class Headers: continue self.values.append(header) - def update(self, dict values: dict[bytes, bytes]): + def update(self, values: Mapping[_HeaderName, bytes | str]): for key, value in values.items(): self[key] = value @@ -133,13 +180,14 @@ cdef class Headers: def __iter__(self): yield from self.values - def __setitem__(self, bytes key, bytes value): + def __setitem__(self, key: str | bytes, value: str | bytes): + """Replace matching entries, encoding text names as ASCII and values as Latin-1.""" # Not obvious, but here we make the decision that setter removes existing headers with matching name: # it feels more natural with syntax: headers[b'X-Foo'] = b'Something' - self.set(key, value) + self.set(_encode_header_name(key), _encode_header_value(value)) - def __getitem__(self, bytes item): - return self.get(item) + def __getitem__(self, item: str | bytes): + return self.get(_encode_header_name(item)) cpdef tuple keys(self): cdef bytes name, value @@ -179,11 +227,11 @@ cdef class Headers: return True return False - def __delitem__(self, bytes key): - self.remove(key) + def __delitem__(self, key: str | bytes): + self.remove(_encode_header_name(key)) - def __contains__(self, bytes key): - return self.contains(key) + def __contains__(self, key: str | bytes): + return self.contains(_encode_header_name(key)) def __repr__(self): return f'' diff --git a/blacksheep/testing/helpers.py b/blacksheep/testing/helpers.py index a45298a3..9894ddd5 100644 --- a/blacksheep/testing/helpers.py +++ b/blacksheep/testing/helpers.py @@ -1,6 +1,9 @@ from typing import Sequence from urllib.parse import quote, urlencode +from blacksheep.common.types import HeadersType as HeaderValues +from blacksheep.common.types import normalize_headers + _DEFAULT_AGENT = ( b"Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:63.0) Gecko/20100101 Firefox/63.0" ) @@ -19,7 +22,7 @@ def _get_tuple(value: list | tuple[str, int]) -> tuple[str, int]: return tuple(value) # type: ignore -HeadersType = None | Sequence[tuple[bytes, bytes]] | dict[str, str] +HeadersType = HeaderValues | None CookiesType = None | Sequence[tuple[bytes, bytes]] | dict[str, str] QueryType = None | bytes | str | dict | list @@ -64,9 +67,16 @@ def get_example_scope( host = f"{server[0]}{port_part}" if isinstance(extra_headers, dict): + # Preserve the testing client's historical UTF-8 mapping convention. extra_headers = [ - (key.encode(), value.encode()) for key, value in extra_headers.items() + ( + key.encode("utf-8") if isinstance(key, str) else key, + value.encode("utf-8") if isinstance(value, str) else value, + ) + for key, value in extra_headers.items() ] + else: + extra_headers = normalize_headers(extra_headers) query_string: bytes = b"" if query: diff --git a/tests/client/test_headers.py b/tests/client/test_headers.py index 80e8e801..b6461b10 100644 --- a/tests/client/test_headers.py +++ b/tests/client/test_headers.py @@ -1,5 +1,11 @@ -from blacksheep import Response, TextContent +import pytest +from h2.config import H2Configuration +from h2.connection import H2Connection +from h2.events import RequestReceived + +from blacksheep import Headers, Request, Response, TextContent from blacksheep.client import ClientSession +from blacksheep.client.connection import HTTP2Connection from . import FakePools @@ -64,3 +70,68 @@ async def middleware_for_assertions(request, next_handler): default_headers=[(b"Hello", b"World"), (b"Foo", b"Power")], ) as client: await client.get(b"/", headers=[(b"Hello", b"Kitty")]) + + +@pytest.mark.parametrize( + "method", ["get", "post", "put", "delete", "trace", "head", "patch", "options"] +) +async def test_client_text_headers_share_encoding_and_preserve_duplicates(method): + fake_pools = FakePools([Response(200)]) + + async def middleware_for_assertions(request, next_handler): + assert request.headers["x-example"] == (b"caf\xe9", b"\xff") + assert request.headers["x-default"] == (b"default=\xe9",) + assert request.headers.get_text_values("x-example") == ("caf\xe9", "\xff") + request.headers.add_text("X-Example", "another") + assert len(request.headers["x-example"]) == 3 + return await next_handler(request) + + async with ClientSession( + base_url="http://localhost:8080", + pools=fake_pools, + middlewares=[middleware_for_assertions], + default_headers={"X-Example": "overridden", "X-Default": "default=\xe9"}, + ) as client: + await getattr(client, method)( + "/", headers=[("x-example", "caf\xe9"), (b"X-Example", b"\xff")] + ) + + +def test_http2_header_serialization_preserves_raw_values_and_duplicates(): + pools = FakePools([]) + connection = HTTP2Connection(pools, "localhost", 443) + request = Request( + "POST", + b"https://localhost/example?q=test", + Headers.from_values( + [ + ("X-Example", "caf\xe9"), + (b"X-Example", b"\xff"), + ("Expect", "100-continue"), + ("Host", "ignored"), + ("Connection", "keep-alive"), + ("Transfer-Encoding", "chunked"), + ] + ).values, + ) + request.with_content(TextContent("hello")) + headers = connection._convert_request_to_h2_headers(request) + assert (b":path", b"/example?q=test") in headers + assert (b"expect", b"100-continue") in headers + assert (b"content-type", b"text/plain; charset=utf-8") in headers + assert not any( + name in (b"host", b"connection", b"transfer-encoding") for name, _ in headers + ) + + peer = H2Connection(config=H2Configuration(client_side=False)) + peer.initiate_connection() + connection.h2_conn.initiate_connection() + connection.h2_conn.send_headers(1, headers, end_stream=False) + events = peer.receive_data(connection.h2_conn.data_to_send()) + received = next(event for event in events if isinstance(event, RequestReceived)) + assert [ + (name, value) for name, value in received.headers if name == b"x-example" + ] == [ + (b"x-example", b"caf\xe9"), + (b"x-example", b"\xff"), + ] diff --git a/tests/test_headers.py b/tests/test_headers.py index 0dcae347..4a4c411c 100644 --- a/tests/test_headers.py +++ b/tests/test_headers.py @@ -1,6 +1,115 @@ +from types import MappingProxyType + import pytest -from blacksheep import Header, Headers +from blacksheep import Header, Headers, Request, Response +from blacksheep.common.types import normalize_headers +from blacksheep.scribe import send_asgi_response +from blacksheep.testing.helpers import get_example_scope + + +@pytest.mark.parametrize("name", ["X-EXAMPLE", b"X-EXAMPLE"]) +def test_dictionary_operations_accept_text_and_bytes(name): + headers = Headers([(b"x-example", b"first"), (b"X-Example", b"second")]) + assert name in headers + assert headers[name] == (b"first", b"second") + del headers[name] + assert name not in headers + assert headers[name] == () + del headers[name] + + +@pytest.mark.parametrize("name", [None, 123, "X-\xe9"]) +@pytest.mark.parametrize("operation", ["lookup", "contains", "delete"]) +def test_dictionary_operations_reject_invalid_names_without_mutation(name, operation): + headers = Headers([(b"x-example", b"value")]) + error = UnicodeEncodeError if isinstance(name, str) else TypeError + with pytest.raises(error): + if operation == "lookup": + headers[name] + elif operation == "contains": + name in headers + else: + del headers[name] + assert headers.values == [(b"x-example", b"value")] + + +@pytest.mark.parametrize("name", ["Set-Cookie", b"Set-Cookie"]) +@pytest.mark.parametrize("value", ["b=\xe9", b"b=\xe9", "", b""]) +def test_add_text_preserves_duplicate_entries(name, value): + headers = Headers([(b"set-cookie", b"a=1")]) + headers.add_text(name, value) + raw_value = value.encode("latin-1") if isinstance(value, str) else value + assert headers["SET-cookie"] == (b"a=1", raw_value) + + +@pytest.mark.parametrize( + "name,value,error", + [ + ("X-\xe9", "value", UnicodeEncodeError), + ("X-Example", "\u20ac", UnicodeEncodeError), + (None, "value", TypeError), + ("X-Example", None, TypeError), + (123, "value", TypeError), + ("X-Example", 123, TypeError), + ], +) +def test_add_text_and_factory_reject_invalid_inputs(name, value, error): + headers = Headers([(b"X-Example", b"original")]) + with pytest.raises(error): + headers.add_text(name, value) + assert headers.values == [(b"X-Example", b"original")] + with pytest.raises(error): + Headers.from_values([(name, value)]) + with pytest.raises(error): + normalize_headers([(name, value)]) + + +@pytest.mark.parametrize( + "values", + [ + {"X-Example": "caf\xe9", "X-Other": b"\xff"}, + MappingProxyType({"X-Example": "caf\xe9", "X-Other": b"\xff"}), + [(b"X-Example", "caf\xe9"), ("X-Other", b"\xff")], + ], +) +def test_factory_and_normalization_share_encoding_policy(values): + expected = [(b"X-Example", b"caf\xe9"), (b"X-Other", b"\xff")] + assert Headers.from_values(values).values == expected + assert normalize_headers(values) == expected + + +def test_factory_and_normalization_preserve_generator_duplicates_and_copy_storage(): + original = [(b"Set-Cookie", b"a=1"), ("set-cookie", "b=\xe9")] + for create in (Headers.from_values, normalize_headers): + values = create(item for item in original) + if isinstance(values, Headers): + values = values.values + assert values == [(b"Set-Cookie", b"a=1"), (b"set-cookie", b"b=\xe9")] + assert values is not original + assert original[1] == ("set-cookie", "b=\xe9") + assert Headers.from_values({}).values == [] + assert normalize_headers([]) == [] + assert normalize_headers(None) is None + + +def test_testing_scope_iterable_uses_same_header_encoding_policy(): + scope = get_example_scope( + "GET", "/", extra_headers=[("X-Example", "caf\xe9"), (b"X-Example", b"\xff")] + ) + assert (b"X-Example", b"caf\xe9") in scope["headers"] + assert (b"X-Example", b"\xff") in scope["headers"] + + +def test_testing_scope_mapping_preserves_legacy_utf8_encoding(): + scope = get_example_scope( + "GET", + "/", + extra_headers={"X-Caf\xe9": "caf\xe9", "X-Euro": "\u20ac", b"X-Raw": b"\xff"}, + ) + assert ("X-Caf\xe9".encode("utf-8"), "caf\xe9".encode("utf-8")) in scope["headers"] + assert (b"X-Euro", "\u20ac".encode("utf-8")) in scope["headers"] + assert (b"X-Raw", b"\xff") in scope["headers"] @pytest.mark.parametrize( @@ -65,6 +174,140 @@ def test_http_header_collection_item_setter(): assert example == (b"Hello, World",) +@pytest.mark.parametrize("name", ["X-Example", b"X-Example"]) +@pytest.mark.parametrize("value", ["caf\xe9", b"caf\xe9", "", b""]) +def test_text_assignment_stores_bytes_and_replaces_all_matching_entries(name, value): + headers = Headers( + [(b"x-example", b"old"), (b"X-EXAMPLE", b"other"), (b"Other", b"kept")] + ) + headers[name] = value + + raw_value = value.encode("latin-1") if isinstance(value, str) else value + assert headers.values == [(b"Other", b"kept"), (b"X-Example", raw_value)] + assert headers[b"x-example"] == (raw_value,) + assert headers.get_text_values("X-EXAMPLE") == (raw_value.decode("latin-1"),) + + +def test_update_accepts_text_and_bytes(): + headers = Headers() + headers.update({"X-Text": "value", b"X-Bytes": b"raw", "X-Mixed": b"mixed"}) + + assert list(headers.items()) == [ + (b"X-Text", b"value"), + (b"X-Bytes", b"raw"), + (b"X-Mixed", b"mixed"), + ] + + +@pytest.mark.parametrize("name", ["SET-cookie", b"SET-cookie"]) +def test_get_text_values_preserves_multiple_values_and_raw_storage(name): + values = [(b"Set-Cookie", b"a=1"), (b"set-cookie", b"b=\xff")] + headers = Headers(values) + + assert headers.get_text_values(name) == ("a=1", "b=\xff") + assert headers.get_text_values("missing") == () + assert headers.values is values + assert headers[b"set-cookie"] == (b"a=1", b"b=\xff") + + +@pytest.mark.parametrize( + "method", ["get_text_values", "get_first_text", "get_single_text"] +) +def test_text_reads_support_explicit_encoding_and_strict_decoding(method): + headers = Headers([(b"X-Text", "caf\xe9".encode("utf-8"))]) + read = getattr(headers, method) + expected = ("caf\xe9",) if method == "get_text_values" else "caf\xe9" + assert read("X-Text", encoding="utf-8") == expected + + headers[b"X-Text"] = b"\xff" + with pytest.raises(UnicodeDecodeError): + read("X-Text", encoding="utf-8") + + +@pytest.mark.parametrize("name", ["X-EXAMPLE", b"X-EXAMPLE"]) +@pytest.mark.parametrize("value", [b"", b"caf\xe9"]) +def test_single_value_text_reads(name, value): + headers = Headers([(b"x-example", value)]) + assert headers.get_first_text(name) == value.decode("latin-1") + assert headers.get_single_text(name) == value.decode("latin-1") + assert headers.get_first_text("missing") is None + with pytest.raises(ValueError, match="does not contain one header"): + headers.get_single_text("missing") + + +def test_first_text_only_decodes_first_value_and_single_text_rejects_duplicates(): + headers = Headers([(b"X-Example", b"first"), (b"x-example", b"\xff")]) + assert headers.get_first_text("X-Example", encoding="utf-8") == "first" + with pytest.raises(ValueError, match="more than one header"): + headers.get_single_text("X-Example", encoding="utf-8") + + +@pytest.mark.parametrize( + "name,value", + [("X-\xe9", "value"), ("X-Example", "\u20ac")], +) +def test_text_assignment_encoding_errors_do_not_mutate_headers(name, value): + headers = Headers([(b"X-Example", b"original")]) + with pytest.raises(UnicodeEncodeError): + headers[name] = value + assert headers.values == [(b"X-Example", b"original")] + + +@pytest.mark.parametrize( + "method", ["get_text_values", "get_first_text", "get_single_text"] +) +def test_text_reads_reject_non_ascii_names(method): + with pytest.raises(UnicodeEncodeError): + getattr(Headers(), method)("X-\xe9") + + +@pytest.mark.parametrize( + "name,value", + [(123, "value"), ("X-Example", 123), (None, "value"), ("X-Example", None)], +) +def test_assignment_rejects_invalid_types_without_mutating_headers(name, value): + headers = Headers([(b"X-Example", b"original")]) + with pytest.raises(TypeError): + headers[name] = value + assert headers.values == [(b"X-Example", b"original")] + + +@pytest.mark.parametrize("name", [123, None]) +@pytest.mark.parametrize( + "method", ["get_text_values", "get_first_text", "get_single_text"] +) +def test_text_reads_reject_invalid_name_type(name, method): + with pytest.raises(TypeError): + getattr(Headers(), method)(name) + + +def test_request_text_headers_share_raw_storage(): + request = Request("GET", b"/", [(b"X-Example", b"original")]) + assert request.headers.get_text_values("x-example") == ("original",) + request.headers["X-Example"] = "updated" + assert request.get_first_header(b"x-example") == b"updated" + + +async def test_response_text_headers_are_sent_as_bytes(): + response = Response(200) + response.headers["X-Example"] = "caf\xe9" + response.headers.update({"X-Other": "value"}) + messages = [] + + async def send(message): + messages.append(message) + + await send_asgi_response(response, send) + + assert messages[0]["type"] == "http.response.start" + assert (b"X-Example", b"caf\xe9") in messages[0]["headers"] + assert (b"X-Other", b"value") in messages[0]["headers"] + assert all( + isinstance(name, bytes) and isinstance(value, bytes) + for name, value in messages[0]["headers"] + ) + + def test_http_header_collection_item_get_single_case_insensitive(): headers = Headers() headers[b"example"] = b"Hello, World"