starlette 1.6.0: datastructures.MutableHeaders, testclient, types.ASGIApp

Clean-room public code sample — generation instructions

Write a brand-new, minimal, self-contained code sample in this clean-room directory.
Do not copy, paraphrase, or reference any existing project source. Work only from this spec.

A csx.json manifest scaffold already exists. Do not recreate it from memory. Preserve its case.goal, packages and symbols; fill its empty case.contract with exact assertions and correct its environment, commands and verifierAdapter for the files you generate.

Goal: verify starlette in pkg:pypi/starlette@1.6.0
Kind: HOW

Use EXACTLY these public packages and versions:
  - pkg:pypi/starlette@1.6.0
Demonstrate these symbols/APIs:
  - starlette
  - starlette.datastructures.MutableHeaders
  - starlette.testclient
  - starlette.types.ASGIApp

Rules:
  - One focused purpose; the smallest project that proves the goal.
  - Include a contract test (test/contract.*) that runs OFFLINE and exits 0 exactly when the goal behavior works.
  - Pin every dependency with a lockfile so resolution is reproducible.
  - No secrets, credentials, or tokens. No real URLs (only example.com or localhost). No absolute paths.
  - No personal names, emails, company names, or project identifiers of any kind.
  - No binaries and no generated output (node_modules, dist, target, venv, .git, .env).
  - Keep it under 200 files and 256KB packed.

{"case":{"caseId":"case:sha256:c3e6140583c14401bce360c6e950d0a7e8721cbc0530b09ae5ce2b38ed2959a1","contract":["MutableHeaders wraps ASGI scope headers and modifies raw headers in-place with lowercase normalization and insertion order preservation","Pure ASGI middleware conforming to ASGIApp signature intercepts requests and stamps headers using MutableHeaders","TestClient dispatches requests offline against http://testserver through the full ASGI pipeline","Starlette TestClient binds httpx2 by default when httpx2 is installed, returning httpx2.Response","Calling raise_for_status on a TestClient error response raises httpx2.HTTPStatusError rather than httpx.HTTPStatusError","Importing starlette.testclient without httpx2 falls back to httpx and emits StarletteDeprecationWarning, which subclasses UserWarning"],"goal":"verify starlette in pkg:pypi/starlette@1.6.0","kind":"HOW","packages":["pkg:pypi/starlette@1.6.0"],"schemaVersion":1,"symbols":["starlette","starlette.datastructures.MutableHeaders","starlette.testclient","starlette.types.ASGIApp"]},"contractCommand":["python","test/contract.py"],"environment":{"arch":"x64","distro":"ubuntu","ecosystem":"pypi","language":"python","libc":"glibc","libcVersion":"2.39","os":"linux","osVersionBucket":"24","packageManager":"pip","runtime":"python","runtimeVersion":"3.12","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/starlette@1.6.0"],"schemaVersion":1,"subject":"pkg:pypi/starlette@1.6.0","symbols":["starlette","starlette.datastructures.MutableHeaders","starlette.testclient","starlette.types.ASGIApp"],"verifierAdapter":"python@1"}

starlette==1.6.0
anyio==4.14.2
idna==3.19
typing-extensions==4.16.0
httpx2==2.10.0
httpcore2==2.10.0
truststore==0.10.4
httpx==0.28.1
httpcore==1.0.9
h11==0.16.0
certifi==2026.7.22

"""Sample demonstrating Starlette ASGI application, pure ASGI middleware, MutableHeaders, and TestClient."""

from starlette.applications import Starlette
from starlette.datastructures import MutableHeaders
from starlette.responses import JSONResponse, PlainTextResponse
from starlette.routing import Route
from starlette.types import ASGIApp, Message, Receive, Scope, Send


class HeaderMutationMiddleware:
    """Pure ASGI middleware demonstrating ASGIApp protocol and MutableHeaders manipulation."""

    def __init__(self, app: ASGIApp, header_key: str = "x-processed-by", header_val: str = "starlette-app") -> None:
        self.app = app
        self.header_key = header_key
        self.header_val = header_val

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        async def send_wrapper(message: Message) -> None:
            if message["type"] == "http.response.start":
                headers = MutableHeaders(scope=message)
                headers[self.header_key] = self.header_val
            await send(message)

        await self.app(scope, receive, send_wrapper)


async def home_endpoint(request) -> JSONResponse:
    """Endpoint returning a JSON payload."""
    return JSONResponse({"status": "ok", "framework": "starlette"})


async def echo_header_endpoint(request) -> PlainTextResponse:
    """Endpoint echoing a request header."""
    val = request.headers.get("x-custom-input", "none")
    return PlainTextResponse(f"received: {val}")


async def error_endpoint(request) -> JSONResponse:
    """Endpoint returning 404 error."""
    return JSONResponse({"error": "resource not found"}, status_code=404)


def create_app() -> ASGIApp:
    """Create a Starlette application wrapped with pure ASGI middleware."""
    routes = [
        Route("/", home_endpoint, methods=["GET"]),
        Route("/echo", echo_header_endpoint, methods=["GET"]),
        Route("/not-found", error_endpoint, methods=["GET"]),
    ]
    raw_app = Starlette(debug=False, routes=routes)
    return HeaderMutationMiddleware(raw_app, header_key="x-custom-marker", header_val="verified-v1")


if __name__ == "__main__":
    app = create_app()
    print("Starlette sample application configured.")

{
  "schemaVersion": 1,
  "goal": "verify starlette in pkg:pypi/starlette@1.6.0",
  "kind": "HOW",
  "packages": [
    "pkg:pypi/starlette@1.6.0"
  ],
  "symbols": [
    "starlette",
    "starlette.datastructures.MutableHeaders",
    "starlette.testclient",
    "starlette.types.ASGIApp"
  ]
}

"""Contract verification tests for Starlette 1.6.0, MutableHeaders, ASGIApp, and TestClient."""

import importlib
import os
import sys
import warnings
from pathlib import Path

# Ensure project root is on sys.path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

import httpx
import httpx2
import starlette
import starlette.testclient
from starlette.datastructures import MutableHeaders
from starlette.exceptions import StarletteDeprecationWarning
from starlette.testclient import TestClient
from starlette.types import ASGIApp

from sample import HeaderMutationMiddleware, create_app


def test_mutable_headers_operations() -> None:
    """Assertion 1: MutableHeaders wraps ASGI scope headers and modifies raw headers in-place with lowercase normalization and insertion order preservation."""
    # Test dictionary-backed and scope-backed MutableHeaders
    message = {
        "type": "http.response.start",
        "status": 200,
        "headers": [
            (b"content-type", b"text/plain"),
            (b"x-trace-id", b"12345"),
        ],
    }

    headers = MutableHeaders(scope=message)

    # Key lookup is case-insensitive
    assert headers["X-Trace-ID"] == "12345"
    assert headers["content-type"] == "text/plain"

    # In-place replacement retains insertion order
    headers["X-Trace-ID"] = "67890"
    assert message["headers"][1] == (b"x-trace-id", b"67890")

    # Setting new key appends to raw headers
    headers["X-New-Header"] = "hello"
    assert (b"x-new-header", b"hello") in message["headers"]

    # Deleting header removes it from raw headers list
    del headers["x-new-header"]
    assert (b"x-new-header", b"hello") not in message["headers"]

    # Test setdefault and append
    val = headers.setdefault("X-Default-Key", "default-val")
    assert val == "default-val"
    assert headers["x-default-key"] == "default-val"

    headers.append("x-multi", "first")
    headers.append("x-multi", "second")
    assert headers.getlist("x-multi") == ["first", "second"]


def test_asgi_app_middleware_interception() -> None:
    """Assertion 2: Pure ASGI middleware conforming to ASGIApp signature intercepts requests and stamps headers using MutableHeaders."""
    app: ASGIApp = create_app()
    assert callable(app)

    client = TestClient(app)
    resp = client.get("/")
    assert resp.status_code == 200
    assert resp.json() == {"status": "ok", "framework": "starlette"}
    assert resp.headers["x-custom-marker"] == "verified-v1"


def test_testclient_offline_dispatch() -> None:
    """Assertion 3: TestClient dispatches requests offline against http://testserver through the full ASGI pipeline."""
    app = create_app()
    client = TestClient(app)

    # TestClient targets http://testserver without binding host ports
    assert str(client.base_url) == "http://testserver"

    # Test request passing headers and receiving response
    resp = client.get("/echo", headers={"x-custom-input": "offline-test"})
    assert resp.status_code == 200
    assert resp.text == "received: offline-test"
    assert resp.headers["x-custom-marker"] == "verified-v1"


def test_testclient_binds_httpx2_by_default() -> None:
    """Assertion 4: Starlette TestClient binds httpx2 by default when httpx2 is installed, returning httpx2.Response."""
    app = create_app()
    client = TestClient(app)

    assert starlette.testclient.httpx is httpx2

    resp = client.get("/")
    assert isinstance(resp, httpx2.Response)
    assert not isinstance(resp, httpx.Response)


def test_testclient_raise_for_status_exception_type() -> None:
    """Assertion 5: Calling raise_for_status on a TestClient error response raises httpx2.HTTPStatusError rather than httpx.HTTPStatusError."""
    app = create_app()
    client = TestClient(app)

    err_resp = client.get("/not-found")
    assert err_resp.status_code == 404

    try:
        err_resp.raise_for_status()
    except httpx2.HTTPStatusError as exc:
        assert not isinstance(exc, httpx.HTTPStatusError)
        assert exc.response.status_code == 404
    else:
        raise AssertionError("Expected httpx2.HTTPStatusError was not raised")


def test_testclient_fallback_to_httpx_with_warning() -> None:
    """Assertion 6: Importing starlette.testclient without httpx2 falls back to httpx and emits StarletteDeprecationWarning, which subclasses UserWarning."""
    class BlockHttpx2Import:
        def find_spec(self, name, path=None, target=None):
            if name == "httpx2":
                raise ModuleNotFoundError("No module named 'httpx2'", name="httpx2")
            return None

    blocker = BlockHttpx2Import()
    sys.meta_path.insert(0, blocker)

    # Remove cached modules to force reload
    saved_httpx2 = sys.modules.pop("httpx2", None)
    saved_testclient = sys.modules.pop("starlette.testclient", None)

    try:
        with warnings.catch_warnings(record=True) as caught:
            warnings.simplefilter("always")
            fallback_module = importlib.import_module("starlette.testclient")

            assert fallback_module.httpx is httpx
            assert len(caught) >= 1
            warning_obj = caught[0].message
            assert isinstance(warning_obj, StarletteDeprecationWarning)
            assert issubclass(StarletteDeprecationWarning, UserWarning)
            assert not issubclass(StarletteDeprecationWarning, DeprecationWarning)
            assert "httpx2" in str(warning_obj)
    finally:
        sys.meta_path.remove(blocker)
        if saved_httpx2 is not None:
            sys.modules["httpx2"] = saved_httpx2
        if saved_testclient is not None:
            sys.modules["starlette.testclient"] = saved_testclient
        else:
            importlib.import_module("starlette.testclient")


def main() -> None:
    test_mutable_headers_operations()
    test_asgi_app_middleware_interception()
    test_testclient_offline_dispatch()
    test_testclient_binds_httpx2_by_default()
    test_testclient_raise_for_status_exception_type()
    test_testclient_fallback_to_httpx_with_warning()
    print("Contract verification succeeded for starlette 1.6.0.")


if __name__ == "__main__":
    main()