from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.responses import JSONResponse
+from starlette.routing import Route
from starlette.websockets import WebSocket, WebSocketDisconnect
if sys.version_info >= (3, 7): # pragma: no cover
asyncio_current_task = asyncio.Task.current_task
from contextlib2 import asynccontextmanager
-mock_service = Starlette()
-
-@mock_service.route("/")
def mock_service_endpoint(request):
return JSONResponse({"mock": "example"})
+mock_service = Starlette(
+ routes=[
+ Route("/", endpoint=mock_service_endpoint),
+ ]
+)
+
+
def current_task():
# anyio's TaskInfo comparisons are invalid after their associated native
# task object is GC'd https://github.com/agronholm/anyio/issues/324
raise RuntimeError(f"unsupported asynclib={asynclib_name}") # pragma: no cover
-startup_error_app = Starlette()
-
-
-@startup_error_app.on_event("startup")
def startup():
raise RuntimeError()
+startup_error_app = Starlette(on_startup=[startup])
+
+
def test_use_testclient_in_endpoint(test_client_factory):
"""
We should be able to use the test client within applications.
during tests or in development.
"""
- app = Starlette()
-
- @app.route("/")
def homepage(request):
client = test_client_factory(mock_service)
response = client.get("/")
return JSONResponse(response.json())
+ app = Starlette(routes=[Route("/", endpoint=homepage)])
+
client = test_client_factory(app)
response = client.get("/")
assert response.json() == {"mock": "example"}
shutdown_task = current_task()
shutdown_loop = get_identity()
- app = Starlette(lifespan=lifespan_context)
-
- @app.route("/loop_id")
async def loop_id(request):
return JSONResponse(get_identity())
+ app = Starlette(
+ lifespan=lifespan_context,
+ routes=[Route("/loop_id", endpoint=loop_id)],
+ )
+
client = test_client_factory(app)
with client: