uvicorn 0.52.1: run

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 uvicorn.run in pkg:pypi/uvicorn@0.52.1
Kind: HOW

Use EXACTLY these public packages and versions:
  - pkg:pypi/uvicorn@0.52.1
Demonstrate these symbols/APIs:
  - uvicorn.run

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:d1e30d363b69c356cea7cf38a26fc28ee68a4da602d35edb5f5cd9f5d391623c","contract":["assert uvicorn.run initializes and executes ASGI application serving HTTP requests on configured host and port","assert uvicorn.run handles multiple HTTP methods and paths returning valid status codes and response bodies","assert uvicorn.run supports factory callable pattern with factory=True to instantiate application on startup","assert uvicorn.Config correctly configures host, port, headers, and resolves application targets via load()"],"goal":"verify uvicorn.run in pkg:pypi/uvicorn@0.52.1","kind":"HOW","packages":["pkg:pypi/uvicorn@0.52.1"],"schemaVersion":1,"symbols":["uvicorn.run"]},"contractCommand":["python","test/contract.py"],"environment":{"arch":"x64","distro":"ubuntu","ecosystem":"pypi","libc":"glibc","libcVersion":"2.39","os":"linux","osVersionBucket":"24","packageManager":"pip","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/uvicorn@0.52.1"],"schemaVersion":1,"subject":"pkg:pypi/uvicorn@0.52.1","symbols":["uvicorn.run"],"verifierAdapter":"python@1"}

[project]
name = "verify-uvicorn-run"
version = "0.1.0"
dependencies = [
    "click==8.1.8",
    "h11==0.16.0",
    "uvicorn==0.52.1",
]

click==8.1.8
h11==0.16.0
uvicorn==0.52.1

"""Sample demonstrating ASGI application serving with uvicorn.run."""

from typing import Any, Callable, Dict, MutableMapping
import uvicorn

Scope = MutableMapping[str, Any]
Receive = Callable[[], Any]
Send = Callable[[Dict[str, Any]], Any]


async def asgi_app(scope: Scope, receive: Receive, send: Send) -> None:
    """Minimal ASGI application handling HTTP requests."""
    if scope["type"] == "http":
        path = scope.get("path", "/")
        if path == "/health":
            body = b'{"status": "ok"}'
            headers = [
                (b"content-type", b"application/json"),
                (b"content-length", str(len(body)).encode("ascii")),
            ]
            await send({
                "type": "http.response.start",
                "status": 200,
                "headers": headers,
            })
            await send({"type": "http.response.body", "body": body})
        elif path == "/echo":
            method = scope.get("method", "GET")
            body = f"Method: {method}".encode("utf-8")
            headers = [
                (b"content-type", b"text/plain"),
                (b"content-length", str(len(body)).encode("ascii")),
            ]
            await send({
                "type": "http.response.start",
                "status": 200,
                "headers": headers,
            })
            await send({"type": "http.response.body", "body": body})
        else:
            body = b"Not Found"
            headers = [
                (b"content-type", b"text/plain"),
                (b"content-length", str(len(body)).encode("ascii")),
            ]
            await send({
                "type": "http.response.start",
                "status": 404,
                "headers": headers,
            })
            await send({"type": "http.response.body", "body": body})


def app_factory() -> Any:
    """Factory function returning an ASGI application."""
    return asgi_app


def start_server(host: str = "127.0.0.1", port: int = 8000) -> None:
    """Run the ASGI application using uvicorn.run."""
    uvicorn.run(
        asgi_app,
        host=host,
        port=port,
        log_level="info",
        access_log=False,
    )


if __name__ == "__main__":
    start_server()

{
  "schemaVersion": 1,
  "goal": "verify uvicorn.run in pkg:pypi/uvicorn@0.52.1",
  "kind": "HOW",
  "packages": [
    "pkg:pypi/uvicorn@0.52.1"
  ],
  "symbols": [
    "uvicorn.run"
  ]
}

"""Contract verification tests for uvicorn.run in pkg:pypi/uvicorn@0.52.1."""

import http.client
import json
import os
import socket
import sys
import threading
import time
from typing import Any, List

# Ensure repository root is on sys.path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

import uvicorn
import sample


def get_free_port() -> int:
    """Find an available TCP port on localhost."""
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind(("127.0.0.1", 0))
        return int(s.getsockname()[1])


def test_uvicorn_run_and_http_serving() -> None:
    """Assertion 1 & 2: uvicorn.run executes ASGI app serving HTTP requests on configured host and port, handling multiple methods and paths."""
    port = get_free_port()
    active_servers: List[uvicorn.Server] = []
    orig_server_init = uvicorn.Server.__init__

    def tracking_server_init(self: uvicorn.Server, *args: Any, **kwargs: Any) -> None:
        orig_server_init(self, *args, **kwargs)
        active_servers.append(self)

    uvicorn.Server.__init__ = tracking_server_init
    try:
        server_thread = threading.Thread(
            target=uvicorn.run,
            args=(sample.asgi_app,),
            kwargs={
                "host": "127.0.0.1",
                "port": port,
                "log_level": "error",
                "access_log": False,
                "lifespan": "off",
            },
            daemon=True,
        )
        server_thread.start()

        # Wait for server to start
        timeout = 10.0
        start_time = time.time()
        while not active_servers or not getattr(active_servers[-1], "started", False):
            if time.time() - start_time > timeout:
                raise TimeoutError("Server failed to start within timeout")
            time.sleep(0.02)

        server = active_servers[-1]

        # Test GET /health
        conn = http.client.HTTPConnection("127.0.0.1", port, timeout=5)
        conn.request("GET", "/health")
        resp = conn.getresponse()
        assert resp.status == 200, f"Expected 200, got {resp.status}"
        data = json.loads(resp.read().decode("utf-8"))
        assert data == {"status": "ok"}
        conn.close()

        # Test POST /echo
        conn = http.client.HTTPConnection("127.0.0.1", port, timeout=5)
        conn.request("POST", "/echo")
        resp = conn.getresponse()
        assert resp.status == 200, f"Expected 200, got {resp.status}"
        body = resp.read().decode("utf-8")
        assert body == "Method: POST"
        conn.close()

        # Test 404 /unknown
        conn = http.client.HTTPConnection("127.0.0.1", port, timeout=5)
        conn.request("GET", "/unknown")
        resp = conn.getresponse()
        assert resp.status == 404, f"Expected 404, got {resp.status}"
        body = resp.read().decode("utf-8")
        assert body == "Not Found"
        conn.close()

        # Signal shutdown and join thread
        server.should_exit = True
        server_thread.join(timeout=5)
        assert not server_thread.is_alive(), "Server thread failed to terminate"
    finally:
        uvicorn.Server.__init__ = orig_server_init


def test_uvicorn_run_with_factory() -> None:
    """Assertion 3: uvicorn.run supports factory callable pattern with factory=True."""
    port = get_free_port()
    active_servers: List[uvicorn.Server] = []
    orig_server_init = uvicorn.Server.__init__
    factory_called: List[bool] = []

    def tracking_server_init(self: uvicorn.Server, *args: Any, **kwargs: Any) -> None:
        orig_server_init(self, *args, **kwargs)
        active_servers.append(self)

    def test_factory() -> Any:
        factory_called.append(True)
        return sample.asgi_app

    uvicorn.Server.__init__ = tracking_server_init
    try:
        server_thread = threading.Thread(
            target=uvicorn.run,
            args=(test_factory,),
            kwargs={
                "host": "127.0.0.1",
                "port": port,
                "factory": True,
                "log_level": "error",
                "access_log": False,
                "lifespan": "off",
            },
            daemon=True,
        )
        server_thread.start()

        timeout = 10.0
        start_time = time.time()
        while not active_servers or not getattr(active_servers[-1], "started", False):
            if time.time() - start_time > timeout:
                raise TimeoutError("Server with factory failed to start within timeout")
            time.sleep(0.02)

        assert len(factory_called) > 0, "Application factory should have been called"

        server = active_servers[-1]

        conn = http.client.HTTPConnection("127.0.0.1", port, timeout=5)
        conn.request("GET", "/health")
        resp = conn.getresponse()
        assert resp.status == 200, f"Expected 200, got {resp.status}"
        conn.close()

        server.should_exit = True
        server_thread.join(timeout=5)
        assert not server_thread.is_alive(), "Server thread failed to terminate"
    finally:
        uvicorn.Server.__init__ = orig_server_init


def test_uvicorn_config_and_load() -> None:
    """Assertion 4: uvicorn.Config correctly configures host, port, headers, and resolves application targets via load()."""
    config = uvicorn.Config(
        sample.asgi_app,
        host="127.0.0.1",
        port=9090,
        headers=[("x-service-name", "sample-service")],
        log_level="warning",
        access_log=False,
    )
    assert config.host == "127.0.0.1"
    assert config.port == 9090
    assert ("x-service-name", "sample-service") in config.headers
    assert config.access_log is False

    # Test import string loading via Config.load()
    import_config = uvicorn.Config("sample:asgi_app", host="127.0.0.1", port=9091)
    import_config.load()
    assert import_config.loaded is True
    assert callable(import_config.loaded_app)


def main() -> int:
    test_uvicorn_run_and_http_serving()
    test_uvicorn_run_with_factory()
    test_uvicorn_config_and_load()
    print("All uvicorn.run contract verification tests passed successfully.")
    return 0


if __name__ == "__main__":
    sys.exit(main())