Nexus File Manager
v2.0
🏠
Dashboard
⚡
System
🌐
›
opt
›
alt
›
python38
›
lib64
›
python3.8
›
test
›
__pycache__
Quick:
⬆️ Parent
🌐 Root
🏠 Home
🌍 WWW
📁 Temp
⚙️ Etc
📤 Upload
📁 Create
⚡ WordPress Admin
🔄 Refresh
✏️ test_threading.cpython-38.pyc
← Back
U i�f� � @ s4 d Z ddlZddlmZmZmZmZ ddlmZm Z ddl Z ddl Z ddl Z ddl Z ddlZddlZddlZddlZddlZddlZddlmZ ddlmZ dZG dd � d e�ZG d d � d e j�ZG d d � d ej�Z G dd� de �Z G dd� de �Z G dd� de �Z G dd� de �Z G dd� de j�Z!G dd� de �Z"G dd� de �Z#G d d � d ej$�Z$G d d � d ej%�Z&e�'e j(dkd �G d!d"� d"ej%��Z)G d#d$� d$ej*�Z*G d%d&� d&ej%�Z+G d'd(� d(ej,�Z,G d)d*� d*ej-�Z-G d+d,� d,ej.�Z.G d-d.� d.ej/�Z/G d/d0� d0ej�Z0G d1d2� d2ej�Z1e2d3k�r0e�3� dS )4z! Tests for the threading module. � N)�verbose� import_module� cpython_only�requires_type_collecting)�assert_python_ok�assert_python_failure)� lock_tests)�support)Znetbsd5zhp-ux11c @ s, e Zd Zdd� Zdd� Zdd� Zdd� Zd S ) �Counterc C s d| _ d S �Nr ��value��self� r �8/opt/alt/python38/lib64/python3.8/test/test_threading.py�__init__" s zCounter.__init__c C s | j d7 _ d S �N� r r r r r �inc$ s z Counter.incc C s | j d8 _ d S r r r r r r �dec& s z Counter.decc C s | j S �Nr r r r r �get( s z Counter.getN)�__name__� __module__� __qualname__r r r r r r r r r ! s r c @ s e Zd Zdd� Zdd� ZdS )� TestThreadc C s, t jj| |d� || _|| _|| _|| _d S )N��name)� threading�Threadr �testcase�sema�mutex�nrunning)r r r! r" r# r$ r r r r , s zTestThread.__init__c C s� t � � d }tr&td| j|d f � | j�� | j�8 | j�� trTt| j�� d� | j � | j�� d� W 5 Q R X t � |� tr�td| jd� | j�@ | j� � | j �| j�� d� tr�td | j| j�� f � W 5 Q R X W 5 Q R X d S ) Ng ��@z task %s will run for %.1f usecg ��.Aztasks are running� Ztask�doner z$%s is finished. %d tasks are running)�randomr �printr r" r# r$ r r r! ZassertLessEqual�time�sleepr ZassertGreaterEqual)r Zdelayr r r �run3 s* � �zTestThread.runN)r r r r r+ r r r r r + s r c @ s e Zd Zdd� Zdd� ZdS )� BaseTestCasec C s t j�� | _d S r )�testr Zthreading_setup�_threadsr r r r �setUpM s zBaseTestCase.setUpc C s t jj| j� t j�� d S r )r- r Zthreading_cleanupr. Z reap_childrenr r r r �tearDownP s zBaseTestCase.tearDownN)r r r r/ r0 r r r r r, L s r, c @ sn e Zd Zdd� Zdd� Zdd� Zdd� Zd d � Zd d � Zd d� Z dd� Z dd� Z dd� Z dd� Z dd� Zdd� Zdd � Zd d � Ze�eed �d �d!d"� �Ze�eed �d#�d$d%� �Zd&d'� Ze�eed �d(�e�eed)�d*�d+d,� ��Ze�ej e kd-�e�eed �d(�e�eed)�d*�d.d/� ���Z e d0d1� �Z d2d3� Z!d4d5� Z"d6d7� Z#d8d9� Z$e%d:d;� �Z&e%d<d=� �Z'd>S )?� ThreadTestsc C s0 d}t jdd�}t �� }t� }g }t|�D ]F}td| | |||�}|�|� | �|j� | � t |�d� |� � q*t t d�r�t dd� |D ��t �� hB }| �d |� | �t|�|d � tr�td � |D ]D}|�� | �|�� � | �|jd � | �|j� | � t |�d � q�t�r td � | �|�� d � d S )N� r% r z <thread %d>z ^<TestThread\(.*, initial\)>$� get_native_idc s s | ] }|j V qd S r )Z native_id)�.0�tr r r � <genexpr>m s z/ThreadTests.test_various_ops.<locals>.<genexpr>r z!waiting for all tasks to completer z#^<TestThread\(.*, stopped -?\d+\)>$zall tasks done)r �BoundedSemaphore�RLockr �ranger �append� assertIsNone�ident� assertRegex�repr�start�hasattr�setr3 � assertNotIn� assertEqual�lenr r( �join� assertFalse�is_alive�assertNotEqual�assertIsNotNoner ) r ZNUMTASKSr" r# Z numrunning�threads�ir5 Z native_idsr r r �test_various_opsY s4 z ThreadTests.test_various_opsc sr | � t�� j� � �fdd�}t�� � g �t�� �* t�|d�}� � � | � �d |� W 5 Q R X tj �d = d S )Nc s �� t�� j� � �� d S r )r: r � currentThreadr< rA r �r& r< r r �f� s z9ThreadTests.test_ident_of_no_threading_threads.<locals>.fr r ) rI r rM r< �Eventr �wait_threads_exit�_thread�start_new_thread�waitrC �_active)r rO �tidr rN r �"test_ident_of_no_threading_threads} s z.ThreadTests.test_ident_of_no_threading_threadsc C sR t r td� zt�d� W n tjk r: t�d��Y nX | �� t�d� d S )Nz!with 256 KiB thread stack size...i �4platform does not support changing thread stack sizer � r r( r � stack_sizerR �error�unittestZSkipTestrL r r r r � test_various_ops_small_stack� s � z(ThreadTests.test_various_ops_small_stackc C sR t r td� zt�d� W n tjk r: t�d��Y nX | �� t�d� d S )Nz with 1 MiB thread stack size...i rX r rY r r r r � test_various_ops_large_stack� s � z(ThreadTests.test_various_ops_large_stackc C s� dd� }t �� }|�� t�� � t�||f�}|�� W 5 Q R X | �|t j� | � t j| t j � | � t j| � � � | � tt j| �d� t j|= d S )Nc S s t �� | �� d S r )r �current_thread�release)r# r r r rO � s z*ThreadTests.test_foreign_thread.<locals>.f� _DummyThread)r �Lock�acquirer rQ rR rS �assertInrU �assertIsInstancera � assertTruerG r= r> )r rO r# rV r r r �test_foreign_thread� s z ThreadTests.test_foreign_threadc s� t d�}|jj}|j|jf|_G dd� dt�� |�� �}t�� }| � |t � | � |d� z|||�}qdW n � k r| Y n X | � d� z| � |d� W n tk r� Y nX t�� �t�� �G � ��fdd�dtj�}|� }d |_|�� tr�td � t�rtd � |d |�}| � |d� t�r*td � ��� }| �|� t�rJtd� | �|j� t�rdtd� ||j|�}| � |d� t�r�td� �jdd� | �|j� t�r�td� |j�r�|�� d S )N�ctypesc @ s e Zd ZdS )z<ThreadTests.test_PyThreadState_SetAsyncExc.<locals>.AsyncExcN)r r r r r r r �AsyncExc� s ri r zAsyncExc not raisedr c s e Zd Z� ��fdd�ZdS )z:ThreadTests.test_PyThreadState_SetAsyncExc.<locals>.Workerc sP t �� | _d| _z��� t�d� qW n" � k rJ d| _��� Y nX d S )NFg�������?T)r � get_ident�id�finishedrA r) r* r �ri Zworker_saw_exceptionZworker_startedr r r+ � s z>ThreadTests.test_PyThreadState_SetAsyncExc.<locals>.Worker.runN�r r r r+ r rm r r �Worker� s ro Tz started worker threadz trying nonsensical thread id���z, waiting for worker thread to get startedz" verifying worker hasn't exitedz2 attempting to raise asynch exception in workerz5 waiting for worker to say it caught the exceptionr2 �Ztimeoutz all OK -- joining worker)r Z pythonapiZPyThreadState_SetAsyncExcZc_ulongZ py_objectZargtypes� Exceptionr rj re �intZ assertGreaterZfailrC �UnboundLocalErrorrP r �daemonr? r r( rT rf rF rl rk rE ) r rh Z set_async_excZ exceptionrV �resultro r5 �retr rm r � test_PyThreadState_SetAsyncExc� sb z*ThreadTests.test_PyThreadState_SetAsyncExcc C sX dd� }t j}|t _z6t jdd� d�}| �t j|j� | �|t jkd� W 5 |t _X d S )Nc W s t �� �d S r )r � ThreadError��argsr r r �fail_new_thread s z7ThreadTests.test_limbo_cleanup.<locals>.fail_new_threadc S s d S r r r r r r �<lambda> � z0ThreadTests.test_limbo_cleanup.<locals>.<lambda>��targetz:Failed to cleanup _limbo map on failure of Thread.start().)r �_start_new_threadr � assertRaisesry r? rF Z_limbo)r r| r� r5 r r r �test_limbo_cleanup s �z ThreadTests.test_limbo_cleanupc C s( t d� tdd�\}}}| �|d� d S )Nrh �-caN if 1: import ctypes, sys, time, _thread # This lock is used as a simple event variable. ready = _thread.allocate_lock() ready.acquire() # Module globals are cleared before __del__ is run # So we save the functions in class dict class C: ensure = ctypes.pythonapi.PyGILState_Ensure release = ctypes.pythonapi.PyGILState_Release def __del__(self): state = self.ensure() self.release(state) def waitingThread(): x = C() ready.release() time.sleep(100) _thread.start_new_thread(waitingThread, ()) ready.acquire() # Be sure the other thread is waiting. sys.exit(42) �* )r r rC �r �rc�out�errr r r � test_finalize_runnning_thread" s z)ThreadTests.test_finalize_runnning_threadc C s t dd� d S )Nr� aP if 1: import sys, threading # A deadlock-killer, to prevent the # testsuite to hang forever def killer(): import os, time time.sleep(2) print('program blocked; aborting') os._exit(2) t = threading.Thread(target=killer) t.daemon = True t.start() # This is the trace function def func(frame, event, arg): threading.current_thread() return func sys.settrace(func) )r r r r r �test_finalize_with_traceC s z$ThreadTests.test_finalize_with_tracec C s0 t dd�\}}}| �|�� d� | �|d� d S )Nr� a� if 1: import threading from time import sleep def child(): sleep(1) # As a non-daemon thread we SHOULD wake up and nothing # should be torn down yet print("Woke up, sleep function is:", sleep) threading.Thread(target=child).start() raise SystemExit s5 Woke up, sleep function is: <built-in function sleep>r~ )r rC �stripr� r r r � test_join_nondaemon_on_shutdown\ s �z+ThreadTests.test_join_nondaemon_on_shutdownc C s~ t j}t�� }z^tdd�D ]N}t�|d � t jdd� d�}|�� |�� |� }| � ||d||f � qW 5 t�|� X d S )Nr �d g-C ��6*?c S s d S r r r r r r r} x r~ z7ThreadTests.test_enumerate_after_join.<locals>.<lambda>r z� triggered after %d trials: %s) r � enumerate�sys�getswitchinterval�setswitchintervalr9 r r? rE rB )r �enum� old_intervalrK r5 �lr r r �test_enumerate_after_joinp s � z%ThreadTests.test_enumerate_after_joinc C s� G dd� dt �}|dd�}t�|�}|j�� ~| j|� dt�|� � d� |dd�}t�|�}|j�� ~| j|� dt�|� � d� d S )Nc @ s e Zd Zdd� Zdd� ZdS )zDThreadTests.test_no_refcycle_through_target.<locals>.RunSelfFunctionc S s. || _ tj| j| fd| id�| _| j�� d S )N� yet_another)r� r{ �kwargs)� should_raiser r �_run�threadr? )r r� r r r r � s �zMThreadTests.test_no_refcycle_through_target.<locals>.RunSelfFunction.__init__c S s | j r t�d S r )r� � SystemExit)r Z other_refr� r r r r� � s zIThreadTests.test_no_refcycle_through_target.<locals>.RunSelfFunction._runN)r r r r r� r r r r �RunSelfFunction� s r� F)r� z%d references still around)�msgT)�object�weakref�refr� rE r; r� � getrefcount)r r� Z cyclic_objectZweak_cyclic_objectZraising_cyclic_objectZweak_raising_cyclic_objectr r r � test_no_refcycle_through_target� s& �� ��z+ThreadTests.test_no_refcycle_through_targetc C sh t �� }|�� |�d� |�� |�d� | �td�� |�� W 5 Q R X t � � }|� � t � � d S )NTr zuse is_alive()) r r ZisDaemonZ setDaemonZgetNameZsetNameZassertWarnsRegex�DeprecationWarningZisAliverP ZisSet� activeCount)r r5 �er r r �test_old_threading_api� s z"ThreadTests.test_old_threading_apic C s2 t �� }| �dt|�� d|_| �dt|�� d S �Nru T)r r rB r> ru rd �r r5 r r r �test_repr_daemon� s z ThreadTests.test_repr_daemonc C sH t �� }| �|j� t jdd�}| �|j� t jdd�}| �|j� d S )NF�ru T)r r rF ru rf r� r r r �test_daemon_param� s z ThreadTests.test_daemon_param�forkztest needs fork()c C s0 d}t d|�\}}}| �|d� | �|d� d S )Na� if 1: import _thread, threading, os, time def background_thread(evt): # Creates and registers the _DummyThread instance threading.current_thread() evt.set() time.sleep(10) evt = threading.Event() _thread.start_new_thread(background_thread, (evt,)) evt.wait() assert threading.active_count() == 2, threading.active_count() if os.fork() == 0: assert threading.active_count() == 1, threading.active_count() os._exit(0) else: os.wait() r� r~ �r rC )r �code�_r� r� r r r � test_dummy_thread_after_fork� s z(ThreadTests.test_dummy_thread_after_fork�needs os.fork()c C s� t �� }| �t j|� tj�d� td�D ]~}tjdd� d�}|� � t � � }|dkrnt � |� � rfdnd� q*|�� t �|d�\}}| �t �|�� | �dt �|�� q*d S ) Ng���ư>� c S s d S r r r r r r r} � r~ z6ThreadTests.test_is_alive_after_fork.<locals>.<lambda>r r � r2 )r� r� � addCleanupr� r- r r9 r r r? �osr� �_exitrG rE �waitpidrf � WIFEXITEDrC � WEXITSTATUS)r r� rK r5 �pid�statusr r r �test_is_alive_after_fork� s z$ThreadTests.test_is_alive_after_forkc sh t �� }� �|jd� � �|jt �� j� � �|jt �� � � fdd�}t j|d�}|�� |� � d S )NZ MainThreadc s � � t�� jt�� j� d S r )rH r � main_threadr< r_ r r r r rO � s �z'ThreadTests.test_main_thread.<locals>.fr ) r r� rC r r< r_ rj r r? rE )r �mainrO Zthr r r �test_main_thread� s z ThreadTests.test_main_threadztest needs os.fork()r� ztest needs os.waitpid()c C s@ d}t d|�\}}}|�� �dd�}| �|d� | �|d� d S )Nak if 1: import os, threading pid = os.fork() if pid == 0: main = threading.main_thread() print(main.name) print(main.ident == threading.current_thread().ident) print(main.ident == threading.get_ident()) else: os.waitpid(pid, 0) r� � � r~ zMainThread True True �r �decode�replacerC �r r� r� r� r� �datar r r �test_main_thread_after_fork� s z'ThreadTests.test_main_thread_after_fork�due to known OS bugc C s@ d}t d|�\}}}|�� �dd�}| �|d� | �|d� d S )Na� if 1: import os, threading, sys def f(): pid = os.fork() if pid == 0: main = threading.main_thread() print(main.name) print(main.ident == threading.current_thread().ident) print(main.ident == threading.get_ident()) # stdout is fully buffered because not a tty, # we have to flush before exit. sys.stdout.flush() else: os.waitpid(pid, 0) th = threading.Thread(target=f) th.start() th.join() r� r� r� r~ zThread-1 True True r� r� r r r �/test_main_thread_after_fork_from_nonmain_thread s z;ThreadTests.test_main_thread_after_fork_from_nonmain_threadc C sB d}t d|�\}}}|�� }| �|d� | �|�� dgd � d S )Na� if 1: import gc, threading main_thread = threading.current_thread() assert main_thread is threading.main_thread() # sanity check class RefCycle: def __init__(self): self.cycle = self def __del__(self): print("GC:", threading.current_thread() is main_thread, threading.main_thread() is main_thread, threading.enumerate() == [main_thread]) RefCycle() gc.collect() # sanity check x = RefCycle() r� r~ zGC: True True True� )r r� rC � splitlinesr� r r r � test_main_thread_during_shutdown, s �z,ThreadTests.test_main_thread_during_shutdownc C s$ d}t d|�\}}}| �|d� d S )Na� if 1: import os import threading import time import random def random_sleep(): seconds = random.random() * 0.010 time.sleep(seconds) class Sleeper: def __del__(self): random_sleep() tls = threading.local() def f(): # Sleep a bit so that the thread is still running when # Py_Finalize() is called. random_sleep() tls.x = Sleeper() random_sleep() threading.Thread(target=f).start() random_sleep() r� r~ r� )r r� r� r� r� r r r �test_finalization_shutdownJ s z&ThreadTests.test_finalization_shutdownc s� t �� �t �� � ��� � �� � �fdd�}tj|d�}| �|jd � |�� ��� | �|� � � |j}| � |jdd�d� � � � | �|jdd�d� | �|� � � |� � | � |� � � | � |j� |� � d S )Nc s �� � � �� t�d� d S )N�{�G�z�?)r` rc r) r* r �Zfinish�startedr r rO s s z'ThreadTests.test_tstate_lock.<locals>.fr r rq F� )rR � allocate_lockrc r r �assertIs� _tstate_lockr? rf rG rF r` r; rE )r rO r5 � tstate_lockr r� r �test_tstate_lockm s& z ThreadTests.test_tstate_lockc s� t �� �t �� � ��� � �� � �fdd�}tj|d�}|�� ��� | �dt|�� � �� d}t d�D ] }|t|�kr� q�t � d� qn| �|t|�� |� � d S )Nc s �� � � �� d S r )r` rc r r� r r rO � s z(ThreadTests.test_repr_stopped.<locals>.fr r� Zstoppedi� r� ) rR r� rc r r r? rd r> r` r9 r) r* rE )r rO r5 Z LOOKING_FORrK r r� r �test_repr_stopped� s" z ThreadTests.test_repr_stoppedc s� t dd�D ]�}t�|�� � fdd�t |�D �}|D ] }|�� q2|D ] }|�� qD� fdd�t |�D �}|D ] }|�� ql|D ] }|�� q~| �t� j� q d S )Nr r2 c s g | ]}t j� jd ��qS �r )r r rc �r4 r� �Zbsr r � <listcomp>� s �z;ThreadTests.test_BoundedSemaphore_limit.<locals>.<listcomp>c s g | ]}t j� jd ��qS r� )r r r` r� r� r r r� � s �)r9 r r7 r? rE r� � ValueErrorr` )r �limitrJ r5 r r� r �test_BoundedSemaphore_limit� s" � � z'ThreadTests.test_BoundedSemaphore_limitc s� �fdd��dd� �� �fdd�� d � _ t�� }t��� z4t��� dd l}|�� � td�D ] }� � q`W 5 t�|� X d S ) Nc s � S r r )�frame�event�arg)� noop_tracer r r� � s z9ThreadTests.test_frame_tstate_tracing.<locals>.noop_tracec s s dV q d S )N� generatorr r r r r r� � s z8ThreadTests.test_frame_tstate_tracing.<locals>.generatorc s � j d kr�� � _ t� j �S r )�gen�nextr )�callbackr� r r r� � s z7ThreadTests.test_frame_tstate_tracing.<locals>.callbackr r% )r� r� �gettrace�settracer � _testcapiZcall_in_temporary_c_threadr9 )r Z old_tracer� r- r )r� r� r� r �test_frame_tstate_tracing� s z%ThreadTests.test_frame_tstate_tracingc C s� dD ]�}| j |d��l t�� }tj|j|d�}|�� |j}|sP| �|tj� n| � |tj� |� � |� � | � |tj� W 5 Q R X qd S )N)FTr� )r� ru ) ZsubTestr rP r rT r? r� rd Z_shutdown_locksrB rA rE )r ru r� r� r� r r r �test_shutdown_locks� s z ThreadTests.test_shutdown_locksN)(r r r rL rW r] r^ rg rx r� r� r� r� r� r� r� r� r� r\ � skipUnlessr@ r� r� r� r� r� �skipIfr� �platform�platforms_to_skipr� r r� r� r� r� r� r r� r� r r r r r1 U sJ $ X! ## 'r1 c @ s� e Zd Zdd� Zdd� Ze�eed�d�e� e j e kd�dd � ��Z e�eed�d�e� e j e kd�d d � ��Ze� e j e kd�d d � �Ze�eed�d�e� e j e kd�dd� ��Ze�eed�d�dd� �ZdS )�ThreadJoinOnShutdownc C s8 d| }t d|�\}}}|�� �dd�}| �|d� d S )Na� if 1: import sys, os, time, threading # a thread, which waits for the main program to terminate def joiningfunc(mainthread): mainthread.join() print('end of thread') # stdout is fully buffered because not a tty, we have to flush # before exit. sys.stdout.flush() r� r� r� zend of main end of thread r� )r �scriptr� r� r� r� r r r � _run_and_join s � z"ThreadJoinOnShutdown._run_and_joinc C s d}| � |� d S )Nz�if 1: import os t = threading.Thread(target=joiningfunc, args=(threading.current_thread(),)) t.start() time.sleep(0.1) print('end of main') �r� �r r� r r r �test_1_join_on_shutdown s z,ThreadJoinOnShutdown.test_1_join_on_shutdownr� r� r� c C s d}| � |� d S )NaG if 1: childpid = os.fork() if childpid != 0: os.waitpid(childpid, 0) sys.exit(0) t = threading.Thread(target=joiningfunc, args=(threading.current_thread(),)) t.start() print('end of main') r� r� r r r � test_2_join_in_forked_process s z2ThreadJoinOnShutdown.test_2_join_in_forked_processc C s d}| � |� d S )Na: if 1: main_thread = threading.current_thread() def worker(): childpid = os.fork() if childpid != 0: os.waitpid(childpid, 0) sys.exit(0) t = threading.Thread(target=joiningfunc, args=(main_thread,)) print('end of main') t.start() t.join() # Should not block: main_thread is already stopped w = threading.Thread(target=worker) w.start() r� r� r r r �!test_3_join_in_forked_from_thread. s z6ThreadJoinOnShutdown.test_3_join_in_forked_from_threadc C s" d}t d|�\}}}| �|� d S )Nac if True: import os import random import sys import time import threading thread_has_run = set() def random_io(): '''Loop for a while sleeping random tiny amounts and doing some I/O.''' while True: with open(os.__file__, 'rb') as in_f: stuff = in_f.read(200) with open(os.devnull, 'wb') as null_f: null_f.write(stuff) time.sleep(random.random() / 1995) thread_has_run.add(threading.current_thread()) def main(): count = 0 for _ in range(40): new_thread = threading.Thread(target=random_io) new_thread.daemon = True new_thread.start() count += 1 while len(thread_has_run) < count: time.sleep(0.001) # Trigger process shutdown sys.exit(0) main() r� )r rF �r r� r� r� r� r r r �test_4_daemon_threadsG s !z*ThreadJoinOnShutdown.test_4_daemon_threadsc C sN dd� }g }t d�D ]"}tj|d�}|�|� |�� q|D ] }|�� q<d S )Nc S s, t �� } | dkr t �| d� n t �d� d S r )r� r� r� r� )r� r r r �do_fork_and_waitv s zIThreadJoinOnShutdown.test_reinit_tls_after_fork.<locals>.do_fork_and_wait� r )r9 r r r: r? rE )r r� rJ rK r5 r r r �test_reinit_tls_after_forkp s z/ThreadJoinOnShutdown.test_reinit_tls_after_forkc C s� g }t d�D ]&}tjdd� d�}|�|� |�� q t�� }|dkrltt� � �dkr`t� d� q�t� d� n t� |d�\}}| � d|� |D ] }|� � q�d S )Nr� c S s t �d�S )Ng333333�?)r) r* r r r r r} � r~ zKThreadJoinOnShutdown.test_clear_threads_states_after_fork.<locals>.<lambda>r r r )r9 r r r: r? r� r� rD r� �_current_framesr� r� rC rE )r rJ rK r5 r� r� r� r r r �$test_clear_threads_states_after_fork� s z9ThreadJoinOnShutdown.test_clear_threads_states_after_forkN)r r r r� r� r\ r� r@ r� r� r� r� r� r� r� r� r� r� r r r r r� � s ( r� c @ s( e Zd Zdd� Zdd� Zedd� �ZdS )�SubinterpThreadingTestsc C sb t �� \}}| �t j|� | �t j|� d|f }tj�|�}| �|d� | �t �|d�d� d S )Na if 1: import os import random import threading import time def random_sleep(): seconds = random.random() * 0.010 time.sleep(seconds) def f(): # Sleep a bit so that the thread is still running when # Py_EndInterpreter is called. random_sleep() os.write(%d, b"x") threading.Thread(target=f).start() random_sleep() r r � x� r� �piper� �closer- r Zrun_in_subinterprC �read�r �r�wr� rw r r r �test_threads_join� s � z)SubinterpThreadingTests.test_threads_joinc C sb t �� \}}| �t j|� | �t j|� d|f }tj�|�}| �|d� | �t �|d�d� d S )Na� if 1: import os import random import threading import time def random_sleep(): seconds = random.random() * 0.010 time.sleep(seconds) class Sleeper: def __del__(self): random_sleep() tls = threading.local() def f(): # Sleep a bit so that the thread is still running when # Py_EndInterpreter is called. random_sleep() tls.x = Sleeper() os.write(%d, b"x") threading.Thread(target=f).start() random_sleep() r r r r r r r r �test_threads_join_2� s � z+SubinterpThreadingTests.test_threads_join_2c C sH d}d|f }t j�� � td|�\}}}W 5 Q R X | �d|�� � d S )NaA if 1: import os import threading import time def f(): # Make sure the daemon thread is still running when # Py_EndInterpreter is called. time.sleep(10) threading.Thread(target=f, daemon=True).start() z[if 1: import _testcapi _testcapi.run_in_subinterp(%r) r� z:Fatal Python error: Py_EndInterpreter: not the last thread)r- r ZSuppressCrashReportr rd r� )r Zsubinterp_coder� r� r� r� r r r � test_daemon_threads_fatal_error� s � �z7SubinterpThreadingTests.test_daemon_threads_fatal_errorN)r r r r r r r r r r r r � s 'r c @ s` e Zd Zdd� Zdd� Zdd� Zdd� Zd d � Zd d � Zd d� Z e dd� �Z dd� Z dd� Z dS )�ThreadingExceptionTestsc C s* t �� }|�� | �t|j� |�� d S r )r r r? r� � RuntimeErrorrE �r r� r r r �test_start_thread_again s z/ThreadingExceptionTests.test_start_thread_againc C s t �� }| �t|j� d S r )r r_ r� r rE )r r_ r r r �test_joining_current_thread s z3ThreadingExceptionTests.test_joining_current_threadc C s t �� }| �t|j� d S r )r r r� r rE r r r r � test_joining_inactive_thread s z4ThreadingExceptionTests.test_joining_inactive_threadc C s. t �� }|�� | �tt|dd� |�� d S r� )r r r? r� r �setattrrE r r r r � test_daemonize_active_thread s z4ThreadingExceptionTests.test_daemonize_active_threadc C s t �� }| �t|j� d S r )r rb r� r r` )r �lockr r r � test_releasing_unacquired_lock s z6ThreadingExceptionTests.test_releasing_unacquired_lockc C sh d}d}t jtjd|gt jt jd�}|�� \}}|�� �dd�}| �|j dd|�� � | �||� d S ) Na if True: import threading def recurse(): return recurse() def outer(): try: recurse() except RecursionError: pass w = threading.Thread(target=outer) w.start() w.join() print('end of main thread') zend of main thread r� )�stdout�stderrr� r� r zUnexpected error: ) � subprocess�Popenr� � executable�PIPEZ communicater� r� rC � returncode)r r� Zexpected_output�pr r r� r r r �test_recursion_limit s � z,ThreadingExceptionTests.test_recursion_limitc C s\ d}t d|�\}}}| �|d� |�� }| �d|� | �d|� | �d|� | �d|� d S )Na� if True: import threading import time running = False def run(): global running running = True while running: time.sleep(0.01) 1/0 t = threading.Thread(target=run) t.start() while not running: time.sleep(0.01) running = False t.join() r� r~ �Exception in thread�"Traceback (most recent call last):�ZeroDivisionError�Unhandled exception�r rC r� rd rB r� r r r �test_print_exception; s z,ThreadingExceptionTests.test_print_exceptionc C s\ d}t d|�\}}}| �|d� |�� }| �d|� | �d|� | �d|� | �d|� d S )Na� if True: import sys import threading import time running = False def run(): global running running = True while running: time.sleep(0.01) 1/0 t = threading.Thread(target=run) t.start() while not running: time.sleep(0.01) sys.stderr = None running = False t.join() r� r~ r r r! r" r# r� r r r �%test_print_exception_stderr_is_none_1V s z=ThreadingExceptionTests.test_print_exception_stderr_is_none_1c C s4 d}t d|�\}}}| �|d� | �d|�� � d S )Na� if True: import sys import threading import time running = False def run(): global running running = True while running: time.sleep(0.01) 1/0 sys.stderr = None t = threading.Thread(target=run) t.start() while not running: time.sleep(0.01) running = False t.join() r� r~ r" )r rC rB r� r� r r r �%test_print_exception_stderr_is_none_2t s z=ThreadingExceptionTests.test_print_exception_stderr_is_none_2c sX dd� � G � fdd�dt j�}|� }|�� |�� | �|j� | �|jt� d |_d S )Nc S s � d S r r r r r r � bare_raise� s zOThreadingExceptionTests.test_bare_raise_in_brand_new_thread.<locals>.bare_raisec s e Zd ZdZ� fdd�ZdS )zOThreadingExceptionTests.test_bare_raise_in_brand_new_thread.<locals>.Issue27558Nc s8 z � � W n( t k r2 } z || _W 5 d }~X Y nX d S r )rr �exc)r r( �r' r r r+ � s zSThreadingExceptionTests.test_bare_raise_in_brand_new_thread.<locals>.Issue27558.run)r r r r( r+ r r) r r � Issue27558� s r* )r r r? rE rI r( re r )r r* r� r r) r �#test_bare_raise_in_brand_new_thread� s z;ThreadingExceptionTests.test_bare_raise_in_brand_new_threadN)r r r r r r r r r r$ r r% r& r+ r r r r r s r c @ s e Zd Zdd� ZdS )� ThreadRunFailc C s t d��d S )N� run failed�r� r r r r r+ � s zThreadRunFail.runNrn r r r r r, � s r, c @ s: e Zd Zdd� Zejdd� �Zdd� Zdd� Zd d � Z d S ) �ExceptHookTestsc C sz t �d�� }tdd�}|�� |�� W 5 Q R X |�� �� }| �d|j� d�|� | �d|� | �d|� | �d|� d S ) Nr zexcepthook threadr �Exception in thread �: �#Traceback (most recent call last): z raise ValueError("run failed")zValueError: run failed) r �captured_outputr, r? rE �getvaluer� rd r )r r r� r r r �test_excepthook� s z ExceptHookTests.test_excepthookc C s� t �d��^}z td��W nJ tk rb } z,t�t�� d��}zt�|� W 5 d }X W 5 d }~X Y nX W 5 Q R X |� � � � }| � dt� � � d�|� | � d|� | � d|� | � d|� d S ) Nr Zbugr0 r1 r2 z raise ValueError("bug")zValueError: bug)N) r r3 r� rr r ZExceptHookArgsr� �exc_info� excepthookr4 r� rd rj )r r r( r{ r r r �test_excepthook_thread_None� s " z+ExceptHookTests.test_excepthook_thread_Nonec C sR G dd� dt j�}t�d�� }|� }|�� |�� W 5 Q R X | �|�� d� d S )Nc @ s e Zd Zdd� ZdS )z4ExceptHookTests.test_system_exit.<locals>.ThreadExitc S s t �d� d S r )r� �exitr r r r r+ � s z8ExceptHookTests.test_system_exit.<locals>.ThreadExit.runNrn r r r r � ThreadExit� s r: r r� )r r r r3 r? rE rC r4 )r r: r r� r r r �test_system_exit� s z ExceptHookTests.test_system_exitc s� d � � fdd�}ztt �td|�� t� }|�� |�� W 5 Q R X | �� jt� | �t � j �d� | �� j � j j � | � � j|� W 5 d � X d S )Nc s | � d S r r )Z hook_argsrz r r �hook� s z4ExceptHookTests.test_custom_excepthook.<locals>.hookr7 r- )r � swap_attrr r, r? rE rC �exc_typer� �str� exc_value� exc_traceback� __traceback__r� r� )r r< r� r rz r �test_custom_excepthook� s z&ExceptHookTests.test_custom_excepthookc s� dd� }d � � fdd�}t �td|��L t �td|��2 t �d�� }t� }|�� |�� W 5 Q R X W 5 Q R X W 5 Q R X | �|� � d� | �� d� d S ) Nc S s t d��d S )N�threading_hook failedr. rz r r r �threading_hook� s zCExceptHookTests.test_custom_excepthook_fail.<locals>.threading_hookc s t |�� d S r )r? )r> r@ rA �Zerr_strr r �sys_hook� s z=ExceptHookTests.test_custom_excepthook_fail.<locals>.sys_hookr7 r z#Exception in threading.excepthook: rD ) r r= r r� r3 r, r? rE rC r4 )r rE rG r r� r rF r �test_custom_excepthook_fail� s ��& �z+ExceptHookTests.test_custom_excepthook_failN) r r r r5 r r r8 r; rC rH r r r r r/ � s r/ c @ s$ e Zd Zdd� Zdd� Zdd� ZdS )� TimerTestsc C s t �| � g | _t�� | _d S r )r, r/ � callback_argsr rP �callback_eventr r r r r/ s zTimerTests.setUpc C s� t �d| j�}|�� | j�� |j�d� d|jd< | j� � t �d| j�}|�� | j�� | � t | j �d� | � | j di fdi fg� |� � |� � d S )Nr� ZblahZbarZfoor� r )r ZTimer� _callback_spyr? rK rT r{ r: r� �clearrC rD rJ rE )r Ztimer1Ztimer2r r r � test_init_immutable_default_args s z+TimerTests.test_init_immutable_default_argsc O s* | j �|d d � |�� f� | j�� d S r )rJ r: �copyrK rA )r r{ r� r r r rL s zTimerTests._callback_spyN)r r r r/ rN rL r r r r rI s rI c @ s e Zd Zeej�ZdS )� LockTestsN)r r r � staticmethodr rb �locktyper r r r rP s rP c @ s e Zd Zeej�ZdS )� PyRLockTestsN)r r r rQ r Z_PyRLockrR r r r r rS ! s rS zRLock not implemented in Cc @ s e Zd Zeej�ZdS )� CRLockTestsN)r r r rQ r �_CRLockrR r r r r rT $ s rT c @ s e Zd Zeej�ZdS )� EventTestsN)r r r rQ r rP Z eventtyper r r r rV ( s rV c @ s e Zd Zeej�ZdS )�ConditionAsRLockTestsN)r r r rQ r � ConditionrR r r r r rW + s rW c @ s e Zd Zeej�ZdS )�ConditionTestsN)r r r rQ r rX Zcondtyper r r r rY / s rY c @ s e Zd Zeej�ZdS )�SemaphoreTestsN)r r r rQ r Z Semaphore�semtyper r r r rZ 2 s rZ c @ s e Zd Zeej�ZdS )�BoundedSemaphoreTestsN)r r r rQ r r7 r[ r r r r r\ 5 s r\ c @ s e Zd Zeej�ZdS )� BarrierTestsN)r r r rQ r ZBarrierZ barriertyper r r r r] 8 s r] c @ s e Zd Zdd� ZdS )� MiscTestCasec C s& dh}ddh}t j| td||d� d S )Nry rM r� )r rR )�extra� blacklist)r Z check__all__r )r r_ r` r r r � test__all__= s �zMiscTestCase.test__all__N)r r r ra r r r r r^ < s r^ c @ s$ e Zd Zdd� Zdd� Zdd� ZdS )�InterruptMainTestsc C sF dd� }t j|d�}| �t�� |�� |�� W 5 Q R X |�� d S )Nc S s t �� d S r )rR �interrupt_mainr r r r �call_interruptH s zHInterruptMainTests.test_interrupt_main_subthread.<locals>.call_interruptr )r r r� �KeyboardInterruptr? rE )r rd r5 r r r � test_interrupt_main_subthreadE s z0InterruptMainTests.test_interrupt_main_subthreadc C s" | � t�� t�� W 5 Q R X d S r )r� re rR rc r r r r � test_interrupt_main_mainthreadP s z1InterruptMainTests.test_interrupt_main_mainthreadc C sV t �t j�}z4t � t jt j� t�� t � t jt j� t�� W 5 t � t j|� X d S r )�signal� getsignal�SIGINT�SIG_IGNrR rc �SIG_DFL)r Zhandlerr r r �test_interrupt_main_noerrorV s z.InterruptMainTests.test_interrupt_main_noerrorN)r r r rf rg rm r r r r rb D s rb �__main__)4�__doc__Z test.supportr- r r r r Ztest.support.script_helperr r r' r� rR r r) r\ r� r� r rh r r r� r� r r r ZTestCaser, r1 r� r r r, r/ rI rP Z RLockTestsrS r� rU rT rV rW rY rZ r\ r] r^ rb r r� r r r r �<module> s^ ! . %_ #Z
💻
System Info
Current path
/opt/alt/python38/lib64/python3.8/test/__pycache__
Contents
0 folders, 1461 files
Disk free
100.49 GB
PHP version
8.3.30
🚀
Quick Actions
📍 Script location
🌐 Web root
🗑️ Temp
📁 /opt/alt/python38/lib64/python3.8/test/__pycache__
⚡ Nexus File Manager • 1461 items