a zef*@sddlZddlZddlZddlZddlZddlZddlZddlZddlZe ej dkdGdddZ Gddde ej Z Gdd d e ej ZGd d d e ZGd d d eej ZGdddeej ZGddde ZGdddeej ZGdddeej ZedkredS)Nposixztests requires a posix system.c@sReZdZddZddZddZdd d Zd d Zd ZddZ ddZ ddZ dS)TestFileIOSignalInterruptcCs d|_dSN)_processselfrH/opt/bitninja-python-dojo/embedded/lib/python3.9/test/test_file_eintr.pysetUpszTestFileIOSignalInterrupt.setUpcCs:|jr6|jdur6z|jWnty4Yn0dSr)rpollkillOSErrorrrrr tearDowns  z"TestFileIOSignalInterrupt.tearDowncCs d|jS)zReturns the infile = ... line of code for the reader process. subclasseses should override this to test different IO objects. z=import %s as io ;infile = io.FileIO(sys.stdin.fileno(), "rb")modnamerrrr _generate_infile_setup_code$sz5TestFileIOSignalInterrupt._generate_infile_setup_codeTcCs||jdur:tdz|jWnty8Yn0|r\|j\}}||7}||7}|d|||fdS)a;A common way to cleanup and fail with useful debug output. Kills the process if it is still running, collects remaining output and fails the test with an error message including the output. Args: why: Text to go after "Error from IO process" in the message. stdout, stderr: standard output and error from the process so far to include in the error message. communicate: bool, when True we call communicate() on the process after killing it to gather additional output. Ng?z/Error from IO process %s: STDOUT: %sSTDERR: %s ) rr timesleep terminater communicatefaildecode)rwhystdoutstderrrZ stdout_endZ stderr_endrrr fail_with_process_info-s  z0TestFileIOSignalInterrupt.fail_with_process_infoc CsH|}t|dksJdtjtjddd|dd|ddgtjtjtjd |_|jj td }|d kr|j d |d |jj |d }g}|st |jjgddd\}}}|jtj|d7}|dkr|j|dq|jj}|dkr|j d|d |jjdd\} } |jjrD|j d|jj| | dddS)anGeneric buffered read method test harness to validate EINTR behavior. Also validates that Python signal handlers are run during the read. Args: data_to_write: String to write to the child process for reading before sending it a signal, confirming the signal was handled, writing a final newline and closing the infile pipe. read_and_verify_code: Single "line" of code to read from a file object named 'infile' and validate the result. This will be executed as part of a python subprocess fed data_to_write. iz#data_to_write must fit in pipe buf.z-uz-czXimport signal, sys ;signal.signal(signal.SIGINT, lambda s, f: sys.stderr.write("$\n")) ;z ;z"sys.stderr.write("Worm Sign!\n") ;zinfile.close())stdinrrs Worm Sign! zwhile awaiting a sign)rrrg?z,reader process failed to handle our signals.s$ zwhile awaiting signal )inputz exited rc=%dF)rN)rlen subprocessPopensys executablePIPErrreadrrwriteselect send_signalsignalSIGINTr rreadliner returncode) r data_to_writeread_and_verify_codeZinfile_setup_codeZ worm_signZ signals_sentZrlist_Z signal_linerrrrr _test_readingHs^        z'TestFileIOSignalInterrupt._test_readingzgot = infile.{read_method_name}() ;expected = {expected!r} ;assert got == expected, ("{read_method_name} returned wrong data.\n""got data %r\nexpected %r" % (got, expected))cCs|jd|jjdddddS)1readline() must handle signals and not lose data. hello, world!r.shello, world! Zread_method_nameZexpectedr0r1Nr3_READING_CODE_TEMPLATEformatrrrr test_readlinesz'TestFileIOSignalInterrupt.test_readlinecCs"|jd|jjdddgdddS)2readlines() must handle signals and not lose data. hello world! readlinesshello sworld! r6r7Nr8rrrr test_readlinessz(TestFileIOSignalInterrupt.test_readlinescCs8|jd|jjdddd|jd|jjdddddS)z0readall() must handle signals and not lose data.r=readall hello world! r6r7r(Nr8rrrr test_readallsz&TestFileIOSignalInterrupt.test_readallN)rrT) __name__ __module__ __qualname__r rrrr3r9r;r?rBrrrr rs  Irc@seZdZdZdS)CTestFileIOSignalInterrupt_ioNrCrDrErrrrr rFsrFc@seZdZdZdS)PyTestFileIOSignalInterrupt_pyioNrHrrrr rIsrIc@seZdZddZddZdS)TestBufferedIOSignalInterruptcCs d|jS)z?Returns the infile = ... line of code to make a BufferedReader.ziimport %s as io ;infile = io.open(sys.stdin.fileno(), "rb") ;assert isinstance(infile, io.BufferedReader)rrrrr rsz9TestBufferedIOSignalInterrupt._generate_infile_setup_codecCs|jd|jjdddddS)zReturns the infile = ... line of code to make a TextIOWrapper.zvimport %s as io ;infile = io.open(sys.stdin.fileno(), "rt", newline=None) ;assert isinstance(infile, io.TextIOWrapper)rrrrr rsz5TestTextIOSignalInterrupt._generate_infile_setup_codecCs|jd|jjdddddS)r4r5r.zhello, world! r6r7Nr8rrrr r;sz'TestTextIOSignalInterrupt.test_readlinecCs"|jd|jjdddgdddS)r<s hello world!r>zhello zworld! r6r7Nr8rrrr r?sz(TestTextIOSignalInterrupt.test_readlinescCs|jd|jjdddddS)z-read() must handle signals and not lose data.r=r(z hello world! r6r7Nr8rrrr rBsz&TestTextIOSignalInterrupt.test_readallN)rCrDrErr;r?rBrrrr rNsrNc@seZdZdZdS)CTestTextIOSignalInterruptrGNrHrrrr rOsrOc@seZdZdZdS)PyTestTextIOSignalInterruptrJNrHrrrr rPsrP__main__)osr*r,r#r%rZunittestrGrJZ skipUnlessnamerZTestCaserFrIrKrLrMrNrOrPrCmainrrrr  s*