Back to index

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

List of all members.

Public Member Functions

def test_constructor
def test_detach_flush
def test_write
def test_write_overflow
def check_writes
def test_writes
def test_writes_and_flushes
def test_writes_and_seeks
def test_writes_and_truncates
def test_write_non_blocking
def test_write_and_rewind
def test_flush
def test_destructor
def test_truncate
def test_threads
def test_misbehaved_io
def test_max_buffer_size_deprecation
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

 tp = pyio.BufferedWriter
string write_mode = "wb"

Detailed Description

Definition at line 1242 of file test_io.py.


Member Function Documentation

def test.test_io.BufferedWriterTest.check_writes (   self,
  intermediate_func 
) [inherited]

Definition at line 1031 of file test_io.py.

01031 
01032     def check_writes(self, intermediate_func):
01033         # Lots of writes, test the flushed output is as expected.
01034         contents = bytes(range(256)) * 1000
01035         n = 0
01036         writer = self.MockRawIO()
01037         bufio = self.tp(writer, 13)
01038         # Generator of write sizes: repeat each N 15 times then proceed to N+1
01039         def gen_sizes():
01040             for size in count(1):
01041                 for i in range(15):
01042                     yield size
01043         sizes = gen_sizes()
01044         while n < len(contents):
01045             size = min(next(sizes), len(contents) - n)
01046             self.assertEqual(bufio.write(contents[n:n+size]), size)
01047             intermediate_func(bufio)
01048             n += size
01049         bufio.flush()
01050         self.assertEqual(contents, b"".join(writer._write_stack))

Here is the call graph for this function:

Here is the caller graph for this function:

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

Definition at line 989 of file test_io.py.

00989 
00990     def test_constructor(self):
00991         rawio = self.MockRawIO()
00992         bufio = self.tp(rawio)
00993         bufio.__init__(rawio)
00994         bufio.__init__(rawio, buffer_size=1024)
00995         bufio.__init__(rawio, buffer_size=16)
00996         self.assertEqual(3, bufio.write(b"abc"))
00997         bufio.flush()
00998         self.assertRaises(ValueError, bufio.__init__, rawio, buffer_size=0)
00999         self.assertRaises(ValueError, bufio.__init__, rawio, buffer_size=-16)
01000         self.assertRaises(ValueError, bufio.__init__, rawio, buffer_size=-1)
01001         bufio.__init__(rawio)
01002         self.assertEqual(3, bufio.write(b"ghi"))
01003         bufio.flush()
01004         self.assertEqual(b"".join(rawio._write_stack), b"abcghi")

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 1121 of file test_io.py.

01121 
01122     def test_destructor(self):
01123         writer = self.MockRawIO()
01124         bufio = self.tp(writer, 8)
01125         bufio.write(b"abc")
01126         del bufio
01127         support.gc_collect()
01128         self.assertEqual(b"abc", writer._write_stack[0])

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 1005 of file test_io.py.

01005 
01006     def test_detach_flush(self):
01007         raw = self.MockRawIO()
01008         buf = self.tp(raw)
01009         buf.write(b"howdy!")
01010         self.assertFalse(raw._write_stack)
01011         buf.detach()
01012         self.assertEqual(raw._write_stack, [b"howdy!"])

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:

def test.test_io.BufferedWriterTest.test_flush (   self) [inherited]

Definition at line 1114 of file test_io.py.

01114 
01115     def test_flush(self):
01116         writer = self.MockRawIO()
01117         bufio = self.tp(writer, 8)
01118         bufio.write(b"abc")
01119         bufio.flush()
01120         self.assertEqual(b"abc", writer._write_stack[0])

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:

Definition at line 1197 of file test_io.py.

01197 
01198     def test_max_buffer_size_deprecation(self):
01199         with support.check_warnings(("max_buffer_size is deprecated",
01200                                      DeprecationWarning)):
01201             self.tp(self.MockRawIO(), 8, 12)
01202 

Reimplemented in test.test_io.BufferedRandomTest.

Definition at line 1190 of file test_io.py.

01190 
01191     def test_misbehaved_io(self):
01192         rawio = self.MisbehavedRawIO()
01193         bufio = self.tp(rawio, 5)
01194         self.assertRaises(IOError, bufio.seek, 0)
01195         self.assertRaises(IOError, bufio.tell)
01196         self.assertRaises(IOError, bufio.write, b"abcdef")

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 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 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 1141 of file test_io.py.

01141 
01142     def test_threads(self):
01143         try:
01144             # Write out many bytes from many threads and test they were
01145             # all flushed.
01146             N = 1000
01147             contents = bytes(range(256)) * N
01148             sizes = cycle([1, 19])
01149             n = 0
01150             queue = deque()
01151             while n < len(contents):
01152                 size = next(sizes)
01153                 queue.append(contents[n:n+size])
01154                 n += size
01155             del contents
01156             # We use a real file object because it allows us to
01157             # exercise situations where the GIL is released before
01158             # writing the buffer to the raw streams. This is in addition
01159             # to concurrency issues due to switching threads in the middle
01160             # of Python code.
01161             with self.open(support.TESTFN, self.write_mode, buffering=0) as raw:
01162                 bufio = self.tp(raw, 8)
01163                 errors = []
01164                 def f():
01165                     try:
01166                         while True:
01167                             try:
01168                                 s = queue.popleft()
01169                             except IndexError:
01170                                 return
01171                             bufio.write(s)
01172                     except Exception as e:
01173                         errors.append(e)
01174                         raise
01175                 threads = [threading.Thread(target=f) for x in range(20)]
01176                 for t in threads:
01177                     t.start()
01178                 time.sleep(0.02) # yield
01179                 for t in threads:
01180                     t.join()
01181                 self.assertFalse(errors,
01182                     "the following exceptions were caught: %r" % errors)
01183                 bufio.close()
01184             with self.open(support.TESTFN, "rb") as f:
01185                 s = f.read()
01186             for i in range(256):
01187                 self.assertEqual(s.count(bytes([i])), N)
01188         finally:
01189             support.unlink(support.TESTFN)

Here is the call graph for this function:

Definition at line 1129 of file test_io.py.

01129 
01130     def test_truncate(self):
01131         # Truncate implicitly flushes the buffer.
01132         with self.open(support.TESTFN, self.write_mode, buffering=0) as raw:
01133             bufio = self.tp(raw, 8)
01134             bufio.write(b"abcdef")
01135             self.assertEqual(bufio.truncate(3), 3)
01136             self.assertEqual(bufio.tell(), 6)
01137         with self.open(support.TESTFN, "rb", buffering=0) as f:
01138             self.assertEqual(f.read(), b"abc")

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:

def test.test_io.BufferedWriterTest.test_write (   self) [inherited]

Definition at line 1013 of file test_io.py.

01013 
01014     def test_write(self):
01015         # Write to the buffered IO but don't overflow the buffer.
01016         writer = self.MockRawIO()
01017         bufio = self.tp(writer, 8)
01018         bufio.write(b"abc")
01019         self.assertFalse(writer._write_stack)

Here is the call graph for this function:

Definition at line 1101 of file test_io.py.

01101 
01102     def test_write_and_rewind(self):
01103         raw = io.BytesIO()
01104         bufio = self.tp(raw, 4)
01105         self.assertEqual(bufio.write(b"abcdef"), 6)
01106         self.assertEqual(bufio.tell(), 6)
01107         bufio.seek(0, 0)
01108         self.assertEqual(bufio.write(b"XY"), 2)
01109         bufio.seek(6, 0)
01110         self.assertEqual(raw.getvalue(), b"XYcdef")
01111         self.assertEqual(bufio.write(b"123456"), 6)
01112         bufio.flush()
01113         self.assertEqual(raw.getvalue(), b"XYcdef123456")

Here is the call graph for this function:

Definition at line 1074 of file test_io.py.

01074 
01075     def test_write_non_blocking(self):
01076         raw = self.MockNonBlockWriterIO()
01077         bufio = self.tp(raw, 8)
01078 
01079         self.assertEqual(bufio.write(b"abcd"), 4)
01080         self.assertEqual(bufio.write(b"efghi"), 5)
01081         # 1 byte will be written, the rest will be buffered
01082         raw.block_on(b"k")
01083         self.assertEqual(bufio.write(b"jklmn"), 5)
01084 
01085         # 8 bytes will be written, 8 will be buffered and the rest will be lost
01086         raw.block_on(b"0")
01087         try:
01088             bufio.write(b"opqrwxyz0123456789")
01089         except self.BlockingIOError as e:
01090             written = e.characters_written
01091         else:
01092             self.fail("BlockingIOError should have been raised")
01093         self.assertEqual(written, 16)
01094         self.assertEqual(raw.pop_written(),
01095             b"abcdefghijklmnopqrwxyz")
01096 
01097         self.assertEqual(bufio.write(b"ABCDEFGHI"), 9)
01098         s = raw.pop_written()
01099         # Previously buffered bytes were flushed
01100         self.assertTrue(s.startswith(b"01234567A"), s)

Here is the call graph for this function:

Definition at line 1020 of file test_io.py.

01020 
01021     def test_write_overflow(self):
01022         writer = self.MockRawIO()
01023         bufio = self.tp(writer, 8)
01024         contents = b"abcdefghijklmnop"
01025         for n in range(0, len(contents), 3):
01026             bufio.write(contents[n:n+3])
01027         flushed = b"".join(writer._write_stack)
01028         # At least (total - 8) bytes were implicitly flushed, perhaps more
01029         # depending on the implementation.
01030         self.assertTrue(flushed.startswith(contents[:-8]), flushed)

Here is the call graph for this function:

def test.test_io.BufferedWriterTest.test_writes (   self) [inherited]

Definition at line 1051 of file test_io.py.

01051 
01052     def test_writes(self):
01053         self.check_writes(lambda bufio: None)

Here is the call graph for this function:

Definition at line 1054 of file test_io.py.

01054 
01055     def test_writes_and_flushes(self):
01056         self.check_writes(lambda bufio: bufio.flush())

Here is the call graph for this function:

Definition at line 1057 of file test_io.py.

01057 
01058     def test_writes_and_seeks(self):
01059         def _seekabs(bufio):
01060             pos = bufio.tell()
01061             bufio.seek(pos + 1, 0)
01062             bufio.seek(pos - 1, 0)
01063             bufio.seek(pos, 0)
01064         self.check_writes(_seekabs)
01065         def _seekrel(bufio):
01066             pos = bufio.seek(0, 1)
01067             bufio.seek(+1, 1)
01068             bufio.seek(-1, 1)
01069             bufio.seek(pos, 0)
01070         self.check_writes(_seekrel)

Here is the call graph for this function:

Definition at line 1071 of file test_io.py.

01071 
01072     def test_writes_and_truncates(self):
01073         self.check_writes(lambda bufio: bufio.truncate(bufio.tell()))

Here is the call graph for this function:


Member Data Documentation

test.test_io.PyBufferedWriterTest.tp = pyio.BufferedWriter [static]

Definition at line 1243 of file test_io.py.

Reimplemented in test.test_io.BufferedRandomTest.

Definition at line 987 of file test_io.py.


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