Back to index

python3.2  3.2.2
Public Member Functions | Public Attributes | Static Public Attributes
test.test_io.PySignalsTest Class Reference
Inheritance diagram for test.test_io.PySignalsTest:
Inheritance graph
[legend]
Collaboration diagram for test.test_io.PySignalsTest:
Collaboration graph
[legend]

List of all members.

Public Member Functions

def setUp
def tearDown
def alarm_interrupt
def check_interrupted_write
def test_interrupted_write_unbuffered
def test_interrupted_write_buffered
def test_interrupted_write_text
def check_reentrant_write
def test_reentrant_write_buffered
def test_reentrant_write_text
def check_interrupted_read_retry
def test_interrupterd_read_retry_buffered
def test_interrupterd_read_retry_text
def check_interrupted_write_retry
def test_interrupterd_write_retry_buffered
def test_interrupterd_write_retry_text

Public Attributes

 oldalrm

Static Public Attributes

 io = pyio
 test_reentrant_write_buffered = None
 test_reentrant_write_text = None

Detailed Description

Definition at line 2872 of file test_io.py.


Member Function Documentation

def test.test_io.SignalsTest.alarm_interrupt (   self,
  sig,
  frame 
) [inherited]

Definition at line 2688 of file test_io.py.

02688 
02689     def alarm_interrupt(self, sig, frame):
02690         1/0

Here is the caller graph for this function:

def test.test_io.SignalsTest.check_interrupted_read_retry (   self,
  decode,
  fdopen_kwargs 
) [inherited]
Check that a buffered read, when it gets interrupted (either
returning a partial result or EINTR), properly invokes the signal
handler and retries if the latter returned successfully.

Definition at line 2776 of file test_io.py.

02776 
02777     def check_interrupted_read_retry(self, decode, **fdopen_kwargs):
02778         """Check that a buffered read, when it gets interrupted (either
02779         returning a partial result or EINTR), properly invokes the signal
02780         handler and retries if the latter returned successfully."""
02781         r, w = os.pipe()
02782         fdopen_kwargs["closefd"] = False
02783         def alarm_handler(sig, frame):
02784             os.write(w, b"bar")
02785         signal.signal(signal.SIGALRM, alarm_handler)
02786         try:
02787             rio = self.io.open(r, **fdopen_kwargs)
02788             os.write(w, b"foo")
02789             signal.alarm(1)
02790             # Expected behaviour:
02791             # - first raw read() returns partial b"foo"
02792             # - second raw read() returns EINTR
02793             # - third raw read() returns b"bar"
02794             self.assertEqual(decode(rio.read(6)), "foobar")
02795         finally:
02796             rio.close()
02797             os.close(w)
02798             os.close(r)

Here is the call graph for this function:

Here is the caller graph for this function:

def test.test_io.SignalsTest.check_interrupted_write (   self,
  item,
  bytes,
  fdopen_kwargs 
) [inherited]
Check that a partial write, when it gets interrupted, properly
invokes the signal handler, and bubbles up the exception raised
in the latter.

Definition at line 2694 of file test_io.py.

02694 
02695     def check_interrupted_write(self, item, bytes, **fdopen_kwargs):
02696         """Check that a partial write, when it gets interrupted, properly
02697         invokes the signal handler, and bubbles up the exception raised
02698         in the latter."""
02699         read_results = []
02700         def _read():
02701             s = os.read(r, 1)
02702             read_results.append(s)
02703         t = threading.Thread(target=_read)
02704         t.daemon = True
02705         r, w = os.pipe()
02706         fdopen_kwargs["closefd"] = False
02707         try:
02708             wio = self.io.open(w, **fdopen_kwargs)
02709             t.start()
02710             signal.alarm(1)
02711             # Fill the pipe enough that the write will be blocking.
02712             # It will be interrupted by the timer armed above.  Since the
02713             # other thread has read one byte, the low-level write will
02714             # return with a successful (partial) result rather than an EINTR.
02715             # The buffered IO layer must check for pending signal
02716             # handlers, which in this case will invoke alarm_interrupt().
02717             self.assertRaises(ZeroDivisionError,
02718                               wio.write, item * (1024 * 1024))
02719             t.join()
02720             # We got one byte, get another one and check that it isn't a
02721             # repeat of the first one.
02722             read_results.append(os.read(r, 1))
02723             self.assertEqual(read_results, [bytes[0:1], bytes[1:2]])
02724         finally:
02725             os.close(w)
02726             os.close(r)
02727             # This is deliberate. If we didn't close the file descriptor
02728             # before closing wio, wio would try to flush its internal
02729             # buffer, and block again.
02730             try:
02731                 wio.close()
02732             except IOError as e:
02733                 if e.errno != errno.EBADF:
02734                     raise

Here is the call graph for this function:

Here is the caller graph for this function:

def test.test_io.SignalsTest.check_interrupted_write_retry (   self,
  item,
  fdopen_kwargs 
) [inherited]
Check that a buffered write, when it gets interrupted (either
returning a partial result or EINTR), properly invokes the signal
handler and retries if the latter returned successfully.

Definition at line 2808 of file test_io.py.

02808 
02809     def check_interrupted_write_retry(self, item, **fdopen_kwargs):
02810         """Check that a buffered write, when it gets interrupted (either
02811         returning a partial result or EINTR), properly invokes the signal
02812         handler and retries if the latter returned successfully."""
02813         select = support.import_module("select")
02814         # A quantity that exceeds the buffer size of an anonymous pipe's
02815         # write end.
02816         N = 1024 * 1024
02817         r, w = os.pipe()
02818         fdopen_kwargs["closefd"] = False
02819         # We need a separate thread to read from the pipe and allow the
02820         # write() to finish.  This thread is started after the SIGALRM is
02821         # received (forcing a first EINTR in write()).
02822         read_results = []
02823         write_finished = False
02824         def _read():
02825             while not write_finished:
02826                 while r in select.select([r], [], [], 1.0)[0]:
02827                     s = os.read(r, 1024)
02828                     read_results.append(s)
02829         t = threading.Thread(target=_read)
02830         t.daemon = True
02831         def alarm1(sig, frame):
02832             signal.signal(signal.SIGALRM, alarm2)
02833             signal.alarm(1)
02834         def alarm2(sig, frame):
02835             t.start()
02836         signal.signal(signal.SIGALRM, alarm1)
02837         try:
02838             wio = self.io.open(w, **fdopen_kwargs)
02839             signal.alarm(1)
02840             # Expected behaviour:
02841             # - first raw write() is partial (because of the limited pipe buffer
02842             #   and the first alarm)
02843             # - second raw write() returns EINTR (because of the second alarm)
02844             # - subsequent write()s are successful (either partial or complete)
02845             self.assertEqual(N, wio.write(item * N))
02846             wio.flush()
02847             write_finished = True
02848             t.join()
02849             self.assertEqual(N, sum(len(x) for x in read_results))
02850         finally:
02851             write_finished = True
02852             os.close(w)
02853             os.close(r)
02854             # This is deliberate. If we didn't close the file descriptor
02855             # before closing wio, wio would try to flush its internal
02856             # buffer, and could block (in case of failure).
02857             try:
02858                 wio.close()
02859             except IOError as e:
02860                 if e.errno != errno.EBADF:
02861                     raise

Here is the call graph for this function:

Here is the caller graph for this function:

def test.test_io.SignalsTest.check_reentrant_write (   self,
  data,
  fdopen_kwargs 
) [inherited]

Definition at line 2744 of file test_io.py.

02744 
02745     def check_reentrant_write(self, data, **fdopen_kwargs):
02746         def on_alarm(*args):
02747             # Will be called reentrantly from the same thread
02748             wio.write(data)
02749             1/0
02750         signal.signal(signal.SIGALRM, on_alarm)
02751         r, w = os.pipe()
02752         wio = self.io.open(w, **fdopen_kwargs)
02753         try:
02754             signal.alarm(1)
02755             # Either the reentrant call to wio.write() fails with RuntimeError,
02756             # or the signal handler raises ZeroDivisionError.
02757             with self.assertRaises((ZeroDivisionError, RuntimeError)) as cm:
02758                 while 1:
02759                     for i in range(100):
02760                         wio.write(data)
02761                         wio.flush()
02762                     # Make sure the buffer doesn't fill up and block further writes
02763                     os.read(r, len(data) * 100)
02764             exc = cm.exception
02765             if isinstance(exc, RuntimeError):
02766                 self.assertTrue(str(exc).startswith("reentrant call"), str(exc))
02767         finally:
02768             wio.close()
02769             os.close(r)

Here is the call graph for this function:

Here is the caller graph for this function:

def test.test_io.SignalsTest.setUp (   self) [inherited]

Definition at line 2682 of file test_io.py.

02682 
02683     def setUp(self):
02684         self.oldalrm = signal.signal(signal.SIGALRM, self.alarm_interrupt)

Here is the caller graph for this function:

def test.test_io.SignalsTest.tearDown (   self) [inherited]

Definition at line 2685 of file test_io.py.

02685 
02686     def tearDown(self):
02687         signal.signal(signal.SIGALRM, self.oldalrm)

Here is the caller graph for this function:

Definition at line 2738 of file test_io.py.

02738 
02739     def test_interrupted_write_buffered(self):
02740         self.check_interrupted_write(b"xy", b"xy", mode="wb")

Here is the call graph for this function:

Definition at line 2741 of file test_io.py.

02741 
02742     def test_interrupted_write_text(self):
02743         self.check_interrupted_write("xy", b"xy", mode="w", encoding="ascii")

Here is the call graph for this function:

Definition at line 2735 of file test_io.py.

02735 
02736     def test_interrupted_write_unbuffered(self):
02737         self.check_interrupted_write(b"xy", b"xy", mode="wb", buffering=0)

Here is the call graph for this function:

Definition at line 2799 of file test_io.py.

02799 
02800     def test_interrupterd_read_retry_buffered(self):
02801         self.check_interrupted_read_retry(lambda x: x.decode('latin1'),
02802                                           mode="rb")

Here is the call graph for this function:

Definition at line 2803 of file test_io.py.

02803 
02804     def test_interrupterd_read_retry_text(self):
02805         self.check_interrupted_read_retry(lambda x: x,
02806                                           mode="r")

Here is the call graph for this function:

Definition at line 2862 of file test_io.py.

02862 
02863     def test_interrupterd_write_retry_buffered(self):
02864         self.check_interrupted_write_retry(b"x", mode="wb")

Here is the call graph for this function:

Definition at line 2865 of file test_io.py.

02865 
02866     def test_interrupterd_write_retry_text(self):
02867         self.check_interrupted_write_retry("x", mode="w", encoding="latin1")
02868 

Here is the call graph for this function:

Definition at line 2770 of file test_io.py.

02770 
02771     def test_reentrant_write_buffered(self):
02772         self.check_reentrant_write(b"xy", mode="wb")

Here is the call graph for this function:

Definition at line 2773 of file test_io.py.

02773 
02774     def test_reentrant_write_text(self):
02775         self.check_reentrant_write("xy", mode="w", encoding="ascii")

Here is the call graph for this function:


Member Data Documentation

Definition at line 2873 of file test_io.py.

Definition at line 2683 of file test_io.py.

Definition at line 2877 of file test_io.py.

Definition at line 2878 of file test_io.py.


The documentation for this class was generated from the following file: