| 339 | |
| 340 | @unittest.skipUnless(pty, "requires pty") |
| 341 | def test_asyncio_repl_is_ok(self): |
| 342 | m, s = pty.openpty() |
| 343 | cmd = [sys.executable, "-I", "-m", "asyncio"] |
| 344 | env = os.environ.copy() |
| 345 | proc = subprocess.Popen( |
| 346 | cmd, |
| 347 | stdin=s, |
| 348 | stdout=s, |
| 349 | stderr=s, |
| 350 | text=True, |
| 351 | close_fds=True, |
| 352 | env=env, |
| 353 | ) |
| 354 | os.close(s) |
| 355 | os.write(m, b"await asyncio.sleep(0)\n") |
| 356 | os.write(m, b"exit()\n") |
| 357 | output = [] |
| 358 | while select.select([m], [], [], SHORT_TIMEOUT)[0]: |
| 359 | try: |
| 360 | data = os.read(m, 1024).decode("utf-8") |
| 361 | if not data: |
| 362 | break |
| 363 | except OSError: |
| 364 | break |
| 365 | output.append(data) |
| 366 | os.close(m) |
| 367 | try: |
| 368 | exit_code = proc.wait(timeout=SHORT_TIMEOUT) |
| 369 | except subprocess.TimeoutExpired: |
| 370 | proc.kill() |
| 371 | exit_code = proc.wait() |
| 372 | |
| 373 | self.assertEqual(exit_code, 0, "".join(output)) |
| 374 | |
| 375 | |
| 376 | @support.force_not_colorized_test_class |