test: Concurrency integration tests for middleware pipeline safety

Verifies request isolation, auth short-circuit, POST body isolation,
and error routing under 50 concurrent requests with random delays.

Author: Rudolf Schmidt

tests/integration/concurrency/__init__.py

@@ -0,0 +1,60 @@
+"""Shared fixtures for concurrency integration tests.
+
+Provides the FastAPI test app built from the fixture directory at
+tests/integration/fixtures/concurrency_app/.  Each test gets a fresh
+copy of the routes in tmp_path for import isolation.
+
+App structure:
+    routes/_middleware.py                    # Root: stamps X-Request-ID, inits trace
+    routes/echo/route.py                    # Echo: root middleware only
+    routes/api/_middleware.py                # API dir: appends "api" to trace
+    routes/api/users/route.py               # Users: file-level middleware
+    routes/api/items/[item_id]/route.py     # Items: handler-level middleware via route()
+    routes/api/protected/_middleware.py      # Auth: short-circuits 401
+    routes/api/protected/route.py           # Protected: returns authenticated user
+    routes/api/messages/route.py            # Messages: POST with JSON body
+    routes/api/tasks/[task_id]/route.py     # Tasks: raises 404 for "missing-*" IDs
+"""
+
+import importlib.util
+import shutil
+from collections.abc import Callable
+from pathlib import Path
+
+import pytest
+from fastapi import FastAPI
+
+CONCURRENT_REQUESTS = 50
+
+FIXTURE_DIR = Path(__file__).parent / "fixtures" / "app"
+
+EXPECTED_TRACES = {
+    "echo": ["root"],
+    "users": ["root", "api", "users-file"],
+    "items": ["root", "api", "items-handler"],
+    "protected": ["root", "api", "auth"],
+    "messages": ["root", "api", "messages-file"],
+    "tasks": ["root", "api"],
+}
+
+
+def _load_create_app() -> Callable[..., FastAPI]:
+    """Import create_app from the fixture's main.py by file path."""
+    spec = importlib.util.spec_from_file_location(
+        "concurrency_app.main", FIXTURE_DIR / "main.py"
+    )
+    assert spec and spec.loader
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module.create_app  # type: ignore[no-any-return]
+
+
+_create_app = _load_create_app()
+
+
+@pytest.fixture
+def app(tmp_path: Path) -> FastAPI:
+    """Build a fresh FastAPI instance with routes copied into tmp_path."""
+    routes_dir = tmp_path / "routes"
+    shutil.copytree(FIXTURE_DIR / "routes", routes_dir)
+    return _create_app(routes_dir)

tests/integration/concurrency/conftest.py

@@ -0,0 +1,27 @@
+"""Concurrency test app — a realistic FastAPI app with file-based routing.
+
+Can be run standalone for manual testing:
+    uvicorn tests.integration.fixtures.concurrency_app.main:app --reload
+
+Routes exercise all three middleware layers with random delays (50ms–1s)
+to surface data leaks and request isolation issues under concurrent load.
+"""
+
+from pathlib import Path
+
+from fastapi import FastAPI
+
+from fastapi_filebased_routing import create_router_from_path
+
+ROUTES_DIR = Path(__file__).parent / "routes"
+
+
+def create_app(routes_dir: Path = ROUTES_DIR) -> FastAPI:
+    """Build a FastAPI instance wired to the given routes directory."""
+    application = FastAPI(title="Concurrency Test App")
+    router = create_router_from_path(routes_dir)
+    application.include_router(router)
+    return application
+
+
+app = create_app()

tests/integration/concurrency/fixtures/app/main.py

@@ -0,0 +1,11 @@
+"""Root middleware — stamps request ID and initializes middleware trace."""
+
+
+async def middleware(request, call_next):  # type: ignore[no-untyped-def]
+    request_id = request.headers.get("x-request-id", "missing")
+    request.state.request_id = request_id
+    request.state.middleware_trace = ["root"]
+    response = await call_next(request)
+    response.headers["X-Request-ID"] = request.state.request_id
+    response.headers["X-Middleware-Trace"] = ",".join(request.state.middleware_trace)
+    return response

tests/integration/concurrency/fixtures/app/routes/_middleware.py

@@ -0,0 +1,6 @@
+"""API directory middleware — appends 'api' to the middleware trace."""
+
+
+async def middleware(request, call_next):  # type: ignore[no-untyped-def]
+    request.state.middleware_trace.append("api")
+    return await call_next(request)

tests/integration/concurrency/fixtures/app/routes/api/_middleware.py

@@ -0,0 +1,26 @@
+"""Items route — handler-level middleware via route metaclass + random delay."""
+
+import asyncio
+import random
+
+from fastapi import Request
+
+from fastapi_filebased_routing.core.middleware import route
+
+
+async def _handler_middleware(request, call_next):  # type: ignore[no-untyped-def]
+    request.state.middleware_trace.append("items-handler")
+    return await call_next(request)
+
+
+class get(route):  # noqa: N801
+    middleware = [_handler_middleware]
+
+    async def handler(request: Request, item_id: str):  # type: ignore[no-untyped-def]  # noqa: N805
+        await asyncio.sleep(random.uniform(0.05, 1.0))
+        return {
+            "request_id": request.state.request_id,
+            "item_id": item_id,
+            "trace": request.state.middleware_trace,
+            "route": "items",
+        }

tests/integration/concurrency/fixtures/app/routes/api/items/[item_id]/route.py

@@ -0,0 +1,35 @@
+"""Messages route — accepts POST with JSON body, echoes it back.
+
+Tests request body isolation: under concurrency, Request A's body
+must never appear in Request B's response.
+"""
+
+import asyncio
+import random
+
+from fastapi import Request
+from pydantic import BaseModel
+
+
+class _MessageBody(BaseModel):
+    sender: str
+    content: str
+
+
+async def _file_middleware(request, call_next):  # type: ignore[no-untyped-def]
+    request.state.middleware_trace.append("messages-file")
+    return await call_next(request)
+
+
+middleware = [_file_middleware]
+
+
+async def post(request: Request, body: _MessageBody):  # type: ignore[no-untyped-def]
+    await asyncio.sleep(random.uniform(0.05, 1.0))
+    return {
+        "request_id": request.state.request_id,
+        "sender": body.sender,
+        "content": body.content,
+        "trace": request.state.middleware_trace,
+        "route": "messages",
+    }

tests/integration/concurrency/fixtures/app/routes/api/messages/route.py

@@ -0,0 +1,19 @@
+"""Auth middleware — short-circuits with 401 for missing/invalid tokens.
+
+This is the security-critical middleware: under concurrency, an authenticated
+response must NEVER leak to an unauthenticated caller, and vice versa.
+"""
+
+from starlette.responses import JSONResponse
+
+
+async def middleware(request, call_next):  # type: ignore[no-untyped-def]
+    token = request.headers.get("authorization", "")
+    if not token.startswith("Bearer "):
+        return JSONResponse(
+            {"error": "unauthorized", "request_id": request.state.request_id},
+            status_code=401,
+        )
+    request.state.user_id = token.removeprefix("Bearer ")
+    request.state.middleware_trace.append("auth")
+    return await call_next(request)

tests/integration/concurrency/fixtures/app/routes/api/protected/_middleware.py

@@ -0,0 +1,19 @@
+"""Protected route — returns the authenticated user's identity.
+
+Only reachable if the auth middleware in _middleware.py allows the request through.
+"""
+
+import asyncio
+import random
+
+from fastapi import Request
+
+
+async def get(request: Request):  # type: ignore[no-untyped-def]
+    await asyncio.sleep(random.uniform(0.05, 1.0))
+    return {
+        "request_id": request.state.request_id,
+        "user_id": request.state.user_id,
+        "trace": request.state.middleware_trace,
+        "route": "protected",
+    }

tests/integration/concurrency/fixtures/app/routes/api/protected/route.py

@@ -0,0 +1,28 @@
+"""Tasks route — raises HTTPException for specific task IDs.
+
+Tests error handling under concurrency: when handler A raises 404,
+handler B (running concurrently) must still get its correct 200 response.
+Task IDs starting with "missing-" trigger a 404 Not Found.
+"""
+
+import asyncio
+import random
+
+from fastapi import HTTPException, Request
+
+
+async def get(request: Request, task_id: str):  # type: ignore[no-untyped-def]
+    await asyncio.sleep(random.uniform(0.05, 1.0))
+
+    if task_id.startswith("missing-"):
+        raise HTTPException(
+            status_code=404,
+            detail=f"Task {task_id} not found",
+        )
+
+    return {
+        "request_id": request.state.request_id,
+        "task_id": task_id,
+        "trace": request.state.middleware_trace,
+        "route": "tasks",
+    }