我们从Python开源项目中，提取了以下18个代码示例，用于说明如何使用asyncio.SubprocessProtocol()。
def test_subprocess_shell_invalid_args(self): @asyncio.coroutine def connect(cmd=None, **kwds): if not cmd: cmd = 'pwd' yield from self.loop.subprocess_shell( asyncio.SubprocessProtocol, cmd, **kwds) with self.assertRaises(ValueError): self.loop.run_until_complete(connect(['ls', '-l'])) with self.assertRaises(ValueError): self.loop.run_until_complete(connect(universal_newlines=True)) with self.assertRaises(ValueError): self.loop.run_until_complete(connect(bufsize=4096)) with self.assertRaises(ValueError): self.loop.run_until_complete(connect(shell=False))
def test_empty(self): f = mock.Mock() p = asyncio.Protocol() self.assertIsNone(p.connection_made(f)) self.assertIsNone(p.connection_lost(f)) self.assertIsNone(p.data_received(f)) self.assertIsNone(p.eof_received()) dp = asyncio.DatagramProtocol() self.assertIsNone(dp.connection_made(f)) self.assertIsNone(dp.connection_lost(f)) self.assertIsNone(dp.error_received(f)) self.assertIsNone(dp.datagram_received(f, f)) sp = asyncio.SubprocessProtocol() self.assertIsNone(sp.connection_made(f)) self.assertIsNone(sp.connection_lost(f)) self.assertIsNone(sp.pipe_data_received(1, f)) self.assertIsNone(sp.pipe_connection_lost(1, f)) self.assertIsNone(sp.process_exited())
def test_cancel_post_init(self): @asyncio.coroutine def cancel_make_transport(): coro = self.loop.subprocess_exec(asyncio.SubprocessProtocol, *PROGRAM_BLOCKED) task = self.loop.create_task(coro) self.loop.call_soon(task.cancel) try: yield from task except asyncio.CancelledError: pass # ignore the log: # "Exception during subprocess creation, kill the subprocess" with test_utils.disable_logger(): self.loop.run_until_complete(cancel_make_transport()) test_utils.run_briefly(self.loop)
def test_subprocess_shell_invalid_args(self): # expected a string, not an int or a list self.assertRaises(TypeError, self.loop.run_until_complete, self.loop.subprocess_shell, asyncio.SubprocessProtocol, 123) self.assertRaises(TypeError, self.loop.run_until_complete, self.loop.subprocess_shell, asyncio.SubprocessProtocol, [sys.executable, '-c', 'pass']) # universal_newlines, shell, bufsize must not be set self.assertRaises(TypeError, self.loop.run_until_complete, self.loop.subprocess_shell, asyncio.SubprocessProtocol, 'exit 0', universal_newlines=True) self.assertRaises(TypeError, self.loop.run_until_complete, self.loop.subprocess_shell, asyncio.SubprocessProtocol, 'exit 0', shell=True) self.assertRaises(TypeError, self.loop.run_until_complete, self.loop.subprocess_shell, asyncio.SubprocessProtocol, 'exit 0', bufsize=4096)
def test_subprocess_exec_invalid_args(self): @asyncio.coroutine def connect(**kwds): yield from self.loop.subprocess_exec( asyncio.SubprocessProtocol, 'pwd', **kwds) with self.assertRaises(ValueError): self.loop.run_until_complete(connect(universal_newlines=True)) with self.assertRaises(ValueError): self.loop.run_until_complete(connect(bufsize=4096)) with self.assertRaises(ValueError): self.loop.run_until_complete(connect(shell=True))
def test_subprocess_exec_invalid_args(self): args = [sys.executable, '-c', 'pass'] # missing program parameter (empty args) self.assertRaises(TypeError, self.loop.run_until_complete, self.loop.subprocess_exec, asyncio.SubprocessProtocol) # expected multiple arguments, not a list self.assertRaises(TypeError, self.loop.run_until_complete, self.loop.subprocess_exec, asyncio.SubprocessProtocol, args) # program arguments must be strings, not int self.assertRaises(TypeError, self.loop.run_until_complete, self.loop.subprocess_exec, asyncio.SubprocessProtocol, sys.executable, 123) # universal_newlines, shell, bufsize must not be set self.assertRaises(TypeError, self.loop.run_until_complete, self.loop.subprocess_exec, asyncio.SubprocessProtocol, *args, universal_newlines=True) self.assertRaises(TypeError, self.loop.run_until_complete, self.loop.subprocess_exec, asyncio.SubprocessProtocol, *args, shell=True) self.assertRaises(TypeError, self.loop.run_until_complete, self.loop.subprocess_exec, asyncio.SubprocessProtocol, *args, bufsize=4096)
def test_close_kill_running(self): @asyncio.coroutine def kill_running(): create = self.loop.subprocess_exec(asyncio.SubprocessProtocol, *PROGRAM_BLOCKED) transport, protocol = yield from create kill_called = False def kill(): nonlocal kill_called kill_called = True orig_kill() proc = transport.get_extra_info('subprocess') orig_kill = proc.kill proc.kill = kill returncode = transport.get_returncode() transport.close() yield from transport._wait() return (returncode, kill_called) # Ignore "Close running child process: kill ..." log with test_utils.disable_logger(): returncode, killed = self.loop.run_until_complete(kill_running()) self.assertIsNone(returncode) # transport.close() must kill the process if it is still running self.assertTrue(killed) test_utils.run_briefly(self.loop)
def test_subprocess_exec_invalid_args(self): args = [sys.executable, '-c', 'pass'] # missing program parameter (empty args) self.assertRaises(TypeError, self.loop.run_until_complete, self.loop.subprocess_exec, asyncio.SubprocessProtocol) # exepected multiple arguments, not a list self.assertRaises(TypeError, self.loop.run_until_complete, self.loop.subprocess_exec, asyncio.SubprocessProtocol, args) # program arguments must be strings, not int self.assertRaises(TypeError, self.loop.run_until_complete, self.loop.subprocess_exec, asyncio.SubprocessProtocol, sys.executable, 123) # universal_newlines, shell, bufsize must not be set self.assertRaises(TypeError, self.loop.run_until_complete, self.loop.subprocess_exec, asyncio.SubprocessProtocol, *args, universal_newlines=True) self.assertRaises(TypeError, self.loop.run_until_complete, self.loop.subprocess_exec, asyncio.SubprocessProtocol, *args, shell=True) self.assertRaises(TypeError, self.loop.run_until_complete, self.loop.subprocess_exec, asyncio.SubprocessProtocol, *args, bufsize=4096)
def __init__(self, stdin=None, stdout=None, stderr=None): self.stdin = stdin self.stdout = stdout self.stderr = stderr self.complete = asyncio.Future() asyncio.SubprocessProtocol.__init__(self)
