<s> """<STR_LIT>""" <EOL> import json <EOL> import os <EOL> pjoin = os . path . join <EOL> import signal <EOL> from subprocess import PIPE <EOL> import sys <EOL> import time <EOL> from unittest import TestCase <EOL> from ipython_genutils . testing import decorators as dec <EOL> from traitlets . config . loader import Config <EOL> from jupyter_core import paths <EOL> from jupyter_client import KernelManager <EOL> from . . manager import start_new_kernel <EOL> from . utils import test_env <EOL> TIMEOUT = <NUM_LIT:30> <EOL> class TestKernelManager ( TestCase ) : <EOL> def setUp ( self ) : <EOL> self . env_patch = test_env ( ) <EOL> self . env_patch . start ( ) <EOL> def tearDown ( self ) : <EOL> self . env_patch . stop ( ) <EOL> def _install_test_kernel ( self ) : <EOL> kernel_dir = pjoin ( paths . jupyter_data_dir ( ) , '<STR_LIT>' , '<STR_LIT>' ) <EOL> os . makedirs ( kernel_dir ) <EOL> with open ( pjoin ( kernel_dir , '<STR_LIT>' ) , '<STR_LIT:w>' ) as f : <EOL> f . write ( json . dumps ( { <EOL> '<STR_LIT>' : [ sys . executable , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> } ) ) <EOL> def _get_tcp_km ( self ) : <EOL> c = Config ( ) <EOL> km = KernelManager ( config = c ) <EOL> return km <EOL> def _get_ipc_km ( self ) : <EOL> c = Config ( ) <EOL> c . KernelManager . transport = '<STR_LIT>' <EOL> c . KernelManager . ip = '<STR_LIT:test>' <EOL> km = KernelManager ( config = c ) <EOL> return km <EOL> def _run_lifecycle ( self , km ) : <EOL> km . start_kernel ( stdout = PIPE , stderr = PIPE ) <EOL> self . assertTrue ( km . is_alive ( ) ) <EOL> km . restart_kernel ( now = True ) <EOL> self . assertTrue ( km . is_alive ( ) ) <EOL> km . interrupt_kernel ( ) <EOL> self . assertTrue ( isinstance ( km , KernelManager ) ) <EOL> km . shutdown_kernel ( now = True ) <EOL> def test_tcp_lifecycle ( self ) : <EOL> km = self . _get_tcp_km ( ) <EOL> self . _run_lifecycle ( km ) <EOL> @ dec . skip_win32 <EOL> def test_ipc_lifecycle ( self ) : <EOL> km = self . _get_ipc_km ( ) <EOL> self . _run_lifecycle ( km ) <EOL> def test_get_connect_info ( self ) : <EOL> km = self . _get_tcp_km ( ) <EOL> cinfo = km . get_connection_info ( ) <EOL> keys = sorted ( cinfo . keys ( ) ) <EOL> expected = sorted ( [ <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT:key>' , '<STR_LIT>' , <EOL> ] ) <EOL> self . assertEqual ( keys , expected ) <EOL> @ dec . skip_win32 <EOL> def test_signal_kernel_subprocesses ( self ) : <EOL> self . _install_test_kernel ( ) <EOL> km , kc = start_new_kernel ( kernel_name = '<STR_LIT>' ) <EOL> def execute ( cmd ) : <EOL> kc . execute ( cmd ) <EOL> reply = kc . get_shell_msg ( TIMEOUT ) <EOL> content = reply [ '<STR_LIT:content>' ] <EOL> self . assertEqual ( content [ '<STR_LIT:status>' ] , '<STR_LIT>' ) <EOL> return content <EOL> self . addCleanup ( kc . stop_channels ) <EOL> self . addCleanup ( km . shutdown_kernel ) <EOL> N = <NUM_LIT:5> <EOL> for i in range ( N ) : <EOL> execute ( "<STR_LIT:start>" ) <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> reply = execute ( '<STR_LIT>' ) <EOL> self . assertEqual ( reply [ '<STR_LIT>' ] [ '<STR_LIT>' ] , [ None ] * N ) <EOL> kc . execute ( '<STR_LIT>' ) <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> km . interrupt_kernel ( ) <EOL> reply = kc . get_shell_msg ( TIMEOUT ) <EOL> content = reply [ '<STR_LIT:content>' ] <EOL> self . assertEqual ( content [ '<STR_LIT:status>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( content [ '<STR_LIT>' ] [ '<STR_LIT>' ] , True ) <EOL> for i in range ( <NUM_LIT:50> ) : <EOL> reply = execute ( '<STR_LIT>' ) <EOL> if reply [ '<STR_LIT>' ] [ '<STR_LIT>' ] != [ - signal . SIGINT ] * N : <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> else : <EOL> break <EOL> self . assertEqual ( reply [ '<STR_LIT>' ] [ '<STR_LIT>' ] , [ - signal . SIGINT ] * N ) <EOL> def test_start_new_kernel ( self ) : <EOL> self . _install_test_kernel ( ) <EOL> km , kc = start_new_kernel ( kernel_name = '<STR_LIT>' ) <EOL> self . addCleanup ( kc . stop_channels ) <EOL> self . addCleanup ( km . shutdown_kernel ) <EOL> self . assertTrue ( km . is_alive ( ) ) <EOL> self . assertTrue ( kc . is_alive ( ) ) </s>
