add tests

Author: kumaraditya303

Doc/library/asyncio-eventloop.rst

@@ -1263,9 +1263,9 @@ Working with pipes
    *protocol_factory* must be a callable returning an
    :ref:`asyncio protocol <asyncio-protocol>` implementation.
 
-   *pipe* is a :term:`file-like object <file object>`.  Not every file-like
-   object is accepted; see :ref:`Supported pipe objects
-   <asyncio-pipe-objects>` below.
+   *pipe* is a :term:`file-like object <file object>`.  See
+   :ref:`Supported pipe objects <asyncio-pipe-objects>` for the objects
+   supported as *pipe*.
 
    Return pair ``(transport, protocol)``, where *transport* supports
    the :class:`ReadTransport` interface and *protocol* is an object
@@ -1282,9 +1282,9 @@ Working with pipes
    *protocol_factory* must be a callable returning an
    :ref:`asyncio protocol <asyncio-protocol>` implementation.
 
-   *pipe* is a :term:`file-like object <file object>`.  Not every file-like
-   object is accepted; see :ref:`Supported pipe objects
-   <asyncio-pipe-objects>` below.
+   *pipe* is a :term:`file-like object <file object>`.  See
+   :ref:`Supported pipe objects <asyncio-pipe-objects>` for the objects
+   supported as *pipe*.
 
    Return pair ``(transport, protocol)``, where *transport* supports
    :class:`WriteTransport` interface and *protocol* is an object
@@ -1297,13 +1297,10 @@ Working with pipes
 
 .. rubric:: Supported pipe objects
 
-Even though the *pipe* argument is a :term:`file-like object <file object>`,
-these methods only work with objects the operating system can poll for
+These methods only work with objects the operating system can poll for
 readiness or perform overlapped I/O on.  Regular files on disk are **not**
-supported on any platform, and neither are :data:`sys.stdin`,
-:data:`sys.stdout` and :data:`sys.stderr` when they have been redirected to
-or from a regular file.  There is no asynchronous file I/O in asyncio; use
-:meth:`loop.run_in_executor` to read and write regular files without
+supported on any platform.  There is no asynchronous file I/O in asyncio;
+use :meth:`loop.run_in_executor` to read and write regular files without
 blocking the event loop.
 
 On Unix, with :class:`SelectorEventLoop`, *pipe* must wrap one of the
@@ -1312,24 +1309,11 @@ following:
 * a pipe, such as an end of an :func:`os.pipe` pair or a FIFO created with
   :func:`os.mkfifo`;
 * a socket;
-* a character device, such as a terminal.  This is why :data:`sys.stdin`
-  works when the program is run interactively but fails when its input is
-  redirected from a file.
-
-Anything else, a regular file in particular, raises :exc:`ValueError`.
+* a character device, such as a terminal.
 
 On Windows, where only :class:`ProactorEventLoop` implements these methods,
 *pipe* must wrap a handle opened for overlapped I/O, since the handle has to
-be associated with an I/O completion port.  In practice this means a named
-pipe, such as the ones created by :func:`!asyncio.windows_utils.pipe` and
-used for the standard streams of a subprocess started by
-:meth:`loop.subprocess_exec`.  Console handles and regular file handles
-cannot be associated with a completion port, so :data:`sys.stdin` and files
-opened with :func:`open` do not work.  Unlike on Unix, this is not reported
-by the method itself: it returns successfully and the resulting
-:exc:`OSError` is later passed to the
-:meth:`event loop exception handler <loop.call_exception_handler>` when the
-transport first reads or writes.
+be associated with an I/O completion port.
 
 .. note::
 

Doc/library/asyncio-platforms.rst

@@ -20,9 +20,7 @@ All Platforms
   cannot be used to monitor file I/O.
 
 * :meth:`loop.connect_read_pipe` and :meth:`loop.connect_write_pipe`
-  cannot be used with regular files, nor with :data:`sys.stdin`,
-  :data:`sys.stdout` or :data:`sys.stderr` when these are redirected to or
-  from a regular file.  See :ref:`Supported pipe objects
+  cannot be used with regular files.  See :ref:`Supported pipe objects
   <asyncio-pipe-objects>` for the objects that are accepted on each
   platform.
 
@@ -71,14 +69,11 @@ All event loops on Windows do not support the following methods:
 
 * :meth:`loop.connect_read_pipe` and :meth:`loop.connect_write_pipe` only
   accept a handle opened for overlapped I/O, which in practice means a named
-  pipe such as those created by :func:`!asyncio.windows_utils.pipe`.
-  Console handles and regular file handles cannot be associated with an I/O
-  completion port, so :data:`sys.stdin`, :data:`sys.stdout`,
-  :data:`sys.stderr` and files opened with :func:`open` are not supported.
-  These methods do not fail immediately in that case: the resulting
-  :exc:`OSError` is passed to the
-  :meth:`event loop exception handler <loop.call_exception_handler>` when the
-  transport first reads or writes.
+  pipe such as those created by :func:`!asyncio.windows_utils.pipe`.  A
+  console handle cannot be associated with an I/O completion port, so
+  :data:`sys.stdin`, :data:`sys.stdout` and :data:`sys.stderr` are not
+  supported.  See :ref:`Supported pipe objects <asyncio-pipe-objects>` for
+  how such a handle is rejected, which differs between the two methods.
 
 The resolution of the monotonic clock on Windows is usually around 15.6
 milliseconds.  The best resolution is 0.5 milliseconds. The resolution depends on the

Lib/test/test_asyncio/test_events.py

@@ -33,6 +33,7 @@
 from multiprocessing.util import _cleanup_tests as multiprocessing_cleanup_tests
 from test.test_asyncio import utils as test_utils
 from test import support
+from test.support import os_helper
 from test.support import socket_helper
 from test.support import threading_helper
 from test.support import ALWAYS_EQ, LARGEST, SMALLEST
@@ -3134,5 +3135,100 @@ def test_get_loop(self):
             events.AbstractServer().get_loop()
 
 
+@unittest.skipIf(sys.platform == 'win32', 'Unix pipe transport semantics')
+class UnixPipeObjectSupportTests(unittest.TestCase):
+    def setUp(self):
+        super().setUp()
+        self.loop = asyncio.SelectorEventLoop()
+        self.addCleanup(self.loop.close)
+
+    def check_accepted(self, connect, pipeobj):
+        lost = self.loop.create_future()
+
+        class Proto(asyncio.Protocol):
+            def connection_lost(self, exc):
+                if not lost.done():
+                    lost.set_result(exc)
+
+        async def run():
+            transport, protocol = await connect(Proto, pipeobj)
+            self.assertIsInstance(protocol, Proto)
+            self.assertFalse(pipeobj.closed)
+            transport.close()
+            self.assertIsNone(await lost)
+
+        self.loop.run_until_complete(run())
+        self.assertTrue(pipeobj.closed)
+
+    def check_rejected(self, connect, pipeobj):
+        self.addCleanup(pipeobj.close)
+        with self.assertRaisesRegex(ValueError, 'Pipe transport is'):
+            self.loop.run_until_complete(connect(asyncio.Protocol, pipeobj))
+
+    def test_read_pipe(self):
+        rfd, wfd = os.pipe()
+        self.addCleanup(os.close, wfd)
+        self.check_accepted(self.loop.connect_read_pipe, open(rfd, 'rb', 0))
+
+    def test_write_pipe(self):
+        rfd, wfd = os.pipe()
+        self.addCleanup(os.close, rfd)
+        self.check_accepted(self.loop.connect_write_pipe, open(wfd, 'wb', 0))
+
+    def test_read_fifo(self):
+        path = os_helper.TESTFN
+        os.mkfifo(path)
+        self.addCleanup(os_helper.unlink, path)
+        rfd = os.open(path, os.O_RDONLY | os.O_NONBLOCK)
+        wfd = os.open(path, os.O_WRONLY)
+        self.addCleanup(os.close, wfd)
+        self.check_accepted(self.loop.connect_read_pipe, open(rfd, 'rb', 0))
+
+    def test_write_fifo(self):
+        path = os_helper.TESTFN
+        os.mkfifo(path)
+        self.addCleanup(os_helper.unlink, path)
+        rfd = os.open(path, os.O_RDONLY | os.O_NONBLOCK)
+        self.addCleanup(os.close, rfd)
+        wfd = os.open(path, os.O_WRONLY)
+        self.check_accepted(self.loop.connect_write_pipe, open(wfd, 'wb', 0))
+
+    def test_read_socket(self):
+        rsock, wsock = socket.socketpair()
+        self.addCleanup(wsock.close)
+        self.check_accepted(self.loop.connect_read_pipe,
+                            open(rsock.detach(), 'rb', 0))
+
+    def test_write_socket(self):
+        rsock, wsock = socket.socketpair()
+        self.addCleanup(rsock.close)
+        self.check_accepted(self.loop.connect_write_pipe,
+                            open(wsock.detach(), 'wb', 0))
+
+    @unittest.skipUnless(hasattr(os, 'openpty'), 'need os.openpty()')
+    def test_read_character_device(self):
+        master, slave = os.openpty()
+        self.addCleanup(os.close, slave)
+        self.check_accepted(self.loop.connect_read_pipe, open(master, 'rb', 0))
+
+    @unittest.skipUnless(hasattr(os, 'openpty'), 'need os.openpty()')
+    def test_write_character_device(self):
+        master, slave = os.openpty()
+        self.addCleanup(os.close, master)
+        self.check_accepted(self.loop.connect_write_pipe, open(slave, 'wb', 0))
+
+    def test_read_regular_file(self):
+        self.addCleanup(os_helper.unlink, os_helper.TESTFN)
+        with open(os_helper.TESTFN, 'wb') as f:
+            f.write(b'spam')
+        self.check_rejected(self.loop.connect_read_pipe,
+                            open(os_helper.TESTFN, 'rb', 0))
+
+    def test_write_regular_file(self):
+        self.addCleanup(os_helper.unlink, os_helper.TESTFN)
+        self.check_rejected(self.loop.connect_write_pipe,
+                            open(os_helper.TESTFN, 'wb', 0))
+
+
 if __name__ == '__main__':
     unittest.main()

Lib/test/test_asyncio/test_windows_events.py

@@ -15,6 +15,8 @@
 
 import asyncio
 from asyncio import windows_events
+from asyncio import windows_utils
+from test.support import os_helper
 from test.test_asyncio import utils as test_utils
 
 
@@ -324,5 +326,89 @@ def threadMain():
         thr.join()
 
 
+class ProactorPipeObjectSupportTests(unittest.TestCase):
+
+    def setUp(self):
+        super().setUp()
+        self.loop = asyncio.ProactorEventLoop()
+        self.addCleanup(self.loop.close)
+        self.errors = []
+        self.loop.set_exception_handler(
+            lambda loop, context: self.errors.append(context))
+
+    def check_read_rejected(self, pipe):
+        self.addCleanup(pipe.close)
+        lost = self.loop.create_future()
+
+        class Proto(asyncio.Protocol):
+            def connection_lost(self, exc):
+                if not lost.done():
+                    lost.set_result(exc)
+
+        async def run():
+            transport, _ = await self.loop.connect_read_pipe(Proto, pipe)
+            await lost
+            transport.close()
+
+        self.loop.run_until_complete(run())
+        self.assertTrue(self.errors, 'no error reached the exception handler')
+        self.assertIsInstance(self.errors[0]['exception'], OSError)
+
+    def check_write_rejected(self, pipe):
+        self.addCleanup(pipe.close)
+        with self.assertRaises(OSError):
+            self.loop.run_until_complete(
+                self.loop.connect_write_pipe(asyncio.BaseProtocol, pipe))
+
+    def test_overlapped_pipe(self):
+        rhandle, whandle = windows_utils.pipe(overlapped=(True, True))
+        rpipe = windows_utils.PipeHandle(rhandle)
+        wpipe = windows_utils.PipeHandle(whandle)
+
+        chunks = []
+        lost = self.loop.create_future()
+
+        class ReadProto(asyncio.Protocol):
+            def data_received(self, data):
+                chunks.append(data)
+
+            def connection_lost(self, exc):
+                if not lost.done():
+                    lost.set_result(exc)
+
+        async def run():
+            rtransport, _ = await self.loop.connect_read_pipe(ReadProto, rpipe)
+            wtransport, _ = await self.loop.connect_write_pipe(
+                asyncio.BaseProtocol, wpipe)
+            wtransport.write(b'spam')
+            wtransport.close()
+            await lost
+            rtransport.close()
+
+        self.loop.run_until_complete(run())
+        self.assertEqual(b''.join(chunks), b'spam')
+        self.assertFalse(self.errors)
+
+    def test_read_non_overlapped_pipe(self):
+        rhandle, whandle = windows_utils.pipe(overlapped=(False, False))
+        self.addCleanup(windows_utils.PipeHandle(whandle).close)
+        self.check_read_rejected(windows_utils.PipeHandle(rhandle))
+
+    def test_write_non_overlapped_pipe(self):
+        rhandle, whandle = windows_utils.pipe(overlapped=(False, False))
+        self.addCleanup(windows_utils.PipeHandle(rhandle).close)
+        self.check_write_rejected(windows_utils.PipeHandle(whandle))
+
+    def test_read_regular_file(self):
+        self.addCleanup(os_helper.unlink, os_helper.TESTFN)
+        with open(os_helper.TESTFN, 'wb') as f:
+            f.write(b'spam')
+        self.check_read_rejected(open(os_helper.TESTFN, 'rb', 0))
+
+    def test_write_regular_file(self):
+        self.addCleanup(os_helper.unlink, os_helper.TESTFN)
+        self.check_write_rejected(open(os_helper.TESTFN, 'wb', 0))
+
+
 if __name__ == '__main__':
     unittest.main()