(self)
| 1022 | self.assertEqual(b''.join(raw._write_stack), b'123\n45') |
| 1023 | |
| 1024 | def test_bufio_write_through(self): |
| 1025 | # Issue #21396: write_through=True doesn't force a flush() |
| 1026 | # on the underlying binary buffered object. |
| 1027 | flush_called, write_called = [], [] |
| 1028 | class BufferedWriter(self.BufferedWriter): |
| 1029 | def flush(self, *args, **kwargs): |
| 1030 | flush_called.append(True) |
| 1031 | return super().flush(*args, **kwargs) |
| 1032 | def write(self, *args, **kwargs): |
| 1033 | write_called.append(True) |
| 1034 | return super().write(*args, **kwargs) |
| 1035 | |
| 1036 | rawio = self.BytesIO() |
| 1037 | data = b"a" |
| 1038 | bufio = BufferedWriter(rawio, len(data)*2) |
| 1039 | textio = self.TextIOWrapper(bufio, encoding='ascii', |
| 1040 | write_through=True) |
| 1041 | # write to the buffered io but don't overflow the buffer |
| 1042 | text = data.decode('ascii') |
| 1043 | textio.write(text) |
| 1044 | |
| 1045 | # buffer.flush is not called with write_through=True |
| 1046 | self.assertFalse(flush_called) |
| 1047 | # buffer.write *is* called with write_through=True |
| 1048 | self.assertTrue(write_called) |
| 1049 | self.assertEqual(rawio.getvalue(), b"") # no flush |
| 1050 | |
| 1051 | write_called = [] # reset |
| 1052 | textio.write(text * 10) # total content is larger than bufio buffer |
| 1053 | self.assertTrue(write_called) |
| 1054 | self.assertEqual(rawio.getvalue(), data * 11) # all flushed |
| 1055 | |
| 1056 | def test_reconfigure_write_through(self): |
| 1057 | raw = self.MockRawIO([]) |
nothing calls this directly
no test coverage detected