starlette 1.6.0: types.Message

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.types.Message 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.types.Message

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:be209a34df5dbd4fadf137c265b38f401665523b9b2b09ce17f04d0543c67212","contract":["starlette.types.Message is a MutableMapping alias representing ASGI event dictionaries","ASGI receive callable produces Message instances that can be inspected and transformed by middleware","Intercepting outgoing http.response.start Message allows mutating status code and response headers","Outgoing http.response.body Message carries response body bytes and more_body continuation flag","Direct invocation of ASGIApp passes Message dicts across receive and send without HTTP network server"],"goal":"verify starlette.types.Message in pkg:pypi/starlette@1.6.0","kind":"HOW","packages":["pkg:pypi/starlette@1.6.0"],"schemaVersion":1,"symbols":["starlette.types.Message"]},"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","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/starlette@1.6.0"],"schemaVersion":1,"subject":"pkg:pypi/starlette@1.6.0","symbols":["starlette.types.Message"],"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

"""Sample demonstrating ASGI Message handling and middleware interception with starlette.types.Message."""

from typing import Callable, List, Optional, Tuple
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 MessageInterceptorMiddleware:
    """Pure ASGI middleware demonstrating incoming and outgoing starlette.types.Message inspection and modification."""

    def __init__(
        self,
        app: ASGIApp,
        on_receive_message: Optional[Callable[[Message], None]] = None,
        on_send_message: Optional[Callable[[Message], None]] = None,
        inject_header: Optional[Tuple[bytes, bytes]] = None,
    ) -> None:
        self.app = app
        self.on_receive_message = on_receive_message
        self.on_send_message = on_send_message
        self.inject_header = inject_header

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

        async def intercepted_receive() -> Message:
            message: Message = await receive()
            if self.on_receive_message is not None:
                self.on_receive_message(message)
            return message

        async def intercepted_send(message: Message) -> None:
            if message["type"] == "http.response.start" and self.inject_header is not None:
                headers = MutableHeaders(scope=message)
                key, val = self.inject_header
                headers[key.decode("latin-1")] = val.decode("latin-1")

            if self.on_send_message is not None:
                self.on_send_message(message)

            await send(message)

        await self.app(scope, intercepted_receive, intercepted_send)


class MessageRecorder:
    """Utility to record all ASGI Message instances exchanged during request-response cycles."""

    def __init__(self) -> None:
        self.received_messages: List[Message] = []
        self.sent_messages: List[Message] = []

    def record_received(self, message: Message) -> None:
        self.received_messages.append(dict(message))

    def record_sent(self, message: Message) -> None:
        self.sent_messages.append(dict(message))


async def echo_message_endpoint(request) -> JSONResponse:
    """Endpoint returning details about the request payload."""
    body_bytes = await request.body()
    return JSONResponse({
        "status": "ok",
        "received_bytes_length": len(body_bytes),
        "body_text": body_bytes.decode("utf-8") if body_bytes else "",
    })


async def text_endpoint(request) -> PlainTextResponse:
    """Endpoint returning plain text response."""
    return PlainTextResponse("Starlette Message Verified")


def create_app(
    on_receive: Optional[Callable[[Message], None]] = None,
    on_send: Optional[Callable[[Message], None]] = None,
    inject_header: Optional[Tuple[bytes, bytes]] = (b"x-message-interceptor", b"active"),
) -> ASGIApp:
    """Create a Starlette application wrapped with ASGI Message interception middleware."""
    routes = [
        Route("/echo", echo_message_endpoint, methods=["POST"]),
        Route("/text", text_endpoint, methods=["GET"]),
    ]
    raw_app = Starlette(debug=False, routes=routes)
    return MessageInterceptorMiddleware(
        raw_app,
        on_receive_message=on_receive,
        on_send_message=on_send,
        inject_header=inject_header,
    )


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

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

"""Contract verification tests for starlette.types.Message in Starlette 1.6.0."""

import asyncio
from pathlib import Path
import sys
from typing import List

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

from starlette.datastructures import MutableHeaders
from starlette.testclient import TestClient
from starlette.types import ASGIApp, Message, Receive, Scope, Send

from sample import MessageInterceptorMiddleware, MessageRecorder, create_app


def test_message_type_conformance() -> None:
    """Assertion 1: starlette.types.Message is a MutableMapping alias representing ASGI event dictionaries."""
    sample_msg: Message = {
        "type": "http.response.start",
        "status": 200,
        "headers": [(b"content-type", b"application/json")],
    }

    # Verify dictionary operations supported by Message type
    assert isinstance(sample_msg, dict)
    assert sample_msg["type"] == "http.response.start"
    assert sample_msg.get("status") == 200

    # In-place mutations on Message dictionary
    sample_msg["status"] = 204
    sample_msg["headers"].append((b"x-custom", b"val"))
    assert sample_msg["status"] == 204
    assert (b"x-custom", b"val") in sample_msg["headers"]


def test_receive_message_interception() -> None:
    """Assertion 2: ASGI receive callable produces Message instances that can be inspected and transformed by middleware."""
    recorder = MessageRecorder()
    app = create_app(on_receive=recorder.record_received, on_send=recorder.record_sent)
    client = TestClient(app)

    payload = "payload-data-alpha"
    resp = client.post("/echo", content=payload.encode("utf-8"))

    assert resp.status_code == 200
    assert resp.json()["received_bytes_length"] == len(payload)
    assert resp.json()["body_text"] == payload

    # Verify recorded receive message
    assert len(recorder.received_messages) >= 1
    req_msg: Message = recorder.received_messages[0]
    assert req_msg["type"] == "http.request"
    assert req_msg["body"] == payload.encode("utf-8")
    assert req_msg.get("more_body", False) is False


def test_send_message_response_start_mutation() -> None:
    """Assertion 3: Intercepting outgoing http.response.start Message allows mutating status code and response headers."""
    recorder = MessageRecorder()
    app = create_app(
        on_receive=recorder.record_received,
        on_send=recorder.record_sent,
        inject_header=(b"x-message-interceptor", b"active-v1"),
    )
    client = TestClient(app)

    resp = client.get("/text")
    assert resp.status_code == 200
    assert resp.text == "Starlette Message Verified"
    assert resp.headers["x-message-interceptor"] == "active-v1"

    # Verify recorded send messages
    start_messages = [m for m in recorder.sent_messages if m["type"] == "http.response.start"]
    assert len(start_messages) == 1
    start_msg: Message = start_messages[0]
    assert start_msg["status"] == 200

    header_dict = dict(start_msg["headers"])
    assert header_dict[b"x-message-interceptor"] == b"active-v1"


def test_send_message_response_body_lifecycle() -> None:
    """Assertion 4: Outgoing http.response.body Message carries response body bytes and more_body continuation flag."""
    recorder = MessageRecorder()
    app = create_app(on_receive=recorder.record_received, on_send=recorder.record_sent)
    client = TestClient(app)

    resp = client.get("/text")
    assert resp.status_code == 200

    body_messages = [m for m in recorder.sent_messages if m["type"] == "http.response.body"]
    assert len(body_messages) >= 1
    body_msg: Message = body_messages[0]
    assert body_msg["type"] == "http.response.body"
    assert body_msg["body"] == b"Starlette Message Verified"
    assert body_msg.get("more_body", False) is False


def test_direct_asgi_raw_message_exchange() -> None:
    """Assertion 5: Direct invocation of ASGIApp passes Message dicts across receive and send without HTTP network server."""
    app: ASGIApp = create_app()

    scope: Scope = {
        "type": "http",
        "http_version": "1.1",
        "method": "POST",
        "path": "/echo",
        "raw_path": b"/echo",
        "query_string": b"",
        "headers": [(b"content-type", b"text/plain")],
    }

    received_output_messages: List[Message] = []

    async def mock_receive() -> Message:
        return {
            "type": "http.request",
            "body": b"direct-asgi-payload",
            "more_body": False,
        }

    async def mock_send(message: Message) -> None:
        received_output_messages.append(message)

    async def run_asgi() -> None:
        await app(scope, mock_receive, mock_send)

    asyncio.run(run_asgi())

    assert len(received_output_messages) == 2
    assert received_output_messages[0]["type"] == "http.response.start"
    assert received_output_messages[0]["status"] == 200
    assert received_output_messages[1]["type"] == "http.response.body"
    assert b"direct-asgi-payload" in received_output_messages[1]["body"]


def main() -> None:
    test_message_type_conformance()
    test_receive_message_interception()
    test_send_message_response_start_mutation()
    test_send_message_response_body_lifecycle()
    test_direct_asgi_raw_message_exchange()
    print("Contract verification succeeded for starlette.types.Message in starlette 1.6.0.")


if __name__ == "__main__":
    main()