forked from jupyterlab/jupyterlab
/
browser_check.py
245 lines (204 loc) · 7.57 KB
/
browser_check.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
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
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
# -*- coding: utf-8 -*-
"""
This module is meant to run JupyterLab in a headless browser, making sure
the application launches and starts up without errors.
"""
import asyncio
import inspect
import logging
import os
import shutil
import subprocess
import sys
import time
from concurrent.futures import ThreadPoolExecutor
from os import path as osp
from jupyter_server.serverapp import aliases, flags
from jupyter_server.utils import pathname2url, urljoin
from tornado.ioloop import IOLoop
from tornado.iostream import StreamClosedError
from tornado.websocket import WebSocketClosedError
from traitlets import Bool
from .labapp import LabApp, get_app_dir
from .tests.test_app import TestEnv
here = osp.abspath(osp.dirname(__file__))
test_flags = dict(flags)
test_flags['core-mode'] = (
{'BrowserApp': {'core_mode': True}},
"Start the app in core mode."
)
test_flags['dev-mode'] = (
{'BrowserApp': {'dev_mode': True}},
"Start the app in dev mode."
)
test_flags['watch'] = (
{'BrowserApp': {'watch': True}},
"Start the app in watch mode."
)
test_aliases = dict(aliases)
test_aliases['app-dir'] = 'BrowserApp.app_dir'
class LogErrorHandler(logging.Handler):
"""A handler that exits with 1 on a logged error."""
def __init__(self):
super().__init__(level=logging.ERROR)
self.errored = False
def filter(self, record):
# Handle known StreamClosedError from Tornado
# These occur when we forcibly close Websockets or
# browser connections during the test.
# https://github.com/tornadoweb/tornado/issues/2834
if hasattr(record, 'exc_info') and not record.exc_info is None and isinstance(record.exc_info[1], (StreamClosedError, WebSocketClosedError)):
return
return super().filter(record)
def emit(self, record):
print(record.msg, file=sys.stderr)
self.errored = True
def run_test(app, func):
"""Synchronous entry point to run a test function.
func is a function that accepts an app url as a parameter and returns a result.
func can be synchronous or asynchronous. If it is synchronous, it will be run
in a thread, so asynchronous is preferred.
"""
IOLoop.current().spawn_callback(run_test_async, app, func)
async def run_test_async(app, func):
"""Run a test against the application.
func is a function that accepts an app url as a parameter and returns a result.
func can be synchronous or asynchronous. If it is synchronous, it will be run
in a thread, so asynchronous is preferred.
"""
handler = LogErrorHandler()
app.log.addHandler(handler)
env_patch = TestEnv()
env_patch.start()
app.log.info('Running async test')
# The entry URL for browser tests is different in notebook >= 6.0,
# since that uses a local HTML file to point the user at the app.
if hasattr(app, 'browser_open_file'):
url = urljoin('file:', pathname2url(app.browser_open_file))
else:
url = app.display_url
# Allow a synchronous function to be passed in.
if inspect.iscoroutinefunction(func):
test = func(url)
else:
app.log.info('Using thread pool executor to run test')
loop = asyncio.get_event_loop()
executor = ThreadPoolExecutor()
task = loop.run_in_executor(executor, func, url)
test = asyncio.wait([task])
try:
await test
except Exception as e:
app.log.critical("Caught exception during the test:")
app.log.error(str(e))
app.log.info("Test Complete")
result = 0
if handler.errored:
result = 1
app.log.critical('Exiting with 1 due to errors')
else:
app.log.info('Exiting normally')
app.log.info('Stopping server...')
try:
app.http_server.stop()
app.io_loop.stop()
env_patch.stop()
except Exception as e:
app.log.error(str(e))
result = 1
finally:
time.sleep(2)
os._exit(result)
async def run_async_process(cmd, **kwargs):
"""Run an asynchronous command"""
proc = await asyncio.create_subprocess_exec(
*cmd,
**kwargs)
stdout, stderr = await proc.communicate()
if proc.returncode != 0:
raise RuntimeError(str(cmd) + ' exited with ' + str(proc.returncode))
return stdout, stderr
async def run_browser(url):
"""Run the browser test and return an exit code.
"""
target = osp.join(get_app_dir(), 'browser_test')
if not osp.exists(osp.join(target, 'node_modules')):
if not osp.exists(target):
os.makedirs(osp.join(target))
await run_async_process(["jlpm", "init", "-y"], cwd=target)
await run_async_process(["jlpm", "add", "playwright@^1.9.2"], cwd=target)
shutil.copy(osp.join(here, 'browser-test.js'), osp.join(target, 'browser-test.js'))
await run_async_process(["node", "browser-test.js", url], cwd=target)
def run_browser_sync(url):
"""Run the browser test and return an exit code.
"""
target = osp.join(get_app_dir(), 'browser_test')
if not osp.exists(osp.join(target, 'node_modules')):
os.makedirs(target)
subprocess.call(["jlpm", "init", "-y"], cwd=target)
subprocess.call(["jlpm", "add", "playwright@^1.9.2"], cwd=target)
shutil.copy(osp.join(here, 'browser-test.js'), osp.join(target, 'browser-test.js'))
return subprocess.check_call(["node", "browser-test.js", url], cwd=target)
class BrowserApp(LabApp):
"""An app the launches JupyterLab and waits for it to start up, checking for
JS console errors, JS errors, and Python logged errors.
"""
name = __name__
open_browser = False
serverapp_config = {
"base_url": "/foo/"
}
default_url = "/lab?reset"
ip = '127.0.0.1'
flags = test_flags
aliases = test_aliases
test_browser = Bool(True)
def initialize_settings(self):
self.settings.setdefault('page_config_data', dict())
self.settings['page_config_data']['browserTest'] = True
self.settings['page_config_data']['buildAvailable'] = False
self.settings['page_config_data']['exposeAppInBrowser'] = True
super().initialize_settings()
def initialize_handlers(self):
func = run_browser if self.test_browser else lambda url: 0
if os.name == 'nt' and func == run_browser:
func = run_browser_sync
run_test(self.serverapp, func)
super().initialize_handlers()
def _jupyter_server_extension_points():
return [
{
'module': __name__,
'app': BrowserApp
}
]
# TODO: remove handling of --notebook arg and the following two
# functions in JupyterLab 4.0
def load_jupyter_server_extension(serverapp):
extension = BrowserApp()
extension.serverapp = serverapp
extension.load_config_file()
extension.update_config(serverapp.config)
extension.parse_command_line(serverapp.extra_args)
extension.initialize()
def _jupyter_server_extension_paths():
return [
{
'module': 'jupyterlab.browser_check'
}
]
if __name__ == '__main__':
skip_options = ["--no-browser-test", "--no-chrome-test"]
for option in skip_options:
if option in sys.argv:
BrowserApp.test_browser = False
sys.argv.remove(option)
if "--notebook" in sys.argv:
from notebook.notebookapp import NotebookApp
NotebookApp.default_url = "/lab"
sys.argv.remove("--notebook")
NotebookApp.nbserver_extensions = {"jupyterlab.browser_check": True}
NotebookApp.open_browser = False
NotebookApp.launch_instance()
else:
BrowserApp.launch_instance()