Nexus File Manager
v2.0
🏠
Dashboard
⚡
System
🌐
›
lib64
›
python2.7
›
test
Quick:
⬆️ Parent
🌐 Root
🏠 Home
🌍 WWW
📁 Temp
⚙️ Etc
📤 Upload
📁 Create
⚡ WordPress Admin
🔄 Refresh
✏️ test_socket.pyc
← Back
� z fc @ s� d d l Z d d l m 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 Z d d l Z y d d l Z Wn e k r� e Z n Xd Z d e j d � Z e j Z d Z e j Z y d d l Z d d l Z Wn e k rYe Z e Z n Xe j Z d Z d e j f d � � YZ d e j f d � � YZ d f d � � YZ d e e f d � � YZ! d e e f d � � YZ" d e! f d � � YZ# d e j e f d � � YZ$ d e j f d � � YZ% e j& e d � d e# f d � � Y� Z' e j& e d � d e" f d � � Y� Z( e j& e d � d e! f d � � Y� Z) e j& e* e d � d � e j& e d � d e$ f d! � � Y� � Z+ e j& e d � d" e! f d# � � Y� Z, e j& e d � d$ e# f d% � � Y� Z- d& e j f d' � � YZ. d( e- f d) � � YZ/ d* e- f d+ � � YZ0 d, e- f d- � � YZ1 d. e2 f d/ � � YZ3 d0 e3 e' f d1 � � YZ4 d2 e j f d3 � � YZ5 e j& e d � d4 e e f d5 � � Y� Z6 e j& e d � d6 e e f d7 � � Y� Z7 d8 e j f d9 � � YZ8 d: e f d; � � YZ9 d< e f d= � � YZ: d> e j f d? � � YZ; e j& e j< d@ k dA � dB e j f dC � � Y� Z= e j& e d � dD e# f dE � � Y� Z> dF Z? dG Z@ dH ZA dI � ZB e j& eB � dJ � dK e j f dL � � Y� ZC e j& eB � dJ � dM e j e f dN � � Y� ZD dO � ZE eF dP k r�eE � n d S(Q i����N( t test_supportg N@i c C s_ y, t j | t j � } | j | | f � Wn t j t j f k rL t SX| j � t Sd S( sZ Try to bind a socket on the given host:port and return True if that has been possible.N( t sockett SOCK_STREAMt bindt errort gaierrort Falset closet True( t hostt portt familyt sock( ( s( /usr/lib64/python2.7/test/test_socket.pyt try_address s s Michael Gilfix was here t SocketTCPTestc B s e Z d � Z d � Z RS( c C sD t j t j t j � | _ t j | j � | _ | j j d � d S( Ni ( R t AF_INETR t servR t bind_portR t listen( t self( ( s( /usr/lib64/python2.7/test/test_socket.pyt setUp7 s c C s | j j � d | _ d S( N( R R t None( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt tearDown< s ( t __name__t __module__R R ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR 5 s t SocketUDPTestc B s e Z d � Z d � Z RS( c C s4 t j t j t j � | _ t j | j � | _ d S( N( R R t SOCK_DGRAMR R R R ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyR B s c C s | j j � d | _ d S( N( R R R ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyR F s ( R R R R ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR @ s t ThreadableTestc B sM e Z d Z d � Z d � Z d � Z d � Z d � Z d � Z d � Z RS( s� Threadable Test class The ThreadableTest class makes it easy to create a threaded client/server pair from an existing unit test. To create a new threaded class from an existing unit test, use multiple inheritance: class NewClass (OldClass, ThreadableTest): pass This class defines two new fixture functions with obvious purposes for overriding: clientSetUp () clientTearDown () Any new test functions within the class must then define tests in pairs, where the test name is preceded with a '_' to indicate the client portion of the test. Ex: def testFoo(self): # Server portion def _testFoo(self): # Client portion Any exceptions raised by the clients during their tests are caught and transferred to the main thread to alert the testing framework. Note, the server setup function cannot call any blocking functions that rely on the client thread during setup, unless serverExplicitReady() is called just before the blocking call (such as in setting up a client/server connection and performing the accept() in setUp(). c C s4 | j | _ | j | _ | j | _ | j | _ d S( N( R t _ThreadableTest__setUpR t _ThreadableTest__tearDownt _setUpt _tearDown( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt __init__p s c C s | j j � d S( s� This method allows the server to explicitly indicate that it wants the client thread to proceed. This is useful if the server is about to execute a blocking routine that is dependent upon the client thread during its setup routine.N( t server_readyt set( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt serverExplicitReadyw s c C s� t j � | _ t j � | _ t j � | _ t j d � | _ | j � } | j d � } | | d } t | d | � } t j | j | f � | _ | j � | j j � s� | j j � n | j j � d S( Ni t .t _( t threadingt EventR! t client_readyt donet Queuet queuet idt rfindt getattrt threadt start_new_threadt clientRunt client_threadR t is_setR" t wait( R t methodnamet it test_method( ( s( /usr/lib64/python2.7/test/test_socket.pyR ~ s c C sI | j � | j j � | j j � sE | j j � } | j | � n d S( N( R R) R4 R+ t emptyt gett fail( R t msg( ( s( /usr/lib64/python2.7/test/test_socket.pyR � s c C s~ | j j � | j � | j j � t | � s? t d � � n y | � Wn# t k ro } | j j | � n X| j � d S( Ns& test_func must be a callable function.( R! R4 t clientSetUpR( R" t callablet TypeErrort ExceptionR+ t putt clientTearDown( R t test_funct strerror( ( s( /usr/lib64/python2.7/test/test_socket.pyR1 � s c C s t d � � d S( Ns clientSetUp must be implemented.( t NotImplementedError( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyR< � s c C s | j j � t j � d S( N( R) R" R/ t exit( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyRA � s ( R R t __doc__R R# R R R1 R< RA ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR J s $ t ThreadedTCPSocketTestc B s& e Z d d � Z d � Z d � Z RS( t runTestc C s$ t j | d | �t j | � d S( Nt methodName( R R R ( R RI ( ( s( /usr/lib64/python2.7/test/test_socket.pyR � s c C s t j t j t j � | _ d S( N( R R R t cli( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyR< � s c C s' | j j � d | _ t j | � d S( N( RJ R R R RA ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyRA � s ( R R R R< RA ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyRG � s t ThreadedUDPSocketTestc B s& e Z d d � Z d � Z d � Z RS( RH c C s$ t j | d | �t j | � d S( NRI ( R R R ( R RI ( ( s( /usr/lib64/python2.7/test/test_socket.pyR � s c C s t j t j t j � | _ d S( N( R R R RJ ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyR< � s c C s' | j j � d | _ t j | � d S( N( RJ R R R RA ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyRA � s ( R R R R< RA ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyRK � s t SocketConnectedTestc B s8 e Z d d � Z d � Z d � Z d � Z d � Z RS( RH c C s t j | d | �d S( NRI ( RG R ( R RI ( ( s( /usr/lib64/python2.7/test/test_socket.pyR � s c C s9 t j | � | j � | j j � \ } } | | _ d S( N( RG R R# R t acceptt cli_conn( R t connt addr( ( s( /usr/lib64/python2.7/test/test_socket.pyR � s c C s' | j j � d | _ t j | � d S( N( RN R R RG R ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyR � s c C s6 t j | � | j j t | j f � | j | _ d S( N( RG R< RJ t connectt HOSTR t serv_conn( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyR< � s c C s' | j j � d | _ t j | � d S( N( RS R R RG RA ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyRA � s ( R R R R R R< RA ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyRL � s t SocketPairTestc B s8 e Z d d � Z d � Z d � Z d � Z d � Z RS( RH c C s' t j j | d | �t j | � d S( NRI ( t unittestt TestCaseR R ( R RI ( ( s( /usr/lib64/python2.7/test/test_socket.pyR � s c C s t j � \ | _ | _ d S( N( R t socketpairR RJ ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyR � s c C s | j j � d | _ d S( N( R R R ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyR � s c C s d S( N( ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyR< � s c C s' | j j � d | _ t j | � d S( N( RJ R R R RA ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyRA � s ( R R R R R R< RA ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyRT � s t GeneralModuleTestsc B s$ e Z e j e d. k d � d � � Z d � Z d � Z d � Z d � Z d � Z d � Z e j e e d � d � d � � Z d � Z d � Z d � Z d � Z d � Z e j e e d � d � d � � Z e j e e d � d � d � � Z e j e e d � d � d � � Z e j e e d � d � d � � Z e j e e d � d � d � � Z d d � Z d � Z d � Z d � Z d � Z d! � Z d" � Z! e j e" j# d# k d$ � d% � � Z$ d& � Z% d' � Z& d( � Z' d) � Z( d* � Z) e* j+ d+ � � Z, e j e- d, � d- � � Z. RS(/ s need _socket modulec C s� t j t j t j � } z? d | j � | j | j | j f } | j t | � | � Wd | j � Xd | j | j | j f } | j t | � | � d S( Ns7 <socket object, fd=%s, family=%s, type=%s, protocol=%s>s7 <socket object, fd=-1, family=%s, type=%s, protocol=%s>( t _socketR R R t filenoR t typet protot assertEqualt reprR ( R t st expected( ( s( /usr/lib64/python2.7/test/test_socket.pyt test_csocket_repr� s " c C s� t j t j t j � } t j | � } | j | j � | j � � | j � d } y | j � Wn t k rt n X| j d � d S( Ns Socket proxy still exists( R R R t weakreft proxyR] RZ R R t ReferenceErrorR: ( R R_ t p( ( s( /usr/lib64/python2.7/test/test_socket.pyt test_weakref s c C sR t j � j } t j | � } | j | � | � ~ t j � | j | � � d S( N( R t _sockRb t reft assertIsR t gc_collectt assertIsNone( R R_ t w( ( s( /usr/lib64/python2.7/test/test_socket.pyt test_weakref__sock s c C sa d � } d � } d � } | j t j | d � | j t j | d � | j t j | d � d S( Nc _ s t j � d S( N( R R ( t argst kwargs( ( s( /usr/lib64/python2.7/test/test_socket.pyt raise_error# s c _ s t j � d S( N( R t herror( Rn Ro ( ( s( /usr/lib64/python2.7/test/test_socket.pyt raise_herror% s c _ s t j � d S( N( R R ( Rn Ro ( ( s( /usr/lib64/python2.7/test/test_socket.pyt raise_gaierror' s s Error raising socket exception.( t assertRaisesR R ( R Rp Rr Rs ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testSocketError! s c C s� t j t j t j � } | j | j � | j d � | j � } | j t � � | j d | � Wd QX| j t � � } | j d | � Wd QX| j d t | j � � | j t � � } | j d d � Wd QX| j d t | j � � | j t � � | j d d | � Wd QX| j t � � } | j d d | � Wd QX| j d t | j � � | j t � � } | j d d d � Wd QX| j d t | j � � | j t � � } | j d d | � Wd QX| j d t | j � � | j t � � } | j d d d � Wd QX| j d t | j � � | j t � � } | j d � Wd QX| j d t | j � � | j t � � } | j d d | d � Wd QX| j d t | j � � d S( Nt i u ☠y @s not complext foos not NoneTypet bars an integer is requireds (1 given)i s (4 given)( Rv i ( R R R t addCleanupR R t getsocknameRt t UnicodeEncodeErrort sendtoR> t assertInt strt exceptionR ( R R_ t socknamet cm( ( s( /usr/lib64/python2.7/test/test_socket.pyt testSendtoErrors0 s@ c C s< t j t j t j t j t j t j t j t j d S( N( R R R R t SOCK_RAWt SOCK_RDMt SOCK_SEQPACKETt SOL_SOCKETt SO_REUSEADDR( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testCrucialConstantsW s c C s� t j � } y t j | � } Wn! t j k rB | j d � n X| j | j d � d k d � y t j | � \ } } } Wn! t j k r� | j d � n X| | g | } t j | � } | | k r� | j d | t | � f � n d S( Ns name lookup failureR$ i s Error resolving host to ip.s address lookup failures= Error testing host resolution mechanisms. (fqdn: %s, all: %s)( R t gethostnamet gethostbynameR t skipTestt assertTruet findt gethostbyaddrt getfqdnR: R^ ( R t hostnamet ipt hnamet aliasest ipaddrst all_host_namest fqhn( ( s( /usr/lib64/python2.7/test/test_socket.pyt testHostnameResb s t getrefcounts test needs sys.getrefcount()c C sW y# t j t � } t j t d � Wn- t k rR | j t j t � | d � n Xd S( Ni s$ socket.getnameinfo loses a reference( t sysR� R R t getnameinfoR> R] ( R t orig( ( s( /usr/lib64/python2.7/test/test_socket.pyt testRefCountGetNameInfou s c C s/ y t j d d � Wn t j k r* n Xd S( Nt xi ( R� i i i ( R R� R ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testInterpreterCrash� s c C s� i d t j 6d t j 6d t j 6d t j 6} x� | j � D]� \ } } d | >d } xK d d d d d d d f D]. } | j | | @| | | | @� � | @� qq W| | � } | j | | @| � | j t | d � q; Wd S( Ni i l i i i�� i igE#i2Tvi" i ��l ( R t htonlt ntohlt htonst ntohst itemsR] Rt t OverflowError( R t sizest funct sizet maskR6 t swapped( ( s( /usr/lib64/python2.7/test/test_socket.pyt testNtoH� s ", c C s� d d d d d d g } d d d d d d g } xB | D]: } t j | � t j | � t j | � t j | � q7 Wxf | D]^ } | j t t j | � | j t t j | � | j t t j | � | j t t j | � q| Wd S( Ni i i l l l i����i����i����l���� l���� l���� ( R R� R� R� R� Rt R� ( R t good_valuest bad_valuest k( ( s( /usr/lib64/python2.7/test/test_socket.pyt testNtoHErrors� s c C s� | j } t j j d � sN t j j d � sN t j j d � sN t j d k rW d } n d } xE | D]4 } y t j | d � } PWqd t j k r� qd Xqd Wt j � t j | � } | | | � y t j | d � } Wn t j k r� d } n X| | | � | t j | � | � | t j | d � | � | d k rX| t j | d � | � n | j t t j d � | j t t j d � d S( Nt linuxt freebsdt netbsdt darwint daytimet qotdt domaint echot tcpt udpi����i ( R� R� R� ( R� R� R� ( R] R� t platformt startswithR t getservbynameR R t getservbyportRt R� ( R t eqt servicest serviceR t port2t udpport( ( s( /usr/lib64/python2.7/test/test_socket.pyt testGetServBy� s6 c C s | j t j � d � t j � } | j | j � d � | j � t j d � | j t j � d � t j � } | j | j � d � | j � t j d � | j t j � d � t j � } | j | j � d � | j � | j t t j d � | j t t j d � d S( Ni i����t spam( R] R t getdefaulttimeoutR t gettimeoutR t setdefaulttimeoutRt t ValueErrorR> ( R R_ ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testDefaultTimeout� s t inet_atons test needs socket.inet_aton()c C s6 | j d t j d � � | j d t j d � � d S( Ns i s 0.0.0.0s �s 255.255.255.255s s ����( R] R R� ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testIPv4_inet_aton_fourbytes� s t inet_ptons test needs socket.inet_pton()c s� d d l m } m � m � � � f d � } | j d | d � � | j d | d � � | j d | d � � | j d | d � � | j d | d � � | j d | d � � | j d | d � � | j d | d � � | j d | d � � d S( Ni����( R� R� R c s � � | � S( N( ( t a( R R� ( s( /usr/lib64/python2.7/test/test_socket.pyt <lambda>� Rv s s 0.0.0.0s � � s 255.0.255.0s ����s 170.170.170.170s s 1.2.3.4s ����s 255.255.255.255( R R� R� R R] ( R t ft g( ( R R� s( /usr/lib64/python2.7/test/test_socket.pyt testIPv4toString� s c s� y6 d d l m � m � m } | s5 | j d � n Wn t k rV | j d � n X� � f d � } | j d | d � � | j d | d � � | j d d | d � � | j d | d � � d S( Ni����( R� t AF_INET6t has_ipv6s IPv6 not availables+ could not import needed symbols from socketc s � � | � S( N( ( R� ( R� R� ( s( /usr/lib64/python2.7/test/test_socket.pyR� Rv s i s ::s 0::0s i s 1::s E�v� V�� �$��s$ 45ef:76cb:1a:56ef:afeb:bac:1924:aeaes s s ( R R� R� R� R� t ImportErrorR] ( R R� R� ( ( R� R� s( /usr/lib64/python2.7/test/test_socket.pyt testIPv6toString s t inet_ntops test needs socket.inet_ntop()c s� d d l m } m � m � � � f d � } | j d | d � � | j d | d � � | j d | d � � | j d | d � � | j d | d � � | j d | d � � | j d | d � � d S( Ni����( t inet_ntoaR� R c s � � | � S( N( ( R� ( R R� ( s( /usr/lib64/python2.7/test/test_socket.pyR� Rv s 1.0.1.0s s 170.85.170.85s �U�Us 255.255.255.255s ����s 1.2.3.4s ( R R� R� R R] ( R R� R� ( ( R R� s( /usr/lib64/python2.7/test/test_socket.pyt testStringToIPv4 s c s� y6 d d l m � m � m } | s5 | j d � n Wn t k rV | j d � n X� � f d � } | j d | d � � | j d | d � � | j d | d � � d S( Ni����( R� R� R� s IPv6 not availables+ could not import needed symbols from socketc s � � | � S( N( ( R� ( R� R� ( s( /usr/lib64/python2.7/test/test_socket.pyR� . Rv s ::s i s ::1i s s! aef:b01:506:1001:ffff:9997:55:170s � ���� Ups s s ( R R� R� R� R� R� R] ( R R� R� ( ( R� R� s( /usr/lib64/python2.7/test/test_socket.pyt testStringToIPv6% s s 0.0.0.0c C s? t j � } | j | d f � | j � \ } } | j � | S( s� Use a temporary socket to elicit an unused ephemeral port. Args: bind_address: Hostname or IP address to search for a port on. Returns: A most likely to be unused port. i ( R R Rz R ( R t bind_addresst tempsockR R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _get_unused_port9 s c C s� | j � } t j t j t j � } | j | j � | j d | f � | j � } y t j t j � � } Wn! t j k r� | j d � n X| j | d d | f d | d � | j | d | � d S( Ns 0.0.0.0s name lookup failurei s %s invalidi ( R� R R R Ry R R Rz R� R� R R� R} R] ( R R R t namet my_ip_addr( ( s( /usr/lib64/python2.7/test/test_socket.pyt testSockNameG s %c C sZ t j t j t j � } | j | j � | j t j t j � } | j | d k d � d S( Ni s initial mode is reuse( R R R Ry R t getsockoptR� R� t assertFalse( R R t reuse( ( s( /usr/lib64/python2.7/test/test_socket.pyt testGetSockOptY s c C ss t j t j t j � } | j | j � | j t j t j d � | j t j t j � } | j | d k d � d S( Ni i s failed to set reuse mode( R R R Ry R t setsockoptR� R� R� R� ( R R R� ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testSetSockOpta s c C sL t j t j t j � } | j d � | j � | j t j | j d � d S( Ni R� ( R R R t settimeoutR Rt R t send( R R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testSendAfterClosei s c C se t j t j t j � } | j | j t j � | j | j t j � | j | j d � | j � d S( Ni ( R R R R] R R[ R\ R ( R R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testNewAttributesp s c C s� t j � } | j | j � t j � } | d } | d } | j t | j t | f � | j t | j t | f � xr t j � D]d } t j � } y | j t | f � Wn7 t k r� } | j t j k s� | d k r� � q� q� XPq� Wd S( Ni i ( R Ry R R t find_unused_portRt R� R RR t itertoolst countt OSErrort errnot EADDRINUSE( R R R t big_portt neg_portR6 t e( ( s( /usr/lib64/python2.7/test/test_socket.pyt test_getsockaddrargx s t nts Windows specificc C s� | j t t j d � � | j t t d � � | j t t d � � | j t t d � � | j t t d � � t j � } | j | j � | j t | j d d � | j t j d � d S( Nt ioctlt SIO_RCVALLt RCVALL_ONt RCVALL_OFFt SIO_KEEPALIVE_VALSi����i id ( i id id ( R� t hasattrR Ry R Rt R� R� R R� ( R R_ ( ( s( /usr/lib64/python2.7/test/test_socket.pyt test_sock_ioctl� s c C sW y t j d d � Wn; t j k rQ } | j t j k rK | j d � n � n Xx0 t j t d � D] } | j t | � d � qe Wt j d d � t j d d � t j d d � t r� t j d d � n t j t d � t j t d � t j t d � t j t d � t j t d t j � } x0 | D]( \ } } } } } | j | t j � q-Wt j t d d t j � } x0 | D]( \ } } } } } | j | t j � q{Wt j t d d d t j � t j t d d d d t j � t j d d t j t j d t j � t t d � rSy# t j d d d d d t j � WqSt j k rOqSXn d S( Nt localhostiP s buggy libc versioni s 127.0.0.1s ::1t httpl P i t AI_NUMERICSERV( R t getaddrinfoR R� t EAI_SERVICER� RR R R] t lent SUPPORTS_IPV6R R t SOL_TCPt AI_PASSIVEt AF_UNSPECR� R� ( R t errt infot infosR R% t socktype( ( s( /usr/lib64/python2.7/test/test_socket.pyt testGetaddrinfo� sB c sK t t d � s t t d � r0 � j d � n � f d � } � f d � } t j � \ } } t j t j | � } z� | r� | j d � n � j t � �& t j d � | j d t j � Wd QX| rt j t j | � t j d � � j t j | j d t j � n Wd t j d � t j t j | � | j � | j � Xd S( Nt alarmRW s9 signal.alarm and socket.socketpair required for this testc s � j t t j d � d S( Ni ( Rt R� t matht acosh( Rn ( R ( s( /usr/lib64/python2.7/test/test_socket.pyt ok_handler� s c s" � j t t j d � d d d S( Ni i ( Rt R� R R ( Rn ( R ( s( /usr/lib64/python2.7/test/test_socket.pyt raising_handler� s g �?i R� i ( R� t signalR R� RW t SIGALRMR� Rt t ZeroDivisionErrorR t sendallR t SOCK_MAX_SIZEt timeoutR ( R t with_timeoutR R t cR_ t old_alarm( ( R s( /usr/lib64/python2.7/test/test_socket.pyt check_sendall_interrupted� s* c C s | j t � d S( N( R R ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt test_sendall_interrupted� s c C s | j t � d S( N( R R ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt% test_sendall_interrupted_with_timeout� s c C sW xP d D]H } t j t j t j � } | j t d f � | j | � | j � q Wd S( Ni i����( i i����( R R R R RR R R ( R t backlogt srv( ( s( /usr/lib64/python2.7/test/test_socket.pyt test_listen_backlog� s c C sb d d l } t j t j t j � } | j t d f � | j t | j | j d � | j � d S( Ni����i i ( t _testcapiR R R R RR Rt R� R t INT_MAXR ( R R R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt test_listen_backlog_overflow� s s IPv6 required for this test.c C s] | j t t j d d � t j t j t j � } z | j t | j d � Wd | j � Xd S( Ns ::1i I���� i����( s ::1i I���� ( s ::1i i����( Rt R� R R� R� R R R ( R R_ ( ( s( /usr/lib64/python2.7/test/test_socket.pyt test_flowinfo� s N(/ R R RU t skipUnlessRY R Ra Rf Rm Ru R� R� R� R� R� R� R� R� R� R� R� R R� R� R� R� R� R� R� R� R� R� R� R� t osR� R� R R R R R R t cpython_onlyR R R! ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyRX � sN ! ' ) $ 1 s! Threading required for this test.t BasicTCPTestc B s� e Z d d � Z d � Z d � Z d � Z d � Z d � Z d � Z d � Z d � Z d � Z d � Z e j e e d � d � d � � Z d � Z d � Z d � Z d � Z d � Z e j e � Z e j d � � Z RS( RH c C s t j | d | �d S( NRI ( RL R ( R RI ( ( s( /usr/lib64/python2.7/test/test_socket.pyR s c C s& | j j d � } | j | t � d S( Ni ( RN t recvR] t MSG( R R; ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testRecv s c C s | j j t � d S( N( RS R� R' ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testRecv s c C sL | j j t t � d � } | j j d � } | | } | j | t � d S( Ni i ( RN R& R R' R] ( R t seg1t seg2R; ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testOverFlowRecv s c C s | j j t � d S( N( RS R� R' ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testOverFlowRecv s c C s, | j j d � \ } } | j | t � d S( Ni ( RN t recvfromR] R' ( R R; RP ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testRecvFrom! s c C s | j j t � d S( N( RS R� R' ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testRecvFrom& s c C sX | j j t t � d � \ } } | j j d � \ } } | | } | j | t � d S( Ni i ( RN R. R R' R] ( R R* RP R+ R; ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testOverFlowRecvFrom) s " c C s | j j t � d S( N( RS R� R' ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testOverFlowRecvFrom0 s c C sK d } x* | j j d � } | s% Pn | | 7} q W| j | d d � d S( NRv i R� i ( RN R& R] ( R R; t read( ( s( /usr/lib64/python2.7/test/test_socket.pyt testSendAll3 s c C s d d } | j j | � d S( NR� i ( RS R ( R t big_chunk( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testSendAll= s t fromfds socket.fromfd not availablec C s] | j j � } t j | t j t j � } | j | j � | j d � } | j | t � d S( Ni ( RN RZ R R7 R R Ry R R& R] R' ( R t fdR R; ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testFromFdA s c C s | j j t � d S( N( RS R� R' ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testFromFdK s c C sB | j j � } | j | j � | j d � } | j | t � d S( Ni ( RN t dupRy R R& R] R' ( R R R; ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testDupN s c C s | j j t � d S( N( RS R� R' ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testDupU s c C s3 | j j d � } | j | t � | j j � d S( Ni ( RN R& R] R' R) R4 ( R R; ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testShutdownX s c C s$ | j j t � | j j d � d S( Ni ( RS R� R' t shutdown( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testShutdowna s c C st d d l } | j j t � | j t | j j | j d � | j t | j j d | j d � | j j d � d S( Ni����i i ( R RS R� R' Rt R� R? R t UINT_MAX( R R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testShutdown_overflowg s ( R R R R( R) R, R- R/ R0 R1 R2 R4 R6 RU R" R� R R9 R: R< R= R> R@ R R$ t testShutdown_overflowRB ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR% s( t BasicUDPTestc B sJ e Z d d � Z d � Z d � Z d � Z d � Z d � Z d � Z RS( RH c C s t j | d | �d S( NRI ( RK R ( R RI ( ( s( /usr/lib64/python2.7/test/test_socket.pyR u s c C s, | j j t t � � } | j | t � d S( N( R R& R R' R] ( R R; ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testSendtoAndRecvx s c C s# | j j t d t | j f � d S( Ni ( RJ R| R' RR R ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testSendtoAndRecv} s c C s2 | j j t t � � \ } } | j | t � d S( N( R R. R R' R] ( R R; RP ( ( s( /usr/lib64/python2.7/test/test_socket.pyR/ � s c C s# | j j t d t | j f � d S( Ni ( RJ R| R' RR R ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyR0 � s c C s | j t | j j d � d S( Ni����( Rt R� R R. ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testRecvFromNegative� s c C s# | j j t d t | j f � d S( Ni ( RJ R| R' RR R ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testRecvFromNegative� s ( R R R RE RF R/ R0 RG RH ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyRD r s t TCPCloserTestc B s e Z d � Z d � Z RS( c C s| | j j � \ } } | j � | j } t j | g g g d � \ } } } | j | | g � | j | j d � d � d S( Ng �?i Rv ( R RM R RJ t selectR] R& ( R RO RP t sdR3 t writeR ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testClose� s $c C s* | j j t | j f � t j d � d S( Ng �?( RJ RQ RR R t timet sleep( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testClose� s ( R R RM RP ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyRI � s RW s test needs socket.socketpair()t BasicSocketPairTestc B s8 e Z d d � Z d � Z d � Z d � Z d � Z RS( RH c C s t j | d | �d S( NRI ( RT R ( R RI ( ( s( /usr/lib64/python2.7/test/test_socket.pyR � s c C s& | j j d � } | j | t � d S( Ni ( R R& R] R' ( R R; ( ( s( /usr/lib64/python2.7/test/test_socket.pyR( � s c C s | j j t � d S( N( RJ R� R' ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyR) � s c C s | j j t � d S( N( R R� R' ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testSend� s c C s& | j j d � } | j | t � d S( Ni ( RJ R& R] R' ( R R; ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testSend� s ( R R R R( R) RR RS ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyRQ � s t NonBlockingTCPTestsc B s} e Z d d � Z d � Z d � Z e j d � � Z e j e � Z d � Z d � Z d � Z d � Z d � Z d � Z RS( RH c C s& t j � | _ t j | d | �d S( NRI ( R& R' t eventRG R ( R RI ( ( s( /usr/lib64/python2.7/test/test_socket.pyR � s c C s� | j j t � | j | j j � � | j j t � | j | j j � d � t j � } y | j j � Wn t j k r� n Xt j � } | j | | d k d � d S( Ng g �?s Error setting non-blocking mode.( R t setblockingR Rk R� R R] RN RM R R R� ( R t startt end( ( s( /usr/lib64/python2.7/test/test_socket.pyt testSetBlocking� s c C s d S( N( ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testSetBlocking� s c C s� d d l } | j | j k r. | j d � n | j j t � | j | j j � d � | j j | j d � | j | j j � � d S( Ni����s needs UINT_MAX < ULONG_MAXg i ( R RA t ULONG_MAXR� R RV R R] R� Rk ( R R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testSetBlocking_overflow� s c C s� | j j d � | j t j � � | j j � \ } } Wd QX| j j � t j | j g g g t � \ } } } | j | k r� | j d � n | j j � \ } } | j | j � | j | j � � d S( Ni s' Error trying to do accept after select.( R RV Rt R R RM RU R" RJ t MAIN_TIMEOUTR: Ry R Rk R� ( R RO RP R3 RL R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testAccept� s 'c C s* | j j � | j j t | j f � d S( N( RU R4 RJ RQ RR R ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testAccept� s c C s# | j j � \ } } | j � d S( N( R RM R ( R RO RP ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testConnect� s c C s- | j j d � | j j t | j f � d S( Ni ( RJ R� RQ RR R ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testConnect� s c C s� | j j � \ } } | j | j � | j d � | j t j � � | j t t � � } Wd QX| j j � t j | g g g t � \ } } } | | k r� | j d � n | j t t � � } | j | t � d S( Ni s0 Error during select call to non-blocking socket.( R RM Ry R RV Rt R R R& R R' RU R" RJ R] R: R] ( R RO RP R; R3 RL R ( ( s( /usr/lib64/python2.7/test/test_socket.pyR( � s $ c C s: | j j t | j f � | j j � | j j t � d S( N( RJ RQ RR R RU R4 R R' ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyR) s ( R R R RY RZ R R$ R\ t _testSetBlocking_overflowR^ R_ R` Ra R( R) ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyRT � s t FileObjectClassTestCasec B s� e Z d Z d d � Z d � Z d � Z d � Z d � Z d � Z d � Z d � Z d � Z d � Z d � Z d � Z d � Z d � Z d � Z d � Z d � Z d � Z d � Z RS( i����RH c C s t j | d | �d S( NRI ( RL R ( R RI ( ( s( /usr/lib64/python2.7/test/test_socket.pyR s c C s, t j | � | j j d | j � | _ d S( Nt rb( RL R RN t makefilet bufsizet serv_file( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyR # s c C s: | j j � | j | j j � t j | � d | _ d S( N( Rg R R� t closedRL R R ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyR ' s c C s& t j | � | j j d � | _ d S( Nt wb( RL R< RS Re t cli_file( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyR< - s c C s: | j j � | j | j j � d | _ t j | � d S( N( Rj R R� Rh R RL RA ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyRA 1 s c C sL | j j t t � d � } | j j d � } | | } | j | t � d S( Ni ( Rg R3 R R' R] ( R t first_segt second_segR; ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testSmallRead7 s c C s! | j j t � | j j � d S( N( Rj RL R' t flush( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testSmallRead> s c C s# | j j � } | j | t � d S( N( Rg R3 R] R' ( R R; ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testFullReadB s c C s! | j j t � | j j � d S( N( Rj RL R' R ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testFullReadG s c C sG d } x* | j j d � } | s% Pn | | 7} q W| j | t � d S( NRv i ( Rg R3 R] R' ( R t buft char( ( s( /usr/lib64/python2.7/test/test_socket.pyt testUnbufferedReadK s c C s! | j j t � | j j � d S( N( Rj RL R' Rn ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testUnbufferedReadU s c C s# | j j � } | j | t � d S( N( Rg t readlineR] R' ( R t line( ( s( /usr/lib64/python2.7/test/test_socket.pyt testReadlineY s c C s! | j j t � | j j � d S( N( Rj RL R' Rn ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testReadline^ s c C s� | j j t d � � } | j d | � | j j t d � � } | j d | � | j j � } | j d | � | j j � } | j d | � | j j � } | j t | � d S( Ns A baloo iss a bears s A BALOO IS A BEAR. ( Rg R3 R R] Rv R' ( R t a_baloo_ist _a_bearRw ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testReadlineAfterReadb s c C sA | j j d � | j j d � | j j t � | j j � d S( Ns A baloo is a bear s A BALOO IS A BEAR. ( Rj RL R' Rn ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testReadlineAfterReadn s c C sK | j j t d � � } | j d | � | j j � } | j d | � d S( Ns End Of t Line( Rg R3 R R] Rv ( R t end_of_Rw ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testReadlineAfterReadNoNewlinet s c C s | j j d � d S( Ns End Of Line( Rj RL ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testReadlineAfterReadNoNewlinez s c C s | j | j j � d S( N( R� Rg Rh ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testClosedAttr} s c C s | j | j j � d S( N( R� Rj Rh ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testClosedAttr� s ( R R Rf R R R R< RA Rm Ro Rp Rq Rt Ru Rx Ry R| R} R� R� R� R� ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyRc s( t FileObjectInterruptedTestCasec B sr e Z d Z d e f d � � YZ e d � � Z d d � Z d d � Z d � Z d � Z d d � Z d � Z RS( s= Test that the file object correctly handles EINTR internally.t MockSocketc B s e Z d d � Z d � Z RS( c C s t | � | _ d S( N( t itert _recv_step( R t recv_funcs( ( s( /usr/lib64/python2.7/test/test_socket.pyR � s c C s | j j � � S( N( R� t next( R R� ( ( s( /usr/lib64/python2.7/test/test_socket.pyR& � s ( ( R R R R& ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR� � s c C s t j t j � � d S( N( R R R� t EINTR( ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _raise_eintr� s i����c K sr | j d d � | j d � d � g � } t j | | � } | j | j | � d � | j | j | � d � d S( NR� c S s d S( Ns" This is the first line And the sec( ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR� � Rv c S s d S( Ns ond line is here ( ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR� � Rv c S s d S( NRv ( ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR� � Rv s This is the first line s And the second line is here ( R� R� R t _fileobjectR] Rv ( R R� Ro t mock_sockt fo( ( s( /usr/lib64/python2.7/test/test_socket.pyt _test_readline� s c K sY | j d d � | j d � d � g � } t j | | � } | j | j | � d � d S( NR� c S s d S( Ns" This is the first line And the sec( ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR� � Rv c S s d S( Ns ond line is here ( ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR� � Rv c S s d S( NRv ( ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR� � Rv s3 This is the first line And the second line is here ( R� R� R R� R] R3 ( R R� Ro R� R� ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _test_read� s c C s8 | j � | j d d � | j � | j d d � d S( NR� id ( R� R� ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt test_default� s c C sP | j d d � | j d d d d � | j d d � | j d d d d � d S( NRf i R� id ( R� R� ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt test_with_1k_buffer� s c C s� | j d d � d � d � | j d � d � g � } t j | d d �} | j | j | � d � | j | j | � d � d S( NR� c S s d S( Nt aa( ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR� � Rv c S s d S( Ns ( ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR� � Rv c S s d S( Nt BB( ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR� � Rv c S s d S( Nt bb( ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR� � Rv c S s d S( NRv ( ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR� � Rv Rf i s aa t BBbb( R� R� R R� R] Rv ( R R� R� R� ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _test_readline_no_buffer� s c C sD | j � | j d d � | j d d � | j d d d d � d S( NR� i Rf i id ( R� R� ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt test_no_buffer� s ( R R RF t objectR� t staticmethodR� R� R� R� R� R� R� ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR� � s t! UnbufferedFileObjectClassTestCasec B s& e Z d Z d Z d � Z d � Z RS( s� Repeat the tests from FileObjectClassTestCase with bufsize==0. In this case (and in this case only), it should be possible to create a file object, read a line from it, create another file object, read another line from it, without loss of data in the first file object's buffer. Note that httplib relies on this when reading multiple requests from the same socket.i c C sb | j j � } | j | d t � | j j d d � | _ | j j � } | j | d t � d S( Ns A. Rd i s B. ( Rg Rv R] R' RN Re ( R Rw ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testUnbufferedReadline� s c C s9 | j j d t � | j j d t � | j j � d S( Ns A. s B. ( Rj RL R' Rn ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testUnbufferedReadline� s ( R R RF Rf R� R� ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR� � s t# LineBufferedFileObjectClassTestCasec B s? e Z d Z d e f d � � YZ d � Z d � Z d � Z RS( i t SocketMemoc B sA e Z d Z d � Z d d � Z d d � Z d � Z d � Z RS( sG A wrapper to keep track of sent data, needed to examine write behaviourc C s | | _ g | _ d S( N( Rg t sent( R R ( ( s( /usr/lib64/python2.7/test/test_socket.pyR � s i c C s- | j j | | � } | j j | | � | S( N( Rg R� R� t append( R t datat flagst n( ( s( /usr/lib64/python2.7/test/test_socket.pyR� � s c C s' | j j | | � | j j | � d S( N( Rg R R� R� ( R R� R� ( ( s( /usr/lib64/python2.7/test/test_socket.pyR � s c C s t | j | � S( N( R. Rg ( R t attr( ( s( /usr/lib64/python2.7/test/test_socket.pyt __getattr__� s c C s5 g | j D]' } t | t � r+ | j � n | ^ q S( N( R� t isinstancet memoryviewt tobytes( R R� ( ( s( /usr/lib64/python2.7/test/test_socket.pyt getsent� s ( R R RF R R� R R� R� ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR� � s c C s, t j | � | j | j j � | j _ d S( N( Rc R R� Rg Rg ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyR � s c C s� t j � } | j | I| j | IJ| j | I| j | I| j | IJ| j d IJ| j j � d | | f } d | | | f } d } | j | j j j � | | | g � d S( NRv s %s %s s %s %s %s s ( R' t stripRg Rn R] Rg R� ( R R; t msg1t msg2t msg3( ( s( /usr/lib64/python2.7/test/test_socket.pyt testLinebufferedWrite s c C s� t j � } d | | f } d | | | f } d } | j j � } | j | | � | j j � } | j | | � | j j � } | j | | � d S( Ns %s %s s %s %s %s s ( R' R� Rj Rv R] ( R R; R� R� R� t l1t l2t l3( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testLinebufferedWrite s ( R R Rf R� R� R R� R� ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR� � s t$ SmallBufferedFileObjectClassTestCasec B s e Z d Z RS( i ( R R Rf ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR� " s t NetworkConnectionTestc B s e Z d Z d � Z RS( s Prove network connection.c C s+ t j t | j f � | _ | j | _ d S( N( R t create_connectionRR R RJ RS ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyR< ) s ( R R RF R< ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR� ' s t BasicTCPTest2c B s e Z d Z RS( sL Tests that NetworkConnection does not break existing TCP functionality. ( R R RF ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR� / s t NetworkConnectionNoServerc B sN e Z d e j f d � � YZ e j d � � Z d � Z d � Z d � Z RS( R� c B s e Z d � Z RS( c G s t j d � � d S( Ns timed out( R R ( R Rn ( ( s( /usr/lib64/python2.7/test/test_socket.pyRQ 5 s ( R R RQ ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR� 4 s c c s/ t j } | j t _ z d VWd | t _ Xd S( s* Return a socket which times out on connectN( R R� ( R t old_socket( ( s( /usr/lib64/python2.7/test/test_socket.pyt mocked_socket_module8 s c C s t j � } t j t j t j � } | j | j � | j t j � � } | j t | f � Wd QX| j | j j t j � d S( N( R R� R R R Ry R Rt R RQ RR R] R R� t ECONNREFUSED( R R RJ R� ( ( s( /usr/lib64/python2.7/test/test_socket.pyt test_connectB s c C s� t j � } | j t j � � } t j t | f � Wd QXt j g } t t d � rh | j t j � n t t d � r� | j t j � n | j | j j | � d S( Nt ENETUNREACHt EADDRNOTAVAIL( R R� Rt R R R� RR R� R� R� R� R� R� R} R ( R R R� t expected_errnos( ( s( /usr/lib64/python2.7/test/test_socket.pyt test_create_connectionJ s c C sC | j � �1 | j t j � � t j t d f � Wd QXWd QXd S( Ni� ( R� Rt R R R� RR ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt test_create_connection_timeouti s ( R R R R� t contextlibt contextmanagerR� R� R� R� ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR� 3 s t NetworkConnectionAttributesTestc B s� e Z d d � Z d � Z d � Z d � Z e Z d � Z e Z d � Z e Z d � Z e Z d � Z e Z d � Z e Z d � Z RS( RH c C s$ t j | d | �t j | � d S( NRI ( R R R ( R RI ( ( s( /usr/lib64/python2.7/test/test_socket.pyR t s c C s t j � | _ d S( N( R R� t source_port( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyR< x s c C s' | j j � d | _ t j | � d S( N( RJ R R R RA ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyRA { s c C s# | j j � \ } } | j � d S( N( R RM R ( R RO RP ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _justAccept� s c C sN t j t | j f d d �| _ | j | j j � | j | j j d � d S( NR i i ( R R� RR R RJ Ry R R] R ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testFamily� s !c C sg t j t | j f d d d d | j f �| _ | j | j j � | j | j j � d | j � d S( NR i t source_addressRv i ( R R� RR R R� RJ Ry R R] Rz ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testSourceAddress� s c C s� | j t j � d k � t j d � z2 t j t | j f � | _ | j | j j � Wd t j d � X| j | j j � d � d S( Ni* ( R� R R� R R� R� RR R RJ Ry R R] R� ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testTimeoutDefault� s c C s� | j t j � d k � t j d � z8 t j t | j f d d �| _ | j | j j � Wd t j d � X| j | j j � d � d S( Ni R ( R� R R� R R� R� RR R RJ Ry R R] R� ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testTimeoutNone� s !c C s> t j t | j f d d �| _ | j | j j � d � d S( NR i ( R R� RR R RJ R] R� ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testTimeoutValueNamed� s !c C sN t j t | j f d � | _ | j | j j � | j | j j � d � d S( Ni ( R R� RR R RJ Ry R R] R� ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testTimeoutValueNonamed� s ( R R R R< RA R� t testFamilyR� t testSourceAddressR� t testTimeoutDefaultR� t testTimeoutNoneR� t testTimeoutValueNamedR� t testTimeoutValueNonamedR� ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR� q s t NetworkConnectionBehaviourTestc B sG e Z d d � Z d � Z d � Z d � Z e Z d � Z d � Z RS( RH c C s$ t j | d | �t j | � d S( NRI ( R R R ( R RI ( ( s( /usr/lib64/python2.7/test/test_socket.pyR � s c C s d S( N( ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyR< � s c C s' | j j � d | _ t j | � d S( N( RJ R R R RA ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyRA � s c C sC | j j � \ } } | j | j � t j d � | j d � d S( Ni s done!( R RM Ry R RN RO R� ( R RO RP ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testInsideTimeout� s c C sB t j t | j f � | _ } | j d � } | j | d � d S( Ni s done!( R R� RR R RJ R& R] ( R R R� ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testInsideTimeout� s c sE t j t | j f d d �| _ � | j t j � f d � � d S( NR i c s � j d � S( Ni ( R& ( ( R ( s( /usr/lib64/python2.7/test/test_socket.pyR� � Rv ( R R� RR R RJ Rt R ( R ( ( R s( /usr/lib64/python2.7/test/test_socket.pyt _testOutsideTimeout� s %( R R R R< RA R� t testOutsideTimeoutR� R� ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR� � s t Urllib2FileobjectTestc B s e Z d � Z RS( c C s� d d d � � Y} | � } t j | � } | j � | j | j � | � } t j | d t �} | j � | j | j � d S( NR� c B s e Z e Z d � Z d � Z RS( c S s d S( N( ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyRn � Rv c S s t | _ d S( N( R Rh ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyR � Rv ( R R R Rh Rn R ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR� � s R ( ( R R� R R� Rh R ( R R� R_ R� ( ( s( /usr/lib64/python2.7/test/test_socket.pyRM � s ( R R RM ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR� � s t TCPTimeoutTestc B s> e Z d � Z d � Z e j e e d � d � d � � Z RS( c s) � f d � } � j t j | d � d S( Nc s! � j j d � � j j � d S( Ng �?( R R� RM ( Rn Ro ( R ( s( /usr/lib64/python2.7/test/test_socket.pyt raise_timeout� s s* Error generating a timeout exception (TCP)( Rt R R ( R R� ( ( R s( /usr/lib64/python2.7/test/test_socket.pyt testTCPTimeout� s c C s� t } y# | j j d � | j j � } WnM t j k rL | j d � n- t j k re t } n | j d � n X| s� | j d � n d S( Ng s% caught timeout instead of error (TCP)s! caught unexpected exception (TCP)s3 accept() returned success when we did not expect it( R R R� RM R R R: R R ( R t okRw ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testTimeoutZero� s R s test needs signal.alarm()c s2 | j j d � d t f d � � Y� � f d � } t j t j | � } z� y� z� y t j d � | j j � } Wn_ t j k r� | j d � nL � k r� n< | j d t j � d t j � f � n X| j d � Wd t j d � XWn � k r| j d � n XWd t j t j | � Xd S( Ng @t Alarmc B s e Z RS( ( R R ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR� s c s � � d S( N( ( R t frame( R� ( s( /usr/lib64/python2.7/test/test_socket.pyt alarm_handler s i s caught timeout instead of Alarms3 caught other exception instead of Alarm: %s(%s): %ss nothing caughti s got Alarm in wrong place( R R� R? R R R RM R R R: R� t exc_infot tracebackt format_exc( R R� R Rw ( ( R� s( /usr/lib64/python2.7/test/test_socket.pyt testInterruptedTimeout s, # ( R R R� R� RU R" R� R R� ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR� � s t UDPTimeoutTestc B s e Z d � Z d � Z RS( c s) � f d � } � j t j | d � d S( Nc s$ � j j d � � j j d � d S( Ng �?i ( R R� R& ( Rn Ro ( R ( s( /usr/lib64/python2.7/test/test_socket.pyR� * s s* Error generating a timeout exception (UDP)( Rt R R ( R R� ( ( R s( /usr/lib64/python2.7/test/test_socket.pyt testUDPTimeout) s c C s� t } y& | j j d � | j j d � } WnM t j k rO | j d � n- t j k rh t } n | j d � n X| s� | j d � n d S( Ng i s% caught timeout instead of error (UDP)s! caught unexpected exception (UDP)s1 recv() returned success when we did not expect it( R R R� R& R R R: R R ( R R� Rw ( ( s( /usr/lib64/python2.7/test/test_socket.pyR� 0 s ( R R R� R� ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR� ' s t TestExceptionsc B s e Z d � Z RS( c C sq | j t t j t � � | j t t j t j � � | j t t j t j � � | j t t j t j � � d S( N( R� t issubclassR R R? Rq R R ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testExceptionTree@ s ( R R R� ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR� > s R� s Linux specific testt TestLinuxAbstractNamespacec B s) e Z d Z d � Z d � Z d � Z RS( il c C s� d } t j t j t j � } | j | � | j d � t j t j t j � } | j | j � � | j � | j | j � | � | j | j � | � d S( Ns python-test-hello �i ( R t AF_UNIXR R R RQ Rz RM R] t getpeername( R t addresst s1t s2( ( s( /usr/lib64/python2.7/test/test_socket.pyt testLinuxAbstractNamespaceK s c C sT d d | j d } t j t j t j � } | j | � | j | j � | � d S( Ns t hi ( t UNIX_PATH_MAXR R� R R R] Rz ( R R� R_ ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testMaxNameV s c C sF d d | j } t j t j t j � } | j t j | j | � d S( Ns R� ( R� R R� R Rt R R ( R R� R_ ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testNameOverflow\ s ( R R R� R� R� R� ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR� F s t BufferIOTestc B s� e Z d Z d d � Z d � Z d � Z d � Z e Z d � Z e Z d � Z d � Z d � Z e Z d � Z e Z d � Z d � Z d � Z e Z RS( sF Test the buffer versions of socket.recv() and socket.send(). RH c C s t j | d | �d S( NRI ( RL R ( R RI ( ( s( /usr/lib64/python2.7/test/test_socket.pyR g s c C sh t j d d d � } | j j | � } | j | t t � � | j � t t � } | j | t � d S( NR t i ( t arrayRN t recv_intoR] R R' t tostring( R Rr t nbytesR; ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testRecvIntoArrayj s c C s3 t j � � t t � } Wd QX| j j | � d S( N( R t check_py3k_warningst bufferR' RS R� ( R Rr ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testRecvIntoArrayq s c C sX t d � } | j j | � } | j | t t � � | t t � } | j | t � d S( Ni ( t bytearrayRN R� R] R R' ( R Rr R� R; ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testRecvIntoBytearrayv s c C s^ t d � } | j j t | � � } | j | t t � � | t t � } | j | t � d S( Ni ( R RN R� R� R] R R' ( R Rr R� R; ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testRecvIntoMemoryview s c C sn t j d d d � } | j j | � \ } } | j | t t � � | j � t t � } | j | t � d S( NR R� i ( R� RN t recvfrom_intoR] R R' R� ( R Rr R� RP R; ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testRecvFromIntoArray� s c C s3 t j � � t t � } Wd QX| j j | � d S( N( R R� R� R' RS R� ( R Rr ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testRecvFromIntoArray� s c C s^ t d � } | j j | � \ } } | j | t t � � | t t � } | j | t � d S( Ni ( R RN R R] R R' ( R Rr R� RP R; ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testRecvFromIntoBytearray� s c C sd t d � } | j j t | � � \ } } | j | t t � � | t t � } | j | t � d S( Ni ( R RN R R� R] R R' ( R Rr R� RP R; ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testRecvFromIntoMemoryview� s c C s, t d � } | j t | j j | d � d S( Ni i ( R Rt R� RN R ( R Rr ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testRecvFromIntoSmallBuffer� s c C s3 t j � � t t � } Wd QX| j j | � d S( N( R R� R� R' RS R� ( R Rr ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testRecvFromIntoSmallBuffer� s c C s0 t � } | j j | � | j j | d � d S( Ni ( R RN R ( R Rr ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testRecvFromIntoEmptyBuffer� s ( R R RF R R� R� R t _testRecvIntoBytearrayR t _testRecvIntoMemoryviewR R R t _testRecvFromIntoBytearrayR t _testRecvFromIntoMemoryviewR R R t _testRecvFromIntoEmptyBuffer( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR� b s$ i� i� i� c C s� t t d � s t Sy t d � } Wn; t k r` } | j t j t j t j f k rZ t S� n X| �) x! | D] } | j d � ro t Sqo WWd QXt S( s� Check if the TIPC module is loaded The TIPC module is not loaded automatically on Ubuntu and probably other Linux distros. t AF_TIPCs /proc/moduless tipc N( R� R R t opent IOErrorR� t ENOENTt EISDIRt EACCESR� R ( R� R� Rw ( ( s( /usr/lib64/python2.7/test/test_socket.pyt isTipcAvailable� s ! s6 TIPC module is not loaded, please 'sudo modprobe tipc't TIPCTestc B s e Z d � Z RS( c C s� t j t j t j � } t j t j t j � } | j t j t j d � t j t t t f } | j | � t j t t t t d d f } | j t | � | j d � \ } } | j | j � | � | j | t � d S( Ni i i i ( R R R� R� R� R� t TIPC_ADDR_NAMESEQt TIPC_STYPEt TIPC_LOWERt TIPC_UPPERR t TIPC_ADDR_NAMER| R' R. R] Rz ( R R RJ t srvaddrt sendaddrR; t recvaddr( ( s( /usr/lib64/python2.7/test/test_socket.pyt testRDM� s ( R R R ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR � s t TIPCThreadableTestc B s8 e Z d d � Z d � Z d � Z d � Z d � Z RS( RH c C s' t j j | d | �t j | � d S( NRI ( RU RV R R ( R RI ( ( s( /usr/lib64/python2.7/test/test_socket.pyR � s c C s� t j t j t j � | _ | j j t j t j d � t j t t t f } | j j | � | j j d � | j � | j j � \ | _ | _ d S( Ni i ( R R R R R� R� R� R R R R R R R# RM RO t connaddr( R R ( ( s( /usr/lib64/python2.7/test/test_socket.pyR � s c C so t j d � t j t j t j � | _ t j t t t t d d f } | j j | � | j j � | _ d S( Ng�������?i i ( RN RO R R R RJ R R R R RQ Rz t cliaddr( R RP ( ( s( /usr/lib64/python2.7/test/test_socket.pyR< � s c C s< | j j d � } | j | t � | j | j | j � d S( Ni ( RO R& R] R' R# R" ( R R; ( ( s( /usr/lib64/python2.7/test/test_socket.pyt testStream s c C s! | j j t � | j j � d S( N( RJ R� R' R ( R ( ( s( /usr/lib64/python2.7/test/test_socket.pyt _testStream s ( R R R R R< R$ R% ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyR! � s c C s� t t t t t t t t t g } | j t t t t t t t t t t g � | j t � | j t � | j t t g � t j � } t j | � t j | � d S( N( RX R% RI R� R� R� R� RD R� t extendRT Rc R� R� R� R� R� R� R� R� R� RQ R� R R! R t threading_setupt run_unittestt threading_cleanup( t testst thread_info( ( s( /usr/lib64/python2.7/test/test_socket.pyt test_main s( t __main__(G RU t testR R� R� R RJ RN R� R* R� R# R� R� R R Rb RY R� R R] R R RR R' t IPV6_ENABLEDR R/ R& RV R R R RG RK RL RT RX R" R% RD RI R� RQ RT Rc 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 ( ( ( s( /usr/lib64/python2.7/test/test_socket.pyt <module> s� b � � h ehF?>D 9U $
💻
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