(self, code: str)
| 124 | # byte, so we don't have to worry about atomicity of the shared |
| 125 | # stdout stream) and then exits. |
| 126 | def run_subproc(self, code: str) -> Tuple[str, str]: |
| 127 | try: |
| 128 | result = subprocess.run( |
| 129 | [sys.executable, "-Werror::DeprecationWarning"], |
| 130 | capture_output=True, |
| 131 | input=code, |
| 132 | encoding="utf8", |
| 133 | check=True, |
| 134 | ) |
| 135 | except subprocess.CalledProcessError as e: |
| 136 | raise RuntimeError( |
| 137 | f"Process returned {e.returncode} stdout={e.stdout} stderr={e.stderr}" |
| 138 | ) from e |
| 139 | return result.stdout, result.stderr |
| 140 | |
| 141 | def test_listen_single(self): |
| 142 | # As a sanity check, run the single-process version through this test |
no test coverage detected