U /e@sUdZddlZddlZddlZddlmZddlmZddlm Z ddl m Z ddl m Z ddl mZdd lmZd Zd Zdaejed <daeed <ddZd(ddZedfddZd)ddZddZeddZddZddZd*d d!Zd"d#Z d$d%Z!Gd&d'd'e Z"dS)+z%utilities for testing IPython kernelsN)contextmanager)Empty)STDOUT)TemporaryDirectory)time)manager)BlockingKernelClient<dKMKCc KsNt|d<zddl}||d<Wnttfk r8YnXtjfdti|S)zostart a new kernel, and return its Manager and Client Integrates with our output capturing for tests. stderrrNstdoutZstartup_timeout)rnoseiptest_stdstreams_fileno ImportErrorAttributeErrorrstart_new_kernelSTARTUP_TIMEOUT)kwargsrr9/tmp/pip-unpacked-wheel-g43y689f/ipykernel/tests/utils.pyrsrc Cs`ddlm}|dkrt}|j|jfD]6}z|dd}Wntk rNYq$Yq(X||q(q$dS)z'flush any messages waiting on the queuevalidate_messageNg?timeout)test_message_specrr Z get_shell_msg get_iopub_msgr)kcrget_msgmsgrrrflush_channels(s  r"shellcCsdt}t|d|d}||d}|dd|kr4q`td|d|t}|||8}|}q|S)Nget__msgrZ parent_headermsg_idzIgnoring reply not to z: )rgetattrprint)rr&rZchannelt0r replyt1rrr get_reply8s  r,cKsddlm}|dkrt}|jfd|i|}t||t}||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 requestrrNcodeZ execute_replyrstatuscontentexecution_statebusyZsilent execute_input traceback )file) rrr executer,TIMEOUTrAssertionErrorgetr(joinsysr )r.rrrr&r*r2r3rrrr7Gs         r7cCs*tdkrt\aattntttS)zCstart the global kernel (if it isn't running) and return its clientN)r rr atexitregisterstop_global_kernelr"rrrrstart_global_kernel`s   r@ccs tVdS)zContext manager for the global kernel instance Should be used for most kernel tests Returns ------- kernel_client: connected KernelClient instance N)r@rrrrkernelks rAcs fdd}j|_j|_|S)z.Decorator for tests that use the global kernelc st}|W5QRXdSN)rA)rtest_frr wrapped_test{sz!uses_kernel..wrapped_test)__doc____name__)rDrErrCr uses_kernelxs rHcCs,tdatdkrdStjdddadS)z4Stop the global shared kernel instance, if it existsNT)now)r Z stop_channelsr Zshutdown_kernelrrrrr?s  r?c CsVdti}zddl}||d<Wnttfk r8YnX|dk rJ||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 rNrZextra_arguments)rrrrrrZ run_kernel)argvrrrrr new_kernels rKcCsd}d}|dd}|d}|d}|dkr:|ddkr:qq|dd kr|d d kr`||d 7}q|d d krz||d 7}qtd|d qq||fS)z%assemble stdout/err from an executionr-rrmsg_typer0r/r1idlestreamnamertextr zbad stream: %r)KeyError)r rr r!rLr0rrrassemble_outputs    rRcCs8|jdd}|d}|d}|dkr|ddkrq4qdS)NrrrLr0r/r1rM)r)rr!rLr0rrr wait_for_idles  rScs,eZdZdZfddZfddZZS)TemporaryWorkingDirectoryz Creates a temporary directory and sets the cwd to that directory. Automatically reverts to previous cwd upon cleanup. Usage example: with TemporaryWorkingDirectory() as tmpdir: ... cs t|_t|jtSrB)osgetcwdold_wdchdirrOsuper __enter__)self __class__rrrZs  z#TemporaryWorkingDirectory.__enter__cst|jt|||SrB)rUrXrWrY__exit__)r[excvaluetbr\rrr^s z"TemporaryWorkingDirectory.__exit__)rG __module__ __qualname__rFrZr^ __classcell__rrr\rrTs rT)N)r-N)N)#rFr=rUr< contextlibrqueuer subprocessrtempfilerrZjupyter_clientrZjupyter_client.blocking.clientrrr8r Z KernelManager__annotations__r rr"r,r7r@rArHr?rKrRrSrTrrrrs6