a <+b/@sdZddlZddlZddlZddlZddlmZddlmZddlm Z ddl m Z ddl m Z ddlmZd Zd Zdadad d Zd%d dZedfddZd&ddZddZe ddZddZddZd'ddZdd Zd!d"ZGd#d$d$eZdS)(z%utilities for testing IPython kernelsN)TemporaryDirectory)time)contextmanager)Empty)STDOUT)manager<dc KsLt|d<zddl}||d<Wnttfy6Yn0tjfdti|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)kwargsr r4lib/python3.9/site-packages/ipykernel/tests/utils.pyrsrc Cs^ddlm}|durt}|j|jfD]4}z|dd}WntyLYq$Yq(0||q(q$dS)z'flush any messages waiting on the queuevalidate_messageNg?timeout)test_message_specrKCZ get_shell_msg get_iopub_msgr)kcrget_msgmsgrrrflush_channels)s   r shellcCsdt}t|d|d}||d}|dd|kr4q`td|d|t}|||8}|}q|S)NZget_Z_msgrZ parent_headermsg_idzIgnoring reply not to z: )rgetattrprint)rr"rZchannelZt0rreplyt1rrr get_reply9s  r'cKsddlm}|durt}|jfd|i|}t||t}||d||jtd}||d||dd d kspJ|d s|jtd}||d ||dd|ksJ|dd rtd |dd t j d||dfS)zBwrapper for doing common steps for validating an execution requestrrNcodeZ execute_replyrstatuscontentexecution_statebusyZsilent execute_input traceback )file) rrrexecuter'TIMEOUTrgetr$joinsysr )r)rrrr"r%r-r.rrrr2Hs         r2cCs*tdurt\aattntttS)zCstart the global kernel (if it isn't running) and return its clientN)KMrratexitregisterstop_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;rrrrkerneljs r<cs fdd}j|_j|_|S)z.Decorator for tests that use the global kernelcs2t}|Wdn1s$0YdSN)r<)rtest_frr wrapped_testxsz!uses_kernel..wrapped_test)__doc____name__)r?r@rr>r uses_kernelvs rCcCs,tdatdurdStjdddadS)z4Stop the global shared kernel instance, if it existsNT)Znow)rZ stop_channelsr7Zshutdown_kernelrrrrr:s  r:c CsXdti}zddl}||d<Wnttfy6Yn0|durH||d<tjfi|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 rNr Zextra_arguments)rr r rrrZ run_kernel)argvrr rrr new_kernels rEcCsd}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_typer+r*r,idlestreamnamer textr zbad stream: %r)KeyError)rr r rrFr+rrrassemble_outputs    rLcCs8|jdd}|d}|d}|dkr|ddkrq4qdS)NrrrFr+r*r,rG)r)rrrFr+rrr wait_for_idles  rMcs,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|jtSr=)osgetcwdold_wdchdirrIsuper __enter__)self __class__rrrTs  z#TemporaryWorkingDirectory.__enter__cst|jt|||Sr=)rOrRrQrS__exit__)rUexcvaluetbrVrrrXs z"TemporaryWorkingDirectory.__exit__)rB __module__ __qualname__rArTrX __classcell__rrrVrrNs rN)N)r(N)N) rAr8rOplatformr6Ztempfilerr contextlibrZqueuer subprocessrZjupyter_clientrrr3r7rrr r'r2r;r<rCr:rErLrMrNrrrrs6