-
Notifications
You must be signed in to change notification settings - Fork 5
Expand file tree
/
Copy pathipynb_test.py
More file actions
159 lines (131 loc) · 4.57 KB
/
ipynb_test.py
File metadata and controls
159 lines (131 loc) · 4.57 KB
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
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
from __future__ import annotations
import contextlib
import inspect
import os
import subprocess
import sys
import tempfile
import warnings
from pathlib import Path
import nbformat
import pytest
from htmlbook.book_name import get_project_name
from nbconvert.exporters import PythonExporter
_REPO_ROOT = Path(__file__).resolve().parents[2]
def _prepare_nbconvert_template() -> None:
temp_jupyter_dir = Path("/tmp/jupyter_templates")
template_dir = temp_jupyter_dir / "nbconvert" / "templates" / "python"
template_dir.mkdir(parents=True, exist_ok=True)
template_file = template_dir / "index.py.j2"
if not template_file.exists():
template_file.write_text(
"""# coding: utf-8
{%- for cell in nb.cells -%}
{%- if cell.cell_type == 'code' -%}
{% for line in cell.source.splitlines() %}
{{ line }}
{% endfor %}
{% if not loop.last %}
{% endif %}
{%- endif -%}
{%- endfor -%}
""",
encoding="utf-8",
)
os.environ.setdefault("JUPYTER_DATA_DIR", str(temp_jupyter_dir))
os.environ.setdefault("JUPYTER_CONFIG_DIR", "/tmp")
def _startup_prelude() -> str:
startup = _REPO_ROOT / "book/htmlbook/tools/jupyter/startup.py"
startup_code = startup.read_text(encoding="utf-8")
project_module = get_project_name().replace("-", "_")
startup_code = startup_code.replace(
'if "/usr/lib/python" in sys.modules["mpl_toolkits"].__file__:',
'mpl_toolkits_file = getattr(sys.modules["mpl_toolkits"], "__file__", "") or ""\n'
' if "/usr/lib/python" in mpl_toolkits_file:',
)
return (
startup_code
+ "\n\n"
+ "try:\n"
+ f" from {project_module}.utils import _set_running_as_test\n"
+ " _set_running_as_test(True)\n"
+ "except ModuleNotFoundError:\n"
+ " pass\n\n"
)
@contextlib.contextmanager
def _chdir(path: Path):
old = Path.cwd()
os.chdir(path)
try:
yield
finally:
os.chdir(old)
def _notebook_source(notebook_path: Path) -> str:
warnings.filterwarnings("ignore", category=SyntaxWarning)
with contextlib.suppress(Exception):
from pydrake.common.deprecation import DrakeDeprecationWarning
warnings.simplefilter("error", DrakeDeprecationWarning)
_prepare_nbconvert_template()
notebook = nbformat.read(notebook_path, as_version=4)
exporter = PythonExporter()
source, _ = exporter.from_notebook_node(notebook)
return _startup_prelude() + source
def _run_notebook(notebook_path: Path) -> None:
source = _notebook_source(notebook_path)
with tempfile.NamedTemporaryFile(
mode="w",
suffix=".py",
prefix=f"ipynb_{notebook_path.stem}_",
delete=False,
encoding="utf-8",
) as f:
f.write(source)
source_path = Path(f.name)
env = os.environ.copy()
env["PYTHONFAULTHANDLER"] = "1"
env.setdefault("OMP_NUM_THREADS", "1")
env.setdefault("OPENBLAS_NUM_THREADS", "1")
env.setdefault("MKL_NUM_THREADS", "1")
env.setdefault("NUMEXPR_NUM_THREADS", "1")
pythonpath = [str(_REPO_ROOT), str(_REPO_ROOT / "book"), str(notebook_path.parent)]
if env.get("PYTHONPATH"):
pythonpath.append(env["PYTHONPATH"])
env["PYTHONPATH"] = os.pathsep.join(pythonpath)
cmd = [sys.executable, str(source_path)]
workdir_cm = tempfile.TemporaryDirectory(prefix=f"ipynb_cwd_{notebook_path.stem}_")
try:
with workdir_cm as run_cwd:
result = subprocess.run(
cmd,
cwd=run_cwd,
env=env,
capture_output=True,
text=True,
)
finally:
source_path.unlink(missing_ok=True)
if result.returncode != 0:
details = [
f"Notebook execution failed: {notebook_path}",
f"Exit code: {result.returncode}",
]
if result.stdout:
details.append(f"STDOUT:\n{result.stdout}")
if result.stderr:
details.append(f"STDERR:\n{result.stderr}")
pytest.fail("\n\n".join(details))
def ipynb_test(path: str, *, name: str | None = None) -> None:
frame = inspect.currentframe()
assert frame is not None
caller = frame.f_back
assert caller is not None
module_globals = caller.f_globals
caller_file = Path(module_globals["__file__"]).resolve()
notebook_path = (caller_file.parent / path).resolve()
test_name = name or f"test_{notebook_path.stem}"
@pytest.mark.notebook
def _test() -> None:
_run_notebook(notebook_path)
_test.__name__ = test_name
_test.__qualname__ = test_name
module_globals[test_name] = _test