diff --git a/Lib/asyncio/unix_events.py b/Lib/asyncio/unix_events.py
index 28fb4918645179..058d9a3a418c7c 100644
--- a/Lib/asyncio/unix_events.py
+++ b/Lib/asyncio/unix_events.py
@@ -522,6 +522,9 @@ def resume_reading(self):
         if self._loop.get_debug():
             logger.debug("%r resumes reading", self)
 
+    def is_reading(self):
+        return not self._paused and not self._closing
+
     def set_protocol(self, protocol):
         self._protocol = protocol
 
diff --git a/Lib/test/test_asyncio/test_unix_events.py b/Lib/test/test_asyncio/test_unix_events.py
index 5487b7afef8326..7037b2091af88d 100644
--- a/Lib/test/test_asyncio/test_unix_events.py
+++ b/Lib/test/test_asyncio/test_unix_events.py
@@ -740,6 +740,20 @@ def test_resume_reading(self, m_read):
         tr.resume_reading()
         self.loop.assert_reader(5, tr._read_ready)
 
+    @mock.patch('os.read')
+    def test_is_reading(self, m_read):
+        tr = self.read_pipe_transport()
+        tr._paused = False
+        tr._closing = False
+        self.assertTrue(tr.is_reading())
+        tr._paused = True
+        self.assertFalse(tr.is_reading())
+        tr._paused = False
+        tr._closing = True
+        self.assertFalse(tr.is_reading())
+        tr._closing = False
+        self.assertTrue(tr.is_reading())
+
     @mock.patch('os.read')
     def test_close(self, m_read):
         tr = self.read_pipe_transport()
diff --git a/Lib/test/test_asyncio/test_unix_pipes.py b/Lib/test/test_asyncio/test_unix_pipes.py
new file mode 100644
index 00000000000000..c3826d5c69c38b
--- /dev/null
+++ b/Lib/test/test_asyncio/test_unix_pipes.py
@@ -0,0 +1,103 @@
+"""
+Functional tests for Unix transport pipes, designed around opening two sockets
+on the localhost and sending information between
+"""
+
+import unittest
+import sys
+import os
+import tempfile
+import threading
+
+if sys.platform == 'win32':
+    raise unittest.SkipTest('UNIX only')
+
+import asyncio
+from asyncio import unix_events
+
+
+def tearDownModule():
+    asyncio.set_event_loop_policy(None)
+
+
+class UnixReadPipeTransportFuncTests(unittest.TestCase):
+    """
+    Verify that transports on Unix can facilitate reading and have access to
+    all state methods: verify using class internals
+    """
+
+    async def register_read_handle(self):
+        """
+        Wait for read_handle and then register it to the loop
+        """
+        self.transport, self.protocol = await self.loop.connect_read_pipe(
+            asyncio.BaseProtocol,
+            self.read_handle,
+        )
+
+    def setup_read_handle(self):
+        """
+        Open the read handle and record it in an attribute
+        """
+        self.read_handle = open(self.pipe, "r")
+
+    def setup_write_handle(self):
+        """
+        Open the write handle and record it in an attribute
+        """
+        self.write_handle = open(self.pipe, "w")
+
+    def setUp(self):
+        """
+        Create the UNIX pipe and register the read end to the loop, and connect
+        a write handle asynchronously
+        """
+        self.loop = asyncio.get_event_loop()
+        self.temp_dir = tempfile.TemporaryDirectory(suffix="async_unix_events")
+        self.pipe = os.path.join(self.temp_dir.name, "unix_pipe")
+        os.mkfifo(self.pipe)
+
+        # Set the threads to open the handles going
+        r_handle_thread = threading.Thread(target=self.setup_read_handle)
+        r_handle_thread.start()
+        w_handle_thread = threading.Thread(target=self.setup_write_handle)
+        w_handle_thread.start()
+
+        # Wait for pipe pair to connect
+        r_handle_thread.join()
+        w_handle_thread.join()
+
+        # Once pipe is connected, get the read transport
+        self.loop.run_until_complete(self.register_read_handle())
+
+        self.assertIsInstance(self.transport,
+                              unix_events._UnixReadPipeTransport)
+
+    def tearDown(self):
+        """
+        Destroy the read transport and the pipe
+        """
+        self.transport.close()
+        self.write_handle.close()
+        self.read_handle.close()
+        self.loop._run_once()
+        os.unlink(self.pipe)
+        self.temp_dir.cleanup()
+        self.loop.close()
+
+    def test_is_reading(self):
+        """
+        Verify that is_reading returns True unless transport is closed/closing
+        or paused
+        """
+        self.assertTrue(self.transport.is_reading())
+        self.transport.pause_reading()
+        self.assertFalse(self.transport.is_reading())
+        self.transport.resume_reading()
+        self.assertTrue(self.transport.is_reading())
+        self.transport.close()
+        self.assertFalse(self.transport.is_reading())
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/Misc/NEWS.d/next/Library/2019-11-04-17-20-43.bpo-38314.zWz6_P.rst b/Misc/NEWS.d/next/Library/2019-11-04-17-20-43.bpo-38314.zWz6_P.rst
new file mode 100644
index 00000000000000..9cfdfc7e75af14
--- /dev/null
+++ b/Misc/NEWS.d/next/Library/2019-11-04-17-20-43.bpo-38314.zWz6_P.rst
@@ -0,0 +1,2 @@
+The read pipe in :mod:`asyncio` UNIX transports now has access to the
+`is_reading()` API.