Fix Stream.readuntil with non-bytes buffer objects

PR python#16429 introduced support for an iterable of separators in
Stream.readuntil. Since bytes-like types are themselves iterable, this
can introduce ambiguities in deciding whether the argument is an
iterator of separators or a singleton separator. In python#16429, only 'bytes'
was considered a singleton, but this will break code that passes other
buffer object types.

The Python library docs don't indicate what separator types were
permitted in Python <=3.12, but comments in typeshed indicate that it
would work with types that implement the buffer protocol and provide a
len(). To keep those cases working the way they did before, I've changed
the detection logic to consider any instance of collections.abc.Buffer
as a singleton separator.

There may still be corner cases where this doesn't do what the user
wants e.g. a numpy array of byte strings will implement the buffer
protocol and hence be treated as a singleton; but at least those corner
cases should behave the same in 3.13 as they did in 3.12.

Relates to python#81322.

Author: bmerry

Lib/asyncio/streams.py

@@ -2,7 +2,7 @@
     'StreamReader', 'StreamWriter', 'StreamReaderProtocol',
     'open_connection', 'start_server')
 
-import collections
+import collections.abc
 import socket
 import sys
 import warnings
@@ -597,7 +597,7 @@ async def readuntil(self, separator=b'\n'):
         the shortest possible separator is considered to be the one that
         matched.
         """
-        if isinstance(separator, bytes):
+        if isinstance(separator, collections.abc.Buffer):
             separator = [separator]
         else:
             # Makes sure shortest matches wins, and supports arbitrary iterables

Lib/test/test_asyncio/test_streams.py

@@ -512,6 +512,13 @@ def test_readuntil_multi_separator_negative_offset(self):
         self.assertEqual(b'dataZA', data)
         self.assertEqual(b'aaa', stream._buffer)
 
+    def test_readuntil_bytearray(self):
+        stream = asyncio.StreamReader(loop=self.loop)
+        stream.feed_data(b'some data\r\n')
+        data = self.loop.run_until_complete(stream.readuntil(bytearray(b'\r\n')))
+        self.assertEqual(b'some data\r\n', data)
+        self.assertEqual(b'', stream._buffer)
+
     def test_readexactly_zero_or_less(self):
         # Read exact number of bytes (zero or less).
         stream = asyncio.StreamReader(loop=self.loop)