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
|
import pytest
from .fixtures import make_app_client
@pytest.mark.parametrize("trace_debug", (True, False))
def test_trace(trace_debug):
with make_app_client(settings={"trace_debug": trace_debug}) as client:
response = client.get("/fixtures/simple_primary_key.json?_trace=1")
assert response.status == 200
data = response.json
if not trace_debug:
assert "_trace" not in data
return
assert "_trace" in data
trace_info = data["_trace"]
assert isinstance(trace_info["request_duration_ms"], float)
assert isinstance(trace_info["sum_trace_duration_ms"], float)
assert isinstance(trace_info["num_traces"], int)
assert isinstance(trace_info["traces"], list)
traces = trace_info["traces"]
assert len(traces) == trace_info["num_traces"]
for trace in traces:
assert isinstance(trace["type"], str)
assert isinstance(trace["start"], float)
assert isinstance(trace["end"], float)
assert trace["duration_ms"] == (trace["end"] - trace["start"]) * 1000
assert isinstance(trace["traceback"], list)
assert isinstance(trace["database"], str)
assert isinstance(trace["sql"], str)
assert isinstance(trace.get("params"), (list, dict, None.__class__))
sqls = [trace["sql"] for trace in traces if "sql" in trace]
# There should be a mix of different types of SQL statement
expected = (
"CREATE TABLE ",
"PRAGMA ",
"INSERT OR REPLACE INTO ",
"INSERT INTO",
"select ",
)
for prefix in expected:
assert any(
sql.startswith(prefix) for sql in sqls
), "No trace beginning with: {}".format(prefix)
# Should be at least one executescript
assert any(trace for trace in traces if trace.get("executescript"))
# And at least one executemany
execute_manys = [trace for trace in traces if trace.get("executemany")]
assert execute_manys
assert all(isinstance(trace["count"], int) for trace in execute_manys)
def test_trace_parallel_queries():
with make_app_client(settings={"trace_debug": True}) as client:
response = client.get("/parallel-queries?_trace=1")
assert response.status == 200
data = response.json
assert data["one"] == 1
assert data["two"] == 2
trace_info = data["_trace"]
traces = [trace for trace in trace_info["traces"] if "sql" in trace]
one, two = traces
# "two" should have started before "one" ended
assert two["start"] < one["end"]
|
