<s> """<STR_LIT>""" <EOL> from subprocess import PIPE <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 . . localinterfaces import localhost <EOL> from jupyter_client import KernelManager <EOL> from jupyter_client . multikernelmanager import MultiKernelManager <EOL> class TestKernelManager ( TestCase ) : <EOL> def _get_tcp_km ( self ) : <EOL> c = Config ( ) <EOL> km = MultiKernelManager ( 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 = MultiKernelManager ( config = c ) <EOL> return km <EOL> def _run_lifecycle ( self , km ) : <EOL> kid = km . start_kernel ( stdout = PIPE , stderr = PIPE ) <EOL> self . assertTrue ( km . is_alive ( kid ) ) <EOL> self . assertTrue ( kid in km ) <EOL> self . assertTrue ( kid in km . list_kernel_ids ( ) ) <EOL> self . assertEqual ( len ( km ) , <NUM_LIT:1> ) <EOL> km . restart_kernel ( kid , now = True ) <EOL> self . assertTrue ( km . is_alive ( kid ) ) <EOL> self . assertTrue ( kid in km . list_kernel_ids ( ) ) <EOL> km . interrupt_kernel ( kid ) <EOL> k = km . get_kernel ( kid ) <EOL> self . assertTrue ( isinstance ( k , KernelManager ) ) <EOL> km . shutdown_kernel ( kid , now = True ) <EOL> self . assertTrue ( not kid in km ) <EOL> def _run_cinfo ( self , km , transport , ip ) : <EOL> kid = km . start_kernel ( stdout = PIPE , stderr = PIPE ) <EOL> k = km . get_kernel ( kid ) <EOL> cinfo = km . get_connection_info ( kid ) <EOL> self . assertEqual ( transport , cinfo [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( ip , cinfo [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( '<STR_LIT>' in cinfo ) <EOL> self . assertTrue ( '<STR_LIT>' in cinfo ) <EOL> stream = km . connect_iopub ( kid ) <EOL> stream . close ( ) <EOL> self . assertTrue ( '<STR_LIT>' in cinfo ) <EOL> stream = km . connect_shell ( kid ) <EOL> stream . close ( ) <EOL> self . assertTrue ( '<STR_LIT>' in cinfo ) <EOL> stream = km . connect_hb ( kid ) <EOL> stream . close ( ) <EOL> km . shutdown_kernel ( kid , now = True ) <EOL> def test_tcp_lifecycle ( self ) : <EOL> km = self . _get_tcp_km ( ) <EOL> self . _run_lifecycle ( km ) <EOL> def test_shutdown_all ( self ) : <EOL> km = self . _get_tcp_km ( ) <EOL> kid = km . start_kernel ( stdout = PIPE , stderr = PIPE ) <EOL> self . assertIn ( kid , km ) <EOL> km . shutdown_all ( ) <EOL> self . assertNotIn ( kid , km ) <EOL> km . shutdown_all ( ) <EOL> def test_tcp_cinfo ( self ) : <EOL> km = self . _get_tcp_km ( ) <EOL> self . _run_cinfo ( km , '<STR_LIT>' , localhost ( ) ) <EOL> @ dec . skip_win32 <EOL> def test_ipc_lifecycle ( self ) : <EOL> km = self . _get_ipc_km ( ) <EOL> self . _run_lifecycle ( km ) <EOL> @ dec . skip_win32 <EOL> def test_ipc_cinfo ( self ) : <EOL> km = self . _get_ipc_km ( ) <EOL> self . _run_cinfo ( km , '<STR_LIT>' , '<STR_LIT:test>' ) </s>
