Nexus File Manager
v2.0
🏠
Dashboard
⚡
System
🌐
›
lib64
›
python2.7
›
test
Quick:
⬆️ Parent
🌐 Root
🏠 Home
🌍 WWW
📁 Temp
⚙️ Etc
📤 Upload
📁 Create
⚡ WordPress Admin
🔄 Refresh
✏️ test_socketserver.pyo
← Back
� z fc @ s6 d Z d d l Z d d l Z d d l Z d d l Z d d l Z d d l Z d d l Z d d l Z d d l Z d d l Z d d l Z d d l Z d d l m Z m Z m Z y d d l Z Wn e k r� e Z n Xe j j d � d Z e j j Z e e d � Z e j e d � Z e e d � o:e j d k Z e j e d � Z d � Z e j Z d d � Z e r�d e j! e j" f d � � YZ# d e j! e j$ f d � � YZ% n e j& d � � Z' d � Z( e j e d � d e j) f d � � Y� Z* d e j) f d � � YZ+ d � Z, e- d k r2e, � n d S( s! Test suite for SocketServer.py. i����N( t reap_childrent reap_threadst verboset networks hello world t AF_UNIXs requires Unix socketst forkt os2s requires forkingc C s# t t d � r t j | � n d S( s7 Call signal.alarm when it exists (i.e. not on Windows).t alarmN( t hasattrt signalR ( t n( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt signal_alarm# s i c C sN t | g g g | � \ } } } | | k r: | j | � St d | f � d S( Ns timed out on %r( t _real_selectt recvt RuntimeError( t sockR t timeoutt rt wt x( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt receive+ s ! t ForkingUnixStreamServerc B s e Z RS( ( t __name__t __module__( ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyR 3 s t ForkingUnixDatagramServerc B s e Z RS( ( R R ( ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyR 7 s c c si t j � } | d k r( t j d � n d Vt j | d � \ } } | j | | � | j d | � d S( Ni iH i i H ( t osR t _exitt Nonet waitpidt assertEqual( t testcaset pidt pid2t status( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt simple_subprocess< s c C so | j � t | d � rk x? | j j � D]. } y t j | d � Wq) t k rV q) Xq) W| j j � n d S( Nt active_childreni ( t server_closeR R# t copyR R t ChildProcessErrort clear( t serverR ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt close_serverH s s! Threading required for this test.t SocketServerTestc B s. e Z d Z d � Z d � Z d � Z d � Z e d � � Z d � Z d � Z d � Z d � Z e d � � Z e d � � Z e d � � Z e e d � � � Z d � Z d � Z e d � � Z e j d � � Z d � Z e d � � Z e d � � Z e e d � � � Z e d � � Z d � Z RS( s Test all socket servers.c C s0 | j t d � t d � d | _ g | _ d S( Ni i< ( t addCleanupR t port_seedt test_files( t self( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt setUpZ s c C s[ | j � t � x9 | j D]. } y t j | � Wq t j k rH q Xq Wg | j (d S( N( t doCleanupsR R- R t removet error( R. t fn( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt tearDown` s c C s | t j k r t d f Sd } t j d k r7 d } n t j d d d | � } t j d k r� | d d k r{ | d } n | d t j t j f k r� | d } n t j d k r� | j t j t j � } q� | j t j t j � } n | j j | � | Sd S( Ni R s \sockett prefixs unix_socket.t diri t :i t /( t sockett AF_INETt HOSTR R t namet tempfilet mktempt sept altsept replaceR- t append( R. t protoR6 R3 ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt pickaddrk s c C si d | f d � � Y} d | f d � � Y} t r: d GHn | | | � } | j | j | j j � � | S( Nt MyServerc B s e Z d � Z RS( c S s | j | � t | � � d S( N( t close_requestR) ( R. t requestt client_address( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt handle_error� s ( R R RI ( ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyRE � s t MyHandlerc B s e Z d � Z RS( c S s# | j j � } | j j | � d S( N( t rfilet readlinet wfilet write( R. t line( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt handle� s ( R R RP ( ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyRJ � s s creating server( R R t server_addressR9 t getsockname( R. t addrt svrclst hdlrbaseRE RJ R( ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt make_server� s c C s' | j | j | j � | | � } | j } t rJ d GHd G| GHd G| GHn t j d d | d | j d i d d 6� } t | _ | j � t r� d GHn x6 t d � D]( } t r� d G| GHn | | j | � q� Wt r� d GHn | j � | j � t | � | j t j | j j � t r#d GHn d S( Ns server createds ADDR =s CLASS =R< s %s servingt targett kwargsg{�G�z�?t poll_intervals server runningi s test clients waiting for servert done( RV RD t address_familyRQ R t threadingt Threadt serve_forevert Truet daemont startt ranget shutdownt joinR) t assertRaisesR9 R2 t fileno( R. RT RU t testfuncR( RS t tt i( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt run_server� s6 c C s� t j | t j � } | j | � | j t � t | d � } } x/ | rs d | k rs t | d � } | | 7} qE W| j | t � | j � d S( Nid s ( R9 t SOCK_STREAMt connectt sendallt TEST_STRR R t close( R. RC RS t st buft data( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt stream_examine� s c C s� t j | t j � } t rC | t j k rC | j | j | � � n | j t | � t | d � } } x/ | r� d | k r� t | d � } | | 7} qi W| j | t � | j � d S( Nid s ( R9 t SOCK_DGRAMt HAVE_UNIX_SOCKETSR t bindRD t sendtoRn R R Ro ( R. RC RS Rp Rq Rr ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt dgram_examine� s c C s | j t j t j | j � d S( N( Rj t SocketServert TCPServert StreamRequestHandlerRs ( R. ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt test_TCPServer� s c C s | j t j t j | j � d S( N( Rj Ry t ThreadingTCPServerR{ Rs ( R. ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt test_ThreadingTCPServer� s c C s3 t | � �! | j t j t j | j � Wd QXd S( N( R" Rj Ry t ForkingTCPServerR{ Rs ( R. ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt test_ForkingTCPServer� s c C s | j t j t j | j � d S( N( Rj Ry t UnixStreamServerR{ Rs ( R. ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt test_UnixStreamServer� s c C s | j t j t j | j � d S( N( Rj Ry t ThreadingUnixStreamServerR{ Rs ( R. ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt test_ThreadingUnixStreamServer� s c C s0 t | � � | j t t j | j � Wd QXd S( N( R" Rj R Ry R{ Rs ( R. ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt test_ForkingUnixStreamServer� s c C s | j t j t j | j � d S( N( Rj Ry t UDPServert DatagramRequestHandlerRx ( R. ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt test_UDPServer� s c C s | j t j t j | j � d S( N( Rj Ry t ThreadingUDPServerR� Rx ( R. ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt test_ThreadingUDPServer� s c C s3 t | � �! | j t j t j | j � Wd QXd S( N( R" Rj Ry t ForkingUDPServerR� Rx ( R. ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt test_ForkingUDPServer� s c # sK t j � d d � f d � � Y} | � t _ z t j VWd � t _ Xd S( s< Mocks the select.select() call to raise EINTR for first callt MockSelectc s e Z d � Z � f d � Z RS( c S s d | _ d S( Ni ( t called( R. ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt __init__ s c sP | j d 7_ | j d k rB t j t j t j t j � � � n � | � Sd S( Ni ( R� t selectR2 t errnot EINTRR t strerror( R. t args( t old_select( s. /usr/lib64/python2.7/test/test_socketserver.pyt __call__ s $( R R R� R� ( ( R� ( s. /usr/lib64/python2.7/test/test_socketserver.pyR� s N( ( R� ( R. R� ( ( R� s. /usr/lib64/python2.7/test/test_socketserver.pyt mocked_select_module s c C sJ | j � �8 } | j t j t j | j � } | j | j d � Wd QXd S( Ni ( R� Rj Ry Rz R{ Rs t assertGreaterR� ( R. t mock_selectR ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt test_InterruptServerSelectCall s c C s | j t j t j | j � d S( N( Rj Ry t UnixDatagramServerR� Rx ( R. ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt test_UnixDatagramServer! s c C s | j t j t j | j � d S( N( Rj Ry t ThreadingUnixDatagramServerR� Rx ( R. ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt test_ThreadingUnixDatagramServer' s c C s | j t t j | j � d S( N( Rj R Ry R� Rx ( R. ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt test_ForkingUnixDatagramServer- s c C s d t j f d � � Y} d t j f d � � Y} g } xm t d � D]_ } | t d f | � } t j d d d | j d i d d 6� } t | _ | j | | f � qE Wx( | D] \ } } | j � | j � q� Wx( | D] \ } } | j � t | � q� Wd S( NRE c B s e Z RS( ( R R ( ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyRE 8 s RJ c B s e Z RS( ( R R ( ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyRJ ; s i i R< s MyServer servingRW RX g{�G�z�?RY ( Ry Rz R{ Rb R; R\ R] R^ R_ R` RB Ra Rc Rd R) ( R. RE RJ t threadsRi Rp Rh ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt test_shutdown4 s" c C sJ xC t d � D]5 } | j t � � t j t d f t j � Wd QXq Wd S( Ni i����( Rb Re t OverflowErrorRy Rz R; R{ ( R. Ri ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt test_tcpserver_bind_leakN s ( R R t __doc__R/ R4 RD RV R Rj Rs Rx R| R~ t requires_forkingR� t requires_unix_socketsR� R� R� R� R� R� t contextlibt contextmanagerR� R� R� R� R� R� R� ( ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyR* V s4 t MiscTestCasec B s e Z d � Z RS( c C s� d t j f d � � Y} | t d f t j � } t j | j t j � } | j | j � | j � | j � | j | j d � t | � d S( NRE c B s e Z d � Z d Z d � Z RS( c S s t S( N( t False( R. RG RH ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt verify_request` s i c S s& | j d 7_ t j j | | � d S( Ni ( t shutdown_calledRy Rz t shutdown_request( R. RG ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyR� d s ( R R R� R� R� ( ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyRE _ s i i ( Ry Rz R; R{ R9 R[ Rk Rl RQ Ro t handle_requestR R� R) ( R. RE R( Rp ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt4 test_shutdown_request_called_if_verify_request_false[ s ( R R R� ( ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyR� Y s c C s2 t j � r t j d � � n t j j t � d S( Ns" can't run when import lock is held( t impt lock_heldt unittestt SkipTestt testt test_supportt run_unittestR* ( ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt test_mainq s t __main__(. R� R� R� R R� R R9 R� R= R� Ry t test.test_supportR� R R R R\ t ImportErrorR R� t requiresRn R; R Ru t skipUnlessR� R< t HAVE_FORKINGR� R R R t ForkingMixInR� R R� R R� R" R) t TestCaseR* R� R� R ( ( ( s. /usr/lib64/python2.7/test/test_socketserver.pyt <module> sT �
💻
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