()
| 105 | |
| 106 | |
| 107 | def test_spawn_stdout(): |
| 108 | # type: () -> None |
| 109 | assert ( |
| 110 | "Jane\n" |
| 111 | == SpawnedJob.stdout( |
| 112 | job=spawn_python_job( |
| 113 | args=["-c", "import sys; print(sys.stdin.read())"], |
| 114 | stdin=subprocess.PIPE, |
| 115 | stdout=subprocess.PIPE, |
| 116 | ), |
| 117 | result_func=lambda stdout: stdout.decode("utf-8"), |
| 118 | input=b"Jane", |
| 119 | ).await_result() |
| 120 | ) |
| 121 | |
| 122 | spawned_job = SpawnedJob.stdout(create_error_job(137), lambda output: 42) |
| 123 | with pytest.raises(Job.Error) as exec_info: |
| 124 | spawned_job.await_result() |
| 125 | assert 137 == exec_info.value.exitcode |
nothing calls this directly
no test coverage detected