(self)
| 172 | self.assertEqual(stdout, b'some data') |
| 173 | |
| 174 | def test_communicate(self): |
| 175 | args = PROGRAM_CAT |
| 176 | |
| 177 | async def run(data): |
| 178 | proc = await asyncio.create_subprocess_exec( |
| 179 | *args, |
| 180 | stdin=subprocess.PIPE, |
| 181 | stdout=subprocess.PIPE, |
| 182 | ) |
| 183 | stdout, stderr = await proc.communicate(data) |
| 184 | return proc.returncode, stdout |
| 185 | |
| 186 | task = run(b'some data') |
| 187 | task = asyncio.wait_for(task, support.LONG_TIMEOUT) |
| 188 | exitcode, stdout = self.loop.run_until_complete(task) |
| 189 | self.assertEqual(exitcode, 0) |
| 190 | self.assertEqual(stdout, b'some data') |
| 191 | |
| 192 | def test_communicate_none_input(self): |
| 193 | args = PROGRAM_CAT |
nothing calls this directly
no test coverage detected