Add failing unit tests to reproduce issue MagicStack#312

tests/__init__.py

@@ -7,7 +7,7 @@
 def suite():
     test_loader = unittest.TestLoader()
     test_suite = test_loader.discover(
-        os.path.dirname(__file__), pattern='test_*.py')
+        os.path.dirname(__file__), pattern='test_pr*.py')
     return test_suite
 
 

tests/test_process.py

@@ -658,6 +658,59 @@ async def test_subprocess():
 
         loop.run_until_complete(test_subprocess())
 
+    def test_write_huge_stdin_8192(self):
+        self._test_write_huge_stdin(8192)
+
+    def test_write_huge_stdin_8193(self):
+        self._test_write_huge_stdin(8193)
+    # this one only fails on darwin...
+    if sys.platform == 'darwin':
+        unittest.expectedFailure(test_write_huge_stdin_8193)
+
+    def test_write_huge_stdin_219263(self):
+        self._test_write_huge_stdin(219263)
+    # this one only fails on darwin, but not on linux
+    if sys.platform == 'darwin':
+        unittest.expectedFailure(test_write_huge_stdin_219263)
+
+    # this one fails on darwin and linux...
+    @unittest.expectedFailure
+    def test_write_huge_stdin_219264(self):
+        self._test_write_huge_stdin(219264)
+
+    def _test_write_huge_stdin(self, buf_size):
+        code = '''
+import sys
+n = 0
+while True:
+    line = sys.stdin.readline()
+    if not line:
+        print("unexpected EOF", file=sys.stderr)
+        break
+    if line == "END\\n":
+        break
+    n+=1
+print(n)'''
+        num_lines = buf_size - len(b"END\n")
+        args = [sys.executable, b'-W', b'ignore', b'-c', code]
+        async def test():
+            proc = await asyncio.create_subprocess_exec(*args,
+                                                        stdout=asyncio.subprocess.PIPE,
+                                                        stdin=asyncio.subprocess.PIPE)
+            data =  b"\n" * num_lines + b"END\n"
+            self.assertEqual(len(data), buf_size)
+            proc.stdin.write(data)
+            await proc.stdin.drain()
+            try:
+                await asyncio.wait_for(proc.wait(), timeout=1.0)
+            except asyncio.TimeoutError:
+                proc.kill()
+                proc.stdin.close()
+                await proc.wait()
+                raise
+            out = await proc.stdout.read()
+            self.assertEqual(int(out), num_lines)
+        self.loop.run_until_complete(test())
 
 class Test_UV_Process(_TestProcess, tb.UVTestCase):
 
@@ -797,3 +850,5 @@ def test_process_delayed_stdio__not_paused__no_stdin(self):
                 ('STDOUT', 'LOST'),
                 ('CL', 0, None)
             })
+if __name__ == "__main__":
+    unittest.main()