Add test for server not supporting lifespan state but app requiring it

starlette/routing.py

@@ -679,15 +679,22 @@ async def lifespan(self, scope: Scope, receive: Receive, send: Send) -> None:
         """
         started = False
         app = scope.get("app")
-        state = scope.get("state")
         await receive()
+        lifespan_needs_state = (
+            len(inspect.signature(self.lifespan_context).parameters) == 2
+        )
+        server_supports_state = "state" in scope
+        if lifespan_needs_state and not server_supports_state:
+            raise RuntimeError(
+                'This server does not support "state" in the lifespan scope.'
+                " Please try updating your ASGI server."
+            )
         try:
             lifespan_context: Lifespan
-            if (
-                len(inspect.signature(self.lifespan_context).parameters) == 2
-                and state is not None
-            ):
-                lifespan_context = functools.partial(self.lifespan_context, state=state)
+            if lifespan_needs_state:
+                lifespan_context = functools.partial(
+                    self.lifespan_context, state=scope["state"]
+                )
             else:
                 lifespan_context = typing.cast(StatelessLifespan, self.lifespan_context)
             async with lifespan_context(app):

starlette/types.py

@@ -8,8 +8,8 @@
 
 ASGIApp = typing.Callable[[Scope, Receive, Send], typing.Awaitable[None]]
 
-StatelessLifespan = typing.Callable[[object], typing.AsyncContextManager]
+StatelessLifespan = typing.Callable[[object], typing.AsyncContextManager[typing.Any]]
 StateLifespan = typing.Callable[
-    [typing.Any, typing.Dict[str, typing.Any]], typing.AsyncContextManager
+    [typing.Any, typing.Dict[str, typing.Any]], typing.AsyncContextManager[typing.Any]
 ]
 Lifespan = typing.Union[StatelessLifespan, StateLifespan]

tests/test_routing.py

@@ -717,6 +717,27 @@ async def run_shutdown(state):
     assert shutdown_complete
 
 
+def test_lifespan_state_unsupported(test_client_factory):
+    @contextlib.asynccontextmanager
+    async def lifespan(app, scope):
+        yield None  # pragma: no cover
+
+    app = Router(
+        lifespan=lifespan,
+        routes=[Mount("/", PlainTextResponse("hello, world"))],
+    )
+
+    async def no_state_wrapper(scope, receive, send):
+        del scope["state"]
+        await app(scope, receive, send)
+
+    with pytest.raises(
+        RuntimeError, match='This server does not support "state" in the lifespan scope'
+    ):
+        with test_client_factory(no_state_wrapper):
+            raise AssertionError("Should not be called")  # pragma: no cover
+
+
 def test_lifespan_async_cm(test_client_factory):
     startup_complete = False
     shutdown_complete = False