| 8 | |
| 9 | |
| 10 | def test_lifespan_on(): |
| 11 | startup_complete = False |
| 12 | shutdown_complete = False |
| 13 | |
| 14 | async def app(scope, receive, send): |
| 15 | nonlocal startup_complete, shutdown_complete |
| 16 | message = await receive() |
| 17 | assert message["type"] == "lifespan.startup" |
| 18 | startup_complete = True |
| 19 | await send({"type": "lifespan.startup.complete"}) |
| 20 | message = await receive() |
| 21 | assert message["type"] == "lifespan.shutdown" |
| 22 | shutdown_complete = True |
| 23 | await send({"type": "lifespan.shutdown.complete"}) |
| 24 | |
| 25 | async def test(): |
| 26 | config = Config(app=app, lifespan="on") |
| 27 | lifespan = LifespanOn(config) |
| 28 | |
| 29 | assert not startup_complete |
| 30 | assert not shutdown_complete |
| 31 | await lifespan.startup() |
| 32 | assert startup_complete |
| 33 | assert not shutdown_complete |
| 34 | await lifespan.shutdown() |
| 35 | assert startup_complete |
| 36 | assert shutdown_complete |
| 37 | |
| 38 | loop = asyncio.new_event_loop() |
| 39 | loop.run_until_complete(test()) |
| 40 | loop.close() |
| 41 | |
| 42 | |
| 43 | def test_lifespan_off(): |