-
Notifications
You must be signed in to change notification settings - Fork 464
/
test_asyncio.py
118 lines (90 loc) · 3.17 KB
/
test_asyncio.py
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
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
import asyncio
import sys
import pytest
import pytest_asyncio
import sentry_sdk
from sentry_sdk.consts import OP
from sentry_sdk.integrations.asyncio import AsyncioIntegration
minimum_python_36 = pytest.mark.skipif(
sys.version_info < (3, 6), reason="ASGI is only supported in Python >= 3.6"
)
async def foo():
await asyncio.sleep(0.01)
async def bar():
await asyncio.sleep(0.01)
@pytest_asyncio.fixture(scope="session")
def event_loop(request):
"""Create an instance of the default event loop for each test case."""
loop = asyncio.get_event_loop_policy().new_event_loop()
yield loop
loop.close()
@minimum_python_36
@pytest.mark.asyncio
async def test_create_task(
sentry_init,
capture_events,
event_loop,
):
sentry_init(
traces_sample_rate=1.0,
send_default_pii=True,
debug=True,
integrations=[
AsyncioIntegration(),
],
)
events = capture_events()
with sentry_sdk.start_transaction(name="test_transaction_for_create_task"):
with sentry_sdk.start_span(op="root", description="not so important"):
tasks = [event_loop.create_task(foo()), event_loop.create_task(bar())]
await asyncio.wait(tasks, return_when=asyncio.FIRST_EXCEPTION)
sentry_sdk.flush()
(transaction_event,) = events
assert transaction_event["spans"][0]["op"] == "root"
assert transaction_event["spans"][0]["description"] == "not so important"
assert transaction_event["spans"][1]["op"] == OP.FUNCTION
assert transaction_event["spans"][1]["description"] == "foo"
assert (
transaction_event["spans"][1]["parent_span_id"]
== transaction_event["spans"][0]["span_id"]
)
assert transaction_event["spans"][2]["op"] == OP.FUNCTION
assert transaction_event["spans"][2]["description"] == "bar"
assert (
transaction_event["spans"][2]["parent_span_id"]
== transaction_event["spans"][0]["span_id"]
)
@minimum_python_36
@pytest.mark.asyncio
async def test_gather(
sentry_init,
capture_events,
):
sentry_init(
traces_sample_rate=1.0,
send_default_pii=True,
debug=True,
integrations=[
AsyncioIntegration(),
],
)
events = capture_events()
with sentry_sdk.start_transaction(name="test_transaction_for_gather"):
with sentry_sdk.start_span(op="root", description="not so important"):
await asyncio.gather(foo(), bar(), return_exceptions=True)
sentry_sdk.flush()
(transaction_event,) = events
assert transaction_event["spans"][0]["op"] == "root"
assert transaction_event["spans"][0]["description"] == "not so important"
assert transaction_event["spans"][1]["op"] == OP.FUNCTION
assert transaction_event["spans"][1]["description"] == "foo"
assert (
transaction_event["spans"][1]["parent_span_id"]
== transaction_event["spans"][0]["span_id"]
)
assert transaction_event["spans"][2]["op"] == OP.FUNCTION
assert transaction_event["spans"][2]["description"] == "bar"
assert (
transaction_event["spans"][2]["parent_span_id"]
== transaction_event["spans"][0]["span_id"]
)