-
Notifications
You must be signed in to change notification settings - Fork 0
/
tests.py
121 lines (96 loc) · 3.36 KB
/
tests.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
119
120
121
from functools import partial
from importlib import metadata
from unittest import mock
import muffin
import pytest
from asgi_tools import Request
@pytest.fixture()
def app():
return muffin.Application(SENTRY_DSN="http://public:[email protected]/1")
@pytest.fixture()
def sentry(app):
import muffin_sentry
version = metadata.version("muffin-sentry")
return muffin_sentry.Plugin(
app,
transaction_style="endpoint",
sdk_options={"environment": "tests", "release": version},
)
def test_request_processor(sentry, client):
request = Request(client.build_scope("/", type="http", method="POST"), lambda: None, lambda: None) # type: ignore[]
processor = partial(sentry.process_data, request=request)
assert processor
event = processor({}, {})
assert event
assert event == {
"request": {
"method": "POST",
"query_string": "",
"url": "http://localhost/",
"headers": {
"host": "localhost",
"user-agent": "ASGI-Tools-Test-Client",
},
"env": {
"REMOTE_ADDR": "127.0.0.1",
},
},
}
async def test_muffin_sentry(app, client, sentry):
import sentry_sdk
hub = sentry_sdk.Hub.current
assert hub.client.options["release"]
assert hub.client.options["environment"] == "tests"
@app.route("/success")
async def success(request):
return "OK"
res = await client.get("/success")
assert res.status_code == 200
@app.route("/error")
async def error(request):
raise Exception("Unhandled exception")
@app.middleware
async def md(handler, request, receive, send):
"""Test middleware errors."""
if "md_error" in request.query:
raise Exception("from middleware")
return await handler(request, receive, send)
@sentry.processor
def user(event, hint, request):
scope = request.scope
if "user" in scope:
event["user"] = scope["user"]
return event
await app.lifespan.run("startup")
with mock.patch("sentry_sdk.transport.HttpTransport.capture_envelope") as mocked:
res = await client.get("/error")
assert res.status_code == 500
assert mocked.called
env, *_ = mocked.call_args.args[0]
event = env.get_event()
assert event["transaction"] == "/error"
assert event["request"]
assert event["request"]["url"] == "http://localhost/error"
@app.route("/scope")
async def scope(request):
request["user"] = {"id": "1", "email": "[email protected]"}
scope = sentry.current_scope.get()
scope.set_tag("tests", "passed")
app.plugins["sentry"].capture_message("tests")
return "OK"
res = await client.get("/scope")
assert res.status_code == 200
assert mocked.called
env, *_ = mocked.call_args.args[0]
event = env.get_event()
assert event["user"]
assert event["tags"]
assert event["request"]
mocked.reset_mock()
res = await client.get("/success?md_error=1")
assert res.status_code == 500
assert mocked.called
env, *_ = mocked.call_args.args[0]
event = env.get_event()
assert event["exception"]["values"][0]["mechanism"]
# ruff: noqa: TRY002, ARG001