<s> import os <EOL> import pytest <EOL> import requests <EOL> import subprocess <EOL> import sys <EOL> import time <EOL> from os . path import join , exists , dirname , pardir <EOL> from requests . exceptions import ConnectionError <EOL> from . utils import write <EOL> def pytest_addoption ( parser ) : <EOL> parser . addoption ( <EOL> "<STR_LIT>" , type = int , default = <NUM_LIT> , help = "<STR_LIT>" <EOL> ) <EOL> @ pytest . fixture ( scope = "<STR_LIT>" ) <EOL> def jupyter_notebook ( request , bokeh_server , log_file ) : <EOL> """<STR_LIT>""" <EOL> from jupyter_core import paths <EOL> config_dir = paths . jupyter_config_dir ( ) <EOL> body = """<STR_LIT>""" <EOL> custom = join ( config_dir , "<STR_LIT>" ) <EOL> if not exists ( custom ) : <EOL> os . makedirs ( custom ) <EOL> customjs = join ( custom , "<STR_LIT>" ) <EOL> with open ( customjs , "<STR_LIT:w>" ) as f : <EOL> f . write ( body ) <EOL> def clean_up_customjs ( ) : <EOL> with open ( customjs , "<STR_LIT:w>" ) as f : <EOL> f . write ( "<STR_LIT>" ) <EOL> request . addfinalizer ( clean_up_customjs ) <EOL> notebook_port = pytest . config . option . notebook_port <EOL> env = os . environ . copy ( ) <EOL> env [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> notebook_dir = join ( dirname ( __file__ ) , pardir , pardir ) <EOL> cmd = [ "<STR_LIT>" , "<STR_LIT>" ] <EOL> argv = [ "<STR_LIT>" , "<STR_LIT>" % notebook_port , "<STR_LIT>" % notebook_dir ] <EOL> jupter_notebook_url = "<STR_LIT>" % notebook_port <EOL> try : <EOL> proc = subprocess . Popen ( cmd + argv , env = env , stdout = log_file , stderr = log_file ) <EOL> except OSError : <EOL> write ( "<STR_LIT>" % "<STR_LIT:U+0020>" . join ( cmd + argv ) ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> else : <EOL> def stop_jupyter_notebook ( ) : <EOL> write ( "<STR_LIT>" ) <EOL> proc . kill ( ) <EOL> request . addfinalizer ( stop_jupyter_notebook ) <EOL> def wait_until ( func , timeout = <NUM_LIT> , interval = <NUM_LIT> ) : <EOL> start = time . time ( ) <EOL> while True : <EOL> if func ( ) : <EOL> return True <EOL> if time . time ( ) - start > timeout : <EOL> return False <EOL> time . sleep ( interval ) <EOL> def wait_for_jupyter_notebook ( ) : <EOL> def helper ( ) : <EOL> if proc . returncode is not None : <EOL> return True <EOL> try : <EOL> return requests . get ( jupter_notebook_url ) <EOL> except ConnectionError : <EOL> return False <EOL> return wait_until ( helper ) <EOL> if not wait_for_jupyter_notebook ( ) : <EOL> write ( "<STR_LIT>" % "<STR_LIT:U+0020>" . join ( cmd + argv ) ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> if proc . returncode is not None : <EOL> write ( "<STR_LIT>" + str ( proc . returncode ) ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> return jupter_notebook_url </s>
