o Phz@sdZddlmZmZddlZddlZddlZddlZddlZddl Z ddl Z ddl Z ddl Z ddl Z ddlmZddlZddlZejeddZdZZdZd d gZd d gZe jd rad gZdgZdZe jdrzddejdejddd DZWne!yYn we jd rdZddZ"Gddde#Z$edd fddZ%d-ddZ&ej'd d!Z(d"d#Z)d$d%Z*ej+ed&d'd(Z,ej+ed&d)d*Z-ej+ed&d+d,Z.dS).z)Accessories for automated py.test runner.)with_statementprint_functionN)Terminalzxterm-256color)kinds SEMAPHORE s SEMAPHORE (P TEST_QUICKz)xterm screen ansi vt220 rxvt cons25 linux TEST_FULLcCsg|] }|dddqS)Nrr)split).0_termrT/var/www/html/env_mimamsha/lib/python3.10/site-packages/blessed/tests/accessories.py &sr)toez-aT)stdout close_fdszxterm screen ansi linuxcCsdzddl}Wn tyYdSwtjtjttjtjd}|j|d}| d|| |S)Nrz .coveragerc) config_filezrun:note) coverage ImportErrorospathjoindirname__file__pardirCoverage set_optionstart)run_noter _coveragerccovrrrinit_subproc_coverage3s     r%c@s(eZdZdZdZdZddZddZdS) as_subprocesszThis helper executes test cases in a child process, avoiding a python-internal bug of _curses: setupterm() may not be called more than once per process. rutf8cCs ||_dS)N)func)selfr(rrr__init__Js zas_subprocess.__init__c Os&t}t\}}||jkrtdjt|j||d}z |j|i|Wnptyt \}}} t } t | D] } | | dq=| dd| ddt ||Dtt jd| tt jtt jtt j|dur||td Ynw|dur||td |tkrtd |tt jd td t } t!"|j#} zt$|d}Wn t%yYn w|sn| | &|7} qt'|d \}}t|dd| df}| dksJ|t(|d ksJdS)Nz5@as_subprocess-{pid};{func_name}(*{args}, **{kwargs}))pid func_nameargskwargsutf-8z(-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=asciicSsg|] }|dqS)r/)rstripencode)r_excrrrrasz*as_subprocess.__call__.. rrz&TEST RUNNER HAS FORKED, {0}=>{1}: EXIT)fileTiz!Output in child process: %s %s %sz(========================================))rgetpidptyfork _CHILD_PIDr%formatr( Exceptionsysexc_infolist traceback format_tbappendr1r2extendformat_exception_onlywrite __stdout__filenorclose __stderr__ __stdin__stopsave_exitprintstderrsix text_typecodecsgetincrementaldecoderencodingreadOSErrordecodewaitpid WEXITSTATUS)r)r-r.pid_testrunnerr+ master_fdr$e_typee_valuee_tbo_errline exc_outputdecoderr3statusexc_output_msgrrr__call__Msv          zas_subprocess.__call__N)__name__ __module__ __qualname____doc__r:rTr*rerrrrr&Bs  r&r'cCst}t|}|d}||s7zt|d}Wn ty%Ynw|s)n||j|dd7}||r||sDJd||f|t |dS)aKRead file descriptor ``fd`` until ``semaphore`` is found. Used to ensure the child process is awake and ready. For timing tests; without a semaphore, the time to fork() would be (incorrectly) included in the duration of the test, which can be very length on continuous integration servers (such as Travis-CI). r0rFfinalz5Semaphore not recv before EOF (expected: %r, got: %r)N) rPrQrRrSrW startswithrrUrVlen)fd semaphorerTtimeoutoutprbr3rrrread_until_semaphores&     rrcCsZt|}t} zt|d}Wn tyY|Sw|s# |S||j|dd7}q )z=Read file descriptor ``fd`` until EOF. Return decoded string.TdFrj)rRrSrPrQrrUrVrW)rnrTrbrqr3rrrread_until_eofs  rtc csz.t|}|dtj@|d<t|tj|dVW|dtjB|d<t|tj|dS|dtjB|d<t|tj|w)z@Ensure any bytes written to pty fd are not duplicated as output.N)termios tcgetattrECHO tcsetattrTCSANOW)rnattrsrrrecho_offs r|cCs:zt|}Wn tjyd}Ynw|r|dSdS)z4Return the result of ``tigetstr`` except as Unicode.Nlatin1r6)cursestigetstrerrorrW)capvalrrr unicode_caps rcGsnzt|}Wn tjyd}Ynw|r5z tj|g|R}Wn tjy-d}Ynw|r5|dSdS)z=Return the result of ``tparm(tigetstr())`` except as Unicode.Nr}r6)r~rrtparmrW)rparmsrrrr unicode_parms r)paramscC|jS)z.Common kind values for all kinds of terminals.paramrequestrrr all_termsrcCr)z*Various number of lines for screen height.rrrrr many_linesrrcCr)z+Various number of columns for screen width.rrrrr many_columnsrr)r')/ri __future__rr contextlib subprocess functoolsr@rvrRr~r=r8rblessedrpytestrPpartial TestTerminalSEND_SEMAPHORE SEMAPHORERECV_SEMAPHOREmany_lines_paramsmany_columns_paramsenvirongetr all_terms_paramsPopenPIPE communicate splitlinesrVr%objectr&rrrtcontextmanagerr|rrfixturerrrrrrrsp     R