B U[@sdZddlmZddlZddlZddlZddlmZddlm Z m Z yddl m Z Wn e k rtddlm Z YnXddlZddlmZdZd Zdadad d Zdd d Zd ddZddZeddZddZddZd!ddZddZddZdS)"z%utilities for testing IPython kernels)print_functionN)contextmanager)PIPESTDOUT)Empty)manager<cKsRy t}Wntk r*ttj}YnX|t|tdt j fdt i|S)zostart a new kernel, and return its Manager and Client Integrates with our output capturing for tests. )stdoutstderrZstartup_timeout) noseiptest_stdstreams_filenoAttributeErroropenosdevnullupdatedictrrstart_new_kernelSTARTUP_TIMEOUT)kwargsr r4lib/python3.7/site-packages/ipykernel/tests/utils.pyrs  rc Csjddlm}|dkrt}xL|j|jfD]<}x6y|jddd}Wntk rTPYq,X||q,Wq&WdS)z'flush any messages waiting on the queue)validate_messageNTg?)blocktimeout)test_message_specrKCZ shell_channel iopub_channelget_msgr)kcrZchannelmsgrrrflush_channels,s r#cKsddlm}|dkrt}|jfd|i|}|jtd}||d||jtd}||d||dd d kspt|d s|jtd}||d ||dd|kst|dd rt d |dd t j d||dfS)zBwrapper for doing common steps for validating an execution requestr)rNcode)rZ execute_replystatuscontentexecution_statebusyZsilent execute_input traceback )file) rrrexecuteZ get_shell_msgTIMEOUTZ get_iopub_msgAssertionErrorgetprintjoinsysr )r%r!rrZmsg_idZreplyr)r*rrrr.<s         r.cCs*tdkrt\aattntttS)zCstart the global kernel (if it isn't running) and return its clientN)KMrratexitregisterstop_global_kernelr#rrrrstart_global_kernelTs   r9ccs tVdS)zContext manager for the global kernel instance Should be used for most kernel tests Returns ------- kernel_client: connected KernelClient instance N)r9rrrrkernel^s r:cs fdd}j|_j|_|S)z.Decorator for tests that use the global kernelc st}|WdQRXdS)N)r:)r!)test_frr wrapped_testlsz!uses_kernel..wrapped_test)__doc____name__)r;r<r)r;r uses_kerneljs r?cCs,tdatdkrdStjdddadS)z4Stop the global shared kernel instance, if it existsNT)Znow)rZ stop_channelsr5Zshutdown_kernelrrrrr8ss  r8cCs:ttdttj}t|td}|dk r.||d<tjf|S)zContext manager for a new kernel in a subprocess Should only be used for tests where the kernel must not be re-used. Returns ------- kernel_client: connected KernelClient instance r )r r NZextra_arguments) getattrr rrrrrrZ run_kernel)argvr rrrr new_kernel}s  rBcCsd}d}x|jddd}|d}|d}|dkr@|dd kr@Pq |dd kr |d d krf||d 7}q|d dkr||d 7}qtd|d q q W||fS)z%assemble stdout/err from an executionr$Tr)rrmsg_typer'r&r(idlestreamnamer textr zbad stream: %r)r KeyError)Ziopubr r r"rCr'rrrassemble_outputs    rIcCs@x:|jjddd}|d}|d}|dkr|ddkrPqWdS) NTr)rrrCr'r&r(rD)rr )r!r"rCr'rrr wait_for_idles rJ)N)r$N)N) r=Z __future__rr6rr4 contextlibr subprocessrrZqueuer ImportErrorZQueuer Zjupyter_clientrrr/r5rrr#r.r9r:r?r8rBrIrJrrrrs4