(test_client_factory: TestClientFactory)
| 7 | |
| 8 | |
| 9 | def test_async_task(test_client_factory: TestClientFactory) -> None: |
| 10 | TASK_COMPLETE = False |
| 11 | |
| 12 | async def async_task() -> None: |
| 13 | nonlocal TASK_COMPLETE |
| 14 | TASK_COMPLETE = True |
| 15 | |
| 16 | task = BackgroundTask(async_task) |
| 17 | |
| 18 | async def app(scope: Scope, receive: Receive, send: Send) -> None: |
| 19 | response = Response("task initiated", media_type="text/plain", background=task) |
| 20 | await response(scope, receive, send) |
| 21 | |
| 22 | client = test_client_factory(app) |
| 23 | response = client.get("/") |
| 24 | assert response.text == "task initiated" |
| 25 | assert TASK_COMPLETE |
| 26 | |
| 27 | |
| 28 | def test_sync_task(test_client_factory: TestClientFactory) -> None: |
nothing calls this directly
no test coverage detected