cros_subprocess.py 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404
  1. # Copyright (c) 2012 The Chromium OS Authors.
  2. # Use of this source code is governed by a BSD-style license that can be
  3. # found in the LICENSE file.
  4. #
  5. # Copyright (c) 2003-2005 by Peter Astrand <astrand@lysator.liu.se>
  6. # Licensed to PSF under a Contributor Agreement.
  7. # See http://www.python.org/2.4/license for licensing details.
  8. """Subprocess execution
  9. This module holds a subclass of subprocess.Popen with our own required
  10. features, mainly that we get access to the subprocess output while it
  11. is running rather than just at the end. This makes it easier to show
  12. progress information and filter output in real time.
  13. """
  14. import errno
  15. import os
  16. import pty
  17. import select
  18. import subprocess
  19. import sys
  20. import unittest
  21. # Import these here so the caller does not need to import subprocess also.
  22. PIPE = subprocess.PIPE
  23. STDOUT = subprocess.STDOUT
  24. PIPE_PTY = -3 # Pipe output through a pty
  25. stay_alive = True
  26. class Popen(subprocess.Popen):
  27. """Like subprocess.Popen with ptys and incremental output
  28. This class deals with running a child process and filtering its output on
  29. both stdout and stderr while it is running. We do this so we can monitor
  30. progress, and possibly relay the output to the user if requested.
  31. The class is similar to subprocess.Popen, the equivalent is something like:
  32. Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
  33. But this class has many fewer features, and two enhancement:
  34. 1. Rather than getting the output data only at the end, this class sends it
  35. to a provided operation as it arrives.
  36. 2. We use pseudo terminals so that the child will hopefully flush its output
  37. to us as soon as it is produced, rather than waiting for the end of a
  38. line.
  39. Use CommunicateFilter() to handle output from the subprocess.
  40. """
  41. def __init__(self, args, stdin=None, stdout=PIPE_PTY, stderr=PIPE_PTY,
  42. shell=False, cwd=None, env=None, **kwargs):
  43. """Cut-down constructor
  44. Args:
  45. args: Program and arguments for subprocess to execute.
  46. stdin: See subprocess.Popen()
  47. stdout: See subprocess.Popen(), except that we support the sentinel
  48. value of cros_subprocess.PIPE_PTY.
  49. stderr: See subprocess.Popen(), except that we support the sentinel
  50. value of cros_subprocess.PIPE_PTY.
  51. shell: See subprocess.Popen()
  52. cwd: Working directory to change to for subprocess, or None if none.
  53. env: Environment to use for this subprocess, or None to inherit parent.
  54. kwargs: No other arguments are supported at the moment. Passing other
  55. arguments will cause a ValueError to be raised.
  56. """
  57. stdout_pty = None
  58. stderr_pty = None
  59. if stdout == PIPE_PTY:
  60. stdout_pty = pty.openpty()
  61. stdout = os.fdopen(stdout_pty[1])
  62. if stderr == PIPE_PTY:
  63. stderr_pty = pty.openpty()
  64. stderr = os.fdopen(stderr_pty[1])
  65. super(Popen, self).__init__(args, stdin=stdin,
  66. stdout=stdout, stderr=stderr, shell=shell, cwd=cwd, env=env,
  67. **kwargs)
  68. # If we're on a PTY, we passed the slave half of the PTY to the subprocess.
  69. # We want to use the master half on our end from now on. Setting this here
  70. # does make some assumptions about the implementation of subprocess, but
  71. # those assumptions are pretty minor.
  72. # Note that if stderr is STDOUT, then self.stderr will be set to None by
  73. # this constructor.
  74. if stdout_pty is not None:
  75. self.stdout = os.fdopen(stdout_pty[0])
  76. if stderr_pty is not None:
  77. self.stderr = os.fdopen(stderr_pty[0])
  78. # Insist that unit tests exist for other arguments we don't support.
  79. if kwargs:
  80. raise ValueError("Unit tests do not test extra args - please add tests")
  81. def ConvertData(self, data):
  82. """Convert stdout/stderr data to the correct format for output
  83. Args:
  84. data: Data to convert, or None for ''
  85. Returns:
  86. Converted data, as bytes
  87. """
  88. if data is None:
  89. return b''
  90. return data
  91. def CommunicateFilter(self, output):
  92. """Interact with process: Read data from stdout and stderr.
  93. This method runs until end-of-file is reached, then waits for the
  94. subprocess to terminate.
  95. The output function is sent all output from the subprocess and must be
  96. defined like this:
  97. def Output([self,] stream, data)
  98. Args:
  99. stream: the stream the output was received on, which will be
  100. sys.stdout or sys.stderr.
  101. data: a string containing the data
  102. Note: The data read is buffered in memory, so do not use this
  103. method if the data size is large or unlimited.
  104. Args:
  105. output: Function to call with each fragment of output.
  106. Returns:
  107. A tuple (stdout, stderr, combined) which is the data received on
  108. stdout, stderr and the combined data (interleaved stdout and stderr).
  109. Note that the interleaved output will only be sensible if you have
  110. set both stdout and stderr to PIPE or PIPE_PTY. Even then it depends on
  111. the timing of the output in the subprocess. If a subprocess flips
  112. between stdout and stderr quickly in succession, by the time we come to
  113. read the output from each we may see several lines in each, and will read
  114. all the stdout lines, then all the stderr lines. So the interleaving
  115. may not be correct. In this case you might want to pass
  116. stderr=cros_subprocess.STDOUT to the constructor.
  117. This feature is still useful for subprocesses where stderr is
  118. rarely used and indicates an error.
  119. Note also that if you set stderr to STDOUT, then stderr will be empty
  120. and the combined output will just be the same as stdout.
  121. """
  122. read_set = []
  123. write_set = []
  124. stdout = None # Return
  125. stderr = None # Return
  126. if self.stdin:
  127. # Flush stdio buffer. This might block, if the user has
  128. # been writing to .stdin in an uncontrolled fashion.
  129. self.stdin.flush()
  130. if input:
  131. write_set.append(self.stdin)
  132. else:
  133. self.stdin.close()
  134. if self.stdout:
  135. read_set.append(self.stdout)
  136. stdout = bytearray()
  137. if self.stderr and self.stderr != self.stdout:
  138. read_set.append(self.stderr)
  139. stderr = bytearray()
  140. combined = bytearray()
  141. input_offset = 0
  142. while read_set or write_set:
  143. try:
  144. rlist, wlist, _ = select.select(read_set, write_set, [], 0.2)
  145. except select.error as e:
  146. if e.args[0] == errno.EINTR:
  147. continue
  148. raise
  149. if not stay_alive:
  150. self.terminate()
  151. if self.stdin in wlist:
  152. # When select has indicated that the file is writable,
  153. # we can write up to PIPE_BUF bytes without risk
  154. # blocking. POSIX defines PIPE_BUF >= 512
  155. chunk = input[input_offset : input_offset + 512]
  156. bytes_written = os.write(self.stdin.fileno(), chunk)
  157. input_offset += bytes_written
  158. if input_offset >= len(input):
  159. self.stdin.close()
  160. write_set.remove(self.stdin)
  161. if self.stdout in rlist:
  162. data = b''
  163. # We will get an error on read if the pty is closed
  164. try:
  165. data = os.read(self.stdout.fileno(), 1024)
  166. except OSError:
  167. pass
  168. if not len(data):
  169. self.stdout.close()
  170. read_set.remove(self.stdout)
  171. else:
  172. stdout += data
  173. combined += data
  174. if output:
  175. output(sys.stdout, data)
  176. if self.stderr in rlist:
  177. data = b''
  178. # We will get an error on read if the pty is closed
  179. try:
  180. data = os.read(self.stderr.fileno(), 1024)
  181. except OSError:
  182. pass
  183. if not len(data):
  184. self.stderr.close()
  185. read_set.remove(self.stderr)
  186. else:
  187. stderr += data
  188. combined += data
  189. if output:
  190. output(sys.stderr, data)
  191. # All data exchanged. Translate lists into strings.
  192. stdout = self.ConvertData(stdout)
  193. stderr = self.ConvertData(stderr)
  194. combined = self.ConvertData(combined)
  195. # Translate newlines, if requested. We cannot let the file
  196. # object do the translation: It is based on stdio, which is
  197. # impossible to combine with select (unless forcing no
  198. # buffering).
  199. if self.universal_newlines and hasattr(file, 'newlines'):
  200. if stdout:
  201. stdout = self._translate_newlines(stdout)
  202. if stderr:
  203. stderr = self._translate_newlines(stderr)
  204. self.wait()
  205. return (stdout, stderr, combined)
  206. # Just being a unittest.TestCase gives us 14 public methods. Unless we
  207. # disable this, we can only have 6 tests in a TestCase. That's not enough.
  208. #
  209. # pylint: disable=R0904
  210. class TestSubprocess(unittest.TestCase):
  211. """Our simple unit test for this module"""
  212. class MyOperation:
  213. """Provides a operation that we can pass to Popen"""
  214. def __init__(self, input_to_send=None):
  215. """Constructor to set up the operation and possible input.
  216. Args:
  217. input_to_send: a text string to send when we first get input. We will
  218. add \r\n to the string.
  219. """
  220. self.stdout_data = ''
  221. self.stderr_data = ''
  222. self.combined_data = ''
  223. self.stdin_pipe = None
  224. self._input_to_send = input_to_send
  225. if input_to_send:
  226. pipe = os.pipe()
  227. self.stdin_read_pipe = pipe[0]
  228. self._stdin_write_pipe = os.fdopen(pipe[1], 'w')
  229. def Output(self, stream, data):
  230. """Output handler for Popen. Stores the data for later comparison"""
  231. if stream == sys.stdout:
  232. self.stdout_data += data
  233. if stream == sys.stderr:
  234. self.stderr_data += data
  235. self.combined_data += data
  236. # Output the input string if we have one.
  237. if self._input_to_send:
  238. self._stdin_write_pipe.write(self._input_to_send + '\r\n')
  239. self._stdin_write_pipe.flush()
  240. def _BasicCheck(self, plist, oper):
  241. """Basic checks that the output looks sane."""
  242. self.assertEqual(plist[0], oper.stdout_data)
  243. self.assertEqual(plist[1], oper.stderr_data)
  244. self.assertEqual(plist[2], oper.combined_data)
  245. # The total length of stdout and stderr should equal the combined length
  246. self.assertEqual(len(plist[0]) + len(plist[1]), len(plist[2]))
  247. def test_simple(self):
  248. """Simple redirection: Get process list"""
  249. oper = TestSubprocess.MyOperation()
  250. plist = Popen(['ps']).CommunicateFilter(oper.Output)
  251. self._BasicCheck(plist, oper)
  252. def test_stderr(self):
  253. """Check stdout and stderr"""
  254. oper = TestSubprocess.MyOperation()
  255. cmd = 'echo fred >/dev/stderr && false || echo bad'
  256. plist = Popen([cmd], shell=True).CommunicateFilter(oper.Output)
  257. self._BasicCheck(plist, oper)
  258. self.assertEqual(plist [0], 'bad\r\n')
  259. self.assertEqual(plist [1], 'fred\r\n')
  260. def test_shell(self):
  261. """Check with and without shell works"""
  262. oper = TestSubprocess.MyOperation()
  263. cmd = 'echo test >/dev/stderr'
  264. self.assertRaises(OSError, Popen, [cmd], shell=False)
  265. plist = Popen([cmd], shell=True).CommunicateFilter(oper.Output)
  266. self._BasicCheck(plist, oper)
  267. self.assertEqual(len(plist [0]), 0)
  268. self.assertEqual(plist [1], 'test\r\n')
  269. def test_list_args(self):
  270. """Check with and without shell works using list arguments"""
  271. oper = TestSubprocess.MyOperation()
  272. cmd = ['echo', 'test', '>/dev/stderr']
  273. plist = Popen(cmd, shell=False).CommunicateFilter(oper.Output)
  274. self._BasicCheck(plist, oper)
  275. self.assertEqual(plist [0], ' '.join(cmd[1:]) + '\r\n')
  276. self.assertEqual(len(plist [1]), 0)
  277. oper = TestSubprocess.MyOperation()
  278. # this should be interpreted as 'echo' with the other args dropped
  279. cmd = ['echo', 'test', '>/dev/stderr']
  280. plist = Popen(cmd, shell=True).CommunicateFilter(oper.Output)
  281. self._BasicCheck(plist, oper)
  282. self.assertEqual(plist [0], '\r\n')
  283. def test_cwd(self):
  284. """Check we can change directory"""
  285. for shell in (False, True):
  286. oper = TestSubprocess.MyOperation()
  287. plist = Popen('pwd', shell=shell, cwd='/tmp').CommunicateFilter(oper.Output)
  288. self._BasicCheck(plist, oper)
  289. self.assertEqual(plist [0], '/tmp\r\n')
  290. def test_env(self):
  291. """Check we can change environment"""
  292. for add in (False, True):
  293. oper = TestSubprocess.MyOperation()
  294. env = os.environ
  295. if add:
  296. env ['FRED'] = 'fred'
  297. cmd = 'echo $FRED'
  298. plist = Popen(cmd, shell=True, env=env).CommunicateFilter(oper.Output)
  299. self._BasicCheck(plist, oper)
  300. self.assertEqual(plist [0], add and 'fred\r\n' or '\r\n')
  301. def test_extra_args(self):
  302. """Check we can't add extra arguments"""
  303. self.assertRaises(ValueError, Popen, 'true', close_fds=False)
  304. def test_basic_input(self):
  305. """Check that incremental input works
  306. We set up a subprocess which will prompt for name. When we see this prompt
  307. we send the name as input to the process. It should then print the name
  308. properly to stdout.
  309. """
  310. oper = TestSubprocess.MyOperation('Flash')
  311. prompt = 'What is your name?: '
  312. cmd = 'echo -n "%s"; read name; echo Hello $name' % prompt
  313. plist = Popen([cmd], stdin=oper.stdin_read_pipe,
  314. shell=True).CommunicateFilter(oper.Output)
  315. self._BasicCheck(plist, oper)
  316. self.assertEqual(len(plist [1]), 0)
  317. self.assertEqual(plist [0], prompt + 'Hello Flash\r\r\n')
  318. def test_isatty(self):
  319. """Check that ptys appear as terminals to the subprocess"""
  320. oper = TestSubprocess.MyOperation()
  321. cmd = ('if [ -t %d ]; then echo "terminal %d" >&%d; '
  322. 'else echo "not %d" >&%d; fi;')
  323. both_cmds = ''
  324. for fd in (1, 2):
  325. both_cmds += cmd % (fd, fd, fd, fd, fd)
  326. plist = Popen(both_cmds, shell=True).CommunicateFilter(oper.Output)
  327. self._BasicCheck(plist, oper)
  328. self.assertEqual(plist [0], 'terminal 1\r\n')
  329. self.assertEqual(plist [1], 'terminal 2\r\n')
  330. # Now try with PIPE and make sure it is not a terminal
  331. oper = TestSubprocess.MyOperation()
  332. plist = Popen(both_cmds, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
  333. shell=True).CommunicateFilter(oper.Output)
  334. self._BasicCheck(plist, oper)
  335. self.assertEqual(plist [0], 'not 1\n')
  336. self.assertEqual(plist [1], 'not 2\n')
  337. if __name__ == '__main__':
  338. unittest.main()