Back to index

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

List of all members.

Public Member Functions

def test_constructor
def test_read
def test_read1
def test_readinto
def test_readlines
def test_buffering
def test_read_non_blocking
def test_read_past_eof
def test_read_all
def test_threads
def test_misbehaved_io
def test_no_extraneous_read
def test_detach
def test_fileno
def test_no_fileno
def test_invalid_args
def test_override_destructor
def test_context_manager
def test_error_through_destructor
def test_repr
def test_flush_error_on_close
def test_multi_close
def test_unseekable
def test_readonly_attributes

Static Public Attributes

string read_mode = "rb"

Detailed Description

Definition at line 758 of file test_io.py.


Member Function Documentation

Definition at line 823 of file test_io.py.

00823 
00824     def test_buffering(self):
00825         data = b"abcdefghi"
00826         dlen = len(data)
00827 
00828         tests = [
00829             [ 100, [ 3, 1, 4, 8 ], [ dlen, 0 ] ],
00830             [ 100, [ 3, 3, 3],     [ dlen ]    ],
00831             [   4, [ 1, 2, 4, 2 ], [ 4, 4, 1 ] ],
00832         ]
00833 
00834         for bufsize, buf_read_sizes, raw_read_sizes in tests:
00835             rawio = self.MockFileIO(data)
00836             bufio = self.tp(rawio, buffer_size=bufsize)
00837             pos = 0
00838             for nbytes in buf_read_sizes:
00839                 self.assertEqual(bufio.read(nbytes), data[pos:pos+nbytes])
00840                 pos += nbytes
00841             # this is mildly implementation-dependent
00842             self.assertEqual(rawio.read_history, raw_read_sizes)

Here is the call graph for this function:

Reimplemented in test.test_io.CBufferedRandomTest, test.test_io.BufferedRandomTest, and test.test_io.CBufferedReaderTest.

Definition at line 761 of file test_io.py.

00761 
00762     def test_constructor(self):
00763         rawio = self.MockRawIO([b"abc"])
00764         bufio = self.tp(rawio)
00765         bufio.__init__(rawio)
00766         bufio.__init__(rawio, buffer_size=1024)
00767         bufio.__init__(rawio, buffer_size=16)
00768         self.assertEqual(b"abc", bufio.read())
00769         self.assertRaises(ValueError, bufio.__init__, rawio, buffer_size=0)
00770         self.assertRaises(ValueError, bufio.__init__, rawio, buffer_size=-16)
00771         self.assertRaises(ValueError, bufio.__init__, rawio, buffer_size=-1)
00772         rawio = self.MockRawIO([b"abc"])
00773         bufio.__init__(rawio)
00774         self.assertEqual(b"abc", bufio.read())

Here is the call graph for this function:

Definition at line 692 of file test_io.py.

00692 
00693     def test_context_manager(self):
00694         # Test usability as a context manager
00695         rawio = self.MockRawIO()
00696         bufio = self.tp(rawio)
00697         def _with():
00698             with bufio:
00699                 pass
00700         _with()
00701         # bufio should now be closed, and using it a second time should raise
00702         # a ValueError.
00703         self.assertRaises(ValueError, _with)

Here is the call graph for this function:

Definition at line 640 of file test_io.py.

00640 
00641     def test_detach(self):
00642         raw = self.MockRawIO()
00643         buf = self.tp(raw)
00644         self.assertIs(buf.detach(), raw)
00645         self.assertRaises(ValueError, buf.detach)

Here is the call graph for this function:

Definition at line 704 of file test_io.py.

00704 
00705     def test_error_through_destructor(self):
00706         # Test that the exception state is not modified by a destructor,
00707         # even if close() fails.
00708         rawio = self.CloseFailureIO()
00709         def f():
00710             self.tp(rawio).xyzzy
00711         with support.captured_output("stderr") as s:
00712             self.assertRaises(AttributeError, f)
00713         s = s.getvalue().strip()
00714         if s:
00715             # The destructor *may* have printed an unraisable error, check it
00716             self.assertEqual(len(s.splitlines()), 1)
00717             self.assertTrue(s.startswith("Exception IOError: "), s)
00718             self.assertTrue(s.endswith(" ignored"), s)

Here is the call graph for this function:

Definition at line 646 of file test_io.py.

00646 
00647     def test_fileno(self):
00648         rawio = self.MockRawIO()
00649         bufio = self.tp(rawio)
00650 
00651         self.assertEqual(42, bufio.fileno())

Here is the call graph for this function:

Definition at line 729 of file test_io.py.

00729 
00730     def test_flush_error_on_close(self):
00731         raw = self.MockRawIO()
00732         def bad_flush():
00733             raise IOError()
00734         raw.flush = bad_flush
00735         b = self.tp(raw)
00736         self.assertRaises(IOError, b.close) # exception not swallowed

Here is the call graph for this function:

Definition at line 657 of file test_io.py.

00657 
00658     def test_invalid_args(self):
00659         rawio = self.MockRawIO()
00660         bufio = self.tp(rawio)
00661         # Invalid whence
00662         self.assertRaises(ValueError, bufio.seek, 0, -1)
00663         self.assertRaises(ValueError, bufio.seek, 0, 3)

Here is the call graph for this function:

Reimplemented in test.test_io.BufferedRandomTest.

Definition at line 914 of file test_io.py.

00914 
00915     def test_misbehaved_io(self):
00916         rawio = self.MisbehavedRawIO((b"abc", b"d", b"efg"))
00917         bufio = self.tp(rawio)
00918         self.assertRaises(IOError, bufio.seek, 0)
00919         self.assertRaises(IOError, bufio.tell)

Here is the call graph for this function:

Definition at line 737 of file test_io.py.

00737 
00738     def test_multi_close(self):
00739         raw = self.MockRawIO()
00740         b = self.tp(raw)
00741         b.close()
00742         b.close()
00743         b.close()
00744         self.assertRaises(ValueError, b.flush)

Here is the call graph for this function:

Definition at line 920 of file test_io.py.

00920 
00921     def test_no_extraneous_read(self):
00922         # Issue #9550; when the raw IO object has satisfied the read request,
00923         # we should not issue any additional reads, otherwise it may block
00924         # (e.g. socket).
00925         bufsize = 16
00926         for n in (2, bufsize - 1, bufsize, bufsize + 1, bufsize * 2):
00927             rawio = self.MockRawIO([b"x" * n])
00928             bufio = self.tp(rawio, bufsize)
00929             self.assertEqual(bufio.read(n), b"x" * n)
00930             # Simple case: one raw read is enough to satisfy the request.
00931             self.assertEqual(rawio._extraneous_reads, 0,
00932                              "failed for {}: {} != 0".format(n, rawio._extraneous_reads))
00933             # A more complex case where two raw reads are needed to satisfy
00934             # the request.
00935             rawio = self.MockRawIO([b"x" * (n - 1), b"x"])
00936             bufio = self.tp(rawio, bufsize)
00937             self.assertEqual(bufio.read(n), b"x" * n)
00938             self.assertEqual(rawio._extraneous_reads, 0,
00939                              "failed for {}: {} != 0".format(n, rawio._extraneous_reads))
00940 

Here is the call graph for this function:

Definition at line 652 of file test_io.py.

00652 
00653     def test_no_fileno(self):
00654         # XXX will we always have fileno() function? If so, kill
00655         # this test. Else, write it.
00656         pass

Definition at line 664 of file test_io.py.

00664 
00665     def test_override_destructor(self):
00666         tp = self.tp
00667         record = []
00668         class MyBufferedIO(tp):
00669             def __del__(self):
00670                 record.append(1)
00671                 try:
00672                     f = super().__del__
00673                 except AttributeError:
00674                     pass
00675                 else:
00676                     f()
00677             def close(self):
00678                 record.append(2)
00679                 super().close()
00680             def flush(self):
00681                 record.append(3)
00682                 super().flush()
00683         rawio = self.MockRawIO()
00684         bufio = MyBufferedIO(rawio)
00685         writable = bufio.writable()
00686         del bufio
00687         support.gc_collect()
00688         if writable:
00689             self.assertEqual(record, [1, 2, 3])
00690         else:
00691             self.assertEqual(record, [1, 2])

Here is the call graph for this function:

Definition at line 775 of file test_io.py.

00775 
00776     def test_read(self):
00777         for arg in (None, 7):
00778             rawio = self.MockRawIO((b"abc", b"d", b"efg"))
00779             bufio = self.tp(rawio)
00780             self.assertEqual(b"abcdefg", bufio.read(arg))
00781         # Invalid args
00782         self.assertRaises(ValueError, bufio.read, -2)

Here is the call graph for this function:

Definition at line 783 of file test_io.py.

00783 
00784     def test_read1(self):
00785         rawio = self.MockRawIO((b"abc", b"d", b"efg"))
00786         bufio = self.tp(rawio)
00787         self.assertEqual(b"a", bufio.read(1))
00788         self.assertEqual(b"b", bufio.read1(1))
00789         self.assertEqual(rawio._reads, 1)
00790         self.assertEqual(b"c", bufio.read1(100))
00791         self.assertEqual(rawio._reads, 1)
00792         self.assertEqual(b"d", bufio.read1(100))
00793         self.assertEqual(rawio._reads, 2)
00794         self.assertEqual(b"efg", bufio.read1(100))
00795         self.assertEqual(rawio._reads, 3)
00796         self.assertEqual(b"", bufio.read1(100))
00797         self.assertEqual(rawio._reads, 4)
00798         # Invalid args
00799         self.assertRaises(ValueError, bufio.read1, -1)

Here is the call graph for this function:

Definition at line 864 of file test_io.py.

00864 
00865     def test_read_all(self):
00866         rawio = self.MockRawIO((b"abc", b"d", b"efg"))
00867         bufio = self.tp(rawio)
00868 
00869         self.assertEqual(b"abcdefg", bufio.read())

Here is the call graph for this function:

Definition at line 843 of file test_io.py.

00843 
00844     def test_read_non_blocking(self):
00845         # Inject some None's in there to simulate EWOULDBLOCK
00846         rawio = self.MockRawIO((b"abc", b"d", None, b"efg", None, None, None))
00847         bufio = self.tp(rawio)
00848         self.assertEqual(b"abcd", bufio.read(6))
00849         self.assertEqual(b"e", bufio.read(1))
00850         self.assertEqual(b"fg", bufio.read())
00851         self.assertEqual(b"", bufio.peek(1))
00852         self.assertIsNone(bufio.read())
00853         self.assertEqual(b"", bufio.read())
00854 
00855         rawio = self.MockRawIO((b"a", None, None))
00856         self.assertEqual(b"a", rawio.readall())
00857         self.assertIsNone(rawio.readall())

Here is the call graph for this function:

Definition at line 858 of file test_io.py.

00858 
00859     def test_read_past_eof(self):
00860         rawio = self.MockRawIO((b"abc", b"d", b"efg"))
00861         bufio = self.tp(rawio)
00862 
00863         self.assertEqual(b"abcdefg", bufio.read(9000))

Here is the call graph for this function:

Definition at line 800 of file test_io.py.

00800 
00801     def test_readinto(self):
00802         rawio = self.MockRawIO((b"abc", b"d", b"efg"))
00803         bufio = self.tp(rawio)
00804         b = bytearray(2)
00805         self.assertEqual(bufio.readinto(b), 2)
00806         self.assertEqual(b, b"ab")
00807         self.assertEqual(bufio.readinto(b), 2)
00808         self.assertEqual(b, b"cd")
00809         self.assertEqual(bufio.readinto(b), 2)
00810         self.assertEqual(b, b"ef")
00811         self.assertEqual(bufio.readinto(b), 1)
00812         self.assertEqual(b, b"gf")
00813         self.assertEqual(bufio.readinto(b), 0)
00814         self.assertEqual(b, b"gf")

Here is the call graph for this function:

Definition at line 815 of file test_io.py.

00815 
00816     def test_readlines(self):
00817         def bufio():
00818             rawio = self.MockRawIO((b"abc\n", b"d\n", b"ef"))
00819             return self.tp(rawio)
00820         self.assertEqual(bufio().readlines(), [b"abc\n", b"d\n", b"ef"])
00821         self.assertEqual(bufio().readlines(5), [b"abc\n", b"d\n"])
00822         self.assertEqual(bufio().readlines(None), [b"abc\n", b"d\n", b"ef"])

Here is the call graph for this function:

Definition at line 750 of file test_io.py.

00750 
00751     def test_readonly_attributes(self):
00752         raw = self.MockRawIO()
00753         buf = self.tp(raw)
00754         x = self.MockRawIO()
00755         with self.assertRaises(AttributeError):
00756             buf.raw = x
00757 

Here is the call graph for this function:

def test.test_io.CommonBufferedTests.test_repr (   self) [inherited]

Definition at line 719 of file test_io.py.

00719 
00720     def test_repr(self):
00721         raw = self.MockRawIO()
00722         b = self.tp(raw)
00723         clsname = "%s.%s" % (self.tp.__module__, self.tp.__name__)
00724         self.assertEqual(repr(b), "<%s>" % clsname)
00725         raw.name = "dummy"
00726         self.assertEqual(repr(b), "<%s name='dummy'>" % clsname)
00727         raw.name = b"dummy"
00728         self.assertEqual(repr(b), "<%s name=b'dummy'>" % clsname)

Here is the call graph for this function:

Reimplemented in test.test_io.BufferedRandomTest.

Definition at line 872 of file test_io.py.

00872 
00873     def test_threads(self):
00874         try:
00875             # Write out many bytes with exactly the same number of 0's,
00876             # 1's... 255's. This will help us check that concurrent reading
00877             # doesn't duplicate or forget contents.
00878             N = 1000
00879             l = list(range(256)) * N
00880             random.shuffle(l)
00881             s = bytes(bytearray(l))
00882             with self.open(support.TESTFN, "wb") as f:
00883                 f.write(s)
00884             with self.open(support.TESTFN, self.read_mode, buffering=0) as raw:
00885                 bufio = self.tp(raw, 8)
00886                 errors = []
00887                 results = []
00888                 def f():
00889                     try:
00890                         # Intra-buffer read then buffer-flushing read
00891                         for n in cycle([1, 19]):
00892                             s = bufio.read(n)
00893                             if not s:
00894                                 break
00895                             # list.append() is atomic
00896                             results.append(s)
00897                     except Exception as e:
00898                         errors.append(e)
00899                         raise
00900                 threads = [threading.Thread(target=f) for x in range(20)]
00901                 for t in threads:
00902                     t.start()
00903                 time.sleep(0.02) # yield
00904                 for t in threads:
00905                     t.join()
00906                 self.assertFalse(errors,
00907                     "the following exceptions were caught: %r" % errors)
00908                 s = b''.join(results)
00909                 for i in range(256):
00910                     c = bytes(bytearray([i]))
00911                     self.assertEqual(s.count(c), N)
00912         finally:
00913             support.unlink(support.TESTFN)

Here is the call graph for this function:

Definition at line 745 of file test_io.py.

00745 
00746     def test_unseekable(self):
00747         bufio = self.tp(self.MockUnseekableIO(b"A" * 10))
00748         self.assertRaises(self.UnsupportedOperation, bufio.tell)
00749         self.assertRaises(self.UnsupportedOperation, bufio.seek, 0)

Here is the call graph for this function:


Member Data Documentation

Reimplemented in test.test_io.BufferedRandomTest.

Definition at line 759 of file test_io.py.


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