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
46
47
48
49
50
|
import asyncio
from collections.abc import AsyncIterator
import pytest
from graphql import ExecutionResult
from pytest_codspeed.plugin import BenchmarkFixture
from .api import schema
@pytest.mark.benchmark
def test_subscription(benchmark: BenchmarkFixture):
s = """
subscription {
something
}
"""
async def _run():
for _ in range(100):
iterator = await schema.subscribe(s)
value = await iterator.__anext__() # type: ignore[union-attr]
assert value.data is not None
assert value.data["something"] == "Hello World!"
benchmark(lambda: asyncio.run(_run()))
@pytest.mark.benchmark
@pytest.mark.parametrize("count", [1000, 20000])
def test_subscription_long_run(benchmark: BenchmarkFixture, count: int) -> None:
s = """#graphql
subscription LongRunning($count: Int!) {
longRunning(count: $count)
}
"""
async def _run():
i = 0
aiterator: AsyncIterator[ExecutionResult] = await schema.subscribe(
s, variable_values={"count": count}
) # type: ignore[assignment]
async for res in aiterator:
assert res.data is not None
assert res.data["longRunning"] == i
i += 1
benchmark(lambda: asyncio.run(_run()))
|
