python (3.11.7)
e* 8 d dl Z d dlZd dlZd dlZd dlZd dlZd dlZd dlmZ ej s ej
d d dl Z d dl Z ej
e j dk d G d d Z G d d eej Z G d
d eej Z G d d
e Z G d deej Z G d deej Z G d de Z G d deej Z G d deej Zedk r ej dS dS ) N)supportz test module requires subprocessposixz tests requires a posix system.c H e Zd Zd Zd Zd Z d
dZd ZdZd Z d
Z
d Z d S )TestFileIOSignalInterruptc d | _ d S N)_processselfs H/BuggyBox/python/3.11.7/bootstrap/lib/python3.11/test/test_file_eintr.pysetUpz TestFileIOSignalInterrupt.setUp s
c | j rF| j / | j d S # t $ r Y d S w xY wd S d S r )r pollkillOSErrorr
s r tearDownz"TestFileIOSignalInterrupt.tearDown sp = T] / / 1 1 9
""$$$$$
9 9s =
A
A c d| j z S )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")modnamer
s r _generate_infile_setup_codez5TestFileIOSignalInterrupt._generate_infile_setup_code' s
? r r Tc | j ?t j d | j n# t
$ r Y nw xY w|r&| j \ }}||z
}||z
}| d|d| d| d dS )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?zError from IO process z
:
STDOUT:
zSTDERR:
) r r timesleep terminater communicatefaildecode)r whystdoutstderrr
stdout_end
stderr_ends r fail_with_process_infoz0TestFileIOSignalInterrupt.fail_with_process_info0 s = ' JsOOO
''))))
!%)]%>%>%@%@ "J
j F j F 33
: ; ; ; ; ;s A
AAc | }t | dk s
J d t j t j ddd|z dz dz |z dz dz gt j t j t j | _ | j j t d
}|d
k r|
d | | j j | d
}g }|st j
| j j gddd \ }}}| j t j |dz
}|dk r.| j | d || j j }|dk r|
d| | j d \ } }
| j j r(|
d| j j z | |
d dS dS )an Generic 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.
i z#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())stdinr! r" s Worm Sign!
zwhile awaiting a sign)r" r g? z,reader process failed to handle our signals.s $
zwhile awaiting signal
)inputz exited rc=%dF)r N)r len
subprocessPopensys
executablePIPEr r" readr% r' writeselect send_signalsignalSIGINTr r readliner
returncode) r
data_to_writeread_and_verify_codeinfile_setup_code worm_sign signals_sentrlist_ signal_liner! r" s r
_test_readingz'TestFileIOSignalInterrupt._test_readingK sk ! < < > > =!!C''')N''' #( tJ %% (,, 7 7
( (
+/ / "
" !ojo! ( ( (
M(--c/.B.BCC ' ' ' '(?/8
(
:
:
:
!!-000 J - )=(>BMMKE1a M % %fm 4 4 4 A Lc!!
""$$$ HIII
J m*3355 & ' '(?/:
(
<
<
< 222?? = # 7 ' '"T]%==F
(
7
7
7
7
7 7 7r zgot = infile.{read_method_name}() ;expected = {expected!r} ;assert got == expected, ("{read_method_name} returned wrong data.\n""got data %r\nexpected %r" % (got, expected))c h | d| j dd dS )1readline() must handle signals and not lose data.
hello, world!r9 s hello, world!
read_method_nameexpectedr; r< NrC _READING_CODE_TEMPLATEformatr
s r
test_readlinez'TestFileIOSignalInterrupt.test_readline sN .%)%@%G%G)3!3 &H &5 &5 6 6 6 6 6r c l | d| j dddg dS )2readlines() must handle signals and not lose data. hello
world! readliness hello
s world!
rG rJ NrK r
s r test_readlinesz(TestFileIOSignalInterrupt.test_readlines sS .%)%@%G%G)4",k!: &H &< &< = = = = =r c | d| j dd | d| j dd dS )z0readall() must handle signals and not lose data.rQ readall
hello
world!
rG rJ r3 NrK r
s r test_readallz&TestFileIOSignalInterrupt.test_readall s .%)%@%G%G)2!3 &H &5 &5 6 6 6
.%)%@%G%G)/!3 &H &5 &5 6 6 6 6 6r N)r r T) __name__
__module__ __qualname__r
r r r% rC rL rN rS rW r( r r r r s >A+/; ; ; ;6D7 D7 D7R
E 6 6 6= = = 6 6 6 6 6r r c e Zd ZdZdS )CTestFileIOSignalInterrupt_ioNrX rY rZ r r( r r r\ r\ GGGr r\ c e Zd ZdZdS )PyTestFileIOSignalInterrupt_pyioNr^ r( r r ra ra GGGr ra c e Zd Zd Zd ZdS ) TestBufferedIOSignalInterruptc d| j z S )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)r r
s r r z9TestBufferedIOSignalInterrupt._generate_infile_setup_code s ? r c h | d| j dd dS )z<BufferedReader.read() must handle signals and not lose data.rQ r3 rV rG rJ NrK r
s r rW z*TestBufferedIOSignalInterrupt.test_readall sN .%)%@%G%G)/!3 &H &5 &5 6 6 6 6 6r N)rX rY rZ r rW r( r r re re s2 6 6 6 6 6r re c e Zd ZdZdS ) CTestBufferedIOSignalInterruptr] Nr^ r( r r ri ri r_ r ri c e Zd ZdZdS ) PyTestBufferedIOSignalInterruptrb Nr^ r( r r rk rk rc r rk c & e Zd Zd Zd Zd Zd ZdS )TestTextIOSignalInterruptc d| j z S )z>Returns the infile = ... line of code to make a TextIOWrapper.zimport %s as io ;infile = io.open(sys.stdin.fileno(), encoding="utf-8", newline=None) ;assert isinstance(infile, io.TextIOWrapper)r r
s r r z5TestTextIOSignalInterrupt._generate_infile_setup_code s > r c h | d| j dd dS )rE rF r9 zhello, world!
rG rJ NrK r
s r rN z'TestTextIOSignalInterrupt.test_readline sN .%)%@%G%G)3!2 &H &4 &4 5 5 5 5 5r c l | d| j dddg dS )rP s
hello
world!rR zhello
zworld!
rG rJ NrK r
s r rS z(TestTextIOSignalInterrupt.test_readlines sS 0%)%@%G%G)4"+Z!8 &H &: &: ; ; ; ; ;r c h | d| j dd dS )z-read() must handle signals and not lose data.rQ r3 z
hello
world!
rG rJ NrK r
s r rW z&TestTextIOSignalInterrupt.test_readall sN .%)%@%G%G)/!2 &H &4 &4 5 5 5 5 5r N)rX rY rZ r rN rS rW r( r r rm rm sP 5 5 5; ; ;5 5 5 5 5r rm c e Zd ZdZdS )CTestTextIOSignalInterruptr] Nr^ r( r r rs rs r_ r rs c e Zd ZdZdS )PyTestTextIOSignalInterruptrb Nr^ r( r r ru ru rc r ru __main__)osr5 r7 r. r0 r unittesttestr has_subprocess_supportSkipTestr] rb
skipUnlessnamer TestCaser\ ra re ri rk rm rs ru rX mainr( r r <module>r s
% ?
(
=
>
>>
RW')IJJ[6 [6 [6 [6 [6 [6 [6 KJ[6| !:H<M ";X=N
6
6
6
6
6$=
6
6
6 %BHDU &CXEV 5 5 5 5 5 9 5 5 5@ !:H<M ";X=N zHMOOOOO r 
Browse
Usage
Versions