File manager - Edit - /home/newsbmcs.com/public_html/static/img/logo/libregrtest.tar
Back
win_utils.py 0000644 00000010713 15030171176 0007136 0 ustar 00 import _overlapped import _thread import _winapi import math import struct import winreg # Seconds per measurement SAMPLING_INTERVAL = 1 # Exponential damping factor to compute exponentially weighted moving average # on 1 minute (60 seconds) LOAD_FACTOR_1 = 1 / math.exp(SAMPLING_INTERVAL / 60) # Initialize the load using the arithmetic mean of the first NVALUE values # of the Processor Queue Length NVALUE = 5 class WindowsLoadTracker(): """ This class asynchronously reads the performance counters to calculate the system load on Windows. A "raw" thread is used here to prevent interference with the test suite's cases for the threading module. """ def __init__(self): # Pre-flight test for access to the performance data; # `PermissionError` will be raised if not allowed winreg.QueryInfoKey(winreg.HKEY_PERFORMANCE_DATA) self._values = [] self._load = None self._running = _overlapped.CreateEvent(None, True, False, None) self._stopped = _overlapped.CreateEvent(None, True, False, None) _thread.start_new_thread(self._update_load, (), {}) def _update_load(self, # localize module access to prevent shutdown errors _wait=_winapi.WaitForSingleObject, _signal=_overlapped.SetEvent): # run until signaled to stop while _wait(self._running, 1000): self._calculate_load() # notify stopped _signal(self._stopped) def _calculate_load(self, # localize module access to prevent shutdown errors _query=winreg.QueryValueEx, _hkey=winreg.HKEY_PERFORMANCE_DATA, _unpack=struct.unpack_from): # get the 'System' object data, _ = _query(_hkey, '2') # PERF_DATA_BLOCK { # WCHAR Signature[4] 8 + # DWOWD LittleEndian 4 + # DWORD Version 4 + # DWORD Revision 4 + # DWORD TotalByteLength 4 + # DWORD HeaderLength = 24 byte offset # ... # } obj_start, = _unpack('L', data, 24) # PERF_OBJECT_TYPE { # DWORD TotalByteLength # DWORD DefinitionLength # DWORD HeaderLength # ... # } data_start, defn_start = _unpack('4xLL', data, obj_start) data_base = obj_start + data_start defn_base = obj_start + defn_start # find the 'Processor Queue Length' counter (index=44) while defn_base < data_base: # PERF_COUNTER_DEFINITION { # DWORD ByteLength # DWORD CounterNameTitleIndex # ... [7 DWORDs/28 bytes] # DWORD CounterOffset # } size, idx, offset = _unpack('LL28xL', data, defn_base) defn_base += size if idx == 44: counter_offset = data_base + offset # the counter is known to be PERF_COUNTER_RAWCOUNT (DWORD) processor_queue_length, = _unpack('L', data, counter_offset) break else: return # We use an exponentially weighted moving average, imitating the # load calculation on Unix systems. # https://en.wikipedia.org/wiki/Load_(computing)#Unix-style_load_calculation # https://en.wikipedia.org/wiki/Moving_average#Exponential_moving_average if self._load is not None: self._load = (self._load * LOAD_FACTOR_1 + processor_queue_length * (1.0 - LOAD_FACTOR_1)) elif len(self._values) < NVALUE: self._values.append(processor_queue_length) else: self._load = sum(self._values) / len(self._values) def close(self, kill=True): self.__del__() return def __del__(self, # localize module access to prevent shutdown errors _wait=_winapi.WaitForSingleObject, _close=_winapi.CloseHandle, _signal=_overlapped.SetEvent): if self._running is not None: # tell the update thread to quit _signal(self._running) # wait for the update thread to signal done _wait(self._stopped, -1) # cleanup events _close(self._running) _close(self._stopped) self._running = self._stopped = None def getloadavg(self): return self._load __pycache__/runtest.cpython-310.pyc 0000644 00000025102 15030171176 0013162 0 ustar 00 o }�5h1 � @ sB d dl Z d dlZd dlZd dlZd dlZd dlZd dlZd dlZd dlZd dl Z d dl m Z d dl m Z d dlmZ d dlmZ d dlmZmZmZ G dd� d�ZG d d � d e�ZG d d � d e�ZG d d� de�ZG dd� de�ZG dd� de�ZG dd� de�Z G dd� de �Z G dd� de�Z G dd� de�Z G dd � d e�Z G d d � d e�Z!d Z"g d �Z#e$� Z%g Z&d!ed"ed#e'fd$d%�Z(d>d&d'�Z)de#e%fd(d)�Z*d"ed*e+d#e+fd+d,�Z,d"ed*e+d#efd-d.�Z-d"ed*e+d#efd/d0�Z.d1d2� Z/d"ed*e+fd3d4�Z0d"ed*e+d#e'fd5d6�Z1 7d?d"ed*e+d8e'd#efd9d:�Z2d*e+d;e3d#dfd<d=�Z4dS )@� N)�support)� os_helper)� Namespace)�saved_test_environment)� clear_caches�format_duration� print_warningc @ sB e Zd Z d dededee dB ddfdd�Zdefd d �ZdS ) � TestResult� N�name� duration_sec�xml_data�returnc C s || _ || _|| _d S �N)r r r )�selfr r r � r �//usr/lib/python3.10/test/libregrtest/runtest.py�__init__ s zTestResult.__init__c C � | j � d�S )Nz finished�r �r r r r �__str__ � zTestResult.__str__)r N)�__name__� __module__� __qualname__�str�float�listr r r r r r r s ��� � � r c @ � e Zd Zdefdd�ZdS )�Passedr c C r )Nz passedr r r r r r # r zPassed.__str__N�r r r r r r r r r r " � r c sv e Zd Z d dededee dB deeeef dB deeeef dB ddf � fd d � Zdefd d �Z� Z S )�Failedr Nr r r �errors�failuresr c s" t � j|||d� || _|| _d S )N)r r )�superr r$ r% )r r r r r$ r% �� __class__r r r ( s zFailed.__init__c C s� | j r6| jr6t| j �}t| j�}d|dkrdnd }d|dkr!dnd }| j� d|� d|� d|� d|� d � S | j rTt| j �}d|dkrEdnd }| j� d|� d|� d �S | jrrt| j�}d|dkrcdnd }| j� d|� d|� d �S | j� d �S ) N�error� �s� �failurez failed (� z, �)� failed)r$ r% �lenr )r �le�lf�error_s� failure_sr r r r 4 s $ zFailed.__str__)r NNN) r r r r r r �tupler r � __classcell__r r r' r r# ' s& ��� ���� r# c @ r )�UncaughtExceptionr c C r )Nz failed (uncaught exception)r r r r r r J r zUncaughtException.__str__Nr! r r r r r8 I r" r8 c @ r )� EnvChangedr c C r )Nz failed (env changed)r r r r r r O r zEnvChanged.__str__Nr! r r r r r9 N r" r9 c @ r )�RefLeakr c C r )Nz failed (reference leak)r r r r r r T r zRefLeak.__str__Nr! r r r r r: S r" r: c @ r )�Skippedr c C r )Nz skippedr r r r r r Y r zSkipped.__str__Nr! r r r r r; X r" r; c @ r )�ResourceDeniedr c C r )Nz skipped (resource denied)r r r r r r ^ r zResourceDenied.__str__Nr! r r r r r< ] r" r< c @ r )� Interruptedr c C r )Nz interruptedr r r r r r c r zInterrupted.__str__Nr! r r r r r= b r" r= c @ r )� ChildErrorr c C r )Nz crashedr r r r r r h r zChildError.__str__Nr! r r r r r> g r" r> c @ r )� DidNotRunr c C r )Nz ran no testsr r r r r r m r zDidNotRun.__str__Nr! r r r r r? l r" r? c @ r )�Timeoutr c C s | j � dt| j�� d�S )Nz timed out (r/ )r r r r r r r r r s zTimeout.__str__Nr! r r r r r@ q r" r@ g >@) � test_grammar� test_opcodes� test_dict� test_builtin�test_exceptions� test_types� test_unittest� test_doctest� test_doctest2� test_support�result�nsr c C s t | t�r|jS t | t�S r )� isinstancer9 �fail_env_changedr# )rK rL r r r � is_failed� s rO c C s | pt j�t j�t��pt jS r )�os�path�dirname�__file__�curdir)rQ r r r � findtestdir� s rU c C sr t | �} t�| �}g }t|�|B }|D ] }tj�|�\}}|dd� dkr2|dv r2||vr2|�|� q|t|� S )z-Return a list of all applicable test modules.N� �test_)z.pyr, )rU rP �listdir�setrQ �splitext�append�sorted) �testdir�stdtests�nottests�names�tests�othersr �mod�extr r r � findtests� s � re � test_namec C s |� d�s| jr |S d| S )Nztest.)� startswithr] �rL rf r r r �get_abs_module� s ri c sX | j }| jd u}|rtj| jdd� t�� }z�t�| j| j � | j r$g nd t_ }| j r/dt_ |rmdt_ t�� }tj}tj}z'|t_|t_t| |dd�} t| t�s^|�� } |�| � |�� W |t_|t_n|t_|t_w | j t_ t| || j d�} |r�dd lm m� � fdd�|D �| _ t�� | | _ | W |r�t� � d t_ S |r�t� � d t_ w )NT)�exitF)�display_failurer c s g | ] }� � |��d ��qS )zus-ascii)�tostring�decode)�.0�x��ETr r � <listcomp>� s ��z _runtest.<locals>.<listcomp>) �verbose3�timeout� faulthandler�dump_traceback_later�time� perf_counterr �set_match_tests� match_tests� ignore_tests�xmlpath�junit_xml_list�failfast�verbose�io�StringIO�sys�stdout�stderr�_runtest_innerrM r �getvalue�write�flush�xml.etree.ElementTree�etree� ElementTreer r �cancel_dump_traceback_later) rL rf �output_on_failure� use_timeout� start_time�xml_list�stream� orig_stdout� orig_stderrrK �outputr rp r �_runtest� s^ � ��� ��r� c C sJ zt | |�W S | js t�� }td|� d|� �tjdd� t|� Y S )a Run a single test. ns -- regrtest namespace of options test_name -- the name of the test Returns a TestResult sub-class depending on the kind of result received. If ns.xmlpath is not None, xml_data is a list containing each generated testsuite element. �test � crashed -- T��filer� )r� �pgo� traceback� format_exc�printr� r� r# )rL rf �msgr r r �runtest� s � r� c C sH t �� }|�| �}|jD ] }t|tjd� q |jr td��t� |� d S )N)r� zerrors while loading tests) �unittest� TestLoader�loadTestsFromModuler$ r� r� r� � Exceptionr � run_unittest)� the_module�loaderra r) r r r � _test_module� s r� c C s t || j| j| jd�S )N)r� )r r �quietr� rh r r r �save_env s r� c C s t | |�}ztj|= W n ty Y nw t�|�}| jr#ddlm} t |dd �}|d u r3t � t |�}z,t | |�� | jrD|| ||�}n|� d}W d � n1 sSw Y W t|| j� nt|| j� w t�� tjr�dt_t|� dttj�� d�� t�tj� tj�� t�� |S )Nr )�dash_R� test_mainFTz created z uncollectable object(s).)ri r� �modules�KeyError� importlib� import_module� huntrleaks�test.libregrtest.refleakr� �getattr� functools�partialr� r� �cleanup_test_droppingsr r � gc_collect�gc�garbage�environment_alteredr r1 � FOUND_GARBAGE�extend�clear� reap_children)rL rf �abstestr� r� � test_runner�refleakr r r �_runtest_inner2 s< � � �� r� Trk c C sF dt _| jrd}z t� t| |�� t| |�}W d � n1 s!w Y W n� t jyO } z| jsA| jsAt|� d|� �dd� t|�W Y d }~S d }~w t j yv } z| jsh| jsht|� d|� �dd� t |�W Y d }~S d }~w t j y� } z&d|� d�}|r�|� d|� �}t|t jdd� t||j|jd �W Y d }~S d }~w t jy� } z!d|� d�}|r�|� d|� �}t|t jdd� t|�W Y d }~S d }~w t jy� t|� Y S ty� t� t|� Y S | j�s t�� }td|� d |� �t jdd� t|� Y S |�rt|�S t j�r t|�S t |�S ) NFz skipped -- T)r� r� r0 z -- r� )r$ r% r� ) r r� r� r r� r� r< r� r� r� �SkipTestr; �TestFailedWithDetailsr� r� r# r$ r% � TestFailed� TestDidNotRunr? �KeyboardInterruptr= r� r� r8 r: r9 r )rL rf rk r� r� �excr r r r� ? sd �� � � � � � r� r c C s t �� tjfD ]x}tj�|�sqtj�|�r"dd l}d|j }}ntj� |�r/dtj }}nt d|�d���|rHt | � d|� d|��� dt _zdd l}t�||j|jB |jB � ||� W q ty� } zt | � d|� d|�d |� �� W Y d }~qd }~ww d S ) Nr � directoryr� z os.path says z) exists but is neither directory nor filez left behind r. Tz and it couldn't be removed: )r r� r �TESTFNrP rQ �exists�isdir�shutil�rmtree�isfile�unlink� RuntimeErrorr r� �stat�chmod�S_IRWXU�S_IRWXG�S_IRWXOr� )rf r r r� �kind�nukerr� r� r r r r� w s0 ����r� r )T)5ru r� r� r� r� rP r� rw r� r� �testr � test.supportr �test.libregrtest.cmdliner �test.libregrtest.save_envr �test.libregrtest.utilsr r r r r r# r8 r9 r: r; r< r= r> r? r@ �PROGRESS_MIN_TIME�STDTESTSrY �NOTTESTSr� �boolrO rU re r ri r� r� r� r� r� r� �intr� r r r r �<module> sf " 9 4���� �8 __pycache__/save_env.cpython-310.pyc 0000644 00000034677 15030171176 0013305 0 ustar 00 o }�5h�2 � @ sn d dl Z d dlZd dlZd dlZd dlZd dlmZ d dlmZ d dl m Z G dd� de �Z G dd� d�Z dS ) � N)�support)� os_helper)� print_warningc @ s e Zd ZdS )�SkipTestEnvironmentN)�__name__� __module__� __qualname__� r r �0/usr/lib/python3.10/test/libregrtest/save_env.pyr s r c @ s e Zd ZdZd�dd�dd�ZdZdd � Zd d � Zd d � Zdd� Z dd� Z dd� Z dd� Z dd� Z dd� Zdd� Zd d � Zd d � Zd d!� Zd"d#� Zd$d%� Zd&d'� Zd(d)� Zd*d+� Zd,d-� Zd.d/� Zd0d1� Zd2d3� Zd4d5� Z d6d7� Z d8d9� Z d:d;� Z d<d=� Z d>d?� Z!d@dA� Z"dBdC� Z#dDdE� Z$dFdG� Z%dHdI� Z&dJdK� Z'dLdM� Z(dNdO� Z)dPdQ� Z*dRdS� Z+dTdU� Z,dVdW� Z-dXdY� Z.dZd[� Z/d\d]� Z0d^d_� Z1d`da� Z2dbdc� Z3ddde� Z4dfdg� Z5dhdi� Z6djdk� Z7dldm� Z8dndo� Z9dpdq� e:e;�D �Z<drds� Z=dtdu� Z>dvdw� Z?dxdy� Z@dzd{� ZAd|d}� ZBd~d� ZCd�S )��saved_test_environmenta� Save bits of the test environment and restore them at block exit. with saved_test_environment(testname, verbose, quiet): #stuff Unless quiet is True, a warning is printed to stderr if any of the saved items was changed by the test. The support.environment_altered attribute is set to True if a change is detected. If verbose is more than 1, the before and after state of changed items is also printed. r F)�pgoc C s || _ || _|| _|| _d S �N)�testname�verbose�quietr )�selfr r r r r r r �__init__% s z saved_test_environment.__init__)zsys.argv�cwdz sys.stdinz sys.stdoutz sys.stderrz os.environzsys.pathzsys.path_hooks� __import__zwarnings.filterszasyncore.socket_mapzlogging._handlerszlogging._handlerListz sys.gettracezsys.warnoptionsz!multiprocessing.process._danglingzthreading._danglingzsysconfig._CONFIG_VARSzsysconfig._INSTALL_SCHEMES�files�localezwarnings.showwarning�shutil_archive_formats�shutil_unpack_formatsz!asyncio.events._event_loop_policyz urllib.requests._url_tempfileszurllib.requests._openerc C s t j| S r )�sys�modules�r �namer r r � get_moduleD � z!saved_test_environment.get_modulec C s z| � |�W S ty t�w r )r �KeyErrorr r r r r �try_get_moduleH s �z%saved_test_environment.try_get_modulec C s | � d�}t|j�S �Nzurllib.request)r �list�_url_tempfiles�r �urllib_requestr r r �"get_urllib_requests__url_tempfilesO � z9saved_test_environment.get_urllib_requests__url_tempfilesc C s |D ]}t �|� qd S r )r �unlink)r � tempfiles�filenamer r r �&restore_urllib_requests__url_tempfilesR s �z=saved_test_environment.restore_urllib_requests__url_tempfilesc C � | � d�}|jS r! )r �_openerr$ r r r �get_urllib_requests__openerV � z2saved_test_environment.get_urllib_requests__openerc C � | � d�}||_d S r! )r r- )r �openerr% r r r � restore_urllib_requests__openerY r' z6saved_test_environment.restore_urllib_requests__openerc C s | � d� t�� S �N�asyncio)r r �maybe_get_event_loop_policy�r r r r �%get_asyncio_events__event_loop_policy] s z<saved_test_environment.get_asyncio_events__event_loop_policyc C s | � d�}|�|� d S r3 )r �set_event_loop_policy)r �policyr4 r r r �)restore_asyncio_events__event_loop_policy` s z@saved_test_environment.restore_asyncio_events__event_loop_policyc C � t tj�tjtjd d � fS r )�idr �argvr6 r r r � get_sys_argvd � z#saved_test_environment.get_sys_argvc C � |d t _|d t jd d �< d S �N� � )r r= )r � saved_argvr r r �restore_sys_argvf � z'saved_test_environment.restore_sys_argvc C � t �� S r )�os�getcwdr6 r r r �get_cwdj � z saved_test_environment.get_cwdc C � t �|� d S r )rH �chdir)r � saved_cwdr r r � restore_cwdl � z"saved_test_environment.restore_cwdc C � t jS r �r �stdoutr6 r r r �get_sys_stdouto � z%saved_test_environment.get_sys_stdoutc C � |t _d S r rR )r � saved_stdoutr r r �restore_sys_stdoutq � z)saved_test_environment.restore_sys_stdoutc C rQ r �r �stderrr6 r r r �get_sys_stderrt rU z%saved_test_environment.get_sys_stderrc C rV r rZ )r � saved_stderrr r r �restore_sys_stderrv rY z)saved_test_environment.restore_sys_stderrc C rQ r �r �stdinr6 r r r � get_sys_stdiny rU z$saved_test_environment.get_sys_stdinc C rV r r_ )r � saved_stdinr r r �restore_sys_stdin{ rY z(saved_test_environment.restore_sys_stdinc C s t tj�tjttj�fS r )r< rH �environ�dictr6 r r r �get_os_environ~ s z%saved_test_environment.get_os_environc C s( |d t _t j�� t j�|d � d S rA )rH rd �clear�update)r � saved_environr r r �restore_os_environ� s z)saved_test_environment.restore_os_environc C r; r )r< r �pathr6 r r r � get_sys_path� r? z#saved_test_environment.get_sys_pathc C r@ rA )r rk )r � saved_pathr r r �restore_sys_path� rF z'saved_test_environment.restore_sys_pathc C r; r )r< r � path_hooksr6 r r r �get_sys_path_hooks� r? z)saved_test_environment.get_sys_path_hooksc C r@ rA )r ro )r � saved_hooksr r r �restore_sys_path_hooks� rF z-saved_test_environment.restore_sys_path_hooksc C rG r )r �gettracer6 r r r �get_sys_gettrace� rK z'saved_test_environment.get_sys_gettracec C rL r )r �settrace)r � trace_fxnr r r �restore_sys_gettrace� rP z+saved_test_environment.restore_sys_gettracec C rQ r ��builtinsr r6 r r r �get___import__� rU z%saved_test_environment.get___import__c C rV r rx )r �import_r r r �restore___import__� rY z)saved_test_environment.restore___import__c C �& | � d�}t|j�|j|jd d � fS �N�warnings)r r< �filters�r r r r r �get_warnings_filters� s z+saved_test_environment.get_warnings_filtersc C s* | � d�}|d |_|d |jd d �< d S )Nr rB rC )r r� )r � saved_filtersr r r r �restore_warnings_filters� s z/saved_test_environment.restore_warnings_filtersc C s t j�d�}|r |j�� pi S )N�asyncore)r r �get� socket_map�copy)r r� r r r �get_asyncore_socket_map� s z.saved_test_environment.get_asyncore_socket_mapc C s4 t j�d�}|d ur|jdd� |j�|� d S d S )Nr� T)� ignore_all)r r r� � close_allr� rh )r � saved_mapr� r r r �restore_asyncore_socket_map� s �z2saved_test_environment.restore_asyncore_socket_mapc C � | � d�}|j|j�� fS �N�shutil)r �_ARCHIVE_FORMATSr� �r r� r r r �get_shutil_archive_formats� s z1saved_test_environment.get_shutil_archive_formatsc C �2 | � d�}|d |_|j�� |j�|d � d S �Nr� r rB )r r� rg rh �r �savedr� r r r � restore_shutil_archive_formats� � z5saved_test_environment.restore_shutil_archive_formatsc C r� r� )r �_UNPACK_FORMATSr� r� r r r �get_shutil_unpack_formats� � z0saved_test_environment.get_shutil_unpack_formatsc C r� r� )r r� rg rh r� r r r � restore_shutil_unpack_formats� r� z4saved_test_environment.restore_shutil_unpack_formatsc C �" | � d�}t|j�|j|j�� fS �N�logging)r r< � _handlersr� �r r� r r r �get_logging__handlers� s z,saved_test_environment.get_logging__handlersc C � d S r r )r �saved_handlersr r r �restore_logging__handlers� � z0saved_test_environment.restore_logging__handlersc C r} r� )r r< � _handlerListr� r r r �get_logging__handlerList� s z/saved_test_environment.get_logging__handlerListc C r� r r )r �saved_handlerListr r r � restore_logging__handlerList� r� z3saved_test_environment.restore_logging__handlerListc C r; r )r< r � warnoptionsr6 r r r �get_sys_warnoptions� r? z*saved_test_environment.get_sys_warnoptionsc C r@ rA )r r� )r � saved_optionsr r r �restore_sys_warnoptions� rF z.saved_test_environment.restore_sys_warnoptionsc C s t j�� S r )� threading� _danglingr� r6 r r r �get_threading__dangling� r z.saved_test_environment.get_threading__danglingc C s t j�� t j�|� d S r )r� r� rg rh )r r� r r r �restore_threading__dangling� r� z2saved_test_environment.restore_threading__danglingc C s | � d�}|�� |j�� S �Nzmultiprocessing.process)r �_cleanupr� r� )r �multiprocessing_processr r r �%get_multiprocessing_process__dangling� s z<saved_test_environment.get_multiprocessing_process__danglingc C s$ | � d�}|j�� |j�|� d S r� )r r� rg rh )r r� r� r r r �)restore_multiprocessing_process__dangling� s z@saved_test_environment.restore_multiprocessing_process__danglingc C s, | � d�}|�d� t|j�|jt|j�fS )N� sysconfig�prefix)r �get_config_varr< � _CONFIG_VARSre �r r� r r r �get_sysconfig__CONFIG_VARS� s �z1saved_test_environment.get_sysconfig__CONFIG_VARSc C r� �Nr� rB rC )r r� rg rh �r r� r� r r r � restore_sysconfig__CONFIG_VARS� r� z5saved_test_environment.restore_sysconfig__CONFIG_VARSc C r� )Nr� )r r< �_INSTALL_SCHEMESr� r� r r r � get_sysconfig__INSTALL_SCHEMES� s �z5saved_test_environment.get_sysconfig__INSTALL_SCHEMESc C r� r� )r r� rg rh r� r r r �"restore_sysconfig__INSTALL_SCHEMES� r� z9saved_test_environment.restore_sysconfig__INSTALL_SCHEMESc C s t dd� t�� D ��S )Nc s s( � | ]}|t j�|�r d nd V qdS )�/� N)rH rk �isdir)�.0�fnr r r � <genexpr> s � �z3saved_test_environment.get_files.<locals>.<genexpr>)�sortedrH �listdirr6 r r r � get_files s �z saved_test_environment.get_filesc C sZ t j}||vr'|d |vr)tj�|�rt �|� d S tj�|�r+t �|� d S d S d S d S )Nr� )r �TESTFNrH rk �isfiler( r� �rmtree)r � saved_valuer� r r r � restore_files s �z$saved_test_environment.restore_filesc C s g | ] }|� d �rtt|��qS )�LC_)� startswith�getattrr )r� �lcr r r � <listcomp> s �z!saved_test_environment.<listcomp>c C sD g }| j D ]}z |�|t�|d �f� W q ttfy Y qw |S r )�_lc�appendr � setlocale� TypeError� ValueError)r �pairingsr� r r r � get_locale s �z!saved_test_environment.get_localec C s |D ] \}}t �||� qd S r )r r� )r r� r� �settingr r r �restore_locale s �z%saved_test_environment.restore_localec C r, r~ )r � showwarningr� r r r �get_warnings_showwarning r/ z/saved_test_environment.get_warnings_showwarningc C r0 r~ )r r� )r �fxnr r r r � restore_warnings_showwarning r' z3saved_test_environment.restore_warnings_showwarningc c sF � | j D ] }|�dd�}d| }d| }|t| |�t| |�fV qd S )N�.�_�get_�restore_)� resources�replacer� )r r � method_suffix�get_name� restore_namer r r � resource_info# s � �z$saved_test_environment.resource_infoc C sP g | _ | �� D ] \}}}z|� }W n ty Y qw | j �||||f� q| S r )� saved_valuesr� r r� )r r r� �restore�originalr r r � __enter__* s �z saved_test_environment.__enter__c C s� | j }d | _ t�� |D ]3\}}}}|� } | |kr?dt_||� | js?| js?t|� d| j� �� td|� d| � d�t j dd� q dS )NTz was modified by z Before: z After: � )�file�flushF) r� r � gc_collect�environment_alteredr r r r �printr r[ ) r �exc_type�exc_val�exc_tbr� r r� r� r� �currentr r r �__exit__5 s ��z saved_test_environment.__exit__N)r F)Dr r r �__doc__r r� r r r& r+ r. r2 r7 r: r> rE rJ rO rT rX r\ r^ ra rc rf rj rl rn rp rr rt rw rz r| r� r� r� r� r� r� r� r� r� r� r� r� r� r� r� r� r� r� r� r� r� r� r� r� �dirr r� r� r� r� r� r� r� r� r r r r r s� r )ry r rH r r� �testr � test.supportr �test.libregrtest.utilsr � Exceptionr r r r r r �<module> s __pycache__/__init__.cpython-310.pyc 0000644 00000000454 15030171176 0013220 0 ustar 00 o }�5hw � @ s$ d dl mZmZmZ d dlmZ dS )� )� _parse_args�RESOURCE_NAMES� ALL_RESOURCES)�mainN)�test.libregrtest.cmdliner r r �test.libregrtest.mainr � r r �0/usr/lib/python3.10/test/libregrtest/__init__.py�<module> s __pycache__/main.cpython-310.pyc 0000644 00000042220 15030171176 0012402 0 ustar 00 o }�5h�d � @ s d dl Z d dlZd dlZd dlZd dlZd dlZd dlZd dlZd dlZd dl Z d dl Z d dl m Z d dl mZmZmZmZmZmZmZmZmZmZmZmZmZmZm Z d dl m Z d dl m Z d dl!m"Z"m#Z#m$Z$m%Z% d dl&m'Z' d dl(m)Z) d Z*G d d � d �Z+dd d �Z,dS )� N)� _parse_args)� findtests�runtest�get_abs_module� is_failed�STDTESTS�NOTTESTS�PROGRESS_MIN_TIME�Passed�Failed� EnvChanged�Skipped�ResourceDenied� Interrupted� ChildError� DidNotRun)� setup_tests)�setup_pgo_tests)�removepy�count�format_duration� printlist)�support)� os_helperg ^@c @ s� e Zd ZdZdd� Zdd� Zd7dd�Zd8d d �Zd d � Zdd� Z dd� Z dd� Z dd� Z dd� Z dd� Zdd� Zd d � Zd d � Zd d!� Zd"d#� Zd$d%� Zd&d'� Zd(d)� Zd*d+� Zd,d-� Zd.d/� Zd9d1d2�Zd3d4� Zd5d6� Z d0S ):�Regrtestal Execute a test suite. This also parses command-line options and modifies its behavior accordingly. tests -- a list of strings containing test names (optional) testdir -- the directory in which to look for tests (optional) Users other than the Python test suite will certainly want to specify testdir; if it's omitted, the directory containing the Python test suite is searched for. If the tests argument is omitted, the tests listed on the command-line will be used. If that's empty, too, then all *.py files beginning with test_ will be used. The other default arguments (verbose, quiet, exclude, single, randomize, findleaks, use_resources, trace, coverdir, print_slow, and random_seed) allow programmers calling main() directly to set the values that would normally be set by flags on the command line. c C s� d | _ g | _g | _g | _g | _g | _g | _g | _g | _g | _ g | _ d | _ d| _ g | _ d | _t�� | _d| _d| _d | _d | _d | _d | _d | _d | _d S )NF� � )�ns�tests�selected�good�bad�skipped�resource_denieds�environment_changed� run_no_tests� need_rerun�rerun� first_result� interrupted� test_times�tracer�time� monotonic� start_time� test_count�test_count_width�next_single_test�next_single_filename� testsuite_xml�win_load_tracker�tmp_dir�worker_test_name��self� r9 �,/usr/lib/python3.10/test/libregrtest/main.py�__init__6 s0 zRegrtest.__init__c C s< t | j�t | j�B t | j�B t | j�B t | j�B t | j�B S �N)�setr r! r" r# r$ r% r7 r9 r9 r: � get_executeda s ���zRegrtest.get_executedFc C sv |j }t|ttf�s|s| j�|j|f� t|t�r!| j�|� nYt|t �r3| j �|� | j �|� nGt|t �r?| j �|� n;t|t �rK| j�|� n/t|t�r_|s^| j�|� | j�|� nt|t�rk| j�|� nt|t�rtd| _ntd| ��|r�t|ttf�s�| j�|� |j}|r�dd lm m} |D ] }z | j�|� |�� W q� |j y� t |t j d� � w d S d S )NTzinvalid test result: %rr ��file)!�name� isinstancer r r* �append� duration_secr r r r" r# r r r$ r r! r& r r% r) � ValueError�remove�xml_data�xml.etree.ElementTree�etree� ElementTreer3 � fromstring� ParseError�print�sys� __stderr__)r8 �resultr' � test_namerG �ET�er9 r9 r: �accumulate_resultf sH � ��zRegrtest.accumulate_resultr c C s� | }| � � }|d urd|d�d|� �}t�� | j }tt|�d�\}}t|d�\}}d|||f }|� d|� �}|rA|d d� }t|dd� d S ) Nz load avg: z.2f� �< z %d:%02d:%02d���T)�flush)� getloadavgr, r- r. �divmod�intrM )r8 �line�empty�load_avg� test_time�mins�secs�hoursr9 r9 r: �log� s z Regrtest.logc C sh | j jrd S || j� �| j� �}t| j�t| j� }|r'| j js'|� d|� �}| �d|� d|� �� d S )N�/�[z] ) r �quietr0 r/ �lenr! r$ �pgorc )r8 � test_index�textr\ �failsr9 r9 r: �display_progress� s zRegrtest.display_progressc C s� t tjdd � fi |��}|jrg t_| _|j}|d ur0ddlm } ||j�\}}||_|| _ t |j � |j rU|j \}}}|dk sF|dk rUd} t| tjdd� t�d� |jr`tj�|j�|_|| _d S )Nr r )�parse_worker_argszwInvalid values for the --huntrleaks/-R parameters. The number of warmups and repetitions must be at least 1 each (1:1).T�r@ rX � )r rN �argv�xmlpathr �junit_xml_listr3 � worker_args�test.libregrtest.runtest_mprm r6 r �args� huntrleaksrM �stderr�exit�tempdir�os�path� expanduserr ) r8 �kwargsr rs rm rQ �warmup� repetitions�_�msgr9 r9 r: � parse_args� s& zRegrtest.parse_argsc C s� || _ | jjr<tj�| jd�| _z"t| jd��}|� � � � }|g| _ W d � n1 s,w Y W n t y; Y nw | jj r�g | _ t �d�}ttj�tj| jj ���*}|D ] }|�dd�d }|� � }|�|�}|d urv| j �|�� � qWW d � n1 s�w Y t| j � | jjr�t| j� td d � }t�� }| jjr�| jj D ]} | |v r�|� | � |� | � q�g | j_ | jj r�t | jj t!� t"� �} nt | jj ||�} | jj s�| j p�| jj p�| | _#n| j | _#| jj�r| j#d d� | _#z| �$| j#d �} | | d | _%W n t&�y Y nw | jj'�r7z| j#d | j#�$| jj'��= W n t(�y6 t)d| jj' t*j+d� Y nw | jj,�rY| jj-d u �rJt.�/d �| j_-t.�0| jj-� t.�1| j#� d S d S ) N� pynexttest�rz\btest_[a-zA-Z0-9_]+\b�#r r z1Couldn't find starting test (%s), using all testsr? i��� )2r r �singlerz r{ �joinr5 r2 �open�read�strip�OSError�fromfile�re�compiler �SAVEDCWD�split�searchrC �groupr rh r r r �copy�excluderu rF �add�testdirr �listr= r �indexr1 � IndexError�startrE rM rN rw � randomize� random_seed�random� randrange�seed�shuffle) r8 r �fp� next_test�regexr\ �match�stdtests�nottests�arg�alltests�posr9 r9 r: � find_tests� s� �� � �� � � � �� �zRegrtest.find_testsc C s | j D ]}t|� qd S r< )r rM )r8 rA r9 r9 r: � list_tests s �zRegrtest.list_testsc C sX |D ]'}t |tjj�r qt |tj�r| �|� qt |tj�r)t�|�r)t |� � � qd S r< ) rB �unittest�loader� _FailedTest� TestSuite� _list_cases�TestCaser � match_testrM �id)r8 �suite�testr9 r9 r: r� s ��zRegrtest._list_casesc C s� dt _t �| jj| jj� | jD ]%}t| j|�}z tj � |�}| � |� W q tj y5 | j �|� Y qw | j rWttjd� ttt| j �d�dtjd� t| j tjd� d S d S )NFr? r� �skipped:)r �verbose�set_match_testsr � match_tests� ignore_testsr r r� �defaultTestLoader�loadTestsFromNamer� �SkipTestr" rC rM rN rw r rg r )r8 rQ �abstestr� r9 r9 r: � list_cases" s � �zRegrtest.list_casesc C s| d| j _d| j _d| j _| �� | _| �� | �d� t| j�}| j� � |D ]~}|j }| j � |� |j p5g }|jp:g }dd� |D �}dd� |D �}d| j _| j j}|sU|r�| j jd u r_g | j _| j j�|� | j j�|� dd�| j j� } | �d |� d | � d �� n | �d |� d �� t| j |�}|| j _| j|dd � t|t�r� nq&| jr�ttt| j�d�d� t| j� | �� d S )NTFz'Re-running failed tests in verbose modec S � g | ] ^}}|� d �d �qS �rU r �r� ��.0�test_full_namer� r9 r9 r: � <listcomp>D � z/Regrtest.rerun_failed_tests.<locals>.<listcomp>c S r� r� r� r� r9 r9 r: r� E r� z matching: �, z Re-running z in verbose mode (�)z in verbose mode)r' r� z failed again:) r r� �failfast�verbose3�get_tests_resultr( rc r� r& �clearrA r' rC �errors�failuresr� �extendr� r rT rB r r! rM r rg r �display_result) r8 � rerun_listrP rQ r� r� � error_names� failure_names�orig_match_tests�matchingr9 r9 r: �rerun_failed_tests3 sF � zRegrtest.rerun_failed_testsc C s | j jrd S t� td| �� � | jrtd� t| j�| �� }|r4t� ttt |�d�d� t |� | j r_| j j s_t� | j sT| jsT| jsTt | j �dkrTtddd� ttt | j �d�d � | j jr�| jjd d � t� td � | jd d � D ]\}}td|t|�f � qx| j r�t� ttt | j �d�d� t | j � | jr�t� td�tt | j�d��� t | j� | jr�| j j s�t� ttt | j�d�d� t | j� | jr�t� tdtt | j�d� � t | j� | jr�t� ttt | j�d�d� t | j� d S d S )Nz== Tests result: %s ==z(Test suite interrupted by signal SIGINT.r� zomitted:r �AllrU )�endzOK.T)�reversez10 slowest tests:� z- %s: %szfailed:z%{} altered the execution environment:r� z%s:z re-run testz run no tests:)r rh rM r� r) r= r r> r rg r r rf r! r" � print_slowr* �sortr r$ �formatr' r% )r8 �omittedr_ r� r9 r9 r: r� _ sd �� � �zRegrtest.display_resultc C s� | j jrdd l}|jddd�| _tj�� }d}| j jr%|dt| j j� 7 }| � |� d }t | j d�D ]�\}}t � � }|}|rDd||f }| �||� | jrcd } tt� �} | jj| t� | d � | d } n t| j |�} | �| � t| t�ru nAt| �}t � � | } | tkr�d |t| �f }nt| t�r�d }tj�� D ]} | |vr�| �d �r�t� | � q�| j j r�t | | j �r� nq2|r�t |� d S d S )Nr FT)�tracer zRun tests sequentiallyz (timeout: %s)r z%s -- %szDresult = runtest(self.ns, test_name); self.accumulate_result(result))�globals�localsrP z%s in %sztest.) r r� �Tracer+ rN �modules�keys�timeoutr rc � enumerater r, r- rl �dictr� �runctxr� r rT rB r �strr r � startswithr �unloadr� r rM )r8 r� � save_modulesr� � previous_testri rQ r. rj �cmdr rP r_ �moduler9 r9 r: �run_tests_sequential� sP �� �z Regrtest.run_tests_sequentialc c s6 � |D ]}|V | j r d S | jjr| jr d S qqr< )r! r �fail_env_changedr$ )r8 r rQ r9 r9 r: � _test_forever� s ���zRegrtest._test_foreverc C sx t dt�� gtj�� �R � t dtjdd�dtj � t dt�� � t� � }|r-t d|� t dt � d�t� � f � d S ) Nz==T)�aliasedz %s-endianz== cwd:z == CPU count:z == encodings: locale=%s, FS=%sF) rM �platform�python_implementationrN �versionr� � byteorderrz �getcwd� cpu_count�locale�getpreferredencoding�getfilesystemencoding)r8 r� r9 r9 r: �display_header� s � � �zRegrtest.display_headerc C s� g }| j r |�d� n | jjr| jr|�d� nt| j| j | j| j| jf�s+|�d� | jr3|�d� |s:|�d� d� |�}| j rId| j |f }|S )N�FAILUREz ENV CHANGEDz NO TEST RUN� INTERRUPTED�SUCCESSr� z %s then %s) r! rC r r� r$ �anyr r" r) r� r( )r8 rP r9 r9 r: r� � s" � zRegrtest.get_tests_resultc C s� | j js| j js| j js| j js| js| j js| �� | j jr4| j j\}}}|dk r4d}t |t j dd� | j j r?t d| j j � | j jrS| �t| j��| _d| _d| _nt| j�| _d�t| j��| _t| j�d | _| j jr�d d lm} t jd kr�| jd u r�d d l m } z|� | _ W n t y� } z t d |� �� W Y d }~nd }~ww z|| � W | j d ur�| j � � d | _ d S d S | j d ur�| j � � d | _ w | �!� d S )N� zhWARNING: Running tests with --huntrleaks/-R and less than 3 warmup repetitions can give false positives!Trn zUsing random seedr z/{}r r )�run_tests_multiprocess�win32)�WindowsLoadTrackerz%Failed to create WindowsLoadTracker: )"r �headerrh rf r� r ru r� rv rM rN �stdoutr� r� �foreverr� r� r r/ r0 �iterr� rg �use_mprt r r� r6 �test.libregrtest.win_utilsr r4 �PermissionError�closer� )r8 r~ r r� r� r r �errorr9 r9 r: � run_tests� sZ ����� �� � zRegrtest.run_testsc C s� | j r+| jr%t| j d��}|�| jd � W d � n1 s w Y nt�| j � | jr=| j�� }|jdd| j j d� t � t � � | j }t dt|� � t d| �� � | j jrft�dt�� � d S d S )N�w� T)� show_missing�summary�coverdirzTotal duration: %szTests result: %szleaks %d)r2 r1 r� �writerz �unlinkr+ �results� write_resultsr r rM r, r- r. r r� �runleaks�system�getpid)r8 r� r� �durationr9 r9 r: �finalize- s& �� ��zRegrtest.finalizec C s | j js | js d S dd lm m} |�d�}dddd�}| jD ]%}|�|� |D ]}z|| t|� |d��7 < W q) t yD Y q)w q |� � D ] \}}|� |t |�� qJtj�tj| j j�}t|d��}|�|�D ]} |�| � qlW d � d S 1 sw Y d S )Nr � testsuites)r r� r� �wb)r rq r3 rH rI rJ �ElementrC r[ �getrE �itemsr= r� rz r{ r� r r� r� � tostringlistr ) r8 rR �root�totalsr� �k�vrq �f�sr9 r9 r: �save_xml_resultB s, �� �"�zRegrtest.save_xml_resultc C st | j jr | j j| _| js0t�� r+t�d�| _| jd u r!t�d�| _tj�| jd�| _nt � � | _tj� | j�| _d S )N� abs_builddir�srcdir�build) r ry r5 � sysconfig�is_python_build�get_config_varrz r{ r� �tempfile� gettempdir�abspathr7 r9 r9 r: � set_temp_dir[ s zRegrtest.set_temp_dirc C sV t j| jdd� t �� }| jd urd�|�}nd�|�}|tj7 }t j� | j|�}|S )NT)�exist_okztest_python_worker_{}ztest_python_{}) rz �makedirsr5 r r6 r� r � FS_NONASCIIr{ r� )r8 �pid�test_cwdr9 r9 r: �create_temp_dirq s zRegrtest.create_temp_dirc C sz dd l }tj�|�| j�d�}td| j � |� |�D ] }tj�|�r/td| � t� |� qtd| � t� |� qd S )Nr z test_python_*zCleanup %s directoryzRemove directory: %szRemove file: %s) �globrz r{ r� �escaper5 rM �isdirr �rmtreer )r8 r< r{ rA r9 r9 r: �cleanup� s �zRegrtest.cleanupNc K s� | � |� | �� | jjr| �� t�d� | �� }z&tj|dd�� || j_ | � ||� W d � W d S 1 s9w Y W d S t y` } zt j tdd� t�|j� W Y d }~d S d }~ww )Nr T)rf )rx )r� r5 r r@ rN rx r; r �temp_cwdry �_main� SystemExit� faulthandler�dump_traceback_later� EXIT_TIMEOUT�code)r8 r r} r: �excr9 r9 r: �main� s &���z Regrtest.mainc C s. | j d ur | j �� S ttd�rt�� d S d S )NrY r )r4 rY �hasattrrz r7 r9 r9 r: rY � s zRegrtest.getloadavgc C s | j d urddlm} || j| j � | jjrtd� | jjt_| jj t_ t | j� | � |� | jj r;| � � t�d� | jjrH| �� t�d� | �� | �� | jjr[| jr[| �� | �� | �� | jrkt�d� | jrst�d� | jjr| jrt�d� t�d� d S )Nr )�run_tests_workerz Press any key to continue...ro � r )r6 rt rK r �wait�inputrh r �PGO� pgo_extended� PGO_EXTENDEDr r� r� rN rx r� r r� �verbose2r! r� r r+ r) r� r$ )r8 r r} rK r9 r9 r: rB � s8 zRegrtest._main)F)r r< ) �__name__� __module__� __qualname__�__doc__r; r> rT rc rl r� r� r� r� r� r� r� r� r� r� r� r r r+ r5 r; r@ rI rY rB r9 r9 r9 r: r s6 + ' I ,;7 1 r c K s t � jdd| i|�� dS )zRun the Python suite.r Nr9 )r rI )r r} r9 r9 r: rI � s rI r< )-rD r� rz r� r� r� rN r/ r2 r, r� �test.libregrtest.cmdliner �test.libregrtest.runtestr r r r r r r r r r r r r r r �test.libregrtest.setupr �test.libregrtest.pgor �test.libregrtest.utilsr r r r r� r � test.supportr rF r rI r9 r9 r9 r: �<module> s4 D D __pycache__/setup.cpython-310.pyc 0000644 00000007153 15030171176 0012624 0 ustar 00 o }�5h� � @ s� d dl Z d dlZd dlZd dlZd dlZd dlZd dlmZ d dlm Z m Z zd dl Z W n e y7 dZ Y nw d dl mZmZ dZdd� Zdd � Zd d � ZdS ) � N)�support)�TESTFN_UNDECODABLE� FS_NONASCII)�setup_unraisable_hook�setup_threading_excepthook� PYTHONREGRTEST_UNICODE_GUARDc C sj zt j�� }W n ttfy d }Y n-w tjd|d� g }ttd�r)|� tj � ttd�r4|� tj � |D ] }tj |d|d� q6t � t� t�t j� | jr\t j�dtj�| j�� t j�� D ](}t|d�r{t|j�D ] \}}tj�|�|j|< qmt|dd �r�tj�|j �|_ qa| j r�d t j _ t �!� | j"d ur�t�#| j"� | j$d ur�t%�&| j$� t�'| j(o�| j(d k� | j)t_)tt d �r�d d � }t �*|� t+� t,� | j-d u�rt.tj/| j-d �t_/t.tj0| j-d �t_0t1tj2| j-�t_2t1tj3| j-�t_3t1tj/| j-�t_/t1tj0| j-�t_0| j4�rddl5m6} d|_7t8�r'tj9�r'tj:�;t<�=� t8� d S t>�r3tj?�;t<t>� d S d S )NT)� all_threads�file�SIGALRM�SIGUSR1)�chainr r �__path__�__file__F� � addaudithookc S s d S �N� )�name�argsr r �-/usr/lib/python3.10/test/libregrtest/setup.py�_test_audit_hookU s z%setup_tests.<locals>._test_audit_hook�( � )�RegressionTestResult)@�sys� __stderr__�fileno� ValueError�AttributeError� faulthandler�enable�hasattr�signal�appendr r �register�_adjust_resource_limits�replace_stdoutr �record_original_stdout�stdout�testdir�path�insert�os�abspath�modules�values� enumerater �getattrr � huntrleaks�unittest� BaseTestSuite�_cleanup�_deactivate_opcache�memlimit� set_memlimit� threshold�gc� set_threshold�suppress_msvcrt_asserts�verbose� use_resourcesr r r �timeout�max� SHORT_TIMEOUT� LONG_TIMEOUT�min�LOOPBACK_TIMEOUT�INTERNET_TIMEOUT�xmlpath�test.support.testresultr �USE_XMLr �supports_bytes_environ�environb� setdefault�UNICODE_GUARD_ENV�encoder �environ) �ns� stderr_fd�signals�signum�module�indexr* r r r r r � setup_tests sl � � �rU c sZ t j� z� �� } W n ty Y dS w t| d� jdddd�t _� fdd�}t�|� dS ) z�Set stdout encoder error handler to backslashreplace (as stderr error handler) to avoid UnicodeEncodeError when printing a tracebackN�w�backslashreplaceF� )�encoding�errors�closefd�newlinec s t j�� � t _d S r )r r( �closer �r( r r �restore_stdout� s z&replace_stdout.<locals>.restore_stdout)r r( r r �openrY �atexitr$ )�fdr_ r r^ r r& s s �� r& c C s� zddl } ddl m}m} W n ty Y dS w | �|�\}}d}||k rh||k rjt||�}z| �|||f� td|� d|� �� W dS tt fyg } ztd|� d|� d |� d �� W Y d}~dS d}~ww dS dS ) z5Adjust the system resource limits (ulimit) if needed.r N)� RLIMIT_NOFILE� RLIM_INFINITYi zRaised RLIMIT_NOFILE: z -> z#Unable to raise RLIMIT_NOFILE from z to z: �.) �resourcerc rd � ImportError� getrlimitrC � setrlimit�printr �OSError)rf rc rd �fd_limit�max_fds� desired_fds� new_fd_limit�errr r r r% � s, � �����r% )ra r r, r" r r3 �testr �test.support.os_helperr r r: rg �test.libregrtest.utilsr r rL rU r& r% r r r r �<module> s$ �^ __pycache__/refleak.cpython-310.pyc 0000644 00000011465 15030171176 0013076 0 ustar 00 o }�5h � @ s� d dl Z d dlZd dlZd dlZd dlmZ d dlmZ d dlm Z d dl m Z zd dl m Z W n eyA d dlZdd� Z Y nw d d � Zd d � Zd d� ZdS )� N)� isabstract)�support)� os_helper)� clear_caches)� _get_dumpc C s& t dd� | jD ��}|| j| j| jfS )Nc s s � | ]}t �|�V qd S �N)�weakref�ref)�.0�obj� r �//usr/lib/python3.10/test/libregrtest/refleak.py� <genexpr> s � z _get_dump.<locals>.<genexpr>)�set� _abc_registry� _abc_cache�_abc_negative_cache�_abc_negative_cache_version)�cls�registry_weakrefsr r r r s �r c ' s� ddl }ddl� ttd�std��t� tjdd� }|j� � }tj � � }zddl }W n t y6 d}Y nw |j � � }i } � fdd�� jjD �D ]} t| �sPqI| �� | g D ] } t| �d | | < qWqIdd� td d �D ���fd d �} | j\} }}tj�tj|�}| | }tt|��}dg| }dg| }dg| }tj}tj }tj }d } }}| j s�t d |dtj d� t d|d d d|� tj dd� t!||||| � |D ]@}|� t!||||| � |� } |� } |� } | j s�t ddtj dd� | | | �||< | | | �||< | | | �||< | }| }| }q�| j �s t tj d� dd� } dd� } d }!|d | f|d | f|d | ffD ]E\}"}#}$|"| d� }"|$|"��rxd ||"|#t"|"�f }%t |%tj dd� t#|d!��}&t |%|&d� |&�$� W d � n 1 �sqw Y d}!�q4|!S )"z�Run a test multiple times, looking for reference leaks. Returns: False if the test didn't leak references; True if we detected refleaks. r N�gettotalrefcountz9Tracking reference leaks requires a debug build of Pythonc � g | ]}t � j|��qS r ��getattr�abc�r �a�� collectionsr r � <listcomp>4 � zdash_R.<locals>.<listcomp>c S s i | ]}||�qS r r )r �valuer r r � <dictcomp>> s zdash_R.<locals>.<dictcomp>i���i� c s � � | | �S r )� setdefault)r! )�int_poolr r �get_pooled_int? s z dash_R.<locals>.get_pooled_int� beginning� repetitions)�file� 1234567890� � T)r( �flush�.� )�endr( r, c S s t dd� | D ��S )Nc s s � | ]}|d kV qdS )r+ Nr )r �deltar r r r s � z2dash_R.<locals>.check_rc_deltas.<locals>.<genexpr>)�all��deltasr r r �check_rc_deltasr s z dash_R.<locals>.check_rc_deltasc S s t | �S r )�anyr2 r r r �check_fd_deltas� s z dash_R.<locals>.check_fd_deltasF� referencesz memory blockszfile descriptorsz%s leaked %s %s, sum=%sr )%�copyreg�collections.abc�hasattr�sys� Exception� warm_caches�warnings�filters�dispatch_table�copy�path_importer_cache� zipimport� ImportError�_zip_directory_cacher �__all__r �__subclasses__r �range� huntrleaks�os�path�joinr �SAVEDCWD�list�getallocatedblocksr �fd_count�quiet�print�stderr�dash_R_cleanup�sum�openr, )'�ns� test_name� test_funcr8 �fs�ps�picrC �zdc�abcsr r r% �nwarmup�ntracked�fname�repcount� rep_range� rc_deltas� alloc_deltas� fd_deltasrO r rP � rc_before� alloc_before� fd_before�i� alloc_after�rc_after�fd_afterr4 r6 �failedr3 � item_name�checker�msg�refrepr )r r$ r �dash_R s� � � � � � ��rs c s dd l }dd l� | tjd d �< |j�� |j�|� tj�� tj�|� zdd l }W n t y4 Y n w |j �� |j �|� t� � � fdd�� j jD �}tt|�}|D ]%}|�� |g D ]} |�| t� �D ] } | � d uru| �| � � qh| �� q_qVt� d S )Nr c r r r r r r r r � r z"dash_R_cleanup.<locals>.<listcomp>)r8 r9 r> r? r@ �clear�updater; rB rC rD rE �_clear_type_cacher rF �filterr rG �getr �register�_abc_caches_clearr ) rZ r[ r\ r] r^ r8 rC � abs_classesr r r r r r rT � s4 � � � rT c C sN t td��} td�D ] }| ||d � q dd� td�D � ttdd�� d S )N� r+ c S s g | ]}t |��qS r )�chr)r rj r r r r � s z warm_caches.<locals>.<listcomp>���i )�bytesrH rN )�srj r r r r= � s r= )rJ �rer; r> �inspectr �testr � test.supportr �test.libregrtest.utilsr �_abcr rD r rs rT r= r r r r �<module> s$ � " __pycache__/runtest_mp.cpython-310.pyc 0000644 00000032426 15030171176 0013665 0 ustar 00 o }�5hsD � @ sb d dl Z d dlZd dlZd dlZd dlZd dlZd dlZd dlZd dlZd dl Z d dl Z d dl m Z m Z mZmZ d dlmZ d dlmZ d dlmZ d dlmZ d dlmZmZmZm Z m Z m Z m Z d dl m!Z! d d l"m#Z#m$Z$ d Z%e%e ksxJ �d Z&e&e%ks�J �d Z'e(ed �o�e(ed �Z)dedede*fdd�Z+de,ee-f fdd�Z.de-dedej/fdd�Z0dede-de fdd�Z1G dd � d �Z2G d d � d e �Z3e-Z4e,ed e3f e,ed e4f B Z5G d!d"� d"e6�Z7G d#d$� d$ej8�Z9d%e:e9 de:e9 fd&d'�Z;G d(d)� d)�Z<d*eddfd+d,�Z=G d-d.� d.ej>�Z?d/e@e-ef dee@e-ef B fd0d1�ZAdeBeCe fd2d3�ZDdS )4� N)� NamedTuple�NoReturn�Literal�Any)�support)� os_helper)� Namespace)�Regrtest)�runtest� is_failed� TestResult� Interrupted�Timeout� ChildError�PROGRESS_MIN_TIME)� setup_tests)�format_duration� print_warningg >@g �r@�setsid�killpg�result�ns�returnc C s& t | t�rdS |jrt| |�rdS dS )NTF)� isinstancer �failfastr )r r � r �2/usr/lib/python3.10/test/libregrtest/runtest_mp.py� must_stop) s r c C s$ t �| �\}}tdi |��}||fS )Nr )�json�loadsr )� worker_args�ns_dict� test_namer r r r �parse_worker_args1 s r# �testnamec C sx t |�}|| f}t�|�}tjgt�� �d�d�d�d�|�}i }tr'd|d< tj |ftj tj dt j dkt jd�|��S ) Nz-uz-mz test.regrtestz --worker-argsT�start_new_session�nt)�stdout�stderr�universal_newlines� close_fds�cwd)�varsr �dumps�sys� executabler �args_from_interpreter_flags�USE_PROCESS_GROUP� subprocess�Popen�PIPE�os�namer �SAVEDCWD)r$ r r! r �cmd�kwr r r �run_test_in_subprocess7 s4 �������r: r" c C s< t | � t| |�}t� ttj|td�dd� t�d� d S )N)�clsT��flushr )r r �printr r- �EncodeTestResultr. �exit)r r" r r r r �run_tests_workerP s rA c @ s0 e Zd ZdZdd� Zdd� Zdd� Zdd � Zd S ) �MultiprocessIteratorz8A thread-safe iterator over tests for multiprocess mode.c C s t �� | _|| _d S �N)� threading�Lock�lock� tests_iter)�selfrG r r r �__init__a s z MultiprocessIterator.__init__c C s | S rC r �rH r r r �__iter__e s z MultiprocessIterator.__iter__c C sB | j � | jd u r t�t| j�W d � S 1 sw Y d S rC )rF rG � StopIteration�nextrJ r r r �__next__h s $�z MultiprocessIterator.__next__c C s2 | j � d | _W d � d S 1 sw Y d S rC )rF rG rJ r r r �stopn s "�zMultiprocessIterator.stopN)�__name__� __module__� __qualname__�__doc__rI rK rN rO r r r r rB ] s rB c @ s. e Zd ZU eed< eed< eed< eed< dS )�MultiprocessResultr r' r( � error_msgN)rP rQ rR r �__annotations__�strr r r r rT s s rT FTc @ s e Zd ZdS )� ExitThreadN)rP rQ rR r r r r rX ~ s rX c s� e Zd Zdeddddf� fdd� Zdefdd �Zd d d �Zd d d �Z d!de dedede fdd�Z dede eeef fdd�Z dede fdd�Zd dd�Zd dd �Zd eddfd d �Z� ZS )"�TestWorkerProcess� worker_id�runner�MultiprocessTestRunnerr Nc sZ t � �� || _|j| _|j| _|j| _|j| _|j| _d | _ d | _ d | _ d| _ d| _ d S )NF)�superrI rZ �pending�outputr �worker_timeout�timeout�regrtest�current_test_name� start_time�_popen�_killed�_stopped)rH rZ r[ �� __class__r r rI � s zTestWorkerProcess.__init__c C s� d| j � �g}| �� r|�d� n|�d� | j}|r#|�d|� �� | j}|d urBt�� | j }|�d| jj � �dt |�� �f� dd� |� S ) NzTestWorkerProcess #�running�stoppedztest=zpid=ztime=z<%s>� ) rZ �is_alive�appendrc re �time� monotonicrd �extend�pidr �join)rH �info�test�popen�dtr r r �__repr__� s �zTestWorkerProcess.__repr__c C s� | j }|d u r d S | jrd S d| _tr| � d�}n| � }td|� �tjdd� ztr5t�|jt j � W d S |� � W d S t yE Y d S t ya } ztd|� d|��� W Y d }~d S d }~ww )NTz process groupzKill ��filer= zFailed to kill z: )re rf r1 r> r. r( r5 r rr �signal�SIGKILL�kill�ProcessLookupError�OSErrorr )rH rv �what�excr r r �_kill� s( "��zTestWorkerProcess._killc C s d| _ | �� d S )NT)rg r� rJ r r r rO � s zTestWorkerProcess.stop� � test_resultr' r( c C s t �� | j |_t||||�S rC )ro rp rd � duration_secrT )rH r� r' r( �err_msgr r r �mp_result_error� s z!TestWorkerProcess.mp_result_errorr" c C s t �� | _|| _zt|| j�}d| _|| _W n d | _� zez\| jr)| � � t �z|j | j d�\}}|j }|d us<J �W n% tjyV | jrJt �| � � d }d }}Y n tyb | jrat �� w |�� }|�� }|||fW W | �� d | _d | _S | � � � | �� d | _d | _w )NF�ra r� )ro rp rd rc r: r rf re rg r� rX � communicatera � returncoder2 �TimeoutExpiredr �strip�rstrip�_wait_completed)rH r" rv r' r( �retcoder r r � _run_process� sP � �z TestWorkerProcess._run_processc C s� | � |�\}}}|d u r| �t|�||�S d }|dkr d| }n1|�d�\}}}|�� }|s1d}n z tj|td�}W n tyP } z d| }W Y d }~nd }~ww |d ur_| �t |�|||�S t ||||�S )Nr z Exit code %s� z Failed to parse worker stdout)� object_hookz Failed to parse worker JSON: %s) r� r� r � rpartitionr� r r �decode_test_result� Exceptionr rT ) rH r" r� r' r( r� �_r r� r r r �_runtest s* �� �zTestWorkerProcess._runtestc C s� | j sRz,zt| j�}W n ty Y W d S w | �|�}| j�d|f� t|j| j �r.W d S W n t y9 Y d S t yL | j�dt � � f� Y d S w | j rd S d S )NFT)rg rM r^ rL r� r_ �putr r r rX � BaseException� traceback� format_exc)rH r" � mp_resultr r r �run" s& � � ��zTestWorkerProcess.runc C sv | j }|j�� |j�� z|�t� W d S tjtfy: } zt d| � dt t�� d|��� W Y d }~d S d }~ww )NzFailed to wait for z completion (timeout=z): ) re r' �closer( �wait� JOIN_TIMEOUTr2 r� r r r )rH rv r� r r r r� 5 s ����z!TestWorkerProcess._wait_completedrd c C sf | � d� | �� s d S t�� | }| j�d| � dt|�� �� |tkr2td| � dt|�� �� d S q)NTg �?z Waiting for z thread for zFailed to join z in ) rs rm ro rp rb �logr r� r )rH rd rw r r r � wait_stoppedD s ��z TestWorkerProcess.wait_stopped�r N)r� r� N)rP rQ rR �intrI rW rx r� rO r rT r� �tupler� r� r� r� �floatr� � __classcell__r r rh r rY � s, ���� � ; rY �workersc C sN g }| D ] }|j }|s qt�� |j }|tkr$d|t|�f }|�|� q|S )Nz%s (%s))rc ro rp rd r r rn )r� rj �workerrc rw �textr r r � get_running\ s �r� c @ sr e Zd Zdeddfdd�Zddd�Zddd �ZdedB fd d �Zd e ddfd d�Z dede fdd�Z ddd�Z dS )r\ rb r Nc C sj || _ | j j| _|j| _t�� | _t| j j�| _| jj d ur-t | jj d | jj d �| _ nd | _ d | _ d S )Ng �?i, ) rb r� r �queue�Queuer_ rB �testsr^ ra �minr` r� )rH rb r r r rI j s � z MultiprocessTestRunner.__init__c s| � fdd�t d� jjd �D �� _dt� j�� d�}� jjr-|dt� jj�t� j�f 7 }� �|� � jD ]}|� � q5d S )Nc s g | ]}t |� ��qS r )rY )�.0�indexrJ r r � <listcomp>{ s �z8MultiprocessTestRunner.start_workers.<locals>.<listcomp>� z Run tests in parallel using z child processesz" (timeout: %s, worker timeout: %s)) �ranger �use_mpr� �lenra r r` r� �start)rH �msgr� r rJ r � start_workersz s � �� �z$MultiprocessTestRunner.start_workersc C s6 t �� }| jD ]}|�� q| jD ]}|�|� qd S rC )ro rp r� rO r� )rH rd r� r r r � stop_workers� s �z#MultiprocessTestRunner.stop_workersc C s� | j jd u}t}tdd� | jD ��rM|rtjtdd� z| jj |d�W S t j y- Y nw t | j�}|rC| j j sC| �dd�|� � tdd� | jD ��sz| jj dd�W S t j y` Y d S w ) Nc s s � | ]}|� � V qd S rC )rm )r� r� r r r � <genexpr>� s � z5MultiprocessTestRunner._get_result.<locals>.<genexpr>T)r@ r� z running: %s�, r )r ra �PROGRESS_UPDATE�anyr� � faulthandler�dump_traceback_later�MAIN_PROCESS_TIMEOUTr_ �getr� �Emptyr� �pgor� rs )rH �use_faulthandlerra rj r r r � _get_result� s* �� ��z"MultiprocessTestRunner._get_resultr� c C s� |j }t|�}|jd ur|d|j 7 }n|jtkr&| jjs&|dt|j� 7 }t| j �}|r:| jjs:|dd� |� 7 }| j � | j |� d S )Nz (%s)z -- running: %sr� )r rW rU r� r r r� r r� r� rs rb �display_progress� test_index)rH r� r r� rj r r r �display_result� s z%MultiprocessTestRunner.display_result�itemc C s� |d r|d }t d|� �� dS | jd7 _|d }| j�|j� | �|� |jr2t|jdd� |jrB| j j sBt|jt jdd� t |j| j �rKdS dS )z&Returns True if test runner must stop.r r� z regrtest worker thread failed: Tr< ry F) r r� rb �accumulate_resultr r� r' r> r( r r� r. r )rH r� r� r� r r r �_process_result� s z&MultiprocessTestRunner._process_resultc C s� | � � d| _z<z | �� }|d u rn | �|�}|rnq W n ty- t� d| j_Y nw W | jj d ur9t � � | j � � | �� d S | jj d urNt � � | j � � | �� w )Nr T)r� r� r� r� �KeyboardInterruptr> rb � interruptedr ra r� �cancel_dump_traceback_laterr^ rO r� )rH r� rO r r r � run_tests� s2 �� �� � z MultiprocessTestRunner.run_testsr� )rP rQ rR r rI r� r� � QueueOutputr� rT r� �boolr� r� r r r r r\ i s r\ rb c C s t | ��� d S rC )r\ r� )rb r r r �run_tests_multiprocess� s r� c s2 e Zd ZdZdedeeef f� fdd� Z� ZS )r? z7Encode a TestResult (sub)class object into a JSON dict.�or c s. t |t�rt|�}|jj|d< |S t� �|�S )N�__test_result__)r r r, ri rP r] �default)rH r� r rh r r r� � s zEncodeTestResult.default) rP rQ rR rS r �dictrW r� r� r r rh r r? � s &r? �dc C sB d| vr| S | � d�}t� D ]}|j|kr |di | �� S qdS )z7Decode a TestResult (sub)class object from a JSON dict.r� Nr )�pop�get_all_test_result_classesrP )r� �cls_namer; r r r r� � s ��r� c C sT d} t h}t|�| kr(t|�} g }|D ] }|�|�� � q|�|� t|�| ks |S )Nr )r r� rq �__subclasses__�update)� prev_count�classes�to_addr; r r r r� s �r� )E� collectionsr� r r5 r� r{ r2 r. rD ro r� �typingr r r r ru r � test.supportr �test.libregrtest.cmdliner �test.libregrtest.mainr �test.libregrtest.runtestr r r r r r r �test.libregrtest.setupr �test.libregrtest.utilsr r r� r� r� �hasattrr1 r� r r� rW r# r3 r: rA rB rT �ExcStrr� r� rX �ThreadrY �listr� r\ r� � JSONEncoderr? r� r� �set�typer� r r r r �<module> sT $ [ & __pycache__/utils.cpython-310.pyc 0000644 00000010576 15030171176 0012627 0 ustar 00 o }�5h � @ s� d dl Z d dlZd dlZd dlZd dlmZ dd� Zdd� Zdd� Z dd d �Z d d� Z da dd� Z dd� Zdd� Zdadd� Zdd� Zdd� ZdS ) � N)�supportc C s� t �| d �}t|d�\} }t| d�\}} t|d�\}}g }|r'|�d| � |r0|�d| � | rG|r<|�d| � n |�d| |d � |sMd| S |d d � }d �|�S ) Ng @�@i� �< z%s hourz%s minz%s secz%.1f secz%s ms� � )�math�ceil�divmod�append�join)�seconds�ms�minutes�hours�parts� r �-/usr/lib/python3.10/test/libregrtest/utils.py�format_duration s" r c C s> | sd S t | �D ]\}}tj�|�\}}|dkr || |< qd S )Nz.py)� enumerate�os�path�splitext)�names�idx�name�basename�extr r r �removepy! s ��r c C s | dkr d| |f S d| |f S )N� z%d %sz%d %ssr )�n�wordr r r �count* s r �F � c C s: d| }t tjd�dd� t| �D ��|||d�|d� dS )z�Print the elements of iterable x to stdout. Optional arg width (default 70) is the maximum line length. Optional arg indent (default 4) is the number of blanks with which to begin each line. r c s s � | ]}t |�V qd S �N)�str)�.0�eltr r r � <genexpr>; s � z printlist.<locals>.<genexpr>)�initial_indent�subsequent_indent)�fileN)�print�textwrap�fillr �sorted)�x�width�indentr* �blanksr r r � printlist1 s � �r3 c C s t �| � d S r# )r � print_warning)�msgr r r r4 @ s r4 c C s0 t jd ur t j�� t jd urt j�� d S d S r# )�sys�stdout�flush�stderrr r r r �flush_std_streamsG s �r: c C sJ dt _td� tj}zt� tjt_t| � tj�� W |t_d S |t_w )NTzUnraisable exception) r �environment_alteredr4 r6 r9 r: � __stderr__�orig_unraisablehookr8 )� unraisable� old_stderrr r r �regrtest_unraisable_hookN s r@ c C s t jatt _d S r# )r6 �unraisablehookr= r@ r r r r �setup_unraisable_hook\ s rB c C sT dt _td| jj� �� tj}zt� tjt_t | � tj� � W |t_d S |t_w )NTzUncaught thread exception: ) r r; r4 �exc_type�__name__r6 r9 r: r< �orig_threading_excepthookr8 )�argsr? r r r � regrtest_threading_excepthooke s rG c C s dd l } | jat| _d S )Nr )� threading� excepthookrE rG )rH r r r �setup_threading_excepthooks s rJ c C s` t j�� D ] } t| d�r| `qt jt jt jt jfD ] }|d ur#|� � qzt jd }W n t y4 Y nw |j � � zt jd }W n t yJ Y nw |� � zt jd }W n t y_ Y nw |j� � zt jd }W n t yu Y nw |�� zt jd }W n t y� Y nw |�� zt jd }W n t y� Y nw |�� zt jd }W n t y� Y nw |�� zt jd } W n t y� Y nw | j� � zt jd } W n t y� Y nw | �� zt jd } W n t y� Y nw d | _zt jd } W n t �y Y nw | �� zt jd } W n t �y Y n w | jD ]}|� �q#t�� d S )N�__warningregistry__zdistutils.dir_util�re� _strptimez urllib.parsezurllib.request� linecache� mimetypes�filecmp�struct�doctest�ctypes�typing)r6 �modules�values�hasattrrK r7 r9 � __stdout__r< r8 �KeyError� _path_created�clear�purge� _regex_cache� clear_cache� urlcleanup� clearcache�_default_mime_types�_cache� _clearcache�master� _reset_cache� _cleanupsr � gc_collect)�mod�stream�distutils_dir_utilrL rM � urllib_parse�urllib_requestrN rO rP rQ rR rS rT �fr r r � clear_cachesz s� �� � � � � � � � � � ��� rn )r! r" N)r �os.pathr r6 r, �testr r r r r3 r4 r= r: r@ rB rE rG rJ rn r r r r �<module> s$ __pycache__/win_utils.cpython-310.pyc 0000644 00000005074 15030171176 0013501 0 ustar 00 o }�5h� � @ s\ d dl Z d dlZd dlZd dlZd dlZd dlZdZde�ed � ZdZ G dd� d�Z dS )� N� �< � c @ sh e Zd ZdZdd� Zejejfdd�Z e j e j e jfdd�Zdd d �Zejejejfd d �Zd d� ZdS )�WindowsLoadTrackerz� This class asynchronously reads the performance counters to calculate the system load on Windows. A "raw" thread is used here to prevent interference with the test suite's cases for the threading module. c C sP t �t j� g | _d | _t�d ddd �| _t�d ddd �| _t � | j di � d S )NTF� ) �winreg� QueryInfoKey�HKEY_PERFORMANCE_DATA�_values�_load� _overlapped� CreateEvent�_running�_stopped�_thread�start_new_thread� _update_load��selfr r �1/usr/lib/python3.10/test/libregrtest/win_utils.py�__init__ s zWindowsLoadTracker.__init__c C s. || j d�r| �� || j d�s|| j� d S )Ni� )r �_calculate_loadr )r �_wait�_signalr r r r & s �z WindowsLoadTracker._update_loadc C s� ||d�\}}|d|d�\}|d||�\}}|| } || } | | k rC|d|| �\} } } | | 7 } | dkr?| | }|d||�\}n| | k s"d S | j d urX| j t |dt | _ d S t| j�tk rg| j�|� d S t| j�t| j� | _ d S )N�2�L� �4xLL�LL28xL�, g �?)r � LOAD_FACTOR_1�lenr �NVALUE�append�sum)r �_query�_hkey�_unpack�data�_� obj_start� data_start� defn_start� data_base� defn_base�size�idx�offset�counter_offset�processor_queue_lengthr r r r 0 s* � �z"WindowsLoadTracker._calculate_loadTc C s | � � d S �N)�__del__)r �killr r r �closeh s zWindowsLoadTracker.closec C sH | j d ur"|| j � || jd� || j � || j� d | _ | _d S d S )N���)r r )r r �_closer r r r r5 l s �zWindowsLoadTracker.__del__c C s | j S r4 )r r r r r � getloadavg{ s z WindowsLoadTracker.getloadavgN)T)�__name__� __module__� __qualname__�__doc__r �_winapi�WaitForSingleObjectr �SetEventr r � QueryValueExr �struct� unpack_fromr r7 � CloseHandler5 r: r r r r r s � � 8 � r ) r r r? �mathrC r �SAMPLING_INTERVAL�expr r" r r r r r �<module> s __pycache__/cmdline.cpython-310.pyc 0000644 00000034742 15030171176 0013103 0 ustar 00 o }�5hL � @ s� d dl Z d dlZd dlZd dlmZ d dlmZ dZdZdZ dZ e d Z G d d � d e j �Z G d d � d e j �Zd d� Zdd� Zdd� Zdd� Zdd� ZdS )� N)�support)� os_helperz�python -m test [options] [test_name1 [test_name2 ...]] python path/to/Lib/test/regrtest.py [options] [test_name1 [test_name2 ...]] a` Run Python regression tests. If no arguments or options are provided, finds all files matching the pattern "test_*" in the Lib/test subdirectory and runs them in alphabetical order (but see -M and -u, below, for exceptions). For more rigorous testing, it is useful to use the following command line: python -E -Wd -m test [options] [test_name1 ...] a, Additional option details: -r randomizes test execution order. You can use --randseed=int to provide an int seed value for the randomizer; this is useful for reproducing troublesome test orders. -s On the first invocation of regrtest using -s, the first test file found or the first test file given on the command line is run, and the name of the next test is recorded in a file named pynexttest. If run from the Python build directory, pynexttest is located in the 'build' subdirectory, otherwise it is located in tempfile.gettempdir(). On subsequent runs, the test in pynexttest is run, and the next test is written to pynexttest. When the last test has been run, pynexttest is deleted. In this way it is possible to single step through the test files. This is useful when doing memory analysis on the Python interpreter, which process tends to consume too many resources to run the full regression test non-stop. -S is used to continue running tests after an aborted run. It will maintain the order a standard run (ie, this assumes -r is not used). This is useful after the tests have prematurely stopped for some external reason and you want to start running from where you left off rather than starting from the beginning. -f reads the names of tests from the file given as f's argument, one or more test names per line. Whitespace is ignored. Blank lines and lines beginning with '#' are ignored. This is especially useful for whittling down failures involving interactions among tests. -L causes the leaks(1) command to be run just before exit if it exists. leaks(1) is available on Mac OS X and presumably on some other FreeBSD-derived systems. -R runs each test several times and examines sys.gettotalrefcount() to see if the test appears to be leaking references. The argument should be of the form stab:run:fname where 'stab' is the number of times the test is run to let gettotalrefcount settle down, 'run' is the number of times further it is run and 'fname' is the name of the file the reports are written to. These parameters all have defaults (5, 4 and "reflog.txt" respectively), and the minimal invocation is '-R :'. -M runs tests that require an exorbitant amount of memory. These tests typically try to ascertain containers keep working when containing more than 2 billion objects, which only works on 64-bit systems. There are also some tests that try to exhaust the address space of the process, which only makes sense on 32-bit systems with at least 2Gb of memory. The passed-in memlimit, which is a string in the form of '2.5Gb', determines how much memory the tests will limit themselves to (but they may go slightly over.) The number shouldn't be more memory than the machine has (including swap memory). You should also keep in mind that swap memory is generally much, much slower than RAM, and setting memlimit to all available RAM or higher will heavily tax the machine. On the other hand, it is no use running these tests with a limit of less than 2.5Gb, and many require more than 20Gb. Tests that expect to use more than memlimit memory will be skipped. The big-memory tests generally run very, very long. -u is used to specify which special resource intensive tests to run, such as those requiring large file support or network connectivity. The argument is a comma-separated list of words indicating the resources to test. Currently only the following are defined: all - Enable all special resources. none - Disable all special resources (this is the default). audio - Tests that use the audio device. (There are known cases of broken audio drivers that can crash Python or even the Linux kernel.) curses - Tests that use curses and will modify the terminal's state and output modes. largefile - It is okay to run some test that may create huge files. These tests can take a long time and may consume >2 GiB of disk space temporarily. network - It is okay to run tests that use external network resource, e.g. testing SSL support for sockets. decimal - Test the decimal module against a large suite that verifies compliance with standards. cpu - Used for certain CPU-heavy tests. subprocess Run all tests for the subprocess module. urlfetch - It is okay to download files required on testing. gui - Run tests that require a running GUI. tzdata - Run tests that require timezone data. To enable all resources except one, use '-uall,-<resource>'. For example, to run all the tests except for the gui tests, give the option '-uall,-gui'. --matchfile filters tests using a text file, one pattern per line. Pattern examples: - test method: test_stat_attributes - test class: FileTests - test identifier: test_os.FileTests.test_stat_attributes ) �audio�curses� largefile�network�decimal�cpu� subprocess�urlfetch�gui)�extralargefile�tzdatac s e Zd Zd� fdd� Z� ZS )� Namespace�returnNc s� d | _ d| _d| _d| _d| _d| _d | _d| _d| _d | _ d| _ d| _ d| _ d| _ d| _d| _d| _d | _d | _d| _d| _d| _d | _d | _d| _d| _t� jdi |�� d S )Nr F� �coverage� ) �testdir�verbose�quiet�exclude�single� randomize�fromfile� findleaks�fail_env_changed� use_resources�trace�coverdir�runleaks� huntrleaks�verbose2�verbose3� print_slow� random_seed�use_mp�forever�header�failfast� match_tests� ignore_tests�pgo� pgo_extended�super�__init__)�self�kwargs�� __class__r �//usr/lib/python3.10/test/libregrtest/cmdline.pyr/ � s6 zNamespace.__init__)r N)�__name__� __module__� __qualname__r/ � __classcell__r r r2 r4 r � s r c s e Zd Z� fdd�Z� ZS )� _ArgParserc s t � �|d � d S )Nz% Pass -h or --help for complete help.)r. �error)r0 �messager2 r r4 r: � s z_ArgParser.error)r5 r6 r7 r: r8 r r r2 r4 r9 � s r9 c C sl t dtttdtjd�} d}| �d�}|jdddd d � |jd d td d� |jdddd � |jddd� |jdddd| d� | �d�}|jdd d d d � |jd d dd!d � |jd"d#dd$d � |jd%d&dd'd � |jd(d)dd*d+d,� |jd-dd.d � | �d/�}|jd0d1dd2| d � |jd3d4d5t d6d7� |jd8d9d:d;| d� |jd<d=dd>d � |jd?d@ddA| d � |jdBdCdDdEdFdGdH� |jdIdJdDdKdFdLdH� |jdMdNdOdPdQ� |jdRdNdSdTdQ� |jdUdVddWd � |jdXdYdZdFt d[| d\� |jd]d^d_d`| d� |jdadbt dcd� | �dd�}|jdedfdgdhdidjdk� |jdldmddn| d � |jdodpdqt dr| d� |jdsdtdudvt dwd7� |jdxdyddzd{d,� |jd|d}dbt d~d� |jdd�dgd d�d�d�� |jd�d�d�t d�d� |jd�d�dd�d � |jd�d�dd�d � |jd�dd�d � |jd�dd�d � |jd�d�d�dd�d�� |jd�dd�d � |jd�dd�d � |jd�d�dNd�d�� |jd�d�d�d� |jd�dd�d � | S )�Nz regrtest.pyF)�prog�usage� description�epilog�add_help�formatter_classz- See the section at bottom for more details.zGeneral optionsz-hz--help�helpz show this help message and exit)�actionrB z --timeout�TIMEOUTzxdump the traceback and exit if a test takes more than TIMEOUT seconds; disabled if TIMEOUT is negative or equals to zero)�metavar�typerB z--wait� store_truez:wait for user input, e.g., allow a debugger to be attachedz --worker-args�ARGS)rE z-Sz--start�STARTz'the name of the test at which to start.)rE rB � Verbosityz-vz --verbose�countz/run tests in verbose mode with output to stdoutz-wz --verbose2z#re-run failed tests in verbose modez-Wz --verbose3z display test output on failurez-qz--quietz'no output unless one or more tests failz-oz --slowestr$ zprint the slowest 10 tests)rC �destrB z--headerz"print header with interpreter infozSelecting testsz-rz --randomizez randomize test execution order.z --randseed�SEEDr% z5pass a random seed to reproduce a previous random run)rE rL rF rB z-fz --fromfile�FILEz'read names of tests to run from a file.z-xz --excludez arguments are tests to *exclude*z-sz--singlez#single step through a set of tests.z-mz--match�PATr* �appendz2match test cases and methods with glob pattern PAT)rE rL rC rB z-iz--ignorer+ z3ignore test cases and methods with glob pattern PATz --matchfile�FILENAME�match_filenamezJsimilar to --match but get patterns from a text file, one pattern per line)rE rL rB z --ignorefile�ignore_filenamezHsimilar to --matchfile but it receives patterns from text file to ignorez-Gz --failfastz1fail as soon as a test fails (only with -v or -W)z-uz--usez RES1,RES2,...z6specify which special resource intensive tests to run.)rE rC rF rB z-Mz --memlimit�LIMITz&run very large memory-consuming tests.z --testdir�DIRzWexecute test files in the specified directory (instead of the Python stdlib test suite)z Special runsz-lz --findleaks� store_const� r z&deprecated alias to --fail-env-changed)rC �const�defaultrB z-Lz --runleaksz*run the leaks(1) command just before exit.z-Rz --huntrleaks� RUNCOUNTSz:search for reference leaks (needs debug build, very slow).z-jz--multiprocess� PROCESSESr& z run PROCESSES processes at oncez-Tz --coverager z4turn on code coverage tracing using the trace modulez-Dz --coverdirz&directory where coverage files are putz-Nz --nocoverdirr z$put coverage files alongside modules)rC rX rL rB z-tz --threshold� THRESHOLDz call gc.set_threshold(THRESHOLD)z-nz --nowindowsz'suppress error message boxes on Windowsz-Fz --foreverzKrun the specified tests in a loop, until an error happens; imply --failfastz --list-testszAonly write the name of tests that will be run, don't execute themz --list-caseszGonly write the name of test cases that will be run , don't execute themz-Pz--pgor, z1enable Profile Guided Optimization (PGO) training)rL rC rB z--pgo-extendedz.enable extended PGO training (slower training)z--fail-env-changedz>if a test file alters the environment, mark the test as failedz --junit-xml�xmlpathz4writes JUnit-style XML results to the specified file)rL rE rB z --tempdir�PATHz/override the working directory for the test runz --cleanupz$remove old test_python_* directories) r9 �USAGE� DESCRIPTION�EPILOG�argparse�RawDescriptionHelpFormatter�add_argument_group� add_argument�float�int�resources_list�relative_filenamer! )�parser� more_details�groupr r r4 �_create_parser� sJ � � �� �� � � � � �� �� �� � �� � ��� � �� ��� � �� �� � � �� � � ��� ��� ���rm c C s t j�tj| �S )N)�os�path�joinr �SAVEDCWD)�stringr r r4 ri E s ri c C sz | � d�}t|�dvrt�d��|d rt|d �nd}|d r&t|d �nd}t|�dkr6|d r6|d nd }|||fS ) N�:)rW � z&needs 2 or 3 colon-separated argumentsr � r � rW z reflog.txt)�split�lenrb �ArgumentTypeErrorrg )rr �args�nwarmup�ntracked�fnamer r r4 r! K s � r! c C sb dd� | � d�D �}|D ]"}|dks|dkrq |d dkr#|dd � }|tvr.t�d | ��q |S ) Nc S s g | ]}|� � �qS r )�lower)�.0�xr r r4 � <listcomp>W s z"resources_list.<locals>.<listcomp>�,�all�noner �-r zinvalid resource: )rw �RESOURCE_NAMESrb ry )rr �u�rr r r4 rh V s �rh c K s: t � }|�� D ]\}}t||�std| ��t|||� q|jd u r%g |_t� }|j| |d�d |_|jD ]}|� d�rH|� d| � t � d� q5|j dkrQd|_|jr\|jr\|� d� |jd uri|jri|� d� |jrw|jsw|jsw|� d � |jr�|js�|js�|jr�|� d � |jr�d|_|jr�td t jd � |j r�d |_|j d ur�|j d kr�d |_ |jd ur�|jd kr�dt � � p�d |_|j �r|j D ]J}|D ]E}|dkr�t!|jd d �< q�|dkr�|jd d �= q�d} |d dkr�d} |dd � }| �r||jv �r|j�"|� q�||jv�r|j�#|� q�q�|j$d u�rd|_%|j�r d|_&|j'�r2|j�r2d|_tdt jd � |j(�rc|j)d u �r?g |_)t*|j(��} | D ] } |j)�#| �+� � �qGW d � n 1 �s^w Y |j,�r�|j-d u �rpg |_-t*|j,��} | D ] } |j-�#| �+� � �qxW d � n 1 �s�w Y |j.�r�d|_|S )Nz3%r is an invalid keyword argument for this function)rz � namespacer r� zunrecognized arguments: %sTz -s and -f don't go together!z -T and -j don't go together!z#-G/--failfast needs either -v or -Wz--pgo/-v don't go together!z\Warning: the --nowindows (-n) option is deprecated. Use -vv to display assertions in stderr.)�filer rW r� r� FznWARNING: Disable --verbose3 because it's incompatible with --huntrleaks: see http://bugs.python.org/issue27103)/r �items�hasattr� TypeError�setattrr rm �parse_known_argsrz � startswithr: �sys�exitr r r r r& r r) r r# r, r" r- � nowindows�print�stderrr �timeoutrn � cpu_count�use� ALL_RESOURCES�removerP r% r r( r! rR r* �open�striprS r+ r' ) rz r1 �ns�k�vrj �arg�ar� r� �fp�liner r r4 � _parse_argsb s� � � � � �� � � � � �r� )rb rn r� �testr � test.supportr r_ r` ra r� r� r �ArgumentParserr9 rm ri r! rh r� r r r r4 �<module> s$ i __pycache__/pgo.cpython-310.pyc 0000644 00000001632 15030171176 0012245 0 ustar 00 o }�5hM � @ s g d �Z dd� ZdS )),� test_array� test_base64� test_binascii� test_binop� test_bisect� test_bytes�test_bz2� test_cmath� test_codecs�test_collections� test_complex�test_dataclasses� test_datetime� test_decimal� test_difflib� test_embed� test_float� test_fstring�test_functools�test_generators� test_hashlib� test_heapq�test_int�test_itertools� test_json� test_long� test_lzma� test_math�test_memoryview� test_operator�test_ordered_dict� test_patma� test_pickle� test_pprint�test_re�test_set� test_sqlite�test_statistics� test_struct� test_tabnanny� test_time� test_unicode�test_xml_etree�test_xml_etree_cc C s& | j s| jstd d � | _ d S d S d S )N)�args� pgo_extended� PGO_TESTS)�ns� r1 �+/usr/lib/python3.10/test/libregrtest/pgo.py�setup_pgo_tests5 s �r3 N)r/ r3 r1 r1 r1 r2 �<module> s / cmdline.py 0000644 00000046177 15030171176 0006551 0 ustar 00 import argparse import os import sys from test import support from test.support import os_helper USAGE = """\ python -m test [options] [test_name1 [test_name2 ...]] python path/to/Lib/test/regrtest.py [options] [test_name1 [test_name2 ...]] """ DESCRIPTION = """\ Run Python regression tests. If no arguments or options are provided, finds all files matching the pattern "test_*" in the Lib/test subdirectory and runs them in alphabetical order (but see -M and -u, below, for exceptions). For more rigorous testing, it is useful to use the following command line: python -E -Wd -m test [options] [test_name1 ...] """ EPILOG = """\ Additional option details: -r randomizes test execution order. You can use --randseed=int to provide an int seed value for the randomizer; this is useful for reproducing troublesome test orders. -s On the first invocation of regrtest using -s, the first test file found or the first test file given on the command line is run, and the name of the next test is recorded in a file named pynexttest. If run from the Python build directory, pynexttest is located in the 'build' subdirectory, otherwise it is located in tempfile.gettempdir(). On subsequent runs, the test in pynexttest is run, and the next test is written to pynexttest. When the last test has been run, pynexttest is deleted. In this way it is possible to single step through the test files. This is useful when doing memory analysis on the Python interpreter, which process tends to consume too many resources to run the full regression test non-stop. -S is used to continue running tests after an aborted run. It will maintain the order a standard run (ie, this assumes -r is not used). This is useful after the tests have prematurely stopped for some external reason and you want to start running from where you left off rather than starting from the beginning. -f reads the names of tests from the file given as f's argument, one or more test names per line. Whitespace is ignored. Blank lines and lines beginning with '#' are ignored. This is especially useful for whittling down failures involving interactions among tests. -L causes the leaks(1) command to be run just before exit if it exists. leaks(1) is available on Mac OS X and presumably on some other FreeBSD-derived systems. -R runs each test several times and examines sys.gettotalrefcount() to see if the test appears to be leaking references. The argument should be of the form stab:run:fname where 'stab' is the number of times the test is run to let gettotalrefcount settle down, 'run' is the number of times further it is run and 'fname' is the name of the file the reports are written to. These parameters all have defaults (5, 4 and "reflog.txt" respectively), and the minimal invocation is '-R :'. -M runs tests that require an exorbitant amount of memory. These tests typically try to ascertain containers keep working when containing more than 2 billion objects, which only works on 64-bit systems. There are also some tests that try to exhaust the address space of the process, which only makes sense on 32-bit systems with at least 2Gb of memory. The passed-in memlimit, which is a string in the form of '2.5Gb', determines how much memory the tests will limit themselves to (but they may go slightly over.) The number shouldn't be more memory than the machine has (including swap memory). You should also keep in mind that swap memory is generally much, much slower than RAM, and setting memlimit to all available RAM or higher will heavily tax the machine. On the other hand, it is no use running these tests with a limit of less than 2.5Gb, and many require more than 20Gb. Tests that expect to use more than memlimit memory will be skipped. The big-memory tests generally run very, very long. -u is used to specify which special resource intensive tests to run, such as those requiring large file support or network connectivity. The argument is a comma-separated list of words indicating the resources to test. Currently only the following are defined: all - Enable all special resources. none - Disable all special resources (this is the default). audio - Tests that use the audio device. (There are known cases of broken audio drivers that can crash Python or even the Linux kernel.) curses - Tests that use curses and will modify the terminal's state and output modes. largefile - It is okay to run some test that may create huge files. These tests can take a long time and may consume >2 GiB of disk space temporarily. network - It is okay to run tests that use external network resource, e.g. testing SSL support for sockets. decimal - Test the decimal module against a large suite that verifies compliance with standards. cpu - Used for certain CPU-heavy tests. subprocess Run all tests for the subprocess module. urlfetch - It is okay to download files required on testing. gui - Run tests that require a running GUI. tzdata - Run tests that require timezone data. To enable all resources except one, use '-uall,-<resource>'. For example, to run all the tests except for the gui tests, give the option '-uall,-gui'. --matchfile filters tests using a text file, one pattern per line. Pattern examples: - test method: test_stat_attributes - test class: FileTests - test identifier: test_os.FileTests.test_stat_attributes """ ALL_RESOURCES = ('audio', 'curses', 'largefile', 'network', 'decimal', 'cpu', 'subprocess', 'urlfetch', 'gui') # Other resources excluded from --use=all: # # - extralagefile (ex: test_zipfile64): really too slow to be enabled # "by default" # - tzdata: while needed to validate fully test_datetime, it makes # test_datetime too slow (15-20 min on some buildbots) and so is disabled by # default (see bpo-30822). RESOURCE_NAMES = ALL_RESOURCES + ('extralargefile', 'tzdata') class Namespace(argparse.Namespace): def __init__(self, **kwargs) -> None: self.testdir = None self.verbose = 0 self.quiet = False self.exclude = False self.single = False self.randomize = False self.fromfile = None self.findleaks = 1 self.fail_env_changed = False self.use_resources = None self.trace = False self.coverdir = 'coverage' self.runleaks = False self.huntrleaks = False self.verbose2 = False self.verbose3 = False self.print_slow = False self.random_seed = None self.use_mp = None self.forever = False self.header = False self.failfast = False self.match_tests = None self.ignore_tests = None self.pgo = False self.pgo_extended = False super().__init__(**kwargs) class _ArgParser(argparse.ArgumentParser): def error(self, message): super().error(message + "\nPass -h or --help for complete help.") def _create_parser(): # Set prog to prevent the uninformative "__main__.py" from displaying in # error messages when using "python -m test ...". parser = _ArgParser(prog='regrtest.py', usage=USAGE, description=DESCRIPTION, epilog=EPILOG, add_help=False, formatter_class=argparse.RawDescriptionHelpFormatter) # Arguments with this clause added to its help are described further in # the epilog's "Additional option details" section. more_details = ' See the section at bottom for more details.' group = parser.add_argument_group('General options') # We add help explicitly to control what argument group it renders under. group.add_argument('-h', '--help', action='help', help='show this help message and exit') group.add_argument('--timeout', metavar='TIMEOUT', type=float, help='dump the traceback and exit if a test takes ' 'more than TIMEOUT seconds; disabled if TIMEOUT ' 'is negative or equals to zero') group.add_argument('--wait', action='store_true', help='wait for user input, e.g., allow a debugger ' 'to be attached') group.add_argument('--worker-args', metavar='ARGS') group.add_argument('-S', '--start', metavar='START', help='the name of the test at which to start.' + more_details) group = parser.add_argument_group('Verbosity') group.add_argument('-v', '--verbose', action='count', help='run tests in verbose mode with output to stdout') group.add_argument('-w', '--verbose2', action='store_true', help='re-run failed tests in verbose mode') group.add_argument('-W', '--verbose3', action='store_true', help='display test output on failure') group.add_argument('-q', '--quiet', action='store_true', help='no output unless one or more tests fail') group.add_argument('-o', '--slowest', action='store_true', dest='print_slow', help='print the slowest 10 tests') group.add_argument('--header', action='store_true', help='print header with interpreter info') group = parser.add_argument_group('Selecting tests') group.add_argument('-r', '--randomize', action='store_true', help='randomize test execution order.' + more_details) group.add_argument('--randseed', metavar='SEED', dest='random_seed', type=int, help='pass a random seed to reproduce a previous ' 'random run') group.add_argument('-f', '--fromfile', metavar='FILE', help='read names of tests to run from a file.' + more_details) group.add_argument('-x', '--exclude', action='store_true', help='arguments are tests to *exclude*') group.add_argument('-s', '--single', action='store_true', help='single step through a set of tests.' + more_details) group.add_argument('-m', '--match', metavar='PAT', dest='match_tests', action='append', help='match test cases and methods with glob pattern PAT') group.add_argument('-i', '--ignore', metavar='PAT', dest='ignore_tests', action='append', help='ignore test cases and methods with glob pattern PAT') group.add_argument('--matchfile', metavar='FILENAME', dest='match_filename', help='similar to --match but get patterns from a ' 'text file, one pattern per line') group.add_argument('--ignorefile', metavar='FILENAME', dest='ignore_filename', help='similar to --matchfile but it receives patterns ' 'from text file to ignore') group.add_argument('-G', '--failfast', action='store_true', help='fail as soon as a test fails (only with -v or -W)') group.add_argument('-u', '--use', metavar='RES1,RES2,...', action='append', type=resources_list, help='specify which special resource intensive tests ' 'to run.' + more_details) group.add_argument('-M', '--memlimit', metavar='LIMIT', help='run very large memory-consuming tests.' + more_details) group.add_argument('--testdir', metavar='DIR', type=relative_filename, help='execute test files in the specified directory ' '(instead of the Python stdlib test suite)') group = parser.add_argument_group('Special runs') group.add_argument('-l', '--findleaks', action='store_const', const=2, default=1, help='deprecated alias to --fail-env-changed') group.add_argument('-L', '--runleaks', action='store_true', help='run the leaks(1) command just before exit.' + more_details) group.add_argument('-R', '--huntrleaks', metavar='RUNCOUNTS', type=huntrleaks, help='search for reference leaks (needs debug build, ' 'very slow).' + more_details) group.add_argument('-j', '--multiprocess', metavar='PROCESSES', dest='use_mp', type=int, help='run PROCESSES processes at once') group.add_argument('-T', '--coverage', action='store_true', dest='trace', help='turn on code coverage tracing using the trace ' 'module') group.add_argument('-D', '--coverdir', metavar='DIR', type=relative_filename, help='directory where coverage files are put') group.add_argument('-N', '--nocoverdir', action='store_const', const=None, dest='coverdir', help='put coverage files alongside modules') group.add_argument('-t', '--threshold', metavar='THRESHOLD', type=int, help='call gc.set_threshold(THRESHOLD)') group.add_argument('-n', '--nowindows', action='store_true', help='suppress error message boxes on Windows') group.add_argument('-F', '--forever', action='store_true', help='run the specified tests in a loop, until an ' 'error happens; imply --failfast') group.add_argument('--list-tests', action='store_true', help="only write the name of tests that will be run, " "don't execute them") group.add_argument('--list-cases', action='store_true', help='only write the name of test cases that will be run' ' , don\'t execute them') group.add_argument('-P', '--pgo', dest='pgo', action='store_true', help='enable Profile Guided Optimization (PGO) training') group.add_argument('--pgo-extended', action='store_true', help='enable extended PGO training (slower training)') group.add_argument('--fail-env-changed', action='store_true', help='if a test file alters the environment, mark ' 'the test as failed') group.add_argument('--junit-xml', dest='xmlpath', metavar='FILENAME', help='writes JUnit-style XML results to the specified ' 'file') group.add_argument('--tempdir', metavar='PATH', help='override the working directory for the test run') group.add_argument('--cleanup', action='store_true', help='remove old test_python_* directories') return parser def relative_filename(string): # CWD is replaced with a temporary dir before calling main(), so we # join it with the saved CWD so it ends up where the user expects. return os.path.join(os_helper.SAVEDCWD, string) def huntrleaks(string): args = string.split(':') if len(args) not in (2, 3): raise argparse.ArgumentTypeError( 'needs 2 or 3 colon-separated arguments') nwarmup = int(args[0]) if args[0] else 5 ntracked = int(args[1]) if args[1] else 4 fname = args[2] if len(args) > 2 and args[2] else 'reflog.txt' return nwarmup, ntracked, fname def resources_list(string): u = [x.lower() for x in string.split(',')] for r in u: if r == 'all' or r == 'none': continue if r[0] == '-': r = r[1:] if r not in RESOURCE_NAMES: raise argparse.ArgumentTypeError('invalid resource: ' + r) return u def _parse_args(args, **kwargs): # Defaults ns = Namespace() for k, v in kwargs.items(): if not hasattr(ns, k): raise TypeError('%r is an invalid keyword argument ' 'for this function' % k) setattr(ns, k, v) if ns.use_resources is None: ns.use_resources = [] parser = _create_parser() # Issue #14191: argparse doesn't support "intermixed" positional and # optional arguments. Use parse_known_args() as workaround. ns.args = parser.parse_known_args(args=args, namespace=ns)[1] for arg in ns.args: if arg.startswith('-'): parser.error("unrecognized arguments: %s" % arg) sys.exit(1) if ns.findleaks > 1: # --findleaks implies --fail-env-changed ns.fail_env_changed = True if ns.single and ns.fromfile: parser.error("-s and -f don't go together!") if ns.use_mp is not None and ns.trace: parser.error("-T and -j don't go together!") if ns.failfast and not (ns.verbose or ns.verbose3): parser.error("-G/--failfast needs either -v or -W") if ns.pgo and (ns.verbose or ns.verbose2 or ns.verbose3): parser.error("--pgo/-v don't go together!") if ns.pgo_extended: ns.pgo = True # pgo_extended implies pgo if ns.nowindows: print("Warning: the --nowindows (-n) option is deprecated. " "Use -vv to display assertions in stderr.", file=sys.stderr) if ns.quiet: ns.verbose = 0 if ns.timeout is not None: if ns.timeout <= 0: ns.timeout = None if ns.use_mp is not None: if ns.use_mp <= 0: # Use all cores + extras for tests that like to sleep ns.use_mp = 2 + (os.cpu_count() or 1) if ns.use: for a in ns.use: for r in a: if r == 'all': ns.use_resources[:] = ALL_RESOURCES continue if r == 'none': del ns.use_resources[:] continue remove = False if r[0] == '-': remove = True r = r[1:] if remove: if r in ns.use_resources: ns.use_resources.remove(r) elif r not in ns.use_resources: ns.use_resources.append(r) if ns.random_seed is not None: ns.randomize = True if ns.verbose: ns.header = True if ns.huntrleaks and ns.verbose3: ns.verbose3 = False print("WARNING: Disable --verbose3 because it's incompatible with " "--huntrleaks: see http://bugs.python.org/issue27103", file=sys.stderr) if ns.match_filename: if ns.match_tests is None: ns.match_tests = [] with open(ns.match_filename) as fp: for line in fp: ns.match_tests.append(line.strip()) if ns.ignore_filename: if ns.ignore_tests is None: ns.ignore_tests = [] with open(ns.ignore_filename) as fp: for line in fp: ns.ignore_tests.append(line.strip()) if ns.forever: # --forever implies --failfast ns.failfast = True return ns refleak.py 0000644 00000014030 15030171176 0006526 0 ustar 00 import os import re import sys import warnings from inspect import isabstract from test import support from test.support import os_helper from test.libregrtest.utils import clear_caches try: from _abc import _get_dump except ImportError: import weakref def _get_dump(cls): # Reimplement _get_dump() for pure-Python implementation of # the abc module (Lib/_py_abc.py) registry_weakrefs = set(weakref.ref(obj) for obj in cls._abc_registry) return (registry_weakrefs, cls._abc_cache, cls._abc_negative_cache, cls._abc_negative_cache_version) def dash_R(ns, test_name, test_func): """Run a test multiple times, looking for reference leaks. Returns: False if the test didn't leak references; True if we detected refleaks. """ # This code is hackish and inelegant, but it seems to do the job. import copyreg import collections.abc if not hasattr(sys, 'gettotalrefcount'): raise Exception("Tracking reference leaks requires a debug build " "of Python") # Avoid false positives due to various caches # filling slowly with random data: warm_caches() # Save current values for dash_R_cleanup() to restore. fs = warnings.filters[:] ps = copyreg.dispatch_table.copy() pic = sys.path_importer_cache.copy() try: import zipimport except ImportError: zdc = None # Run unmodified on platforms without zipimport support else: zdc = zipimport._zip_directory_cache.copy() abcs = {} for abc in [getattr(collections.abc, a) for a in collections.abc.__all__]: if not isabstract(abc): continue for obj in abc.__subclasses__() + [abc]: abcs[obj] = _get_dump(obj)[0] # bpo-31217: Integer pool to get a single integer object for the same # value. The pool is used to prevent false alarm when checking for memory # block leaks. Fill the pool with values in -1000..1000 which are the most # common (reference, memory block, file descriptor) differences. int_pool = {value: value for value in range(-1000, 1000)} def get_pooled_int(value): return int_pool.setdefault(value, value) nwarmup, ntracked, fname = ns.huntrleaks fname = os.path.join(os_helper.SAVEDCWD, fname) repcount = nwarmup + ntracked # Pre-allocate to ensure that the loop doesn't allocate anything new rep_range = list(range(repcount)) rc_deltas = [0] * repcount alloc_deltas = [0] * repcount fd_deltas = [0] * repcount getallocatedblocks = sys.getallocatedblocks gettotalrefcount = sys.gettotalrefcount fd_count = os_helper.fd_count # initialize variables to make pyflakes quiet rc_before = alloc_before = fd_before = 0 if not ns.quiet: print("beginning", repcount, "repetitions", file=sys.stderr) print(("1234567890"*(repcount//10 + 1))[:repcount], file=sys.stderr, flush=True) dash_R_cleanup(fs, ps, pic, zdc, abcs) for i in rep_range: test_func() dash_R_cleanup(fs, ps, pic, zdc, abcs) # dash_R_cleanup() ends with collecting cyclic trash: # read memory statistics immediately after. alloc_after = getallocatedblocks() rc_after = gettotalrefcount() fd_after = fd_count() if not ns.quiet: print('.', end='', file=sys.stderr, flush=True) rc_deltas[i] = get_pooled_int(rc_after - rc_before) alloc_deltas[i] = get_pooled_int(alloc_after - alloc_before) fd_deltas[i] = get_pooled_int(fd_after - fd_before) alloc_before = alloc_after rc_before = rc_after fd_before = fd_after if not ns.quiet: print(file=sys.stderr) # These checkers return False on success, True on failure def check_rc_deltas(deltas): # Checker for reference counters and memory blocks. # # bpo-30776: Try to ignore false positives: # # [3, 0, 0] # [0, 1, 0] # [8, -8, 1] # # Expected leaks: # # [5, 5, 6] # [10, 1, 1] return all(delta >= 1 for delta in deltas) def check_fd_deltas(deltas): return any(deltas) failed = False for deltas, item_name, checker in [ (rc_deltas, 'references', check_rc_deltas), (alloc_deltas, 'memory blocks', check_rc_deltas), (fd_deltas, 'file descriptors', check_fd_deltas) ]: # ignore warmup runs deltas = deltas[nwarmup:] if checker(deltas): msg = '%s leaked %s %s, sum=%s' % ( test_name, deltas, item_name, sum(deltas)) print(msg, file=sys.stderr, flush=True) with open(fname, "a") as refrep: print(msg, file=refrep) refrep.flush() failed = True return failed def dash_R_cleanup(fs, ps, pic, zdc, abcs): import copyreg import collections.abc # Restore some original values. warnings.filters[:] = fs copyreg.dispatch_table.clear() copyreg.dispatch_table.update(ps) sys.path_importer_cache.clear() sys.path_importer_cache.update(pic) try: import zipimport except ImportError: pass # Run unmodified on platforms without zipimport support else: zipimport._zip_directory_cache.clear() zipimport._zip_directory_cache.update(zdc) # clear type cache sys._clear_type_cache() # Clear ABC registries, restoring previously saved ABC registries. abs_classes = [getattr(collections.abc, a) for a in collections.abc.__all__] abs_classes = filter(isabstract, abs_classes) for abc in abs_classes: for obj in abc.__subclasses__() + [abc]: for ref in abcs.get(obj, set()): if ref() is not None: obj.register(ref()) obj._abc_caches_clear() clear_caches() def warm_caches(): # char cache s = bytes(range(256)) for i in range(256): s[i:i+1] # unicode cache [chr(i) for i in range(256)] # int cache list(range(-5, 257)) pgo.py 0000644 00000002515 15030171176 0005707 0 ustar 00 # Set of tests run by default if --pgo is specified. The tests below were # chosen based on the following criteria: either they exercise a commonly used # C extension module or type, or they run some relatively typical Python code. # Long running tests should be avoided because the PGO instrumented executable # runs slowly. PGO_TESTS = [ 'test_array', 'test_base64', 'test_binascii', 'test_binop', 'test_bisect', 'test_bytes', 'test_bz2', 'test_cmath', 'test_codecs', 'test_collections', 'test_complex', 'test_dataclasses', 'test_datetime', 'test_decimal', 'test_difflib', 'test_embed', 'test_float', 'test_fstring', 'test_functools', 'test_generators', 'test_hashlib', 'test_heapq', 'test_int', 'test_itertools', 'test_json', 'test_long', 'test_lzma', 'test_math', 'test_memoryview', 'test_operator', 'test_ordered_dict', 'test_patma', 'test_pickle', 'test_pprint', 'test_re', 'test_set', 'test_sqlite', 'test_statistics', 'test_struct', 'test_tabnanny', 'test_time', 'test_unicode', 'test_xml_etree', 'test_xml_etree_c', ] def setup_pgo_tests(ns): if not ns.args and not ns.pgo_extended: # run default set of tests for PGO training ns.args = PGO_TESTS[:] setup.py 0000644 00000013662 15030171176 0006267 0 ustar 00 import atexit import faulthandler import os import signal import sys import unittest from test import support from test.support.os_helper import TESTFN_UNDECODABLE, FS_NONASCII try: import gc except ImportError: gc = None from test.libregrtest.utils import (setup_unraisable_hook, setup_threading_excepthook) UNICODE_GUARD_ENV = "PYTHONREGRTEST_UNICODE_GUARD" def setup_tests(ns): try: stderr_fd = sys.__stderr__.fileno() except (ValueError, AttributeError): # Catch ValueError to catch io.UnsupportedOperation on TextIOBase # and ValueError on a closed stream. # # Catch AttributeError for stderr being None. stderr_fd = None else: # Display the Python traceback on fatal errors (e.g. segfault) faulthandler.enable(all_threads=True, file=stderr_fd) # Display the Python traceback on SIGALRM or SIGUSR1 signal signals = [] if hasattr(signal, 'SIGALRM'): signals.append(signal.SIGALRM) if hasattr(signal, 'SIGUSR1'): signals.append(signal.SIGUSR1) for signum in signals: faulthandler.register(signum, chain=True, file=stderr_fd) _adjust_resource_limits() replace_stdout() support.record_original_stdout(sys.stdout) if ns.testdir: # Prepend test directory to sys.path, so runtest() will be able # to locate tests sys.path.insert(0, os.path.abspath(ns.testdir)) # Some times __path__ and __file__ are not absolute (e.g. while running from # Lib/) and, if we change the CWD to run the tests in a temporary dir, some # imports might fail. This affects only the modules imported before os.chdir(). # These modules are searched first in sys.path[0] (so '' -- the CWD) and if # they are found in the CWD their __file__ and __path__ will be relative (this # happens before the chdir). All the modules imported after the chdir, are # not found in the CWD, and since the other paths in sys.path[1:] are absolute # (site.py absolutize them), the __file__ and __path__ will be absolute too. # Therefore it is necessary to absolutize manually the __file__ and __path__ of # the packages to prevent later imports to fail when the CWD is different. for module in sys.modules.values(): if hasattr(module, '__path__'): for index, path in enumerate(module.__path__): module.__path__[index] = os.path.abspath(path) if getattr(module, '__file__', None): module.__file__ = os.path.abspath(module.__file__) if ns.huntrleaks: unittest.BaseTestSuite._cleanup = False sys._deactivate_opcache() if ns.memlimit is not None: support.set_memlimit(ns.memlimit) if ns.threshold is not None: gc.set_threshold(ns.threshold) support.suppress_msvcrt_asserts(ns.verbose and ns.verbose >= 2) support.use_resources = ns.use_resources if hasattr(sys, 'addaudithook'): # Add an auditing hook for all tests to ensure PySys_Audit is tested def _test_audit_hook(name, args): pass sys.addaudithook(_test_audit_hook) setup_unraisable_hook() setup_threading_excepthook() if ns.timeout is not None: # For a slow buildbot worker, increase SHORT_TIMEOUT and LONG_TIMEOUT support.SHORT_TIMEOUT = max(support.SHORT_TIMEOUT, ns.timeout / 40) support.LONG_TIMEOUT = max(support.LONG_TIMEOUT, ns.timeout / 4) # If --timeout is short: reduce timeouts support.LOOPBACK_TIMEOUT = min(support.LOOPBACK_TIMEOUT, ns.timeout) support.INTERNET_TIMEOUT = min(support.INTERNET_TIMEOUT, ns.timeout) support.SHORT_TIMEOUT = min(support.SHORT_TIMEOUT, ns.timeout) support.LONG_TIMEOUT = min(support.LONG_TIMEOUT, ns.timeout) if ns.xmlpath: from test.support.testresult import RegressionTestResult RegressionTestResult.USE_XML = True # Ensure there's a non-ASCII character in env vars at all times to force # tests consider this case. See BPO-44647 for details. if TESTFN_UNDECODABLE and os.supports_bytes_environ: os.environb.setdefault(UNICODE_GUARD_ENV.encode(), TESTFN_UNDECODABLE) elif FS_NONASCII: os.environ.setdefault(UNICODE_GUARD_ENV, FS_NONASCII) def replace_stdout(): """Set stdout encoder error handler to backslashreplace (as stderr error handler) to avoid UnicodeEncodeError when printing a traceback""" stdout = sys.stdout try: fd = stdout.fileno() except ValueError: # On IDLE, sys.stdout has no file descriptor and is not a TextIOWrapper # object. Leaving sys.stdout unchanged. # # Catch ValueError to catch io.UnsupportedOperation on TextIOBase # and ValueError on a closed stream. return sys.stdout = open(fd, 'w', encoding=stdout.encoding, errors="backslashreplace", closefd=False, newline='\n') def restore_stdout(): sys.stdout.close() sys.stdout = stdout atexit.register(restore_stdout) def _adjust_resource_limits(): """Adjust the system resource limits (ulimit) if needed.""" try: import resource from resource import RLIMIT_NOFILE, RLIM_INFINITY except ImportError: return fd_limit, max_fds = resource.getrlimit(RLIMIT_NOFILE) # On macOS the default fd limit is sometimes too low (256) for our # test suite to succeed. Raise it to something more reasonable. # 1024 is a common Linux default. desired_fds = 1024 if fd_limit < desired_fds and fd_limit < max_fds: new_fd_limit = min(desired_fds, max_fds) try: resource.setrlimit(RLIMIT_NOFILE, (new_fd_limit, max_fds)) print(f"Raised RLIMIT_NOFILE: {fd_limit} -> {new_fd_limit}") except (ValueError, OSError) as err: print(f"Unable to raise RLIMIT_NOFILE from {fd_limit} to " f"{new_fd_limit}: {err}.") save_env.py 0000644 00000031217 15030171176 0006731 0 ustar 00 import builtins import locale import os import sys import threading from test import support from test.support import os_helper from test.libregrtest.utils import print_warning class SkipTestEnvironment(Exception): pass # Unit tests are supposed to leave the execution environment unchanged # once they complete. But sometimes tests have bugs, especially when # tests fail, and the changes to environment go on to mess up other # tests. This can cause issues with buildbot stability, since tests # are run in random order and so problems may appear to come and go. # There are a few things we can save and restore to mitigate this, and # the following context manager handles this task. class saved_test_environment: """Save bits of the test environment and restore them at block exit. with saved_test_environment(testname, verbose, quiet): #stuff Unless quiet is True, a warning is printed to stderr if any of the saved items was changed by the test. The support.environment_altered attribute is set to True if a change is detected. If verbose is more than 1, the before and after state of changed items is also printed. """ def __init__(self, testname, verbose=0, quiet=False, *, pgo=False): self.testname = testname self.verbose = verbose self.quiet = quiet self.pgo = pgo # To add things to save and restore, add a name XXX to the resources list # and add corresponding get_XXX/restore_XXX functions. get_XXX should # return the value to be saved and compared against a second call to the # get function when test execution completes. restore_XXX should accept # the saved value and restore the resource using it. It will be called if # and only if a change in the value is detected. # # Note: XXX will have any '.' replaced with '_' characters when determining # the corresponding method names. resources = ('sys.argv', 'cwd', 'sys.stdin', 'sys.stdout', 'sys.stderr', 'os.environ', 'sys.path', 'sys.path_hooks', '__import__', 'warnings.filters', 'asyncore.socket_map', 'logging._handlers', 'logging._handlerList', 'sys.gettrace', 'sys.warnoptions', # multiprocessing.process._cleanup() may release ref # to a thread, so check processes first. 'multiprocessing.process._dangling', 'threading._dangling', 'sysconfig._CONFIG_VARS', 'sysconfig._INSTALL_SCHEMES', 'files', 'locale', 'warnings.showwarning', 'shutil_archive_formats', 'shutil_unpack_formats', 'asyncio.events._event_loop_policy', 'urllib.requests._url_tempfiles', 'urllib.requests._opener', ) def get_module(self, name): # function for restore() methods return sys.modules[name] def try_get_module(self, name): # function for get() methods try: return self.get_module(name) except KeyError: raise SkipTestEnvironment def get_urllib_requests__url_tempfiles(self): urllib_request = self.try_get_module('urllib.request') return list(urllib_request._url_tempfiles) def restore_urllib_requests__url_tempfiles(self, tempfiles): for filename in tempfiles: os_helper.unlink(filename) def get_urllib_requests__opener(self): urllib_request = self.try_get_module('urllib.request') return urllib_request._opener def restore_urllib_requests__opener(self, opener): urllib_request = self.get_module('urllib.request') urllib_request._opener = opener def get_asyncio_events__event_loop_policy(self): self.try_get_module('asyncio') return support.maybe_get_event_loop_policy() def restore_asyncio_events__event_loop_policy(self, policy): asyncio = self.get_module('asyncio') asyncio.set_event_loop_policy(policy) def get_sys_argv(self): return id(sys.argv), sys.argv, sys.argv[:] def restore_sys_argv(self, saved_argv): sys.argv = saved_argv[1] sys.argv[:] = saved_argv[2] def get_cwd(self): return os.getcwd() def restore_cwd(self, saved_cwd): os.chdir(saved_cwd) def get_sys_stdout(self): return sys.stdout def restore_sys_stdout(self, saved_stdout): sys.stdout = saved_stdout def get_sys_stderr(self): return sys.stderr def restore_sys_stderr(self, saved_stderr): sys.stderr = saved_stderr def get_sys_stdin(self): return sys.stdin def restore_sys_stdin(self, saved_stdin): sys.stdin = saved_stdin def get_os_environ(self): return id(os.environ), os.environ, dict(os.environ) def restore_os_environ(self, saved_environ): os.environ = saved_environ[1] os.environ.clear() os.environ.update(saved_environ[2]) def get_sys_path(self): return id(sys.path), sys.path, sys.path[:] def restore_sys_path(self, saved_path): sys.path = saved_path[1] sys.path[:] = saved_path[2] def get_sys_path_hooks(self): return id(sys.path_hooks), sys.path_hooks, sys.path_hooks[:] def restore_sys_path_hooks(self, saved_hooks): sys.path_hooks = saved_hooks[1] sys.path_hooks[:] = saved_hooks[2] def get_sys_gettrace(self): return sys.gettrace() def restore_sys_gettrace(self, trace_fxn): sys.settrace(trace_fxn) def get___import__(self): return builtins.__import__ def restore___import__(self, import_): builtins.__import__ = import_ def get_warnings_filters(self): warnings = self.try_get_module('warnings') return id(warnings.filters), warnings.filters, warnings.filters[:] def restore_warnings_filters(self, saved_filters): warnings = self.get_module('warnings') warnings.filters = saved_filters[1] warnings.filters[:] = saved_filters[2] def get_asyncore_socket_map(self): asyncore = sys.modules.get('asyncore') # XXX Making a copy keeps objects alive until __exit__ gets called. return asyncore and asyncore.socket_map.copy() or {} def restore_asyncore_socket_map(self, saved_map): asyncore = sys.modules.get('asyncore') if asyncore is not None: asyncore.close_all(ignore_all=True) asyncore.socket_map.update(saved_map) def get_shutil_archive_formats(self): shutil = self.try_get_module('shutil') # we could call get_archives_formats() but that only returns the # registry keys; we want to check the values too (the functions that # are registered) return shutil._ARCHIVE_FORMATS, shutil._ARCHIVE_FORMATS.copy() def restore_shutil_archive_formats(self, saved): shutil = self.get_module('shutil') shutil._ARCHIVE_FORMATS = saved[0] shutil._ARCHIVE_FORMATS.clear() shutil._ARCHIVE_FORMATS.update(saved[1]) def get_shutil_unpack_formats(self): shutil = self.try_get_module('shutil') return shutil._UNPACK_FORMATS, shutil._UNPACK_FORMATS.copy() def restore_shutil_unpack_formats(self, saved): shutil = self.get_module('shutil') shutil._UNPACK_FORMATS = saved[0] shutil._UNPACK_FORMATS.clear() shutil._UNPACK_FORMATS.update(saved[1]) def get_logging__handlers(self): logging = self.try_get_module('logging') # _handlers is a WeakValueDictionary return id(logging._handlers), logging._handlers, logging._handlers.copy() def restore_logging__handlers(self, saved_handlers): # Can't easily revert the logging state pass def get_logging__handlerList(self): logging = self.try_get_module('logging') # _handlerList is a list of weakrefs to handlers return id(logging._handlerList), logging._handlerList, logging._handlerList[:] def restore_logging__handlerList(self, saved_handlerList): # Can't easily revert the logging state pass def get_sys_warnoptions(self): return id(sys.warnoptions), sys.warnoptions, sys.warnoptions[:] def restore_sys_warnoptions(self, saved_options): sys.warnoptions = saved_options[1] sys.warnoptions[:] = saved_options[2] # Controlling dangling references to Thread objects can make it easier # to track reference leaks. def get_threading__dangling(self): # This copies the weakrefs without making any strong reference return threading._dangling.copy() def restore_threading__dangling(self, saved): threading._dangling.clear() threading._dangling.update(saved) # Same for Process objects def get_multiprocessing_process__dangling(self): multiprocessing_process = self.try_get_module('multiprocessing.process') # Unjoined process objects can survive after process exits multiprocessing_process._cleanup() # This copies the weakrefs without making any strong reference return multiprocessing_process._dangling.copy() def restore_multiprocessing_process__dangling(self, saved): multiprocessing_process = self.get_module('multiprocessing.process') multiprocessing_process._dangling.clear() multiprocessing_process._dangling.update(saved) def get_sysconfig__CONFIG_VARS(self): # make sure the dict is initialized sysconfig = self.try_get_module('sysconfig') sysconfig.get_config_var('prefix') return (id(sysconfig._CONFIG_VARS), sysconfig._CONFIG_VARS, dict(sysconfig._CONFIG_VARS)) def restore_sysconfig__CONFIG_VARS(self, saved): sysconfig = self.get_module('sysconfig') sysconfig._CONFIG_VARS = saved[1] sysconfig._CONFIG_VARS.clear() sysconfig._CONFIG_VARS.update(saved[2]) def get_sysconfig__INSTALL_SCHEMES(self): sysconfig = self.try_get_module('sysconfig') return (id(sysconfig._INSTALL_SCHEMES), sysconfig._INSTALL_SCHEMES, sysconfig._INSTALL_SCHEMES.copy()) def restore_sysconfig__INSTALL_SCHEMES(self, saved): sysconfig = self.get_module('sysconfig') sysconfig._INSTALL_SCHEMES = saved[1] sysconfig._INSTALL_SCHEMES.clear() sysconfig._INSTALL_SCHEMES.update(saved[2]) def get_files(self): return sorted(fn + ('/' if os.path.isdir(fn) else '') for fn in os.listdir()) def restore_files(self, saved_value): fn = os_helper.TESTFN if fn not in saved_value and (fn + '/') not in saved_value: if os.path.isfile(fn): os_helper.unlink(fn) elif os.path.isdir(fn): os_helper.rmtree(fn) _lc = [getattr(locale, lc) for lc in dir(locale) if lc.startswith('LC_')] def get_locale(self): pairings = [] for lc in self._lc: try: pairings.append((lc, locale.setlocale(lc, None))) except (TypeError, ValueError): continue return pairings def restore_locale(self, saved): for lc, setting in saved: locale.setlocale(lc, setting) def get_warnings_showwarning(self): warnings = self.try_get_module('warnings') return warnings.showwarning def restore_warnings_showwarning(self, fxn): warnings = self.get_module('warnings') warnings.showwarning = fxn def resource_info(self): for name in self.resources: method_suffix = name.replace('.', '_') get_name = 'get_' + method_suffix restore_name = 'restore_' + method_suffix yield name, getattr(self, get_name), getattr(self, restore_name) def __enter__(self): self.saved_values = [] for name, get, restore in self.resource_info(): try: original = get() except SkipTestEnvironment: continue self.saved_values.append((name, get, restore, original)) return self def __exit__(self, exc_type, exc_val, exc_tb): saved_values = self.saved_values self.saved_values = None # Some resources use weak references support.gc_collect() for name, get, restore, original in saved_values: current = get() # Check for changes to the resource's value if current != original: support.environment_altered = True restore(original) if not self.quiet and not self.pgo: print_warning(f"{name} was modified by {self.testname}") print(f" Before: {original}\n After: {current} ", file=sys.stderr, flush=True) return False main.py 0000644 00000062250 15030171176 0006050 0 ustar 00 import faulthandler import locale import os import platform import random import re import sys import sysconfig import tempfile import time import unittest from test.libregrtest.cmdline import _parse_args from test.libregrtest.runtest import ( findtests, runtest, get_abs_module, is_failed, STDTESTS, NOTTESTS, PROGRESS_MIN_TIME, Passed, Failed, EnvChanged, Skipped, ResourceDenied, Interrupted, ChildError, DidNotRun) from test.libregrtest.setup import setup_tests from test.libregrtest.pgo import setup_pgo_tests from test.libregrtest.utils import removepy, count, format_duration, printlist from test import support from test.support import os_helper # bpo-38203: Maximum delay in seconds to exit Python (call Py_Finalize()). # Used to protect against threading._shutdown() hang. # Must be smaller than buildbot "1200 seconds without output" limit. EXIT_TIMEOUT = 120.0 class Regrtest: """Execute a test suite. This also parses command-line options and modifies its behavior accordingly. tests -- a list of strings containing test names (optional) testdir -- the directory in which to look for tests (optional) Users other than the Python test suite will certainly want to specify testdir; if it's omitted, the directory containing the Python test suite is searched for. If the tests argument is omitted, the tests listed on the command-line will be used. If that's empty, too, then all *.py files beginning with test_ will be used. The other default arguments (verbose, quiet, exclude, single, randomize, findleaks, use_resources, trace, coverdir, print_slow, and random_seed) allow programmers calling main() directly to set the values that would normally be set by flags on the command line. """ def __init__(self): # Namespace of command line options self.ns = None # tests self.tests = [] self.selected = [] # test results self.good = [] self.bad = [] self.skipped = [] self.resource_denieds = [] self.environment_changed = [] self.run_no_tests = [] self.need_rerun = [] self.rerun = [] self.first_result = None self.interrupted = False # used by --slow self.test_times = [] # used by --coverage, trace.Trace instance self.tracer = None # used to display the progress bar "[ 3/100]" self.start_time = time.monotonic() self.test_count = '' self.test_count_width = 1 # used by --single self.next_single_test = None self.next_single_filename = None # used by --junit-xml self.testsuite_xml = None # misc self.win_load_tracker = None self.tmp_dir = None self.worker_test_name = None def get_executed(self): return (set(self.good) | set(self.bad) | set(self.skipped) | set(self.resource_denieds) | set(self.environment_changed) | set(self.run_no_tests)) def accumulate_result(self, result, rerun=False): test_name = result.name if not isinstance(result, (ChildError, Interrupted)) and not rerun: self.test_times.append((result.duration_sec, test_name)) if isinstance(result, Passed): self.good.append(test_name) elif isinstance(result, ResourceDenied): self.skipped.append(test_name) self.resource_denieds.append(test_name) elif isinstance(result, Skipped): self.skipped.append(test_name) elif isinstance(result, EnvChanged): self.environment_changed.append(test_name) elif isinstance(result, Failed): if not rerun: self.bad.append(test_name) self.need_rerun.append(result) elif isinstance(result, DidNotRun): self.run_no_tests.append(test_name) elif isinstance(result, Interrupted): self.interrupted = True else: raise ValueError("invalid test result: %r" % result) if rerun and not isinstance(result, (Failed, Interrupted)): self.bad.remove(test_name) xml_data = result.xml_data if xml_data: import xml.etree.ElementTree as ET for e in xml_data: try: self.testsuite_xml.append(ET.fromstring(e)) except ET.ParseError: print(xml_data, file=sys.__stderr__) raise def log(self, line=''): empty = not line # add the system load prefix: "load avg: 1.80 " load_avg = self.getloadavg() if load_avg is not None: line = f"load avg: {load_avg:.2f} {line}" # add the timestamp prefix: "0:01:05 " test_time = time.monotonic() - self.start_time mins, secs = divmod(int(test_time), 60) hours, mins = divmod(mins, 60) test_time = "%d:%02d:%02d" % (hours, mins, secs) line = f"{test_time} {line}" if empty: line = line[:-1] print(line, flush=True) def display_progress(self, test_index, text): if self.ns.quiet: return # "[ 51/405/1] test_tcl passed" line = f"{test_index:{self.test_count_width}}{self.test_count}" fails = len(self.bad) + len(self.environment_changed) if fails and not self.ns.pgo: line = f"{line}/{fails}" self.log(f"[{line}] {text}") def parse_args(self, kwargs): ns = _parse_args(sys.argv[1:], **kwargs) if ns.xmlpath: support.junit_xml_list = self.testsuite_xml = [] worker_args = ns.worker_args if worker_args is not None: from test.libregrtest.runtest_mp import parse_worker_args ns, test_name = parse_worker_args(ns.worker_args) ns.worker_args = worker_args self.worker_test_name = test_name # Strip .py extensions. removepy(ns.args) if ns.huntrleaks: warmup, repetitions, _ = ns.huntrleaks if warmup < 1 or repetitions < 1: msg = ("Invalid values for the --huntrleaks/-R parameters. The " "number of warmups and repetitions must be at least 1 " "each (1:1).") print(msg, file=sys.stderr, flush=True) sys.exit(2) if ns.tempdir: ns.tempdir = os.path.expanduser(ns.tempdir) self.ns = ns def find_tests(self, tests): self.tests = tests if self.ns.single: self.next_single_filename = os.path.join(self.tmp_dir, 'pynexttest') try: with open(self.next_single_filename, 'r') as fp: next_test = fp.read().strip() self.tests = [next_test] except OSError: pass if self.ns.fromfile: self.tests = [] # regex to match 'test_builtin' in line: # '0:00:00 [ 4/400] test_builtin -- test_dict took 1 sec' regex = re.compile(r'\btest_[a-zA-Z0-9_]+\b') with open(os.path.join(os_helper.SAVEDCWD, self.ns.fromfile)) as fp: for line in fp: line = line.split('#', 1)[0] line = line.strip() match = regex.search(line) if match is not None: self.tests.append(match.group()) removepy(self.tests) if self.ns.pgo: # add default PGO tests if no tests are specified setup_pgo_tests(self.ns) stdtests = STDTESTS[:] nottests = NOTTESTS.copy() if self.ns.exclude: for arg in self.ns.args: if arg in stdtests: stdtests.remove(arg) nottests.add(arg) self.ns.args = [] # if testdir is set, then we are not running the python tests suite, so # don't add default tests to be executed or skipped (pass empty values) if self.ns.testdir: alltests = findtests(self.ns.testdir, list(), set()) else: alltests = findtests(self.ns.testdir, stdtests, nottests) if not self.ns.fromfile: self.selected = self.tests or self.ns.args or alltests else: self.selected = self.tests if self.ns.single: self.selected = self.selected[:1] try: pos = alltests.index(self.selected[0]) self.next_single_test = alltests[pos + 1] except IndexError: pass # Remove all the selected tests that precede start if it's set. if self.ns.start: try: del self.selected[:self.selected.index(self.ns.start)] except ValueError: print("Couldn't find starting test (%s), using all tests" % self.ns.start, file=sys.stderr) if self.ns.randomize: if self.ns.random_seed is None: self.ns.random_seed = random.randrange(10000000) random.seed(self.ns.random_seed) random.shuffle(self.selected) def list_tests(self): for name in self.selected: print(name) def _list_cases(self, suite): for test in suite: if isinstance(test, unittest.loader._FailedTest): continue if isinstance(test, unittest.TestSuite): self._list_cases(test) elif isinstance(test, unittest.TestCase): if support.match_test(test): print(test.id()) def list_cases(self): support.verbose = False support.set_match_tests(self.ns.match_tests, self.ns.ignore_tests) for test_name in self.selected: abstest = get_abs_module(self.ns, test_name) try: suite = unittest.defaultTestLoader.loadTestsFromName(abstest) self._list_cases(suite) except unittest.SkipTest: self.skipped.append(test_name) if self.skipped: print(file=sys.stderr) print(count(len(self.skipped), "test"), "skipped:", file=sys.stderr) printlist(self.skipped, file=sys.stderr) def rerun_failed_tests(self): self.ns.verbose = True self.ns.failfast = False self.ns.verbose3 = False self.first_result = self.get_tests_result() self.log() self.log("Re-running failed tests in verbose mode") rerun_list = list(self.need_rerun) self.need_rerun.clear() for result in rerun_list: test_name = result.name self.rerun.append(test_name) errors = result.errors or [] failures = result.failures or [] error_names = [test_full_name.split(" ")[0] for (test_full_name, *_) in errors] failure_names = [test_full_name.split(" ")[0] for (test_full_name, *_) in failures] self.ns.verbose = True orig_match_tests = self.ns.match_tests if errors or failures: if self.ns.match_tests is None: self.ns.match_tests = [] self.ns.match_tests.extend(error_names) self.ns.match_tests.extend(failure_names) matching = "matching: " + ", ".join(self.ns.match_tests) self.log(f"Re-running {test_name} in verbose mode ({matching})") else: self.log(f"Re-running {test_name} in verbose mode") result = runtest(self.ns, test_name) self.ns.match_tests = orig_match_tests self.accumulate_result(result, rerun=True) if isinstance(result, Interrupted): break if self.bad: print(count(len(self.bad), 'test'), "failed again:") printlist(self.bad) self.display_result() def display_result(self): # If running the test suite for PGO then no one cares about results. if self.ns.pgo: return print() print("== Tests result: %s ==" % self.get_tests_result()) if self.interrupted: print("Test suite interrupted by signal SIGINT.") omitted = set(self.selected) - self.get_executed() if omitted: print() print(count(len(omitted), "test"), "omitted:") printlist(omitted) if self.good and not self.ns.quiet: print() if (not self.bad and not self.skipped and not self.interrupted and len(self.good) > 1): print("All", end=' ') print(count(len(self.good), "test"), "OK.") if self.ns.print_slow: self.test_times.sort(reverse=True) print() print("10 slowest tests:") for test_time, test in self.test_times[:10]: print("- %s: %s" % (test, format_duration(test_time))) if self.bad: print() print(count(len(self.bad), "test"), "failed:") printlist(self.bad) if self.environment_changed: print() print("{} altered the execution environment:".format( count(len(self.environment_changed), "test"))) printlist(self.environment_changed) if self.skipped and not self.ns.quiet: print() print(count(len(self.skipped), "test"), "skipped:") printlist(self.skipped) if self.rerun: print() print("%s:" % count(len(self.rerun), "re-run test")) printlist(self.rerun) if self.run_no_tests: print() print(count(len(self.run_no_tests), "test"), "run no tests:") printlist(self.run_no_tests) def run_tests_sequential(self): if self.ns.trace: import trace self.tracer = trace.Trace(trace=False, count=True) save_modules = sys.modules.keys() msg = "Run tests sequentially" if self.ns.timeout: msg += " (timeout: %s)" % format_duration(self.ns.timeout) self.log(msg) previous_test = None for test_index, test_name in enumerate(self.tests, 1): start_time = time.monotonic() text = test_name if previous_test: text = '%s -- %s' % (text, previous_test) self.display_progress(test_index, text) if self.tracer: # If we're tracing code coverage, then we don't exit with status # if on a false return value from main. cmd = ('result = runtest(self.ns, test_name); ' 'self.accumulate_result(result)') ns = dict(locals()) self.tracer.runctx(cmd, globals=globals(), locals=ns) result = ns['result'] else: result = runtest(self.ns, test_name) self.accumulate_result(result) if isinstance(result, Interrupted): break previous_test = str(result) test_time = time.monotonic() - start_time if test_time >= PROGRESS_MIN_TIME: previous_test = "%s in %s" % (previous_test, format_duration(test_time)) elif isinstance(result, Passed): # be quiet: say nothing if the test passed shortly previous_test = None # Unload the newly imported modules (best effort finalization) for module in sys.modules.keys(): if module not in save_modules and module.startswith("test."): support.unload(module) if self.ns.failfast and is_failed(result, self.ns): break if previous_test: print(previous_test) def _test_forever(self, tests): while True: for test_name in tests: yield test_name if self.bad: return if self.ns.fail_env_changed and self.environment_changed: return def display_header(self): # Print basic platform information print("==", platform.python_implementation(), *sys.version.split()) print("==", platform.platform(aliased=True), "%s-endian" % sys.byteorder) print("== cwd:", os.getcwd()) cpu_count = os.cpu_count() if cpu_count: print("== CPU count:", cpu_count) print("== encodings: locale=%s, FS=%s" % (locale.getpreferredencoding(False), sys.getfilesystemencoding())) def get_tests_result(self): result = [] if self.bad: result.append("FAILURE") elif self.ns.fail_env_changed and self.environment_changed: result.append("ENV CHANGED") elif not any((self.good, self.bad, self.skipped, self.interrupted, self.environment_changed)): result.append("NO TEST RUN") if self.interrupted: result.append("INTERRUPTED") if not result: result.append("SUCCESS") result = ', '.join(result) if self.first_result: result = '%s then %s' % (self.first_result, result) return result def run_tests(self): # For a partial run, we do not need to clutter the output. if (self.ns.header or not(self.ns.pgo or self.ns.quiet or self.ns.single or self.tests or self.ns.args)): self.display_header() if self.ns.huntrleaks: warmup, repetitions, _ = self.ns.huntrleaks if warmup < 3: msg = ("WARNING: Running tests with --huntrleaks/-R and less than " "3 warmup repetitions can give false positives!") print(msg, file=sys.stdout, flush=True) if self.ns.randomize: print("Using random seed", self.ns.random_seed) if self.ns.forever: self.tests = self._test_forever(list(self.selected)) self.test_count = '' self.test_count_width = 3 else: self.tests = iter(self.selected) self.test_count = '/{}'.format(len(self.selected)) self.test_count_width = len(self.test_count) - 1 if self.ns.use_mp: from test.libregrtest.runtest_mp import run_tests_multiprocess # If we're on windows and this is the parent runner (not a worker), # track the load average. if sys.platform == 'win32' and self.worker_test_name is None: from test.libregrtest.win_utils import WindowsLoadTracker try: self.win_load_tracker = WindowsLoadTracker() except PermissionError as error: # Standard accounts may not have access to the performance # counters. print(f'Failed to create WindowsLoadTracker: {error}') try: run_tests_multiprocess(self) finally: if self.win_load_tracker is not None: self.win_load_tracker.close() self.win_load_tracker = None else: self.run_tests_sequential() def finalize(self): if self.next_single_filename: if self.next_single_test: with open(self.next_single_filename, 'w') as fp: fp.write(self.next_single_test + '\n') else: os.unlink(self.next_single_filename) if self.tracer: r = self.tracer.results() r.write_results(show_missing=True, summary=True, coverdir=self.ns.coverdir) print() duration = time.monotonic() - self.start_time print("Total duration: %s" % format_duration(duration)) print("Tests result: %s" % self.get_tests_result()) if self.ns.runleaks: os.system("leaks %d" % os.getpid()) def save_xml_result(self): if not self.ns.xmlpath and not self.testsuite_xml: return import xml.etree.ElementTree as ET root = ET.Element("testsuites") # Manually count the totals for the overall summary totals = {'tests': 0, 'errors': 0, 'failures': 0} for suite in self.testsuite_xml: root.append(suite) for k in totals: try: totals[k] += int(suite.get(k, 0)) except ValueError: pass for k, v in totals.items(): root.set(k, str(v)) xmlpath = os.path.join(os_helper.SAVEDCWD, self.ns.xmlpath) with open(xmlpath, 'wb') as f: for s in ET.tostringlist(root): f.write(s) def set_temp_dir(self): if self.ns.tempdir: self.tmp_dir = self.ns.tempdir if not self.tmp_dir: # When tests are run from the Python build directory, it is best practice # to keep the test files in a subfolder. This eases the cleanup of leftover # files using the "make distclean" command. if sysconfig.is_python_build(): self.tmp_dir = sysconfig.get_config_var('abs_builddir') if self.tmp_dir is None: # bpo-30284: On Windows, only srcdir is available. Using # abs_builddir mostly matters on UNIX when building Python # out of the source tree, especially when the source tree # is read only. self.tmp_dir = sysconfig.get_config_var('srcdir') self.tmp_dir = os.path.join(self.tmp_dir, 'build') else: self.tmp_dir = tempfile.gettempdir() self.tmp_dir = os.path.abspath(self.tmp_dir) def create_temp_dir(self): os.makedirs(self.tmp_dir, exist_ok=True) # Define a writable temp dir that will be used as cwd while running # the tests. The name of the dir includes the pid to allow parallel # testing (see the -j option). pid = os.getpid() if self.worker_test_name is not None: test_cwd = 'test_python_worker_{}'.format(pid) else: test_cwd = 'test_python_{}'.format(pid) test_cwd += os_helper.FS_NONASCII test_cwd = os.path.join(self.tmp_dir, test_cwd) return test_cwd def cleanup(self): import glob path = os.path.join(glob.escape(self.tmp_dir), 'test_python_*') print("Cleanup %s directory" % self.tmp_dir) for name in glob.glob(path): if os.path.isdir(name): print("Remove directory: %s" % name) os_helper.rmtree(name) else: print("Remove file: %s" % name) os_helper.unlink(name) def main(self, tests=None, **kwargs): self.parse_args(kwargs) self.set_temp_dir() if self.ns.cleanup: self.cleanup() sys.exit(0) test_cwd = self.create_temp_dir() try: # Run the tests in a context manager that temporarily changes the CWD # to a temporary and writable directory. If it's not possible to # create or change the CWD, the original CWD will be used. # The original CWD is available from os_helper.SAVEDCWD. with os_helper.temp_cwd(test_cwd, quiet=True): # When using multiprocessing, worker processes will use test_cwd # as their parent temporary directory. So when the main process # exit, it removes also subdirectories of worker processes. self.ns.tempdir = test_cwd self._main(tests, kwargs) except SystemExit as exc: # bpo-38203: Python can hang at exit in Py_Finalize(), especially # on threading._shutdown() call: put a timeout faulthandler.dump_traceback_later(EXIT_TIMEOUT, exit=True) sys.exit(exc.code) def getloadavg(self): if self.win_load_tracker is not None: return self.win_load_tracker.getloadavg() if hasattr(os, 'getloadavg'): return os.getloadavg()[0] return None def _main(self, tests, kwargs): if self.worker_test_name is not None: from test.libregrtest.runtest_mp import run_tests_worker run_tests_worker(self.ns, self.worker_test_name) if self.ns.wait: input("Press any key to continue...") support.PGO = self.ns.pgo support.PGO_EXTENDED = self.ns.pgo_extended setup_tests(self.ns) self.find_tests(tests) if self.ns.list_tests: self.list_tests() sys.exit(0) if self.ns.list_cases: self.list_cases() sys.exit(0) self.run_tests() self.display_result() if self.ns.verbose2 and self.bad: self.rerun_failed_tests() self.finalize() self.save_xml_result() if self.bad: sys.exit(2) if self.interrupted: sys.exit(130) if self.ns.fail_env_changed and self.environment_changed: sys.exit(3) sys.exit(0) def main(tests=None, **kwargs): """Run the Python suite.""" Regrtest().main(tests=tests, **kwargs) runtest_mp.py 0000644 00000042163 15030171176 0007325 0 ustar 00 import collections import faulthandler import json import os import queue import signal import subprocess import sys import threading import time import traceback from typing import NamedTuple, NoReturn, Literal, Any from test import support from test.support import os_helper from test.libregrtest.cmdline import Namespace from test.libregrtest.main import Regrtest from test.libregrtest.runtest import ( runtest, is_failed, TestResult, Interrupted, Timeout, ChildError, PROGRESS_MIN_TIME) from test.libregrtest.setup import setup_tests from test.libregrtest.utils import format_duration, print_warning # Display the running tests if nothing happened last N seconds PROGRESS_UPDATE = 30.0 # seconds assert PROGRESS_UPDATE >= PROGRESS_MIN_TIME # Kill the main process after 5 minutes. It is supposed to write an update # every PROGRESS_UPDATE seconds. Tolerate 5 minutes for Python slowest # buildbot workers. MAIN_PROCESS_TIMEOUT = 5 * 60.0 assert MAIN_PROCESS_TIMEOUT >= PROGRESS_UPDATE # Time to wait until a worker completes: should be immediate JOIN_TIMEOUT = 30.0 # seconds USE_PROCESS_GROUP = (hasattr(os, "setsid") and hasattr(os, "killpg")) def must_stop(result: TestResult, ns: Namespace) -> bool: if isinstance(result, Interrupted): return True if ns.failfast and is_failed(result, ns): return True return False def parse_worker_args(worker_args) -> tuple[Namespace, str]: ns_dict, test_name = json.loads(worker_args) ns = Namespace(**ns_dict) return (ns, test_name) def run_test_in_subprocess(testname: str, ns: Namespace) -> subprocess.Popen: ns_dict = vars(ns) worker_args = (ns_dict, testname) worker_args = json.dumps(worker_args) cmd = [sys.executable, *support.args_from_interpreter_flags(), '-u', # Unbuffered stdout and stderr '-m', 'test.regrtest', '--worker-args', worker_args] # Running the child from the same working directory as regrtest's original # invocation ensures that TEMPDIR for the child is the same when # sysconfig.is_python_build() is true. See issue 15300. kw = {} if USE_PROCESS_GROUP: kw['start_new_session'] = True return subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True, close_fds=(os.name != 'nt'), cwd=os_helper.SAVEDCWD, **kw) def run_tests_worker(ns: Namespace, test_name: str) -> NoReturn: setup_tests(ns) result = runtest(ns, test_name) print() # Force a newline (just in case) # Serialize TestResult as dict in JSON print(json.dumps(result, cls=EncodeTestResult), flush=True) sys.exit(0) # We do not use a generator so multiple threads can call next(). class MultiprocessIterator: """A thread-safe iterator over tests for multiprocess mode.""" def __init__(self, tests_iter): self.lock = threading.Lock() self.tests_iter = tests_iter def __iter__(self): return self def __next__(self): with self.lock: if self.tests_iter is None: raise StopIteration return next(self.tests_iter) def stop(self): with self.lock: self.tests_iter = None class MultiprocessResult(NamedTuple): result: TestResult stdout: str stderr: str error_msg: str ExcStr = str QueueOutput = tuple[Literal[False], MultiprocessResult] | tuple[Literal[True], ExcStr] class ExitThread(Exception): pass class TestWorkerProcess(threading.Thread): def __init__(self, worker_id: int, runner: "MultiprocessTestRunner") -> None: super().__init__() self.worker_id = worker_id self.pending = runner.pending self.output = runner.output self.ns = runner.ns self.timeout = runner.worker_timeout self.regrtest = runner.regrtest self.current_test_name = None self.start_time = None self._popen = None self._killed = False self._stopped = False def __repr__(self) -> str: info = [f'TestWorkerProcess #{self.worker_id}'] if self.is_alive(): info.append("running") else: info.append('stopped') test = self.current_test_name if test: info.append(f'test={test}') popen = self._popen if popen is not None: dt = time.monotonic() - self.start_time info.extend((f'pid={self._popen.pid}', f'time={format_duration(dt)}')) return '<%s>' % ' '.join(info) def _kill(self) -> None: popen = self._popen if popen is None: return if self._killed: return self._killed = True if USE_PROCESS_GROUP: what = f"{self} process group" else: what = f"{self}" print(f"Kill {what}", file=sys.stderr, flush=True) try: if USE_PROCESS_GROUP: os.killpg(popen.pid, signal.SIGKILL) else: popen.kill() except ProcessLookupError: # popen.kill(): the process completed, the TestWorkerProcess thread # read its exit status, but Popen.send_signal() read the returncode # just before Popen.wait() set returncode. pass except OSError as exc: print_warning(f"Failed to kill {what}: {exc!r}") def stop(self) -> None: # Method called from a different thread to stop this thread self._stopped = True self._kill() def mp_result_error( self, test_result: TestResult, stdout: str = '', stderr: str = '', err_msg=None ) -> MultiprocessResult: test_result.duration_sec = time.monotonic() - self.start_time return MultiprocessResult(test_result, stdout, stderr, err_msg) def _run_process(self, test_name: str) -> tuple[int, str, str]: self.start_time = time.monotonic() self.current_test_name = test_name try: popen = run_test_in_subprocess(test_name, self.ns) self._killed = False self._popen = popen except: self.current_test_name = None raise try: if self._stopped: # If kill() has been called before self._popen is set, # self._popen is still running. Call again kill() # to ensure that the process is killed. self._kill() raise ExitThread try: stdout, stderr = popen.communicate(timeout=self.timeout) retcode = popen.returncode assert retcode is not None except subprocess.TimeoutExpired: if self._stopped: # kill() has been called: communicate() fails # on reading closed stdout/stderr raise ExitThread # On timeout, kill the process self._kill() # None means TIMEOUT for the caller retcode = None # bpo-38207: Don't attempt to call communicate() again: on it # can hang until all child processes using stdout and stderr # pipes completes. stdout = stderr = '' except OSError: if self._stopped: # kill() has been called: communicate() fails # on reading closed stdout/stderr raise ExitThread raise else: stdout = stdout.strip() stderr = stderr.rstrip() return (retcode, stdout, stderr) except: self._kill() raise finally: self._wait_completed() self._popen = None self.current_test_name = None def _runtest(self, test_name: str) -> MultiprocessResult: retcode, stdout, stderr = self._run_process(test_name) if retcode is None: return self.mp_result_error(Timeout(test_name), stdout, stderr) err_msg = None if retcode != 0: err_msg = "Exit code %s" % retcode else: stdout, _, result = stdout.rpartition("\n") stdout = stdout.rstrip() if not result: err_msg = "Failed to parse worker stdout" else: try: # deserialize run_tests_worker() output result = json.loads(result, object_hook=decode_test_result) except Exception as exc: err_msg = "Failed to parse worker JSON: %s" % exc if err_msg is not None: return self.mp_result_error(ChildError(test_name), stdout, stderr, err_msg) return MultiprocessResult(result, stdout, stderr, err_msg) def run(self) -> None: while not self._stopped: try: try: test_name = next(self.pending) except StopIteration: break mp_result = self._runtest(test_name) self.output.put((False, mp_result)) if must_stop(mp_result.result, self.ns): break except ExitThread: break except BaseException: self.output.put((True, traceback.format_exc())) break def _wait_completed(self) -> None: popen = self._popen # stdout and stderr must be closed to ensure that communicate() # does not hang popen.stdout.close() popen.stderr.close() try: popen.wait(JOIN_TIMEOUT) except (subprocess.TimeoutExpired, OSError) as exc: print_warning(f"Failed to wait for {self} completion " f"(timeout={format_duration(JOIN_TIMEOUT)}): " f"{exc!r}") def wait_stopped(self, start_time: float) -> None: # bpo-38207: MultiprocessTestRunner.stop_workers() called self.stop() # which killed the process. Sometimes, killing the process from the # main thread does not interrupt popen.communicate() in # TestWorkerProcess thread. This loop with a timeout is a workaround # for that. # # Moreover, if this method fails to join the thread, it is likely # that Python will hang at exit while calling threading._shutdown() # which tries again to join the blocked thread. Regrtest.main() # uses EXIT_TIMEOUT to workaround this second bug. while True: # Write a message every second self.join(1.0) if not self.is_alive(): break dt = time.monotonic() - start_time self.regrtest.log(f"Waiting for {self} thread " f"for {format_duration(dt)}") if dt > JOIN_TIMEOUT: print_warning(f"Failed to join {self} in {format_duration(dt)}") break def get_running(workers: list[TestWorkerProcess]) -> list[TestWorkerProcess]: running = [] for worker in workers: current_test_name = worker.current_test_name if not current_test_name: continue dt = time.monotonic() - worker.start_time if dt >= PROGRESS_MIN_TIME: text = '%s (%s)' % (current_test_name, format_duration(dt)) running.append(text) return running class MultiprocessTestRunner: def __init__(self, regrtest: Regrtest) -> None: self.regrtest = regrtest self.log = self.regrtest.log self.ns = regrtest.ns self.output: queue.Queue[QueueOutput] = queue.Queue() self.pending = MultiprocessIterator(self.regrtest.tests) if self.ns.timeout is not None: # Rely on faulthandler to kill a worker process. This timouet is # when faulthandler fails to kill a worker process. Give a maximum # of 5 minutes to faulthandler to kill the worker. self.worker_timeout = min(self.ns.timeout * 1.5, self.ns.timeout + 5 * 60) else: self.worker_timeout = None self.workers = None def start_workers(self) -> None: self.workers = [TestWorkerProcess(index, self) for index in range(1, self.ns.use_mp + 1)] msg = f"Run tests in parallel using {len(self.workers)} child processes" if self.ns.timeout: msg += (" (timeout: %s, worker timeout: %s)" % (format_duration(self.ns.timeout), format_duration(self.worker_timeout))) self.log(msg) for worker in self.workers: worker.start() def stop_workers(self) -> None: start_time = time.monotonic() for worker in self.workers: worker.stop() for worker in self.workers: worker.wait_stopped(start_time) def _get_result(self) -> QueueOutput | None: use_faulthandler = (self.ns.timeout is not None) timeout = PROGRESS_UPDATE # bpo-46205: check the status of workers every iteration to avoid # waiting forever on an empty queue. while any(worker.is_alive() for worker in self.workers): if use_faulthandler: faulthandler.dump_traceback_later(MAIN_PROCESS_TIMEOUT, exit=True) # wait for a thread try: return self.output.get(timeout=timeout) except queue.Empty: pass # display progress running = get_running(self.workers) if running and not self.ns.pgo: self.log('running: %s' % ', '.join(running)) # all worker threads are done: consume pending results try: return self.output.get(timeout=0) except queue.Empty: return None def display_result(self, mp_result: MultiprocessResult) -> None: result = mp_result.result text = str(result) if mp_result.error_msg is not None: # CHILD_ERROR text += ' (%s)' % mp_result.error_msg elif (result.duration_sec >= PROGRESS_MIN_TIME and not self.ns.pgo): text += ' (%s)' % format_duration(result.duration_sec) running = get_running(self.workers) if running and not self.ns.pgo: text += ' -- running: %s' % ', '.join(running) self.regrtest.display_progress(self.test_index, text) def _process_result(self, item: QueueOutput) -> bool: """Returns True if test runner must stop.""" if item[0]: # Thread got an exception format_exc = item[1] print_warning(f"regrtest worker thread failed: {format_exc}") return True self.test_index += 1 mp_result = item[1] self.regrtest.accumulate_result(mp_result.result) self.display_result(mp_result) if mp_result.stdout: print(mp_result.stdout, flush=True) if mp_result.stderr and not self.ns.pgo: print(mp_result.stderr, file=sys.stderr, flush=True) if must_stop(mp_result.result, self.ns): return True return False def run_tests(self) -> None: self.start_workers() self.test_index = 0 try: while True: item = self._get_result() if item is None: break stop = self._process_result(item) if stop: break except KeyboardInterrupt: print() self.regrtest.interrupted = True finally: if self.ns.timeout is not None: faulthandler.cancel_dump_traceback_later() # Always ensure that all worker processes are no longer # worker when we exit this function self.pending.stop() self.stop_workers() def run_tests_multiprocess(regrtest: Regrtest) -> None: MultiprocessTestRunner(regrtest).run_tests() class EncodeTestResult(json.JSONEncoder): """Encode a TestResult (sub)class object into a JSON dict.""" def default(self, o: Any) -> dict[str, Any]: if isinstance(o, TestResult): result = vars(o) result["__test_result__"] = o.__class__.__name__ return result return super().default(o) def decode_test_result(d: dict[str, Any]) -> TestResult | dict[str, Any]: """Decode a TestResult (sub)class object from a JSON dict.""" if "__test_result__" not in d: return d cls_name = d.pop("__test_result__") for cls in get_all_test_result_classes(): if cls.__name__ == cls_name: return cls(**d) def get_all_test_result_classes() -> set[type[TestResult]]: prev_count = 0 classes = {TestResult} while len(classes) > prev_count: prev_count = len(classes) to_add = [] for cls in classes: to_add.extend(cls.__subclasses__()) classes.update(to_add) return classes runtest.py 0000644 00000030425 15030171176 0006627 0 ustar 00 import faulthandler import functools import gc import importlib import io import os import sys import time import traceback import unittest from test import support from test.support import os_helper from test.libregrtest.cmdline import Namespace from test.libregrtest.save_env import saved_test_environment from test.libregrtest.utils import clear_caches, format_duration, print_warning class TestResult: def __init__( self, name: str, duration_sec: float = 0.0, xml_data: list[str] | None = None, ) -> None: self.name = name self.duration_sec = duration_sec self.xml_data = xml_data def __str__(self) -> str: return f"{self.name} finished" class Passed(TestResult): def __str__(self) -> str: return f"{self.name} passed" class Failed(TestResult): def __init__( self, name: str, duration_sec: float = 0.0, xml_data: list[str] | None = None, errors: list[tuple[str, str]] | None = None, failures: list[tuple[str, str]] | None = None, ) -> None: super().__init__(name, duration_sec=duration_sec, xml_data=xml_data) self.errors = errors self.failures = failures def __str__(self) -> str: if self.errors and self.failures: le = len(self.errors) lf = len(self.failures) error_s = "error" + ("s" if le > 1 else "") failure_s = "failure" + ("s" if lf > 1 else "") return f"{self.name} failed ({le} {error_s}, {lf} {failure_s})" if self.errors: le = len(self.errors) error_s = "error" + ("s" if le > 1 else "") return f"{self.name} failed ({le} {error_s})" if self.failures: lf = len(self.failures) failure_s = "failure" + ("s" if lf > 1 else "") return f"{self.name} failed ({lf} {failure_s})" return f"{self.name} failed" class UncaughtException(Failed): def __str__(self) -> str: return f"{self.name} failed (uncaught exception)" class EnvChanged(Failed): def __str__(self) -> str: return f"{self.name} failed (env changed)" class RefLeak(Failed): def __str__(self) -> str: return f"{self.name} failed (reference leak)" class Skipped(TestResult): def __str__(self) -> str: return f"{self.name} skipped" class ResourceDenied(Skipped): def __str__(self) -> str: return f"{self.name} skipped (resource denied)" class Interrupted(TestResult): def __str__(self) -> str: return f"{self.name} interrupted" class ChildError(Failed): def __str__(self) -> str: return f"{self.name} crashed" class DidNotRun(TestResult): def __str__(self) -> str: return f"{self.name} ran no tests" class Timeout(Failed): def __str__(self) -> str: return f"{self.name} timed out ({format_duration(self.duration_sec)})" # Minimum duration of a test to display its duration or to mention that # the test is running in background PROGRESS_MIN_TIME = 30.0 # seconds # small set of tests to determine if we have a basically functioning interpreter # (i.e. if any of these fail, then anything else is likely to follow) STDTESTS = [ 'test_grammar', 'test_opcodes', 'test_dict', 'test_builtin', 'test_exceptions', 'test_types', 'test_unittest', 'test_doctest', 'test_doctest2', 'test_support' ] # set of tests that we don't want to be executed when using regrtest NOTTESTS = set() # used by --findleaks, store for gc.garbage FOUND_GARBAGE = [] def is_failed(result: TestResult, ns: Namespace) -> bool: if isinstance(result, EnvChanged): return ns.fail_env_changed return isinstance(result, Failed) def findtestdir(path=None): return path or os.path.dirname(os.path.dirname(__file__)) or os.curdir def findtests(testdir=None, stdtests=STDTESTS, nottests=NOTTESTS): """Return a list of all applicable test modules.""" testdir = findtestdir(testdir) names = os.listdir(testdir) tests = [] others = set(stdtests) | nottests for name in names: mod, ext = os.path.splitext(name) if mod[:5] == "test_" and ext in (".py", "") and mod not in others: tests.append(mod) return stdtests + sorted(tests) def get_abs_module(ns: Namespace, test_name: str) -> str: if test_name.startswith('test.') or ns.testdir: return test_name else: # Import it from the test package return 'test.' + test_name def _runtest(ns: Namespace, test_name: str) -> TestResult: # Handle faulthandler timeout, capture stdout+stderr, XML serialization # and measure time. output_on_failure = ns.verbose3 use_timeout = (ns.timeout is not None) if use_timeout: faulthandler.dump_traceback_later(ns.timeout, exit=True) start_time = time.perf_counter() try: support.set_match_tests(ns.match_tests, ns.ignore_tests) support.junit_xml_list = xml_list = [] if ns.xmlpath else None if ns.failfast: support.failfast = True if output_on_failure: support.verbose = True stream = io.StringIO() orig_stdout = sys.stdout orig_stderr = sys.stderr try: sys.stdout = stream sys.stderr = stream result = _runtest_inner(ns, test_name, display_failure=False) if not isinstance(result, Passed): output = stream.getvalue() orig_stderr.write(output) orig_stderr.flush() finally: sys.stdout = orig_stdout sys.stderr = orig_stderr else: # Tell tests to be moderately quiet support.verbose = ns.verbose result = _runtest_inner(ns, test_name, display_failure=not ns.verbose) if xml_list: import xml.etree.ElementTree as ET result.xml_data = [ ET.tostring(x).decode('us-ascii') for x in xml_list ] result.duration_sec = time.perf_counter() - start_time return result finally: if use_timeout: faulthandler.cancel_dump_traceback_later() support.junit_xml_list = None def runtest(ns: Namespace, test_name: str) -> TestResult: """Run a single test. ns -- regrtest namespace of options test_name -- the name of the test Returns a TestResult sub-class depending on the kind of result received. If ns.xmlpath is not None, xml_data is a list containing each generated testsuite element. """ try: return _runtest(ns, test_name) except: if not ns.pgo: msg = traceback.format_exc() print(f"test {test_name} crashed -- {msg}", file=sys.stderr, flush=True) return Failed(test_name) def _test_module(the_module): loader = unittest.TestLoader() tests = loader.loadTestsFromModule(the_module) for error in loader.errors: print(error, file=sys.stderr) if loader.errors: raise Exception("errors while loading tests") support.run_unittest(tests) def save_env(ns: Namespace, test_name: str): return saved_test_environment(test_name, ns.verbose, ns.quiet, pgo=ns.pgo) def _runtest_inner2(ns: Namespace, test_name: str) -> bool: # Load the test function, run the test function, handle huntrleaks # and findleaks to detect leaks abstest = get_abs_module(ns, test_name) # remove the module from sys.module to reload it if it was already imported try: del sys.modules[abstest] except KeyError: pass the_module = importlib.import_module(abstest) if ns.huntrleaks: from test.libregrtest.refleak import dash_R # If the test has a test_main, that will run the appropriate # tests. If not, use normal unittest test loading. test_runner = getattr(the_module, "test_main", None) if test_runner is None: test_runner = functools.partial(_test_module, the_module) try: with save_env(ns, test_name): if ns.huntrleaks: # Return True if the test leaked references refleak = dash_R(ns, test_name, test_runner) else: test_runner() refleak = False finally: cleanup_test_droppings(test_name, ns.verbose) support.gc_collect() if gc.garbage: support.environment_altered = True print_warning(f"{test_name} created {len(gc.garbage)} " f"uncollectable object(s).") # move the uncollectable objects somewhere, # so we don't see them again FOUND_GARBAGE.extend(gc.garbage) gc.garbage.clear() support.reap_children() return refleak def _runtest_inner( ns: Namespace, test_name: str, display_failure: bool = True ) -> TestResult: # Detect environment changes, handle exceptions. # Reset the environment_altered flag to detect if a test altered # the environment support.environment_altered = False if ns.pgo: display_failure = False try: clear_caches() with save_env(ns, test_name): refleak = _runtest_inner2(ns, test_name) except support.ResourceDenied as msg: if not ns.quiet and not ns.pgo: print(f"{test_name} skipped -- {msg}", flush=True) return ResourceDenied(test_name) except unittest.SkipTest as msg: if not ns.quiet and not ns.pgo: print(f"{test_name} skipped -- {msg}", flush=True) return Skipped(test_name) except support.TestFailedWithDetails as exc: msg = f"test {test_name} failed" if display_failure: msg = f"{msg} -- {exc}" print(msg, file=sys.stderr, flush=True) return Failed(test_name, errors=exc.errors, failures=exc.failures) except support.TestFailed as exc: msg = f"test {test_name} failed" if display_failure: msg = f"{msg} -- {exc}" print(msg, file=sys.stderr, flush=True) return Failed(test_name) except support.TestDidNotRun: return DidNotRun(test_name) except KeyboardInterrupt: print() return Interrupted(test_name) except: if not ns.pgo: msg = traceback.format_exc() print(f"test {test_name} crashed -- {msg}", file=sys.stderr, flush=True) return UncaughtException(test_name) if refleak: return RefLeak(test_name) if support.environment_altered: return EnvChanged(test_name) return Passed(test_name) def cleanup_test_droppings(test_name: str, verbose: int) -> None: # First kill any dangling references to open files etc. # This can also issue some ResourceWarnings which would otherwise get # triggered during the following test run, and possibly produce failures. support.gc_collect() # Try to clean up junk commonly left behind. While tests shouldn't leave # any files or directories behind, when a test fails that can be tedious # for it to arrange. The consequences can be especially nasty on Windows, # since if a test leaves a file open, it cannot be deleted by name (while # there's nothing we can do about that here either, we can display the # name of the offending test, which is a real help). for name in (os_helper.TESTFN,): if not os.path.exists(name): continue if os.path.isdir(name): import shutil kind, nuker = "directory", shutil.rmtree elif os.path.isfile(name): kind, nuker = "file", os.unlink else: raise RuntimeError(f"os.path says {name!r} exists but is neither " f"directory nor file") if verbose: print_warning(f"{test_name} left behind {kind} {name!r}") support.environment_altered = True try: import stat # fix possible permissions problems that might prevent cleanup os.chmod(name, stat.S_IRWXU | stat.S_IRWXG | stat.S_IRWXO) nuker(name) except Exception as exc: print_warning(f"{test_name} left behind {kind} {name!r} " f"and it couldn't be removed: {exc}") __init__.py 0000644 00000000167 15030171176 0006662 0 ustar 00 from test.libregrtest.cmdline import _parse_args, RESOURCE_NAMES, ALL_RESOURCES from test.libregrtest.main import main utils.py 0000644 00000012031 15030171176 0006254 0 ustar 00 import math import os.path import sys import textwrap from test import support def format_duration(seconds): ms = math.ceil(seconds * 1e3) seconds, ms = divmod(ms, 1000) minutes, seconds = divmod(seconds, 60) hours, minutes = divmod(minutes, 60) parts = [] if hours: parts.append('%s hour' % hours) if minutes: parts.append('%s min' % minutes) if seconds: if parts: # 2 min 1 sec parts.append('%s sec' % seconds) else: # 1.0 sec parts.append('%.1f sec' % (seconds + ms / 1000)) if not parts: return '%s ms' % ms parts = parts[:2] return ' '.join(parts) def removepy(names): if not names: return for idx, name in enumerate(names): basename, ext = os.path.splitext(name) if ext == '.py': names[idx] = basename def count(n, word): if n == 1: return "%d %s" % (n, word) else: return "%d %ss" % (n, word) def printlist(x, width=70, indent=4, file=None): """Print the elements of iterable x to stdout. Optional arg width (default 70) is the maximum line length. Optional arg indent (default 4) is the number of blanks with which to begin each line. """ blanks = ' ' * indent # Print the sorted list: 'x' may be a '--random' list or a set() print(textwrap.fill(' '.join(str(elt) for elt in sorted(x)), width, initial_indent=blanks, subsequent_indent=blanks), file=file) def print_warning(msg): support.print_warning(msg) orig_unraisablehook = None def flush_std_streams(): if sys.stdout is not None: sys.stdout.flush() if sys.stderr is not None: sys.stderr.flush() def regrtest_unraisable_hook(unraisable): global orig_unraisablehook support.environment_altered = True print_warning("Unraisable exception") old_stderr = sys.stderr try: flush_std_streams() sys.stderr = sys.__stderr__ orig_unraisablehook(unraisable) sys.stderr.flush() finally: sys.stderr = old_stderr def setup_unraisable_hook(): global orig_unraisablehook orig_unraisablehook = sys.unraisablehook sys.unraisablehook = regrtest_unraisable_hook orig_threading_excepthook = None def regrtest_threading_excepthook(args): global orig_threading_excepthook support.environment_altered = True print_warning(f"Uncaught thread exception: {args.exc_type.__name__}") old_stderr = sys.stderr try: flush_std_streams() sys.stderr = sys.__stderr__ orig_threading_excepthook(args) sys.stderr.flush() finally: sys.stderr = old_stderr def setup_threading_excepthook(): global orig_threading_excepthook import threading orig_threading_excepthook = threading.excepthook threading.excepthook = regrtest_threading_excepthook def clear_caches(): # Clear the warnings registry, so they can be displayed again for mod in sys.modules.values(): if hasattr(mod, '__warningregistry__'): del mod.__warningregistry__ # Flush standard output, so that buffered data is sent to the OS and # associated Python objects are reclaimed. for stream in (sys.stdout, sys.stderr, sys.__stdout__, sys.__stderr__): if stream is not None: stream.flush() # Clear assorted module caches. # Don't worry about resetting the cache if the module is not loaded try: distutils_dir_util = sys.modules['distutils.dir_util'] except KeyError: pass else: distutils_dir_util._path_created.clear() try: re = sys.modules['re'] except KeyError: pass else: re.purge() try: _strptime = sys.modules['_strptime'] except KeyError: pass else: _strptime._regex_cache.clear() try: urllib_parse = sys.modules['urllib.parse'] except KeyError: pass else: urllib_parse.clear_cache() try: urllib_request = sys.modules['urllib.request'] except KeyError: pass else: urllib_request.urlcleanup() try: linecache = sys.modules['linecache'] except KeyError: pass else: linecache.clearcache() try: mimetypes = sys.modules['mimetypes'] except KeyError: pass else: mimetypes._default_mime_types() try: filecmp = sys.modules['filecmp'] except KeyError: pass else: filecmp._cache.clear() try: struct = sys.modules['struct'] except KeyError: pass else: struct._clearcache() try: doctest = sys.modules['doctest'] except KeyError: pass else: doctest.master = None try: ctypes = sys.modules['ctypes'] except KeyError: pass else: ctypes._reset_cache() try: typing = sys.modules['typing'] except KeyError: pass else: for f in typing._cleanups: f() support.gc_collect()
| ver. 1.4 |
Github
|
.
| PHP 8.2.28 | Generation time: 0.03 |
proxy
|
phpinfo
|
Settings