Updated July 1, 2026

On this page

Reference for fastware's pure-ASGI middleware: CORS, request-ID tracing, request timing, trusted-host validation, and ViteDevProxy backend-first routing.

#Middleware API Reference

All middleware classes are pure ASGI -- no framework dependency beyond fastware's own send_error helper. This makes them streaming-safe (SSE, WebSocket) and avoids the response-buffering issues of BaseHTTPMiddleware-style wrappers.

Built-in middleware is automatically applied by create_app when the corresponding AppConfig fields are set. You can also use these classes directly for custom middleware stacks.

Middleware API Reference
Middleware	Purpose	`create_app` Parameter
`RequestIDMiddleware`	Assigns or propagates a unique X-Request-Id per request	`request_id`
`RequestTimingMiddleware`	Logs method, path, status, and duration; ring buffer for recent requests	`request_timing`
`CORSMiddleware`	Preflight OPTIONS handling and CORS response header injection	`cors_origins`
`TrustedHostMiddleware`	Rejects requests from unlisted Host headers (DNS rebinding protection)	`trusted_hosts`
`ViteDevProxy`	Proxies unmatched requests to a Vite dev server (backend-first routing)	`vite_dev_port`

#`src.fastware.middleware`

Pure ASGI middleware for request tracing, CORS headers, trusted-host validation, and Vite dev proxy routing, all streaming-safe for SSE and WebSocket.

All middleware classes are pure ASGI -- no framework dependency beyond fastware's own send_error helper. This keeps them streaming-safe (SSE, WebSocket) and avoids the response-buffering issues of BaseHTTPMiddleware-style wrappers.

Classes: RequestIDMiddleware — assigns/propagates X-Request-Id per request RequestTimingMiddleware — logs method, path, status, duration; ring buffer CORSMiddleware — preflight OPTIONS + response header injection TrustedHostMiddleware — rejects requests from unlisted Host headers ViteDevProxy — proxies non-API requests to a Vite dev server

#`RequestIDMiddleware`

Assign or propagate a unique request ID per request.

If the incoming request carries an X-Request-Id header, that value is reused. Otherwise a new UUID4 is generated. The ID is stored in scope["state"]["request_id"] and returned as an X-Request-Id response header.

When structlog is available, contextvars are cleared at the start of each request (preventing context leak from a previous request) and the request ID is bound so all log entries within the request include it.

#`RequestTimingMiddleware`

Log every HTTP request with method, path, status, and duration.

Wraps send to capture the status code from http.response.start, then uses try/finally so timing fires even for long-lived SSE streams (when the client disconnects the ASGI handler returns).

Args:

app: The inner ASGI application.
error_log: Optional ErrorLog instance. On 5xx responses the

middleware calls error_log.append(...) to persist the failure for dashboard visibility.

exclude_paths: Iterable of path prefixes to exclude from the ring

buffer (e.g. ["/events"]). Excluded requests are still logged, just not stored.

maxlen: Maximum number of entries in the ring buffer (default 10000).

#`CORSMiddleware`

Add CORS headers to responses and handle preflight OPTIONS requests.

Args:

app: The inner ASGI application.
allow_origins: List of allowed origins (e.g. ["http://localhost:5173"]).

Use ["*"] to allow any origin.

allow_methods: HTTP methods to advertise. Defaults to common methods.
allow_headers: Request headers the client may send. Defaults to

common headers.

allow_credentials: Whether to set Access-Control-Allow-Credentials.

Raises:

ValueError: if allow_origins contains "*" while

allow_credentials is True. Because this middleware echoes the request origin, that combination would grant credentialed cross-origin access to any site.

#`_cors_headers`

python

def _cors_headers(self, origin: str) -> list[tuple[bytes, bytes]]

Build the list of CORS response headers for origin.

#`TrustedHostMiddleware`

Reject requests whose Host header is not in the allow-list.

Prevents DNS rebinding attacks for servers bound to localhost.

Args:

app: The inner ASGI application.
allowed_hosts: List of hostnames (with optional port) to allow.

Use ["*"] to disable the check.

#`ViteDevProxy`

ASGI middleware that proxies unmatched requests to a Vite dev server.

Uses a backend-first routing strategy for HTTP: requests hit the backend first, streaming the response through message-by-message (SSE-safe). Only the http.response.start message is held back until the status is known; if the backend returns 404 (no route matched), the backend's response is discarded and the request is replayed to Vite instead. This means backend routes like /health or /metrics work without being under an API prefix.

Paths matching api_prefix or backend_prefixes always go straight to the backend with no 404-retry — their 404s belong to the client. WebSocket upgrades cannot be retried, so they use the same prefix rule: matching paths go to the backend; everything else is proxied to Vite (for HMR).

#`init`

python

def __init__(self, app: Callable, *, vite_port: int, api_prefix: str='/api', backend_prefixes: list[str] | None=None) -> None

Wrap app with backend-first Vite dev proxying.

Args:

app: The inner ASGI application (the fastware app) handled

first for every request.

vite_port: Port the Vite dev server is listening on; unmatched

requests are proxied there.

api_prefix: Path prefix for backend routes that always go

straight to the backend without a 404-retry (default "/api").

backend_prefixes: Additional backend path prefixes routed to

the app, notably for WebSocket upgrades (default ["/events", "/ws"]). /ws is the conventional app WebSocket path; routing it to the backend does not interfere with Vite HMR, whose websocket connects at / (identified by the vite-hmr subprotocol), not /ws.

#`_is_api_request`

python

def _is_api_request(self, path: str) -> bool

Return True if this path should go to the app, not be proxied.

The reserved /__fastware/ namespace (version endpoint and any future diagnostics) is always a backend prefix -- never proxied to Vite -- so those endpoints behave identically in dev and prod. This also matters for WebSocket upgrades, which cannot be 404-retried.

#`close`

python

async def close(self) -> None

#`_proxy_http`

python

async def _proxy_http(self, scope: Scope, send: Send, *, body: bytes=b'') -> None

Forward an HTTP request to the Vite dev server.

The body parameter contains the pre-captured request body (already consumed from receive by the backend during the try-first phase).

#`_proxy_ws`

python

async def _proxy_ws(self, scope: Scope, receive: Receive, send: Send) -> None

Bidirectional WebSocket proxy to Vite (for HMR).

Uses the websockets library. If it is not installed, or the proxy connection fails, the client connection is closed with code 1011 (internal error) and a reason -- never a clean 1000 close that would hide the failure.

python

class TestCORSMiddleware:

    @pytest.mark.anyio
    async def test_preflight_returns_204(self):
        """Preflight OPTIONS with matching origin returns 204 with CORS headers."""
        app = _make_app()
        mw = CORSMiddleware(app, allow_origins=["http://localhost:5173"])
        async with _client(mw) as client:
            resp = await client.options(
                "/health",
                headers={
                    "origin": "http://localhost:5173",
                    "access-control-request-method": "POST",
                },
            )
            assert resp.status_code == 204
            assert resp.headers["access-control-allow-origin"] == "http://localhost:5173"
            assert "POST" in resp.headers["access-control-allow-methods"]
            assert resp.headers.get("access-control-allow-credentials") == "true"

    @pytest.mark.anyio
    async def test_normal_request_gets_cors_headers(self):
        """A normal GET with matching origin gets CORS headers in the response."""
        app = _make_app()
        mw = CORSMiddleware(app, allow_origins=["http://localhost:5173"])
        async with _client(mw) as client:
            resp = await client.get(
                "/health",
                headers={"origin": "http://localhost:5173"},
            )
            assert resp.status_code == 200
            assert resp.headers["access-control-allow-origin"] == "http://localhost:5173"

    @pytest.mark.anyio
    async def test_unmatched_origin_passes_through(self):
        """A request from an unlisted origin gets no CORS headers."""
        app = _make_app()
        mw = CORSMiddleware(app, allow_origins=["http://allowed.example.com"])
        async with _client(mw) as client:
            resp = await client.get(
                "/health",
                headers={"origin": "http://evil.example.com"},
            )
            assert resp.status_code == 200
            assert "access-control-allow-origin" not in resp.headers

    @pytest.mark.anyio
    async def test_wildcard_origin(self):
        """allow_origins=["*"] (without credentials) matches any origin."""
        app = _make_app()
        mw = CORSMiddleware(app, allow_origins=["*"], allow_credentials=False)
        async with _client(mw) as client:
            resp = await client.get(
                "/health",
                headers={"origin": "http://anything.example.com"},
            )
            assert resp.headers["access-control-allow-origin"] == "http://anything.example.com"

    def test_wildcard_with_credentials_raises(self):
        """allow_origins=["*"] + allow_credentials=True is rejected at construction."""
        app = _make_app()
        with pytest.raises(ValueError, match="allow_credentials"):
            CORSMiddleware(app, allow_origins=["*"], allow_credentials=True)

    def test_wildcard_with_default_credentials_raises(self):
        """The default allow_credentials=True must also be rejected with '*'."""
        app = _make_app()
        with pytest.raises(ValueError, match="allow_credentials"):
            CORSMiddleware(app, allow_origins=["*"])

    @pytest.mark.anyio
    async def test_vary_origin_on_normal_response(self):
        """CORS responses carry Vary: Origin so caches key on the origin."""
        app = _make_app()
        mw = CORSMiddleware(app, allow_origins=["http://localhost:5173"])
        async with _client(mw) as client:
            resp = await client.get(
                "/health",
                headers={"origin": "http://localhost:5173"},
            )
            assert "origin" in resp.headers.get("vary", "").lower()

    @pytest.mark.anyio
    async def test_vary_origin_on_preflight(self):
        """Preflight responses carry Vary: Origin."""
        app = _make_app()
        mw = CORSMiddleware(app, allow_origins=["http://localhost:5173"])
        async with _client(mw) as client:
            resp = await client.options(
                "/health",
                headers={
                    "origin": "http://localhost:5173",
                    "access-control-request-method": "POST",
                },
            )
            assert resp.status_code == 204
            assert "origin" in resp.headers.get("vary", "").lower()

    @pytest.mark.anyio
    async def test_plain_options_reaches_app_route(self):
        """OPTIONS without Access-Control-Request-Method is NOT a preflight.

        App-defined OPTIONS routes must remain reachable; only real
        preflights (which carry Access-Control-Request-Method) are
        short-circuited by the middleware.
        """
        router = Router()

        async def custom_options(req):
            return JSONResponse({"custom": True})

        router.add_route("OPTIONS", "/thing", custom_options)
        app = create_app(router, request_id=False, request_timing=False)
        mw = CORSMiddleware(app, allow_origins=["http://test"])
        async with _client(mw) as client:
            resp = await client.options(
                "/thing",
                headers={"origin": "http://test"},
            )
            assert resp.status_code == 200
            assert resp.json() == {"custom": True}
            # CORS headers still injected on the pass-through response.
            assert resp.headers["access-control-allow-origin"] == "http://test"

    @pytest.mark.anyio
    async def test_no_origin_header_passes_through(self):
        """A request without an Origin header passes through without CORS headers."""
        app = _make_app()
        mw = CORSMiddleware(app, allow_origins=["http://localhost:5173"])
        async with _client(mw) as client:
            resp = await client.get("/health")
            assert resp.status_code == 200
            assert "access-control-allow-origin" not in resp.headers

    @pytest.mark.anyio
    async def test_custom_methods_and_headers(self):
        """Custom allow_methods and allow_headers are returned in preflight."""
        app = _make_app()
        mw = CORSMiddleware(
            app,
            allow_origins=["http://test"],
            allow_methods=["GET", "POST"],
            allow_headers=["x-custom"],
            allow_credentials=False,
        )
        async with _client(mw) as client:
            resp = await client.options(
                "/health",
                headers={
                    "origin": "http://test",
                    "access-control-request-method": "POST",
                },
            )
            assert resp.status_code == 204
            assert resp.headers["access-control-allow-methods"] == "GET, POST"
            assert resp.headers["access-control-allow-headers"] == "x-custom"
            assert "access-control-allow-credentials" not in resp.headers

    @pytest.mark.anyio
    async def test_non_http_passthrough(self):
        """Non-http scope types pass through without modification."""
        called = False

        async def inner_app(scope, receive, send):
            nonlocal called
            called = True

        mw = CORSMiddleware(inner_app, allow_origins=["*"], allow_credentials=False)
        await mw({"type": "websocket"}, None, None)
        assert called

python

class TestRequestIDMiddleware:

    @pytest.mark.anyio
    async def test_generates_request_id(self):
        """A new UUID is generated when no X-Request-Id header is sent."""
        app = _make_app()
        inner = RequestIDMiddleware(app)
        async with _client(inner) as client:
            resp = await client.get("/health")
            assert resp.status_code == 200
            rid = resp.headers.get("x-request-id")
            assert rid is not None
            assert len(rid) == 36  # UUID4 format

    @pytest.mark.anyio
    async def test_propagates_incoming_request_id(self):
        """An incoming X-Request-Id header is reused in the response."""
        app = _make_app()
        inner = RequestIDMiddleware(app)
        async with _client(inner) as client:
            resp = await client.get(
                "/health",
                headers={"x-request-id": "my-custom-id"},
            )
            assert resp.headers["x-request-id"] == "my-custom-id"

    @pytest.mark.anyio
    async def test_duplicate_header_first_value_wins(self):
        """With duplicate X-Request-Id headers, the first value is used."""
        app = _make_app()
        inner = RequestIDMiddleware(app)
        async with _client(inner) as client:
            resp = await client.get(
                "/health",
                headers=[("x-request-id", "first-id"), ("x-request-id", "second-id")],
            )
            assert resp.headers["x-request-id"] == "first-id"

    @pytest.mark.anyio
    async def test_stores_in_scope_state(self):
        """Request ID is stored in scope['state']['request_id']."""
        router = Router()
        captured = {}

        @router.get("/check")
        async def check(req):
            captured["request_id"] = req.state.get("request_id")
            return JSONResponse({"ok": True})

        app = create_app(router, request_id=False, request_timing=False)
        inner = RequestIDMiddleware(app)
        async with _client(inner) as client:
            resp = await client.get(
                "/check",
                headers={"x-request-id": "trace-123"},
            )
            assert resp.status_code == 200
            assert captured["request_id"] == "trace-123"

    @pytest.mark.anyio
    async def test_non_http_passthrough(self):
        """Non-http scope types pass through without modification."""
        called = False

        async def inner_app(scope, receive, send):
            nonlocal called
            called = True

        mw = RequestIDMiddleware(inner_app)
        await mw({"type": "lifespan"}, None, None)
        assert called

python

class TestRequestTimingMiddleware:

    @pytest.mark.anyio
    async def test_request_count_increments(self):
        """request_count increments on each request."""
        app = _make_app()
        mw = RequestTimingMiddleware(app)
        assert mw.request_count == 0
        async with _client(mw) as client:
            await client.get("/health")
        assert mw.request_count == 1

    @pytest.mark.anyio
    async def test_ring_buffer_captures_entry(self):
        """Ring buffer stores (timestamp, method, path, status, duration_ms)."""
        app = _make_app()
        mw = RequestTimingMiddleware(app)
        async with _client(mw) as client:
            await client.get("/health")
        assert len(mw.request_history) == 1
        entry = mw.request_history[0]
        assert entry[1] == "GET"
        assert entry[2] == "/health"
        assert entry[3] == 200
        assert entry[4] >= 0  # duration_ms

    @pytest.mark.anyio
    async def test_ring_buffer_maxlen(self):
        """Ring buffer respects maxlen."""
        app = _make_app()
        mw = RequestTimingMiddleware(app, maxlen=3)
        async with _client(mw) as client:
            for _ in range(5):
                await client.get("/health")
        assert len(mw.request_history) == 3

    @pytest.mark.anyio
    async def test_exclude_paths(self):
        """Excluded paths are not stored in the ring buffer."""
        app = _make_app()
        mw = RequestTimingMiddleware(app, exclude_paths=["/health"])
        async with _client(mw) as client:
            await client.get("/health")
        assert len(mw.request_history) == 0
        assert mw.request_count == 1  # Still counted

    @pytest.mark.anyio
    async def test_error_log_on_5xx(self, tmp_path):
        """On 5xx responses, the error_log.append() is called."""
        error_log = ErrorLog(tmp_path / "errors.db")
        app = _make_app()
        mw = RequestTimingMiddleware(app, error_log=error_log)
        async with _client(mw) as client:
            resp = await client.get("/fail")
            assert resp.status_code == 500
        entries = error_log.recent()
        assert len(entries) == 1
        assert entries[0]["method"] == "GET"
        assert entries[0]["path"] == "/fail"
        assert entries[0]["status_code"] == 500

    @pytest.mark.anyio
    async def test_non_http_passthrough(self):
        """Non-http scope types pass through without modification."""
        called = False

        async def inner_app(scope, receive, send):
            nonlocal called
            called = True

        mw = RequestTimingMiddleware(inner_app)
        await mw({"type": "lifespan"}, None, None)
        assert called

python

class TestTrustedHostMiddleware:

    @pytest.mark.anyio
    async def test_allowed_host_passes(self):
        """A request with a trusted Host header passes through."""
        app = _make_app()
        mw = TrustedHostMiddleware(app, allowed_hosts=["test"])
        async with _client(mw) as client:
            resp = await client.get("/health")
            assert resp.status_code == 200

    @pytest.mark.anyio
    async def test_disallowed_host_returns_400(self):
        """A request from an untrusted Host returns 400."""
        app = _make_app()
        mw = TrustedHostMiddleware(app, allowed_hosts=["trusted.local"])
        async with _client(mw, base_url="http://evil.local") as client:
            resp = await client.get("/health")
            assert resp.status_code == 400
            data = resp.json()
            assert "Invalid host" in data["detail"]

    @pytest.mark.anyio
    async def test_wildcard_allows_all(self):
        """allowed_hosts=["*"] allows any host."""
        app = _make_app()
        mw = TrustedHostMiddleware(app, allowed_hosts=["*"])
        async with _client(mw, base_url="http://anything.example.com") as client:
            resp = await client.get("/health")
            assert resp.status_code == 200

    @pytest.mark.anyio
    async def test_non_http_passthrough(self):
        """Non-http/ws scope types pass through without modification."""
        called = False

        async def inner_app(scope, receive, send):
            nonlocal called
            called = True

        mw = TrustedHostMiddleware(inner_app, allowed_hosts=["localhost"])
        await mw({"type": "lifespan"}, None, None)
        assert called

    @pytest.mark.anyio
    async def test_websocket_disallowed_host_closed(self):
        """DNS-rebinding protection applies to ws:// too — bad Host is rejected."""
        called = False

        async def inner_app(scope, receive, send):
            nonlocal called
            called = True

        mw = TrustedHostMiddleware(inner_app, allowed_hosts=["trusted.local"])
        sent: list[dict] = []
        incoming = [{"type": "websocket.connect"}]

        async def receive():
            return incoming.pop(0)

        async def send(message):
            sent.append(message)

        scope = {
            "type": "websocket",
            "path": "/ws",
            "headers": [(b"host", b"evil.local")],
        }
        await mw(scope, receive, send)
        assert not called
        assert sent[-1]["type"] == "websocket.close"
        assert sent[-1]["code"] == 1008  # policy violation

    @pytest.mark.anyio
    async def test_websocket_allowed_host_passes(self):
        """A websocket with a trusted Host header reaches the app."""
        called = False

        async def inner_app(scope, receive, send):
            nonlocal called
            called = True

        mw = TrustedHostMiddleware(inner_app, allowed_hosts=["trusted.local"])
        sent: list[dict] = []

        async def receive():
            return {"type": "websocket.connect"}

        async def send(message):
            sent.append(message)

        scope = {
            "type": "websocket",
            "path": "/ws",
            "headers": [(b"host", b"trusted.local")],
        }
        await mw(scope, receive, send)
        assert called
        assert sent == []  # middleware did not respond on the app's behalf

#CORS Configuration for a Typical SPA

When building a single-page application with a separate frontend dev server (e.g., Vite on port 5173), you need to configure CORS to allow the frontend origin. The CORSMiddleware handles preflight OPTIONS requests automatically and injects the correct Access-Control-Allow-Origin, Access-Control-Allow-Methods, and Access-Control-Allow-Headers response headers on every cross-origin request:

python

from fastware import Router, create_app, AppConfig

router = Router()

# Option 1: Via AppConfig
app = create_app(
    router,
    config=AppConfig(
        cors_origins=["http://localhost:5173"],
        api_prefix="/api",
        spa_fallback=Path("dist/index.html"),
    ),
)

# Option 2: Using CORSMiddleware directly for full control
from fastware.middleware import CORSMiddleware

app = create_app(
    router,
    middleware=[
        lambda app: CORSMiddleware(
            app,
            allow_origins=["http://localhost:5173", "https://myapp.com"],
            allow_methods=["GET", "POST", "PUT", "DELETE"],
            allow_headers=["authorization", "content-type", "x-csrf-token"],
            allow_credentials=True,
        ),
    ],
)

When cors_origins is set on AppConfig, create_app applies CORSMiddleware automatically with default methods and headers. Use the direct middleware approach when you need to customize allowed methods or headers.

#ViteDevProxy Routing

The ViteDevProxy middleware uses a backend-first routing strategy for HTTP requests, forwarding unmatched paths to the Vite dev server so that frontend assets, HMR WebSocket connections, and backend API routes all work through a single origin without manual proxy configuration:

Every HTTP request hits the fastware backend first.
If the backend returns 404 (no route matched), the request is proxied to the Vite dev server.
This means backend routes like /health or /metrics work without needing an API prefix.

The backend_prefixes parameter controls which additional paths are always routed to the backend for WebSocket connections (since WebSocket upgrades cannot be retried). By default, this includes ["/events"] for SSE endpoints.

python

from fastware import Router, create_app, AppConfig

router = Router()

# Vite dev server on port 5173
# Backend routes: /api/*, /events, /ws/*
app = create_app(
    router,
    config=AppConfig(
        vite_dev_port=5173,
        api_prefix="/api",
    ),
)

# Or with custom backend prefixes for WebSocket routing:
from fastware.middleware import ViteDevProxy

app = create_app(
    router,
    middleware=[
        lambda app: ViteDevProxy(
            app,
            vite_port=5173,
            api_prefix="/api",
            backend_prefixes=["/events", "/ws", "/notifications"],
        ),
    ],
)

For WebSocket connections, prefix-based routing is used: paths matching api_prefix or any backend_prefixes entry go to the backend; everything else is proxied to Vite (for HMR). HTTP requests use the try-backend-first approach regardless of path.

python

class TestViteDevProxyHTTP:

    @pytest.mark.anyio
    async def test_streams_response_chunks_without_buffering(self):
        """SSE-style responses stream through chunk by chunk.

        The inner app blocks after its first chunk until that chunk has
        been observed downstream.  A proxy that buffers the whole response
        never forwards the chunk, so this times out (red before the fix).
        """
        import asyncio

        from fastware.middleware import ViteDevProxy

        proceed = asyncio.Event()

        async def inner_app(scope, receive, send):
            await send({
                "type": "http.response.start",
                "status": 200,
                "headers": [(b"content-type", b"text/event-stream")],
            })
            await send({
                "type": "http.response.body",
                "body": b"data: 1\n\n",
                "more_body": True,
            })
            await proceed.wait()
            await send({
                "type": "http.response.body",
                "body": b"data: 2\n\n",
                "more_body": False,
            })

        proxy = ViteDevProxy(inner_app, vite_port=1)  # port never contacted
        sent: list[dict] = []

        async def receive():
            return {"type": "http.request", "body": b"", "more_body": False}

        async def send(message):
            sent.append(message)

        task = asyncio.ensure_future(proxy(_http_scope("/stream"), receive, send))
        try:

            async def first_chunk_forwarded():
                while not any(m["type"] == "http.response.body" for m in sent):
                    await asyncio.sleep(0.01)

            await asyncio.wait_for(first_chunk_forwarded(), timeout=2)
        finally:
            proceed.set()
        await asyncio.wait_for(task, timeout=2)

        assert sent[0]["type"] == "http.response.start"
        bodies = [m for m in sent if m["type"] == "http.response.body"]
        assert bodies[0]["body"] == b"data: 1\n\n"
        assert bodies[-1]["body"] == b"data: 2\n\n"

    @pytest.mark.anyio
    async def test_backend_prefix_404_not_proxied(self):
        """A 404 from a backend-prefixed path returns as-is, never hits Vite."""
        from fastware.middleware import ViteDevProxy

        async def inner_app(scope, receive, send):
            await send({"type": "http.response.start", "status": 404, "headers": []})
            await send({"type": "http.response.body", "body": b"nope"})

        proxy = ViteDevProxy(inner_app, vite_port=1, backend_prefixes=["/events"])
        sent: list[dict] = []

        async def receive():
            return {"type": "http.request", "body": b"", "more_body": False}

        async def send(message):
            sent.append(message)

        await proxy(_http_scope("/events/missing"), receive, send)
        assert sent[0]["type"] == "http.response.start"
        assert sent[0]["status"] == 404  # not a 502 from a Vite connect attempt
        assert sent[1]["body"] == b"nope"

    @pytest.mark.anyio
    async def test_unread_request_body_replayed_to_vite(self, monkeypatch):
        """A 404ing backend that never reads the body doesn't lose it for Vite."""
        from fastware.middleware import ViteDevProxy

        async def inner_app(scope, receive, send):
            # 404 without ever reading the request body.
            await send({"type": "http.response.start", "status": 404, "headers": []})
            await send({"type": "http.response.body", "body": b""})

        proxy = ViteDevProxy(inner_app, vite_port=1)
        captured: dict = {}

        async def fake_proxy_http(scope, send, *, body=b""):
            captured["body"] = body

        monkeypatch.setattr(proxy, "_proxy_http", fake_proxy_http)

        chunks = [
            {"type": "http.request", "body": b"hello ", "more_body": True},
            {"type": "http.request", "body": b"world", "more_body": False},
        ]

        async def receive():
            return chunks.pop(0)

        async def send(message):
            pass

        await proxy(_http_scope("/app-page", method="POST"), receive, send)
        assert captured["body"] == b"hello world"

    @pytest.mark.anyio
    async def test_404_swallowed_then_proxied(self, monkeypatch):
        """On a non-backend 404, nothing from the backend reaches the client."""
        from fastware.middleware import ViteDevProxy

        async def inner_app(scope, receive, send):
            await send({"type": "http.response.start", "status": 404, "headers": []})
            await send({"type": "http.response.body", "body": b"backend 404 page"})

        proxy = ViteDevProxy(inner_app, vite_port=1)
        proxied = {}

        async def fake_proxy_http(scope, send, *, body=b""):
            proxied["called"] = True

        monkeypatch.setattr(proxy, "_proxy_http", fake_proxy_http)
        sent: list[dict] = []

        async def receive():
            return {"type": "http.request", "body": b"", "more_body": False}

        async def send(message):
            sent.append(message)

        await proxy(_http_scope("/index.html"), receive, send)
        assert proxied.get("called")
        assert sent == []  # backend's 404 messages were discarded

python

class TestViteDevProxyWS:

    @staticmethod
    def _ws_fixtures():
        sent: list[dict] = []
        incoming = [{"type": "websocket.connect"}]

        async def receive():
            return incoming.pop(0)

        async def send(message):
            sent.append(message)

        scope = {
            "type": "websocket",
            "path": "/hmr",
            "query_string": b"",
            "headers": [],
        }
        return scope, receive, send, sent

    @pytest.mark.anyio
    async def test_missing_websockets_closes_with_error_code(self):
        """Without the websockets library, close with 1011 -- not a clean 1000."""
        from fastware.middleware import ViteDevProxy

        async def inner_app(scope, receive, send):
            raise AssertionError("inner app must not be called")

        proxy = ViteDevProxy(inner_app, vite_port=1)
        scope, receive, send, sent = self._ws_fixtures()

        with patch.dict(
            "sys.modules",
            {
                "websockets": None,
                "websockets.asyncio": None,
                "websockets.asyncio.client": None,
            },
        ):
            await proxy._proxy_ws(scope, receive, send)

        close = sent[-1]
        assert close["type"] == "websocket.close"
        assert close["code"] == 1011

    @pytest.mark.anyio
    async def test_vite_unreachable_closes_with_error_code(self):
        """A failed proxy connection closes with 1011, not a clean 1000."""
        pytest.importorskip("websockets")
        from fastware.middleware import ViteDevProxy

        async def inner_app(scope, receive, send):
            raise AssertionError("inner app must not be called")

        proxy = ViteDevProxy(inner_app, vite_port=1)  # nothing listens here
        scope, receive, send, sent = self._ws_fixtures()

        await proxy._proxy_ws(scope, receive, send)

        close = sent[-1]
        assert close["type"] == "websocket.close"
        assert close["code"] == 1011

#Middleware API Reference

#src.fastware.middleware

#RequestIDMiddleware

#RequestTimingMiddleware

#CORSMiddleware

#_cors_headers

#TrustedHostMiddleware

#ViteDevProxy

#__init__

#_is_api_request

#close

#_proxy_http

#_proxy_ws

#CORS Configuration for a Typical SPA

#ViteDevProxy Routing

#`src.fastware.middleware`

#`RequestIDMiddleware`

#`RequestTimingMiddleware`

#`CORSMiddleware`

#`_cors_headers`

#`TrustedHostMiddleware`

#`ViteDevProxy`

#`init`

#`_is_api_request`

#`close`

#`_proxy_http`

#`_proxy_ws`