a <+b:A@sxdZddlZddlZddlZddlZddlZddlZddlm Z ddl m Z ddl Z ddl m Z ddlZddlmZddlmZmZmZmZmZmZmZmZdLd d Zd dZddZe jjddddZe jjddddZ ddZ!ddZ"e dde jj#ejdkp ejdko ej$dkd dd!d"Z%e ddd#d$Z&e dde jj#ejdkprejdkorej$dkd dd%d&Z'd'd(Z(d)d*Z)d+d,Z*d-d.Z+d/d0Z,e jj#ejdkd1dd2d3Z-d4d5Z.d6d7Z/e jj#ej0d8pej0dd9dd:d;Z1e jj#e2dd?Z3d@dAZ4e jj#ej5dBkdCddDdEZ6e jj#dFej7vdGddHdIZ8dJdKZ9dS)Mztest the IPython KernelN)TemporaryDirectory)flaky)version)locate_profile) new_kernelkernelTIMEOUTassemble_outputexecuteflush_channels wait_for_idle get_replyTstdoutcCsNt|ddt|t|d|d\}}t|j\}}|t|ksJJdS)N import syskccodez"print(sys.%s._is_master_process()))r r r get_iopub_msgstriprepr)rexpectedstreammsg_idcontentrstderrr:lib/python3.9/site-packages/ipykernel/tests/test_kernel.py _check_masters  rcCs,|ddkr(ds(Jddg|ddS)z#If status=error, show the tracebackstatuserrorF tracebackN)join)rrrr _check_status#s r%cCsltR}t|dd\}}t|j\}}|dks2J|dks>Jt|ddWdn1s^0YdS) simple print statement in kernelz print('hi')rzhi r!TrN)rr r rr)rrrrrrrrtest_simple_print+s   r(zDCurrently don't capture during test as pytest does its own capturingreasoncCsptV}|j}t|dd\}}t|\}}|dks6J|dksBJt|ddWdn1sb0YdS)r&z#import os; os.system('echo capsys')rzcapsys r!Tr'Nr iopub_channelr r rriopubrrrrrrrtest_capture_fd5s   r/cCsptV}|j}t|dd\}}t|\}}|dks6J|dksBJt|ddWdn1sb0YdS)r!zimport subprocess, sys; subprocess.run(['python', '-c', 'import os; os.system("echo CAP1"); print("CAP2")'], stderr=sys.stderr)rz CAP1 CAP2 Tr'Nr+r-rrr%test_subprocess_peek_at_stream_filenoCs    r0cCsnt.}t|dd\}}t|j\}}Wdn1s:0Ytj|t| }d|vsjJdS)z3test that sys.path doesn't get messed up by default!import sys; print(repr(sys.path))rNr!) rr r rsysrwriteast literal_evalrrrrrrZsys_pathrrr test_sys_pathTs , r7cCsxtdtdg.}t|dd\}}t|j\}}Wdn1sD0Ytj|t | }d|vstJdS)zJtest that sys.path doesn't get messed up when `--profile-dir` is specifiedz --profile-dirdefaultr1rNr!) rrr r rr2rr3r4r5rr6rrrtest_sys_path_profile_dir`s , r9)Zmax_runsZwin32darwin)r:z7subprocess prints fail on Windows and MacOS Python 3.8+c Cst}t|ddt|d}dddd|dd d g}t||d \}}t|j\}}|d |ksnJ|t|D]}|t |d ksvJ|qv|dksJt|ddt|dddWdn1s0YdS)zprinting from forked mp.ProcessTr'r"z import timeimport multiprocessing as mpzHpool = [mp.Process(target=print, args=('hello', i,)) for i in range(%i)]for p in pool: p.start()for p in pool: p.join()ztime.sleep(0.5),rZhellorr!rrrN) rrr r$r r rcountrangestr)rnprrrrrnrrrtest_subprocess_printms(     rGcCstz}d}ddd|ddg}t||d\}}t|j\}}|dksLJ|dksXJt|d d t|d d d Wd n1s0Yd S)z9mp.Process without print doesn't trigger iostream mp_moder=r"r>z?pool = [mp.Process(target=range, args=(i,)) for i in range(%i)]r?r@rr!Tr'rrAN)rr$r r rr)rrErrrrrrrrtest_subprocess_noprints   rHcCstn}dgd}t||d\}}t|j\}}|dks@Jd|vsLJt|ddt|ddd Wd n1sz0Yd S) z!error in mp.Process doesn't crashr")r>z(p = mp.Process(target=int, args=('hi',))z p.start()zp.join()rr! ValueErrorTr'rrAN)rr$r r rr)rrrrrrrrrtest_subprocess_errors   rJc Cst}|j}d}d}djfit}|j|dd}|jtd}|ddd ksVJ|d }|d |ksnJd }|||jtd} | d d dksJt |j \} } | |dksJWdn1s0YdS)z test inputinputzprompt> zprint({input_f}("{theprompt}"))T)Z allow_stdintimeoutheaderZmsg_typeZ input_requestrpromptz some textrokr"N) rr,formatlocalsr Z get_stdin_msgr rK get_shell_msgr r) rr.Zinput_fZ thepromptrrmsgrtextreplyrrrrrtest_raw_inputs    rWc Cst}t}tj|d}td|dt|td|dt|td||d\}}|ddksjJt|dd }|}Wdn1s0Yd|vsJd|vsJWdn1s0YWdn1s0YdS) Nzhist.outza=1ru b="abcþ"z %hist -f rrPzutf-8)encoding) rrospathr$r r openread)rtdfile_rVfrrrrtest_save_historys  & rbcCsrtjddd}tJ}dgd}t||d\}}|ddksPJ|d d Wdn1sd0YdS) NZ faulthandlerzthis test needs faulthandlerr)r")rzimport faulthandlerz import signalzfaulthandler.enable()z(if not sys.platform.startswith("win32"):z) faulthandler.register(signal.SIGTERM)rXrrPr#r!)pytest importorskiprr$r get)Z faulthadlerrrr`rVrrrtest_smoke_faulthandlers rfcCsbtjddddg}tj|ddd}|jdks4J|jd |jvsBJd |jvsPJd |jvs^Jd S) zipython kernel --help-all worksz-mIPythonrz --help-allT)rMZcapture_outputrs TracebacksOptionssClassN)r2 executable subprocessrun returncoderr)cmdprocrrrtest_help_outputs rocCst}|d|jtd}|dddks2J|d|jtd}|dddks\J|d|jtd}|ddd ksJ|dd d ksJ|d |jtd}|dddksJWdn1s0YdS) Nz2+2rLrrcompletez raise = 2Zinvalidz a = [1, 2,Z incompleteindentr!z %%timeit a )rZ is_completerSr )rrVrrrtest_is_completes        rrzonly run on Windowsc Cst>}td|dt|d}|||jtd}Wdn1sJ0Y|d}|ddkslJ|d}|d }|d |dd ksJ|d |ksJ|d }|sJ|D]&}|d|d |}||sJqdS) Na = 1rXzimport IPython b = a.rLrrrPza.Z cursor_endZ cursor_startmatches)rr r rprSr find startswith) rcellrVcstartendrumZ completedrrr test_completes"  * r}cCstjdddtd}dddg}t|ddi|d\}}t||d d}t|d |d d vsjJWdn1s~0YdS) NZ matplotlibzthis test requires matplotlibr)r"z+import matplotlib, matplotlib.pyplot as pltz"backend = matplotlib.get_backend()Zbackend)user_expressionsrr~Zbackend_inlinedataz text/plain)rcrdrr$r r%)rrxr`rVZbackend_bundlerrr test_matplotlib_inline_on_import1s  rc Csd}t}td|d\}}t||dd}d}g}t|D]}|||q@t||D]H\}}|jtd}t|d|dd|ksJ|d d |ks`Jq`Wdn1s0YdS) NdrsrXZexecution_countrza += 1 arLrZ parent_headerr)rr r%rCappend enumeraterSr ) Nrr`rVoffsetrxZmsg_idsirrrrtest_message_orderAs    rZlinuxztest only on windowsc Cstz}tP}tj|d}t|d}|dWdn1sL0Yd}tjtj|d}tj||dd}| d|d|j t d}|d d d ksJt |j \}} ||vsJt||j d d |j t d}|d d d ks Jt |j \}} d|vs&J|j dd |j t d}|d d d ksTJWdn1sj0YWdn1s0YdS)Nzunc.txtzw+z # UNC testz\\localhost\C$rzcd srLrrrPZls)rZcd)rrrZr[r$r\r3 splitdrivedirnamer rSr r rr ) rr^Zdrive_file_pathraZunc_rootZ file_pathZ unc_file_pathrVouterrrrrtest_unc_pathsTs* (      rZPyPyzdoes not work on PyPycCs~td}|j}td|dt||tdD]}|rJtdq2qPq2|r\JWdn1sp0YdS)z*Kernel exits after polite shutdown_requestrsrXi,g?N) rparentr r ZshutdownrCZis_alivetimesleep)rkmrrrr test_shutdownss   rcCsptV}|j}|d}td|ddlm}t||t }||d|Wdn1sb0YdS)z The kernel exits after being interrupted while waiting in input(). input() appears to have issues other functions don't, and it needs to be interruptible in order for pdb to be interruptible. input()rvalidate_message execute_replyN rrr rrinterrupt_kerneltest_message_specrrr rrrrrVrrrtest_interrupt_during_inputs    rntz0Message based interrupt not supported on WindowscCsxt^}|j}d|j_|d}td|ddlm }t ||t }||d|Wdn1sj0YdS)z messagerrrrN) rrZ kernel_specZinterrupt_moder rrrrrrr rrrrtest_interrupt_with_messages    rZ__pypy__z fails on pypycCstx}|j}|d}|d}td|ddlm}t||t }||d|t||t }||d|Wdn1s0YdS)z The kernel exits after being interrupted while waiting in pdb.set_trace(). Merely testing input() isn't enough, pdb has its own issues that need to be handled in addition. This test will fail with versions of IPython < 7.14.0. zimport pdb; pdb.set_trace()z3 + 4rrrNr)rrrZmsg_id2rrVrrr#test_interrupt_during_pdb_set_traces        rc Cs8d}t}|d}t|||d}g}t|D]}||d|q6tdg}t|D]0}|jdi}|j |||ddqht||g}|D]}|t||qg} |D]}| t||d d qWdn1s0Yd d |D} d d | D} | d| dks4JdS)Nr=passz&import asyncio; await asyncio.sleep(2)zi = g?Zkernel_info_requestrNrZcontrol)ZchannelcSsg|]}|ddqSrNdater.0rTrrr z0test_control_thread_priority..cSsg|]}|ddqSrrrrrrrrr) rr rrCrrrZsessionrTZcontrol_channelsend) rrrZ sleep_msg_idZ shell_msg_idsrZcontrol_msg_idsrTZ shell_repliesZcontrol_repliesZ shell_datesZ control_datesrrrtest_control_thread_prioritys0        4r)Tr):__doc__r4os.pathrZplatformrjr2rZtempfilerrrcZ packagingrrgZ IPython.pathsrZutilsrrr r r r r rrr%r(Zmarkskipr/r0r7r9Zskipif version_inforGrHrJrWrbrfrorrr}rrrwrZpython_implementationrrnamerbuiltin_module_namesrrrrrrs    (