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
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
|
import signal
import textwrap
from test.fake_time_util import fake_time
from threading import Thread
from time import sleep
import pytest
# note: IPython should be imported within each test. Importing it in our tests
# seems to cause problems with subsequent tests.
cell_code = """
import time
def function_a():
function_b()
function_c()
def function_b():
function_d()
def function_c():
function_d()
def function_d():
function_e()
def function_e():
time.sleep(0.1)
function_a()
"""
# Tests #
@pytest.mark.ipythonmagic
def test_magics(ip):
from IPython.utils.capture import capture_output as capture_ipython_output
with fake_time():
with capture_ipython_output() as captured:
ip.run_cell_magic("pyinstrument", line="", cell=cell_code)
assert len(captured.outputs) == 1
output = captured.outputs[0]
assert "text/html" in output.data
assert "text/plain" in output.data
assert "function_a" in output.data["text/html"]
assert "<iframe" in output.data["text/html"]
assert "function_a" in output.data["text/plain"]
assert "- 0.200 function_a" in output.data["text/plain"]
assert "- 0.100 FakeClock.sleep" in output.data["text/plain"]
with fake_time():
with capture_ipython_output() as captured:
# this works because function_a was defined in the previous cell
ip.run_line_magic("pyinstrument", line="function_a()")
assert len(captured.outputs) == 1
output = captured.outputs[0]
assert "function_a" in output.data["text/plain"]
assert "- 0.100 FakeClock.sleep" in output.data["text/plain"]
@pytest.mark.ipythonmagic
def test_magic_empty_line(ip):
# check empty line input
ip.run_line_magic("pyinstrument", line="")
@pytest.mark.ipythonmagic
def test_magic_no_variable_expansion(ip, capsys):
ip.run_line_magic("pyinstrument", line="print(\"hello {len('world')}\")")
captured = capsys.readouterr()
assert "hello {len('world')}" in captured.out
assert "hello 5" not in captured.out
@pytest.mark.ipythonmagic
def test_pyinstrument_handles_interrupt_silently(ip, capsys):
from pyinstrument.magic.magic import InterruptSilently
thread = Thread(target=_interrupt_after_1s)
thread.start()
# expect our custom exception to bubble up
with pytest.raises(InterruptSilently):
ip.run_cell_magic("pyinstrument", "", "from time import sleep; sleep(2)")
thread.join()
# nothing should have hit stderr
_, err = capsys.readouterr()
assert err.strip() == ""
@pytest.mark.ipythonmagic
def test_async_cell_with_pyinstrument(ip, capsys):
ip.run_cell_magic(
"pyinstrument",
line="--async_mode=enabled",
cell=textwrap.dedent(
"""
import asyncio
async def function_a():
await asyncio.sleep(0.1)
return 42
a = await function_a()
print("a:", a)
"""
),
)
stdout, stderr = capsys.readouterr()
assert "a: 42" in stdout
# Utils #
@pytest.fixture(scope="module")
def session_ip():
from IPython.testing.globalipapp import start_ipython
yield start_ipython()
def _interrupt_after_1s():
sleep(1)
signal.raise_signal(signal.SIGINT)
@pytest.fixture(scope="function")
def ip(session_ip):
session_ip.run_line_magic(magic_name="load_ext", line="pyinstrument")
yield session_ip
session_ip.run_line_magic(magic_name="reset", line="-f")
|
