(self)
| 141 | |
| 142 | @gen_test |
| 143 | def test_subprocess(self): |
| 144 | if IOLoop.configured_class().__name__.endswith("LayeredTwistedIOLoop"): |
| 145 | # This test fails non-deterministically with LayeredTwistedIOLoop. |
| 146 | # (the read_until('\n') returns '\n' instead of 'hello\n') |
| 147 | # This probably indicates a problem with either TornadoReactor |
| 148 | # or TwistedIOLoop, but I haven't been able to track it down |
| 149 | # and for now this is just causing spurious travis-ci failures. |
| 150 | raise unittest.SkipTest( |
| 151 | "Subprocess tests not compatible with " "LayeredTwistedIOLoop" |
| 152 | ) |
| 153 | subproc = Subprocess( |
| 154 | [sys.executable, "-u", "-i"], |
| 155 | stdin=Subprocess.STREAM, |
| 156 | stdout=Subprocess.STREAM, |
| 157 | stderr=subprocess.STDOUT, |
| 158 | ) |
| 159 | self.addCleanup(lambda: self.term_and_wait(subproc)) |
| 160 | self.addCleanup(subproc.stdout.close) |
| 161 | self.addCleanup(subproc.stdin.close) |
| 162 | yield subproc.stdout.read_until(b">>> ") |
| 163 | subproc.stdin.write(b"print('hello')\n") |
| 164 | data = yield subproc.stdout.read_until(b"\n") |
| 165 | self.assertEqual(data, b"hello\n") |
| 166 | |
| 167 | yield subproc.stdout.read_until(b">>> ") |
| 168 | subproc.stdin.write(b"raise SystemExit\n") |
| 169 | data = yield subproc.stdout.read_until_close() |
| 170 | self.assertEqual(data, b"") |
| 171 | |
| 172 | @gen_test |
| 173 | def test_close_stdin(self): |
nothing calls this directly
no test coverage detected