micropython/streampair: Use RingIO and add async tests.

- Replace deprecated ringbuffer with RingIO name.
- Remove unused collections.deque import.
- Add test_select_poll_compatibility() to verify select.poll().
- Add test_streamreader_direct_usage() for asyncio.StreamReader.

The existing ioctl implementation correctly returns -1 for
unsupported operations, enabling select.poll() to work with pure
Python stream objects. Since asyncio.StreamReader is built on
select.poll(), StreamPair works seamlessly with async I/O.

These tests document that Python objects with proper ioctl()
implementation work with micropython polling and async infrastructure
without requiring C-level integration.

Signed-off-by: Andrew Leech <andrew@alelec.net>
Signed-off-by: Andrew Leech <andrew.leech@planetinnovation.com.au>

Author: pi-anl

micropython/streampair/streampair.py

@@ -1,7 +1,6 @@
 import io
 
-from collections import deque
-from micropython import ringbuffer, const
+from micropython import RingIO, const
 
 try:
     from typing import Union, Tuple
@@ -28,14 +27,14 @@ def streampair(buffer_size: Union[int, Tuple[int, int]]=256):
     except TypeError:
         size_a = size_b = buffer_size
 
-    a = ringbuffer(size_a)
-    b = ringbuffer(size_b)
+    a = RingIO(size_a)
+    b = RingIO(size_b)
     return StreamPair(a, b), StreamPair(b, a)
 
 
 class StreamPair(io.IOBase):
 
-    def __init__(self, own: ringbuffer, other: ringbuffer):
+    def __init__(self, own: RingIO, other: RingIO):
         self.own = own
         self.other = other
         super().__init__()

micropython/streampair/test_streampair.py

@@ -1,5 +1,6 @@
 import asyncio
 import unittest
+import select
 from streampair import streampair
 
 def async_test(f):
@@ -14,8 +15,8 @@ def wrapper(*args, **kwargs):
 
     return wrapper
 
-class StreamPairTestCase(unittest.TestCase):
 
+class StreamPairTestCase(unittest.TestCase):
     def test_streampair(self):
         a, b = streampair()
         assert a.write(b"foo") == 3
@@ -49,6 +50,59 @@ async def test_async_streampair(self):
         assert not b.any()
         assert not a.any()
 
+    def test_select_poll_compatibility(self):
+        """Test that streampair works with select.poll()"""
+        a, b = streampair()
+
+        # Register stream with poll
+        poller = select.poll()
+        poller.register(a, select.POLLIN)
+
+        # No data available initially
+        events = poller.poll(0)
+        assert len(events) == 0, f"Expected no events, got {events}"
+
+        # Write data to b, should be readable from a
+        b.write(b"test data")
+
+        # Should now poll as readable
+        events = poller.poll(0)
+        assert len(events) == 1, f"Expected 1 event, got {events}"
+        assert events[0][0] == a, "Event should be for stream a"
+        assert events[0][1] & select.POLLIN, "Should be readable"
+
+        # Read the data
+        data = a.read()
+        assert data == b"test data", f"Expected b'test data', got {data}"
+
+        # Should no longer poll as readable
+        events = poller.poll(0)
+        assert len(events) == 0, f"Expected no events after read, got {events}"
+
+        poller.unregister(a)
+
+    @async_test
+    async def test_streamreader_direct_usage(self):
+        """Test that streampair can be used directly with asyncio.StreamReader"""
+        a, b = streampair()
+
+        # Create StreamReader directly on the streampair object
+        reader = asyncio.StreamReader(a)
+
+        # Write data in background task
+        async def write_delayed():
+            await asyncio.sleep_ms(10)
+            b.write(b"async test\n")
+
+        task = asyncio.create_task(write_delayed())
+
+        # Should be able to read via StreamReader
+        data = await asyncio.wait_for(reader.readline(), 1.0)
+        assert data == b"async test\n", f"Expected b'async test\\n', got {data}"
+
+        # Wait for background task to complete
+        await task
+
 
 if __name__ == "__main__":
     unittest.main()