1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
|
import threading
from typing import Generator
import pytest
from quart import request, ResponseReturnValue
from quart_trio import QuartTrio
@pytest.fixture(name="app")
def _app() -> QuartTrio:
app = QuartTrio(__name__)
@app.route("/", methods=["GET", "POST"])
def index() -> ResponseReturnValue:
return request.method
@app.route("/gen")
def gen() -> ResponseReturnValue:
def _gen() -> Generator[bytes, None, None]:
yield b"%d" % threading.current_thread().ident
for _ in range(2):
yield b"b"
return _gen(), 200
return app
@pytest.mark.trio
async def test_sync_request_context(app: QuartTrio) -> None:
test_client = app.test_client()
response = await test_client.get("/")
assert b"GET" in (await response.get_data())
response = await test_client.post("/")
assert b"POST" in (await response.get_data())
@pytest.mark.trio
async def test_sync_generator(app: QuartTrio) -> None:
test_client = app.test_client()
response = await test_client.get("/gen")
result = await response.get_data()
assert result[-2:] == b"bb"
assert int(result[:-2]) != threading.current_thread().ident
|
