launchnotebook.py
96 lines
| 3.0 KiB
| text/x-python
|
PythonLexer
Brian E. Granger
|
r13038 | """Base class for notebook tests.""" | |
Brian E. Granger
|
r15080 | from __future__ import print_function | |
Brian E. Granger
|
r13038 | import sys | |
import time | |||
Zachary Sailer
|
r13045 | import requests | |
Thomas Kluyver
|
r13099 | from contextlib import contextmanager | |
Thomas Kluyver
|
r13833 | from subprocess import Popen, STDOUT | |
Brian E. Granger
|
r13038 | from unittest import TestCase | |
Thomas Kluyver
|
r13833 | import nose | |
Brian E. Granger
|
r13038 | from IPython.utils.tempdir import TemporaryDirectory | |
Paul Ivanov
|
r14071 | MAX_WAITTIME = 30 # seconds to wait for notebook server to start | |
POLL_INTERVAL = 0.1 # time between attempts | |||
Brian E. Granger
|
r13038 | class NotebookTestBase(TestCase): | |
"""A base class for tests that need a running notebook. | |||
This creates an empty profile in a temp ipython_dir | |||
and then starts the notebook server with a separate temp notebook_dir. | |||
""" | |||
Zachary Sailer
|
r13046 | port = 12341 | |
Brian E. Granger
|
r13038 | ||
MinRK
|
r13049 | @classmethod | |
def wait_until_alive(cls): | |||
"""Wait for the server to be alive""" | |||
url = 'http://localhost:%i/api/notebooks' % cls.port | |||
Paul Ivanov
|
r14071 | for _ in range(int(MAX_WAITTIME/POLL_INTERVAL)): | |
Zachary Sailer
|
r13045 | try: | |
MinRK
|
r13049 | requests.get(url) | |
Zachary Sailer
|
r13045 | except requests.exceptions.ConnectionError: | |
Thomas Kluyver
|
r14067 | if cls.notebook.poll() is not None: | |
raise RuntimeError("The notebook server exited with status %s" \ | |||
% cls.notebook.poll()) | |||
Paul Ivanov
|
r14071 | time.sleep(POLL_INTERVAL) | |
MinRK
|
r13049 | else: | |
Thomas Kluyver
|
r14067 | return | |
raise TimeoutError("The notebook server didn't start up correctly.") | |||
MinRK
|
r13049 | ||
@classmethod | |||
def wait_until_dead(cls): | |||
Thomas Kluyver
|
r14068 | """Wait for the server process to terminate after shutdown""" | |
Paul Ivanov
|
r14071 | for _ in range(int(MAX_WAITTIME/POLL_INTERVAL)): | |
Thomas Kluyver
|
r14068 | if cls.notebook.poll() is not None: | |
return | |||
Paul Ivanov
|
r14071 | time.sleep(POLL_INTERVAL) | |
MinRK
|
r13049 | ||
Thomas Kluyver
|
r14067 | raise TimeoutError("Undead notebook server") | |
MinRK
|
r13049 | @classmethod | |
def setup_class(cls): | |||
cls.ipython_dir = TemporaryDirectory() | |||
cls.notebook_dir = TemporaryDirectory() | |||
Brian E. Granger
|
r13038 | notebook_args = [ | |
sys.executable, '-c', | |||
'from IPython.html.notebookapp import launch_new_instance; launch_new_instance()', | |||
MinRK
|
r13049 | '--port=%d' % cls.port, | |
Paul Ivanov
|
r14071 | '--port-retries=0', # Don't try any other ports | |
Brian E. Granger
|
r13038 | '--no-browser', | |
MinRK
|
r13049 | '--ipython-dir=%s' % cls.ipython_dir.name, | |
MinRK
|
r13180 | '--notebook-dir=%s' % cls.notebook_dir.name, | |
] | |||
cls.notebook = Popen(notebook_args, | |||
Thomas Kluyver
|
r13839 | stdout=nose.iptest_stdstreams_fileno(), | |
Thomas Kluyver
|
r13833 | stderr=STDOUT, | |
MinRK
|
r13180 | ) | |
MinRK
|
r13049 | cls.wait_until_alive() | |
@classmethod | |||
def teardown_class(cls): | |||
cls.notebook.terminate() | |||
cls.ipython_dir.cleanup() | |||
cls.notebook_dir.cleanup() | |||
cls.wait_until_dead() | |||
@classmethod | |||
def base_url(cls): | |||
return 'http://localhost:%i/' % cls.port | |||
Thomas Kluyver
|
r13099 | ||
@contextmanager | |||
Thomas Kluyver
|
r13105 | def assert_http_error(status, msg=None): | |
Thomas Kluyver
|
r13099 | try: | |
yield | |||
except requests.HTTPError as e: | |||
real_status = e.response.status_code | |||
assert real_status == status, \ | |||
"Expected status %d, got %d" % (real_status, status) | |||
Thomas Kluyver
|
r13105 | if msg: | |
assert msg in str(e), e | |||
Thomas Kluyver
|
r13099 | else: | |
assert False, "Expected HTTP error status" |