(self)
| 3682 | self.assertEqual(b''.join(raw._write_stack), b'123\n45') |
| 3683 | |
| 3684 | def test_bufio_write_through(self): |
| 3685 | # Issue #21396: write_through=True doesn't force a flush() |
| 3686 | # on the underlying binary buffered object. |
| 3687 | flush_called, write_called = [], [] |
| 3688 | class BufferedWriter(self.BufferedWriter): |
| 3689 | def flush(self, *args, **kwargs): |
| 3690 | flush_called.append(True) |
| 3691 | return super().flush(*args, **kwargs) |
| 3692 | def write(self, *args, **kwargs): |
| 3693 | write_called.append(True) |
| 3694 | return super().write(*args, **kwargs) |
| 3695 | |
| 3696 | rawio = self.BytesIO() |
| 3697 | data = b"a" |
| 3698 | bufio = BufferedWriter(rawio, len(data)*2) |
| 3699 | textio = self.TextIOWrapper(bufio, encoding='ascii', |
| 3700 | write_through=True) |
| 3701 | # write to the buffered io but don't overflow the buffer |
| 3702 | text = data.decode('ascii') |
| 3703 | textio.write(text) |
| 3704 | |
| 3705 | # buffer.flush is not called with write_through=True |
| 3706 | self.assertFalse(flush_called) |
| 3707 | # buffer.write *is* called with write_through=True |
| 3708 | self.assertTrue(write_called) |
| 3709 | self.assertEqual(rawio.getvalue(), b"") # no flush |
| 3710 | |
| 3711 | write_called = [] # reset |
| 3712 | textio.write(text * 10) # total content is larger than bufio buffer |
| 3713 | self.assertTrue(write_called) |
| 3714 | self.assertEqual(rawio.getvalue(), data * 11) # all flushed |
| 3715 | |
| 3716 | def test_reconfigure_write_through(self): |
| 3717 | raw = self.MockRawIO([]) |
nothing calls this directly
no test coverage detected