ó [³XMc@sidZddlZddlZddlZddlZddlZddlmZmZm Z m Z m Z m Z m Z ddlmZddlmZddlmZmZde jfd„ƒYZd ejfd „ƒYZd e jfd „ƒYZd e jfd„ƒYZdejfd„ƒYZejedƒsGde_ nej!edƒsede_ ndS(sK Tests for implementations of L{IReactorUNIX} and L{IReactorUNIXDatagram}. iÿÿÿÿN(t interfacestreactortprotocolterrortaddresstdefertutils(tlockfile(tunittest(tMyServerFactorytMyClientFactorytFailedConnectionClientFactorycBseZd„Zd„ZRS(cCs ||_dS(N(tonFail(tselfR ((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyt__init__scCs|jj|ƒdS(N(R terrback(R t connectortreason((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pytclientConnectionFaileds(t__name__t __module__RR(((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyR s tUnixSocketTestCasecBsheZdZd„Zd„Zd„Zd„Zd„Zd„Zd„Z d„Z d „Z d „Z RS( s Test unix sockets. cs¾ˆjƒ}ˆjƒ‰tƒ‰tjƒ}ˆ_tj|ˆƒ}ˆj|jƒt j t j t j ƒ}ˆj|j ƒ|j ˆƒ|j|ƒ‡‡‡fd†}|j|ƒ|S(sè The address passed to the server factory's C{buildProtocol} method and the address returned by the connected protocol's transport's C{getPeer} method match the address the client socket is bound to. csBtjˆƒ}ˆjˆj|gƒˆj|jjƒ|ƒdS(N(Rt UNIXAddresst assertEqualt peerAddressest transporttgetPeer(tprototexpected(R tpeernamet serverFactory(s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyt cbConnMade/s(tmktempR RtDeferredtprotocolConnectionMadeRt listenUNIXt addCleanupt stopListeningtsockettAF_UNIXt SOCK_STREAMtclosetbindtconnectt addCallback(R tfilenametconnMadetunixPortt unixSocketR((R RRs:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyt test_peerBinds      cs·ˆjƒ‰tƒ}tjƒ}||_tjˆ|ƒ}ˆj|jƒt ƒ‰tjƒ}|ˆ_tj ˆˆƒ}tj ||gƒ}‡‡‡fd†}|j |ƒ|S(sŠ L{IReactorUNIX.connectUNIX} can be used to connect a client to a server started with L{IReactorUNIX.listenUNIX}. csI|\}}ˆjˆjtjˆƒgƒ|jjƒ|jjƒdS(N(RRRRRtloseConnection(t.0tserverProtocoltclientProtocol(R R-t clientFactory(s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyt allConnectedGs    ( R R RR!R"RR#R$R%R t connectUNIXt gatherResultsR,(R RtserverConnMadeR/tclientConnMadetctdR7((R R-R6s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyt test_dumber7s        csïˆjƒ‰tƒ}tjƒ}||_tjˆ|dtƒ‰ˆjt j ˆdƒƒt ƒ‰tjƒ}|ˆ_tj ˆˆddƒ}tj ||gƒ}‡‡‡‡fd†}|j|ƒ‡‡fd†}|j|ƒ|S(sà A lockfile is created and locked when L{IReactorUNIX.listenUNIX} is called and released when the Deferred returned by the L{IListeningPort} provider's C{stopListening} method is called back. twantPIDs.locktcheckPIDicsO|\}}ˆjˆjtjˆƒgƒ|jjƒ|jjƒˆjƒS(N(RRRRRR2R%(R3R4t clientProto(R/R R-R6(s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyt _portStuffjs     cs!ˆjtjˆdƒdƒdS(Ns.locktlocked(tfailIfRtisLocked(tignored(R R-(s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyt_checkvs(R R RR!R"RR#tTruet assertTrueRRER R8R9R,(R RR:R;R<R=RBRG((R R-R6R/s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyt test_pidFileTs          cst|jƒ‰tƒ‰tjˆˆdtƒ}|jtjtjˆˆdtƒ‡‡fd†}|jƒj |ƒS(s› L{IReactorUNIX.listenUNIX} raises L{error.CannotListenError} if passed the name of a file on which a server is already listening. R?cs"tjˆˆdtƒ}|jƒS(NR?(RR#RHR%(tignR/(R-R(s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pytstoppedListening‰s( R R RR#RHt assertRaisesRtCannotListenErrorR%R,(R R/RL((R-Rs:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyttest_socketLocking|s  cCsp|jƒ|_d|jf}itjjtjƒd6}tjtj dd|fd|ƒ}|j |ƒ|S(Nsnfrom twisted.internet import protocol, reactor reactor.listenUNIX(%r, protocol.ServerFactory(), wantPID=True) t PYTHONPATHs-us-ctenv( R R-tostpathseptjointsystpathRtgetProcessValuet executableR,(R tcallbacktsourceRQR=((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyt_uncleanSocketTests $ cs‡fd†}ˆj|ƒS(s" If passed C{True} for the C{wantPID} parameter, a server can be started listening with L{IReactorUNIX.listenUNIX} when passed the name of a file on which a previous server which has not exited cleanly has been listening using the C{wantPID} option. cs(tjˆjtƒdtƒ}|jƒS(NR?(RR#R-R RHR%(RKtp(R (s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pytranStupidChild¢s(R[(R R]((R s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyttest_uncleanServerSocketLocking›scs‡fd†}ˆj|ƒS(s¶ If passed C{True} for the C{checkPID} parameter, a client connection attempt made with L{IReactorUNIX.connectUNIX} fails with L{error.BadFileError}. csFtjƒ}t|ƒ}tjˆj|dtƒ}ˆj|tj ƒS(NR@( RR!R RR8R-RHt assertFailureRt BadFileError(RKR=tfR<(R (s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyR]¯s  (R[(R R]((R s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyttest_connectToUncleanServer©scs’ˆjƒ}tj||ƒ‰dˆ|f}ˆjtˆƒ|ƒˆjtˆƒ|ƒtjˆjƒ}‡‡‡fd†}|j |ƒ|S(s~ Test the C{__str__} and C{__repr__} implementations of a UNIX port when used with the given factory. s <%s on %r>cs=dˆf}ˆjtˆƒ|ƒˆjtˆƒ|ƒdS(Ns<%s (not listening)>(Rtreprtstr(RKtunconnectedString(R t factoryNameR/(s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyRLÄs ( R RR#RRcRdRt maybeDeferredR%R,(R RRfR-tconnectedStringR=RL((R RfR/s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyt _reprTest·s  cCs9ddd„ƒY}|j|tjƒ|j|ƒdƒS(s The two string representations of the L{IListeningPort} returned by L{IReactorUNIX.listenUNIX} contains the name of the classic factory class being used and the filename on which the port is listening or indicates that the port is not listening. tClassicFactorycBseZd„Zd„ZRS(cSsdS(N((R ((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pytdoStartÔscSsdS(N((R ((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pytdoStop×s(RRRkRl(((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyRjÓs s%twisted.test.test_unix.ClassicFactory((tassertIsInstancettypest ClassTypeRi(R Rj((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyttest_reprWithClassicFactoryÌscCs9dtfd„ƒY}|j|tƒ|j|ƒdƒS(s! The two string representations of the L{IListeningPort} returned by L{IReactorUNIX.listenUNIX} contains the name of the new-style factory class being used and the filename on which the port is listening or indicates that the port is not listening. tNewStyleFactorycBseZd„Zd„ZRS(cSsdS(N((R ((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyRkéscSsdS(N((R ((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyRlìs(RRRkRl(((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyRqès s&twisted.test.test_unix.NewStyleFactory(tobjectRmttypeRi(R Rq((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyttest_reprWithNewStyleFactoryás( RRt__doc__R1R>RJROR[R^RbRiRpRt(((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyRs   (     t ClientProtocBs<eZeZZdZd„Zd„Zd„Z d„Z RS(cCs"tjƒ|_tjƒ|_dS(N(RR!tdeferredStartedtdeferredGotBack(R ((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyRûscCs t|_dS(N(RHtstopped(R ((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyt stopProtocolÿscCst|_|jjdƒdS(N(RHtstartedRwRYtNone(R ((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyt startProtocols cCs||_|jjdƒdS(N(tgotbackRxRYR|(R tdata((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pytdatagramReceiveds N( RRtFalseR{RyR|R~RRzR}R€(((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyRv÷s     t ServerProtocBs@eZeZZdZZd„Zd„Z d„Z d„Z RS(cCs"tjƒ|_tjƒ|_dS(N(RR!RwtdeferredGotWhat(R ((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyRscCs t|_dS(N(RHRy(R ((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyRzscCst|_|jjdƒdS(N(RHR{RwRYR|(R ((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyR}s cCs9||_|jjd|ƒ||_|jjdƒdS(Nshi back(tgotfromRtwritetgotwhatRƒRYR|(R Rtaddr((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyR€s  N( RRRR{RyR|R†R„RRzR}R€(((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyR‚ s      tDatagramUnixSocketTestCasecBs;eZdZd„Zd„Zd„Zd„Zd„ZRS(s% Test datagram UNIX sockets. cs׈jƒ‰ˆjƒ}tƒ‰tƒ‰tj|ˆƒ}ˆj|jƒtj|ˆdˆƒ}ˆj|jƒtj ˆj ˆj gƒ}‡‡fd†}‡‡‡‡fd†}|j |ƒ|j |ƒ|S(sf Test that a datagram can be sent to and received by a server and vice versa. t bindAddresscs)ˆjjdƒtjˆjˆjgƒS(Nthi(RR…RR9RƒRx(RF(tcptsp(s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyR…4s cs=ˆjdˆjƒˆjˆˆjƒˆjdˆjƒdS(NRŠshi back(tfailUnlessEqualR†R„R~(RF(R‹R RŒt clientaddr(s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyt_cbTestExchange9s( R R‚RvRtlistenUNIXDatagramR$R%tconnectUNIXDatagramRR9RwR,(R t serveraddrtsR<R=R…R((RŽR RŒR‹s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyt test_exchange%s      cCs^|jƒ}tƒ}tj||ƒ}|jtjtj||ƒ|jƒtj |ƒdS(sž L{IReactorUNIXDatagram.listenUNIXDatagram} raises L{error.CannotListenError} if the unix socket specified is already in use. N( R R‚RRtfailUnlessRaisesRRNR%RRtunlink(R R‡R\R“((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyttest_cannotListenCs    cs’ˆjƒ}tj||ƒ‰dˆ|f}ˆjtˆƒ|ƒˆjtˆƒ|ƒtjˆjƒ}‡‡‡fd†}|j |ƒ|S(sˆ Test the C{__str__} and C{__repr__} implementations of a UNIX datagram port when used with the given protocol. s <%s on %r>cs=dˆf}ˆjtˆƒ|ƒˆjtˆƒ|ƒdS(Ns<%s (not listening)>(RRcRd(RKRe(R t protocolNameR/(s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyRL_s ( R RRRRcRdRRgR%R,(R t serverProtoR˜R-Rht stopDeferredRL((R R/R˜s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyRiRs  cCs9ddd„ƒY}|j|tjƒ|j|ƒdƒS(s0 The two string representations of the L{IListeningPort} returned by L{IReactorUNIXDatagram.listenUNIXDatagram} contains the name of the classic protocol class being used and the filename on which the port is listening or indicates that the port is not listening. tClassicProtocolcBseZd„Zd„ZRS(cSsdS(N((R R((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pytmakeConnectionoscSsdS(N((R ((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyRlrs(RRRœRl(((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyR›ns s&twisted.test.test_unix.ClassicProtocol((RmRnRoRi(R R›((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyttest_reprWithClassicProtocolgscCs9dtfd„ƒY}|j|tƒ|j|ƒdƒS(s2 The two string representations of the L{IListeningPort} returned by L{IReactorUNIXDatagram.listenUNIXDatagram} contains the name of the new-style protocol class being used and the filename on which the port is listening or indicates that the port is not listening. tNewStyleProtocolcBseZd„Zd„ZRS(cSsdS(N((R R((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyRœ„scSsdS(N((R ((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyRl‡s(RRRœRl(((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyRžƒs s'twisted.test.test_unix.NewStyleProtocol(RrRmRsRi(R Rž((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyttest_reprWithNewStyleProtocol|s(RRRuR”R—RiRRŸ(((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyRˆ!s     s1This reactor does not support UNIX domain socketss3This reactor does not support UNIX datagram sockets("RutstatRRRURnR&ttwisted.internetRRRRRRRttwisted.pythonRt twisted.trialRttwisted.test.test_tcpR R t ClientFactoryR tTestCaseRtConnectedDatagramProtocolRvtDatagramProtocolR‚Rˆt IReactorUNIXR|tskiptIReactorUNIXDatagram(((s:/usr/lib/python2.7/dist-packages/twisted/test/test_unix.pyts0 4 Üq