a
    (ñ»jj8  ã                   @  sv  d dl mZ d dlZd dlZd dlZd dlZd dlZd dlmZ d dl	m
Z
 d dlZd dlZG dd„ de
ƒZdddœd	d
„Zddœdd„Zddœdd„Zddddœdd„Zdddœdd„Zddddœdd„Zddddœdd„Zddddœdd„Zddddœd d!„Zddddœd"d#„Zddddœd$d%„Zddœd&d'„Zd(dœd)d*„Zejjeƒ  d+d,�ddœd-d.„ƒZddœd/d0„ZdS )1é    )ÚannotationsN)Úpartial)ÚProtocolc                   @  s   e Zd Zddddœdd„ZdS )ÚRawInputÚ Ústr©ÚpromptÚreturnc                 C  s   d S ©N© )Úselfr	   r   r   ú;E:\soccer\ml_env\Lib\site-packages\trio/_tests/test_repl.pyÚ__call__   ó    zRawInput.__call__N)r   )Ú__name__Ú
__module__Ú__qualname__r   r   r   r   r   r      s   r   z	list[str])Úcmdsr
   c                   s(   t | ƒ‰ g ‰ddddœ‡ ‡fdd„}|S )z¨
    Pass in a list of strings.
    Returns a callable that returns each string, each time its called
    When there are not more strings to return, raise EOFError
    r   r   r   c                   s2   ˆ  | ¡ z
tˆ ƒW S  ty,   td ‚Y n0 d S r   )ÚappendÚnextÚStopIterationÚEOFError)r	   ©Z	cmds_iterZpromptsr   r   Ú_raw_helper   s
    

z$build_raw_input.<locals>._raw_helper)r   )Úiter)r   r   r   r   r   Úbuild_raw_input   s    r   ÚNone)r
   c                  C  sL   t dgƒ} | ƒ dksJ ‚t t¡� | ƒ  W d  ƒ n1 s>0    Y  dS )z"Quick test of our helper function.Zcmd1N)r   ÚpytestÚraisesr   )Ú	raw_inputr   r   r   Útest_build_raw_input'   s    
r!   zdict[str, object]c                   C  s   dt iS )NÚ__builtins__)r"   r   r   r   r   Úbuild_locals2   s    r#   zpytest.CaptureFixture[str]zpytest.MonkeyPatch)ÚcapsysÚmonkeypatchr
   c                 Ã  s`   t jjtƒ d�}tg d¢ƒ}| |d|¡ t j |¡I dH  |  ¡ \}}| ¡ g d¢ks\J ‚dS )zŽ
    Run some basic commands through the interpreter while capturing stdout.
    Ensure that the interpreted prints the expected results.
    ©Zrepl_locals)zx = 1zprint(f'{x=}')ú'hello'zdef func():z  print(x + 1)r   zfunc()zasync def afunc():z
  return 4r   zawait afunc()ú
import sysz"sys.stdout.write('hello stdout\n')r    N)zx=1r'   Ú2Ú4zhello stdoutZ13)	ÚtrioÚ_replÚTrioInteractiveConsoler#   r   ÚsetattrÚrun_replÚ
readouterrÚ
splitlines©r$   r%   Úconsoler    ÚoutZ_errr   r   r   Útest_basic_interaction6   s    ÿr5   )r%   r
   c                 Ã  sh   t jjtƒ d�}tdgƒ}|  |d|¡ t t¡�" t j 	|¡I d H  W d   ƒ n1 sZ0    Y  d S )Nr&   zraise SystemExitr    )
r+   r,   r-   r#   r   r.   r   r   Ú
SystemExitr/   )r%   r3   r    r   r   r   Ú"test_system_exits_quit_interpreter[   s    ÿÿr7   c                 Ã  sp   t jjtƒ d�}tg d¢ƒ}| |d|¡ t j |¡I d H  |  ¡ \}}d|v sTJ ‚d|vs`J ‚d|v slJ ‚d S )Nr&   )z"import signal, trio, trio.lowlevelzasync def f():zh  trio.lowlevel.spawn_system_task(    trio.to_thread.run_sync,    signal.raise_signal, signal.SIGINT,  )z  await trio.sleep_forever()z  print('should not see this')r   z	await f()z print('AFTER KeyboardInterrupt')r    ÚKeyboardInterruptZshouldzAFTER KeyboardInterrupt©r+   r,   r-   r#   r   r.   r/   r0   ©r$   r%   r3   r    r4   Úerrr   r   r   Útest_KI_interruptsg   s    ÿr<   c                 Ã  sX   t jjtƒ d�}tg d¢ƒ}| |d|¡ t j |¡I d H  |  ¡ \}}d|v sTJ ‚d S )Nr&   )r(   úif sys.version_info < (3, 11):ú/  from exceptiongroup import BaseExceptionGroupr   z<raise BaseExceptionGroup('', [RuntimeError(), SystemExit()])ú!print('AFTER BaseExceptionGroup')r    úAFTER BaseExceptionGroupr9   r2   r   r   r   Útest_system_exits_in_exc_groupƒ   s    ÿ
rA   c                 Ã  sX   t jjtƒ d�}tg d¢ƒ}| |d|¡ t j |¡I d H  |  ¡ \}}d|v sTJ ‚d S )Nr&   )r(   r=   r>   r   zraise BaseExceptionGroup(z?  '', [BaseExceptionGroup('', [RuntimeError(), SystemExit()])])r?   r    r@   r9   r2   r   r   r   Ú%test_system_exits_in_nested_exc_groupš   s    ÿrB   c                 Ã  sp   t jjtƒ d�}tddgƒ}| |d|¡ t j |¡I d H  |  ¡ \}}d|vsTJ ‚d|vs`J ‚d|v slJ ‚d S )Nr&   zraise BaseExceptionúprint('AFTER BaseException')r    ú_threads.pyú_repl.pyúAFTER BaseExceptionr9   r:   r   r   r   Útest_base_exception_captured²   s    ýÿrG   c                 Ã  sX   t jjtƒ d�}tddgƒ}| |d|¡ t j |¡I d H  |  ¡ \}}d|v sTJ ‚d S )Nr&   z&raise ExceptionGroup('', [KeyError()])zprint('AFTER ExceptionGroup')r    zAFTER ExceptionGroupr9   r2   r   r   r   Útest_exc_group_capturedÆ   s    ýÿrH   c                 Ã  sp   t jjtƒ d�}tg d¢ƒ}| |d|¡ t j |¡I d H  |  ¡ \}}d|vsTJ ‚d|vs`J ‚d|v slJ ‚d S )Nr&   )z-async def async_func_raises_base_exception():z  raise BaseExceptionr   z(await async_func_raises_base_exception()rC   r    rD   rE   rF   r9   r:   r   r   r   Ú*test_base_exception_capture_from_coroutineØ   s    ÿrI   c                  C  s(   t jtjddgdd�} | jdks$J ‚dS )zL
    Basic smoke test when running via the package __main__ entrypoint.
    ú-mr+   s   exit())Úinputr   N)Ú
subprocessÚrunÚsysÚ
executableÚ
returncode)Úreplr   r   r   Útest_main_entrypointð   s    rR   Úboolc                  C  s4   t jdkrdS t d¡} |  ¡ s$dS |  ¡ dkS d S )NÚlinuxFz /proc/sys/dev/tty/legacy_tiocstiTÚ1)rN   ÚplatformÚpathlibÚPathÚexistsÚ	read_text)Zsysctlr   r   r   Úshould_try_newline_injectionø   s    

r[   z"the ioctl we use is disabled in CI)Úreasonc                  C  sz  t jdksJ ‚dd l} |  ¡ \}}|dkrHtjt jgt jdddg¢R Ž  d}| d¡sh|t |d¡7 }qLt	| 
¡ ƒ d}t |d	¡ | d
¡s |t |d¡7 }q„| d¡dks²J ‚t	| 
¡ ƒ d}t |tj¡ | d
¡sì|t |d¡7 }qÐd|v søJ ‚t	| 
¡ ƒ d}t |d¡ t |tj¡ | d
¡�sB|t |d¡7 }�q"d|v �sPJ ‚t	| 
¡ ƒ t |¡ t |d¡d  d S )NÚwin32r   ú-urJ   r+   r   s   import trio
>>> i   ó   print("hello!")
ó   >>> ó   hello!é   ó   KeyboardInterrupts   print("hello!")é   )rN   rV   ÚptyZforkÚosÚexeclprO   ÚendswithÚreadÚprintÚdecodeÚwriteÚcountÚkillÚsignalÚSIGINTÚcloseÚwaitpid)re   ÚpidZpty_fdÚbufferr   r   r   Útest_ki_newline_injection  s<    



ru   c                  Ã  s  t  ¡ 4 I d H �šÞ} |  tt jtjdddgtjtj	tjtj
dkrFtjndd�¡I d H }|j4 I d H �šR d}|j2 z*3 d H W }||7 }| dd	¡ d
¡rr q qr6 t| ¡ ƒ d}|j d¡I d H  |j2 z"3 d H W }||7 }| d¡rÈ qîqÈ6 d|v súJ ‚t| ¡ ƒ tj
dk�rfd}|j d¡I d H  |j2 z(3 d H W }||7 }| d¡�r. �qZ�q.6 t| ¡ ƒ d}|j d¡I d H  |j2 z(3 d H W }||7 }| d¡�r‚ �q®�q‚6 t| ¡ ƒ tj
dk�rÌtjntj}t |j|¡ tj
dk�r |j d	¡I d H  n|j d	¡I d H  d}|j2 z(3 d H W }||7 }| d¡�r �qH�q6 d|v �sVJ ‚t| ¡ ƒ |j d¡I d H  d}d}|j2 zT3 d H W }||7 }| dd	¡ d¡�rÂ|�sÂt |j|¡ d}| d¡�r‚ �qÚ�q‚6 d|v �sèJ ‚d|v �söJ ‚t| ¡ ƒ |j d¡I d H  d}d}|j2 zT3 d H W }||7 }| dd	¡ d¡�rb|�sbt |j|¡ d}| d¡�r" �qz�q"6 d|v �sˆJ ‚d|v �s–J ‚t| ¡ ƒ W d   ƒI d H  �qÐ1 I d H �sÆ0    Y  | j ¡  W d   ƒI d H  �q1 I d H �sþ0    Y  d S )Nr^   rJ   r+   r]   r   )ÚstdoutÚstderrÚstdinÚcreationflagsr   s   
ó   
s   import trio
>>> r_   r`   ra   sI   import ctypes; ctypes.windll.kernel32.SetConsoleCtrlHandler(None, False)
sW   import coverage; trio.lowlevel.enable_ki_protection(coverage.pytracer.PyTracer._trace)
rc   s+   print("READY"); await trio.sleep_forever()
Fs   READY
Ts   trios/   import time; print("READY"); time.sleep(99999)
s	   Traceback)r+   Zopen_nurseryÚstartr   Zrun_processrN   rO   rL   ÚPIPEÚSTDOUTrV   ÚCREATE_NEW_PROCESS_GROUPrv   Úreplacerh   rj   rk   rx   Zsend_allro   ÚCTRL_C_EVENTrp   rf   rn   rs   Zcancel_scopeÚcancel)ZnurseryÚprocrt   ÚpartZsignal_sentZkilledr   r   r   Útest_ki_in_repl<  s¬    úÿ

ÿ
ÿ
ÿ
:r„   ) Ú
__future__r   rf   rW   ro   rL   rN   Ú	functoolsr   Útypingr   r   Z
trio._replr+   r   r   r!   r#   r5   r7   r<   rA   rB   rG   rH   rI   rR   r[   ÚmarkZskipifru   r„   r   r   r   r   Ú<module>   s:   %þ4