\ddZddlZddlZddlZ ddlZddlmZn #e$rdZYnwxYwddl m Z ddl m Z Gdde j ZGd d e j ZGd d e j ZdS) z& Tests for L{twisted.internet.fdesc}. N)fdescznot supported on this platform)untilConcludes)unittestceZdZdZdZdZdS)NonBlockingTestszE Tests for L{fdesc.setNonBlocking} and L{fdesc.setBlocking}. ctj\}}|tj||tj||t j|t jtjztj || t j|t jtjzdS)zR L{fdesc.setNonBlocking} sets a file description to non-blocking. N) ospipe addCleanupclose assertFalsefcntlF_GETFL O_NONBLOCKrsetNonBlocking assertTrueselfrws 7lib/python3.11/site-packages/twisted/test/test_fdesc.pytest_setNonBlockingz$NonBlockingTests.test_setNonBlockingswyy1 !$$$ !$$$ Q 66FGGG Q  Au}55 EFFFFFctj\}}|tj||tj|t j|t j||tj|tj tj zdS)zK L{fdesc.setBlocking} sets a file description to blocking. N) r r r r rr setBlockingr rrrrs rtest_setBlockingz!NonBlockingTests.test_setBlocking'swyy1 !$$$ !$$$ Q ! Q 66FGGGGGrN)__name__ __module__ __qualname____doc__rrrrrrsD G G G H H H H HrrcZeZdZdZdZdZdZdZdZdZ dZ d Z d Z d Z d Zd ZdS)ReadWriteTestsz< Tests for L{fdesc.readFromFD}, L{fdesc.writeToFD}. ctj\|_|_t j|jt j|jdS)zG Create a non-blocking pipe that can be used in tests. N)r r rrrrrs rsetUpzReadWriteTests.setUp8sC TV$$$ TV$$$$$rc tj|jn#t$rYnwxYw tj|jdS#t$rYdSwxYw)z Close pipes. N)r r rOSErrorrr%s rtearDownzReadWriteTests.tearDown@s{  HTV        D   HTV         DD s ))A AAc6tj|j|S)z) Write data to the pipe. )r writeToFDr)rds rwritezReadWriteTests.writeMstvq)))rcdg}tj|j|j}| |r|dSdS|S)z* Read data from the pipe. Nrr)r readFromFDrappend)rlress rreadzReadWriteTests.readSs? tvqx00 ; t sJrc |d}||dk|}|t |||dd||dS)z Test that the number of bytes L{fdesc.writeToFD} reports as written with its return value are seen by L{fdesc.readFromFD}. shellorN)r-rr3 assertEquallen)rnss rtest_writeAndReadz ReadWriteTests.test_writeAndReadasx JJx  A IIKK Q### "1"q)))))rcddz}||}||dkg}d}d}||ks|dkrP|||t |dz }|dz }||kJ|dkPd|}|t ||||d||dS) z} Similar to L{test_writeAndRead}, but use a much larger string to verify the behavior for that case. s 0123456879i'r2rN)r-rr0r3r6joinr5)rorigwrittenresult resultlengthis rtest_writeAndReadLargez%ReadWriteTests.test_writeAndReadLargels u$**T"" ! $$$ W$$B MM$))++ & & & Cr OO +L FA W$$B &!! Vg... hwh00000rcg}tj|j|j}||g||dS)z Verify that reading from a file descriptor with no data does not raise an exception and does not result in the callback function being called. N)rr/rr0r5 assertIsNone)rr1rAs rtest_readFromEmptyz!ReadWriteTests.test_readFromEmptysO !$&!(33 B &!!!!!rctj|j||t jdS)z Test that using L{fdesc.readFromFD} on a cleanly closed file descriptor returns a connection done indicator. N)r r rr5r3rCONNECTION_DONEr%s rtest_readFromCleanClosez&ReadWriteTests.test_readFromCleanClose=  e&;<<<<.eagainWrites))C CIIrrMrcDt}tj|_|rW)r(rXEINTRrZs r eintrWritez3ReadWriteTests.test_writeErrors..eintrWrites))C CIIrN)r r-r5)r oldOsWriter^ras rtest_writeErrorszReadWriteTests.test_writeErrorssX      "   TZZ--q 1 1 1!BHHzBH ! ! ! !     "   TZZ--q 1 1 1!BHHHzBH ! ! ! !s)AA!4)B++B9N)rrrr r&r)r-r3r9rDrGrJrOrRrTrcr!rrr#r#3s%%%   ***    * * *111("""===BBB===BBB"""""rr#c(eZdZdZdZdZdZdZdS)CloseOnExecTestszL Tests for L{fdesc._setCloseOnExec} and L{fdesc._unsetCloseOnExec}. z import os, errno try: os.write(%d, b'lul') except OSError as e: if e.errno == errno.EBADF: os._exit(0) os._exit(5) except BaseException: os._exit(10) else: os._exit(20) ctj}|dkr tjtjtjd|j|fzgdS#t$r0ddl}| tj dYdSwxYwttj |ddS)Nrz-cr=) r forkexecvsys executableprogramfileno BaseException traceback print_exc_exitrwaitpid)rfObjpidros r_execWithFileDescriptorz(CloseOnExecTests._execWithFileDescriptorsgii !88 N^T4<4;;==:J+JK!       ##%%%  ""*c155a8 8sAA$$6BBct|d5}tj|||}|tj|| tj |dddddS#1swxYwYdS)z A file descriptor passed to L{fdesc._setCloseOnExec} is not inherited by a new process image created with one of the exec family of functions. wbrN) openmktempr_setCloseOnExecrmrurr WIFEXITEDr5 WEXITSTATUSrrsstatuss rtest_setCloseOnExecz$CloseOnExecTests.test_setCloseOnExecs $++-- & & 8$  !$++-- 0 0 011$77F OOBL00 1 1 1   R^F33Q 7 7 7  8 8 8 8 8 8 8 8 8 8 8 8 8 8 8 8 8 8sB B;;B?B?ct|d5}tj|tj|||}|tj || tj |dddddS#1swxYwYdS)z A file descriptor passed to L{fdesc._unsetCloseOnExec} is inherited by a new process image created with one of the exec family of functions. rwN) rxryrrzrm_unsetCloseOnExecrurr r{r5r|r}s rtest_unsetCloseOnExecz&CloseOnExecTests.test_unsetCloseOnExecs $++-- & & 9$  !$++-- 0 0 0  #DKKMM 2 2 211$77F OOBL00 1 1 1   R^F33R 8 8 8  9 9 9 9 9 9 9 9 9 9 9 9 9 9 9 9 9 9sB1C!!C%(C%N)rrrr rlrurrr!rrreresR G999$ 8 8 8 9 9 9 9 9rre)r rXr rjrtwisted.internetr ImportErrorskiptwisted.python.utilr twisted.trialrSynchronousTestCaserr#rer!rrrs? 'LLL'&&&&&&,,, +DDD, /.....""""""HHHHHx3HHH8Q"Q"Q"Q"Q"X1Q"Q"Q"h;9;9;9;9;9x3;9;9;9;9;9s %%