.Ph dZddlZddlZddlZddlZddlZddlZddlZddl Z e j Z e j Z ej j Ze jjZe jZe jjZe jjZe jjZe jZdZddZdS)z Helper for testing. NcDtjdfd }|t_dS)z Redirect all warnings to stdout.Nrct||tj||tjdS)N)ospathbasenamesysstdout)msgcatfnamelnofilelineshowwarning_origs N/var/www/html/test/jupyter/venv/lib/python3.11/site-packages/joblib/testing.py showwarningz'warnings_to_stdout..showwarnings4c27#3#3E#:#:D#*MMMMM)Nr)warningsr)rrs @rwarnings_to_stdoutrs@+NNNNNN'Hrc tjtjtj fd} fd} |Utj||}|tjd|z|}| \}} || } } jdkr+d j|| } t| |8tj ||s#td |||8tj || s#td || |*| | dSdS#|)| | wwxYw) aRuns a command in a subprocess with timeout in seconds. A SIGTERM is sent after `timeout` and if it does not terminate, a SIGKILL is sent after `2 * timeout`. Also checks returncode is zero, stdout if stdout_regex is set, and stderr if stderr_regex is set. )r stderrc^tjddS)z Attempt to terminate a leftover process spawned during test execution: ideally this should not be needed but can help avoid clogging the CI workers in case of deadlocks. Timeout running N)rwarn terminatecmdprocsrterminate_processz0check_subprocess_call..terminate_process0s4  .../// rc^tjddS)z Kill a leftover process spawned during test execution: ideally this should not be needed but can help avoid clogging the CI workers in case of deadlocks. rN)rrkillrsr kill_processz+check_subprocess_call..kill_process9s/  .../// rNrz/Non-zero return code: {}. Stdout: {} Stderr: {}z,Unexpected stdout: {!r} does not match: {!r}z,Unexpected stderr: {!r} does not match: {!r}) subprocessPopenPIPE threadingTimerstart communicatedecode returncodeformat ValueErrorresearchcancel) rtimeout stdout_regex stderr_regexr r#terminate_timer kill_timerr rmessagers ` @rcheck_subprocess_callr9%s  C   P P PD  'og7HIIO  ! ! # # #"WlCCJ      ))++&--// ?a  LTTGW%% %  #BIlF,K,K #?FF &   #BIlF,K,K #?FF &     " " $ $ $          7   " " $ $ $        s EF))-G)rNN)__doc__os.pathrr0r%rr(r_pytestpytestraiseswarnsrunnerSkippedSkipTestmarkskipiffixture parametrizer3xfailparamrr9rrrJs     > !   .k% +   '''; ; ; ; ; ; r