add an unit test for stdin/stdout

1 files changed, 40 insertions, 3 deletions

diff --git a/tests/test_subprocess.py b/tests/test_subprocess.py
index 87d0ee4..723afd6 100644
--- a/tests/test_subprocess.py
+++ b/tests/test_subprocess.py
@@ -1,5 +1,5 @@
+from asyncio import subprocess
 import asyncio
-import asyncio.subprocess
 import os
 import sys
 import unittest
@@ -14,16 +14,53 @@ class SubprocessTestCase(unittest.TestCase):
         # ensure that the event loop is passed explicitly in the code
         policy.set_event_loop(None)
 
-    def test_shell(self):
+    def tearDown(self):
+        policy = asyncio.get_event_loop_policy()
+        self.loop.close()
+        policy.set_child_watcher(None)
+
+    def test_call(self):
         args = [sys.executable, '-c', 'pass']
 
         @asyncio.coroutine
         def run():
-            return (yield from asyncio.subprocess.call(*args, loop=self.loop))
+            return (yield from subprocess.call(*args, loop=self.loop))
 
         exitcode = self.loop.run_until_complete(run())
         self.assertEqual(exitcode, 0)
 
+    def test_stdin_stdout(self):
+        code = '; '.join((
+            'import sys',
+            'data = sys.stdin.buffer.read()',
+            'sys.stdout.buffer.write(data)',
+        ))
+        args = [sys.executable, '-c', code]
+
+        @asyncio.coroutine
+        def run(data):
+            proc = yield from asyncio.create_subprocess_exec(
+                                          *args,
+                                          stdin=subprocess.PIPE,
+                                          stdout=subprocess.PIPE,
+                                          loop=self.loop)
+
+            # feed data
+            proc.stdin.write(data)
+            yield from proc.stdin.drain()
+            proc.stdin.close()
+
+            # get output and exitcode
+            data = yield from proc.stdout.read()
+            exitcode = yield from proc.wait()
+            return (exitcode, data)
+
+        task = run(b'some data')
+        task = asyncio.wait_for(task, 10.0, loop=self.loop)
+        exitcode, stdout = self.loop.run_until_complete(task)
+        self.assertEqual(exitcode, 0)
+        self.assertEqual(stdout, b'some data')
+
 
 if __name__ == '__main__':
     unittest.main()