mirror of https://github.com/python/cpython
asyncio: Improve default writelines().
This commit is contained in:
parent
afbf827875
commit
f10345e102
|
@ -1,5 +1,9 @@
|
||||||
"""Abstract Transport class."""
|
"""Abstract Transport class."""
|
||||||
|
|
||||||
|
import sys
|
||||||
|
|
||||||
|
PY34 = sys.version_info >= (3, 4)
|
||||||
|
|
||||||
__all__ = ['ReadTransport', 'WriteTransport', 'Transport']
|
__all__ = ['ReadTransport', 'WriteTransport', 'Transport']
|
||||||
|
|
||||||
|
|
||||||
|
@ -85,11 +89,15 @@ class WriteTransport(BaseTransport):
|
||||||
def writelines(self, list_of_data):
|
def writelines(self, list_of_data):
|
||||||
"""Write a list (or any iterable) of data bytes to the transport.
|
"""Write a list (or any iterable) of data bytes to the transport.
|
||||||
|
|
||||||
The default implementation just calls write() for each item in
|
The default implementation concatenates the arguments and
|
||||||
the list/iterable.
|
calls write() on the result.
|
||||||
"""
|
"""
|
||||||
for data in list_of_data:
|
if not PY34:
|
||||||
self.write(data)
|
# In Python 3.3, bytes.join() doesn't handle memoryview.
|
||||||
|
list_of_data = (
|
||||||
|
bytes(data) if isinstance(data, memoryview) else data
|
||||||
|
for data in list_of_data)
|
||||||
|
self.write(b''.join(list_of_data))
|
||||||
|
|
||||||
def write_eof(self):
|
def write_eof(self):
|
||||||
"""Close the write end after flushing buffered data.
|
"""Close the write end after flushing buffered data.
|
||||||
|
|
|
@ -24,12 +24,18 @@ class TransportTests(unittest.TestCase):
|
||||||
transport = transports.Transport()
|
transport = transports.Transport()
|
||||||
transport.write = unittest.mock.Mock()
|
transport.write = unittest.mock.Mock()
|
||||||
|
|
||||||
transport.writelines(['line1', 'line2', 'line3'])
|
transport.writelines([b'line1',
|
||||||
self.assertEqual(3, transport.write.call_count)
|
bytearray(b'line2'),
|
||||||
|
memoryview(b'line3')])
|
||||||
|
self.assertEqual(1, transport.write.call_count)
|
||||||
|
transport.write.assert_called_with(b'line1line2line3')
|
||||||
|
|
||||||
def test_not_implemented(self):
|
def test_not_implemented(self):
|
||||||
transport = transports.Transport()
|
transport = transports.Transport()
|
||||||
|
|
||||||
|
self.assertRaises(NotImplementedError,
|
||||||
|
transport.set_write_buffer_limits)
|
||||||
|
self.assertRaises(NotImplementedError, transport.get_write_buffer_size)
|
||||||
self.assertRaises(NotImplementedError, transport.write, 'data')
|
self.assertRaises(NotImplementedError, transport.write, 'data')
|
||||||
self.assertRaises(NotImplementedError, transport.write_eof)
|
self.assertRaises(NotImplementedError, transport.write_eof)
|
||||||
self.assertRaises(NotImplementedError, transport.can_write_eof)
|
self.assertRaises(NotImplementedError, transport.can_write_eof)
|
||||||
|
|
Loading…
Reference in New Issue