Nexus File Manager
v2.0
🏠
Dashboard
⚡
System
🌐
›
lib64
›
python2.7
›
test
Quick:
⬆️ Parent
🌐 Root
🏠 Home
🌍 WWW
📁 Temp
⚙️ Etc
📤 Upload
📁 Create
⚡ WordPress Admin
🔄 Refresh
✏️ test_queue.pyo
← Back
� z fc @ sM d d l Z d d l Z d d l Z d d l m Z e j d � Z d Z d e j f d � � YZ d d d � � YZ d e f d � � YZ d e e j f d � � YZ d e e j f d � � YZ d e e j f d � � YZ d e f d � � YZ d e j f d � � YZ d e e j f d � � YZ d � Z e d k rIe � n d S( i����N( t test_supportt threadingi t _TriggerThreadc B s e Z d � Z d � Z RS( c C s5 | | _ | | _ t j � | _ t j j | � d S( N( t fnt argsR t Eventt startedEventt Threadt __init__( t selfR R ( ( s' /usr/lib64/python2.7/test/test_queue.pyR s c C s. t j d � | j j � | j | j � d S( Ng�������?( t timet sleepR t setR R ( R ( ( s' /usr/lib64/python2.7/test/test_queue.pyt run s ( t __name__t __module__R R ( ( ( s' /usr/lib64/python2.7/test/test_queue.pyR s t BlockingTestMixinc B s# e Z d � Z d � Z d � Z RS( c C s d | _ d S( N( t Nonet t( R ( ( s' /usr/lib64/python2.7/test/test_queue.pyt tearDown. s c C s� t | | � | _ | j j � | | � | _ | j j j � sT | j d | � n | j j d � | j j � r� | j d | � n | j S( Ns, blocking function '%r' appeared not to blocki s, trigger function '%r' appeared to not return( R R t startt resultR t is_sett failt joint is_alive( R t block_funct block_argst trigger_funct trigger_args( ( s' /usr/lib64/python2.7/test/test_queue.pyt do_blocking_test1 s c C s� t | | � | _ | j j � z: y | | � Wn | k rF � n X| j d | � Wd | j j d � | j j � r� | j d | � n | j j j � s� | j d � n Xd S( Ns expected exception of kind %ri s, trigger function '%r' appeared to not returns( trigger thread ended but event never set( R R R R R R R R ( R R R R R t expected_exception_class( ( s' /usr/lib64/python2.7/test/test_queue.pyt do_exceptional_blocking_test@ s ( R R R R R ( ( ( s' /usr/lib64/python2.7/test/test_queue.pyR , s t BaseQueueTestc B sG e Z d � Z d � Z d � Z d � Z d � Z d � Z d � Z RS( c C s d | _ t j � | _ d S( Ni ( t cumR t Lockt cumlock( R ( ( s' /usr/lib64/python2.7/test/test_queue.pyt setUpV s c C s� | j � s t d � n | j d � | j d � | j d � t d d d d g d d d d g d d d d g � } | j � | j � | j � g } | j | | | j j d � x< t t d � D]* } | j | � | j | j � d � q� W| j | j � d � d t } d t } | j | � | j | j � d � y$ | j | d d �| j d � Wn t j k rzn Xy$ | j | d d �| j d � Wn t j k r�n X| j | j | f | j d � | j | j | t d f | j d � x t t � D] } | j � qW| j | j � d � y! | j d d � | j d � Wn t j k rhn Xy! | j d d � | j d � Wn t j k r�n X| j | j d | j d � | j | j t d f | j d � d S( Ns& Call this function with an empty queueio iM i� t Queuet LifoQueuet PriorityQueues& Didn't seem to queue the correct data!i s Queue should not be emptys Queue should not be fulli i s Queue should be fullt blocki s( Didn't appear to block with a full queuet timeoutg{�G�z�?s+ Didn't appear to time-out with a full queuei s Queue should be emptys* Didn't appear to block with an empty queues- Didn't appear to time-out with an empty queuet emptyi ( ( ( ( R+ ( R+ ( R+ t RuntimeErrort putt dictt gett assertEqualt __class__R t ranget QUEUE_SIZEt assertTruet fullR R& t FullR t Truet Empty( R t qt target_ordert actual_ordert it lastR5 ( ( s' /usr/lib64/python2.7/test/test_queue.pyt simple_queue_testZ s\ ! % c C s` xY t r[ | j � } | d k r/ | j � d S| j � | j | 7_ Wd QX| j � q Wd S( N( R7 R/ R t task_doneR$ R" ( R R9 t x( ( s' /usr/lib64/python2.7/test/test_queue.pyt worker� s c C s� d | _ x0 d D]( } t j d | j d | f � j � q Wx! t d � D] } | j | � qI W| j � | j | j t t d � � d � x d D] } | j d � q� W| j � d S( Ni i t targetR id s0 q.join() did not block until all tasks were done( i i ( i i ( R" R R RA R t xrangeR- R R0 t sumR2 R ( R R9 R< ( ( s' /usr/lib64/python2.7/test/test_queue.pyt queue_join_test� s & c C s? | j � } y | j � Wn t k r- n X| j d � d S( Ns( Did not detect task count going negative( t type2testR? t ValueErrorR ( R R9 ( ( s' /usr/lib64/python2.7/test/test_queue.pyt test_queue_task_done� s c C sY | j � } | j | � | j | � y | j � Wn t k rG n X| j d � d S( Ns( Did not detect task count going negative( RF RE R? RG R ( R R9 ( ( s' /usr/lib64/python2.7/test/test_queue.pyt test_queue_join� s c C s- | j t � } | j | � | j | � d S( N( RF R3 R> ( R R9 ( ( s' /usr/lib64/python2.7/test/test_queue.pyt test_simple_queue� s ( R R R% R> RA RE RH RI RJ ( ( ( s' /usr/lib64/python2.7/test/test_queue.pyR! U s 5 t QueueTestc B s e Z e j Z RS( ( R R R& RF ( ( ( s' /usr/lib64/python2.7/test/test_queue.pyRK � s t LifoQueueTestc B s e Z e j Z RS( ( R R R& R' RF ( ( ( s' /usr/lib64/python2.7/test/test_queue.pyRL � s t PriorityQueueTestc B s e Z e j Z RS( ( R R R& R( RF ( ( ( s' /usr/lib64/python2.7/test/test_queue.pyRM � s t FailingQueueExceptionc B s e Z RS( ( R R ( ( ( s' /usr/lib64/python2.7/test/test_queue.pyRN � s t FailingQueuec B s# e Z d � Z d � Z d � Z RS( c G s) t | _ t | _ t j j | | � d S( N( t Falset fail_next_putt fail_next_getR& R ( R R ( ( s' /usr/lib64/python2.7/test/test_queue.pyR � s c C s1 | j r t | _ t d � n t j j | | � S( Ns You Lose( RQ RP RN R& t _put( R t item( ( s' /usr/lib64/python2.7/test/test_queue.pyRS � s c C s. | j r t | _ t d � n t j j | � S( Ns You Lose( RR RP RN R& t _get( R ( ( s' /usr/lib64/python2.7/test/test_queue.pyRU � s ( R R R RS RU ( ( ( s' /usr/lib64/python2.7/test/test_queue.pyRO � s t FailingQueueTestc B s e Z d � Z d � Z RS( c C s� | j � s t d � n x% t t d � D] } | j | � q) Wt | _ y$ | j d d d �| j d � Wn t k r� n Xt | _ y$ | j d d d �| j d � Wn t k r� n X| j d � | j | j � d � t | _ y- | j | j d | j d � | j d � Wn t k r.n X| j d � t | _ y9 | j | j d t d f | j d t � | j d � Wn t k r�n X| j d � | j | j � d � | j � | j | j � d � | j d � | j | j � d � | j | j d | j d � x t t � D] } | j � q"W| j | j � d � | j d � t | _ y | j � | j d � Wn t k r�n X| j | j � d � t | _ y! | j d d � | j d � Wn t k r�n X| j | j � d � | j � | j | j � d � t | _ y0 | j | j d | j d t � | j d � Wn t k rin X| j | j � d � | j � | j | j � d � d S( Ns& Call this function with an empty queuei t oopsR) i s) The queue didn't fail when it should haveR* g�������?R= s Queue should be fullR5 i s Queue should not be fulls Queue should be emptyt firsts Queue should not be emptyR+ ( R5 ( ( ( R5 ( ( ( R+ ( R+ R, R2 R3 R- R7 RQ R RN R4 R5 R R/ R RR ( R R9 R< ( ( s' /usr/lib64/python2.7/test/test_queue.pyt failing_queue_test� s� ! c C s* t t � } | j | � | j | � d S( N( RO R3 RY ( R R9 ( ( s' /usr/lib64/python2.7/test/test_queue.pyt test_failing_queue7 s ( R R RY RZ ( ( ( s' /usr/lib64/python2.7/test/test_queue.pyRV � s Qc C s t j t t t t � d S( N( R t run_unittestRK RL RM RV ( ( ( s' /usr/lib64/python2.7/test/test_queue.pyt test_main? s t __main__( ( R& R t unittestt testR t import_moduleR R3 R R R R! t TestCaseRK RL RM t ExceptionRN RO RV R\ R ( ( ( s' /usr/lib64/python2.7/test/test_queue.pyt <module> s" )p [
💻
System Info
Current path
/lib64/python2.7/test
Contents
10 folders, 1437 files
Disk free
100.01 GB
PHP version
8.3.30
🚀
Quick Actions
📍 Script location
🌐 Web root
🗑️ Temp
📁 /lib64/python2.7/test
⚡ Nexus File Manager • 1447 items