(interp)
| 52 | |
| 53 | @contextlib.contextmanager |
| 54 | def _running(interp): |
| 55 | r, w = os.pipe() |
| 56 | def run(): |
| 57 | _interpreters.run_string(interp, dedent(f""" |
| 58 | # wait for "signal" |
| 59 | with open({r}, encoding="utf-8") as rpipe: |
| 60 | rpipe.read() |
| 61 | """)) |
| 62 | |
| 63 | t = threading.Thread(target=run) |
| 64 | t.start() |
| 65 | _wait_for_interp_to_run(interp) |
| 66 | |
| 67 | yield |
| 68 | |
| 69 | with open(w, 'w', encoding="utf-8") as spipe: |
| 70 | spipe.write('done') |
| 71 | t.join() |
| 72 | |
| 73 | |
| 74 | def clean_up_interpreters(): |
no test coverage detected
searching dependent graphs…