Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/pexpect/spawnbase.py: 18%
233 statements
« prev ^ index » next coverage.py v7.4.4, created at 2024-04-20 06:09 +0000
1from io import StringIO, BytesIO
2import codecs
3import os
4import sys
5import re
6import errno
7from .exceptions import ExceptionPexpect, EOF, TIMEOUT
8from .expect import Expecter, searcher_string, searcher_re
10PY3 = (sys.version_info[0] >= 3)
11text_type = str if PY3 else unicode
13class _NullCoder(object):
14 """Pass bytes through unchanged."""
15 @staticmethod
16 def encode(b, final=False):
17 return b
19 @staticmethod
20 def decode(b, final=False):
21 return b
23class SpawnBase(object):
24 """A base class providing the backwards-compatible spawn API for Pexpect.
26 This should not be instantiated directly: use :class:`pexpect.spawn` or
27 :class:`pexpect.fdpexpect.fdspawn`.
28 """
29 encoding = None
30 pid = None
31 flag_eof = False
33 def __init__(self, timeout=30, maxread=2000, searchwindowsize=None,
34 logfile=None, encoding=None, codec_errors='strict'):
35 self.stdin = sys.stdin
36 self.stdout = sys.stdout
37 self.stderr = sys.stderr
39 self.searcher = None
40 self.ignorecase = False
41 self.before = None
42 self.after = None
43 self.match = None
44 self.match_index = None
45 self.terminated = True
46 self.exitstatus = None
47 self.signalstatus = None
48 # status returned by os.waitpid
49 self.status = None
50 # the child file descriptor is initially closed
51 self.child_fd = -1
52 self.timeout = timeout
53 self.delimiter = EOF
54 self.logfile = logfile
55 # input from child (read_nonblocking)
56 self.logfile_read = None
57 # output to send (send, sendline)
58 self.logfile_send = None
59 # max bytes to read at one time into buffer
60 self.maxread = maxread
61 # Data before searchwindowsize point is preserved, but not searched.
62 self.searchwindowsize = searchwindowsize
63 # Delay used before sending data to child. Time in seconds.
64 # Set this to None to skip the time.sleep() call completely.
65 self.delaybeforesend = 0.05
66 # Used by close() to give kernel time to update process status.
67 # Time in seconds.
68 self.delayafterclose = 0.1
69 # Used by terminate() to give kernel time to update process status.
70 # Time in seconds.
71 self.delayafterterminate = 0.1
72 # Delay in seconds to sleep after each call to read_nonblocking().
73 # Set this to None to skip the time.sleep() call completely: that
74 # would restore the behavior from pexpect-2.0 (for performance
75 # reasons or because you don't want to release Python's global
76 # interpreter lock).
77 self.delayafterread = 0.0001
78 self.softspace = False
79 self.name = '<' + repr(self) + '>'
80 self.closed = True
82 # Unicode interface
83 self.encoding = encoding
84 self.codec_errors = codec_errors
85 if encoding is None:
86 # bytes mode (accepts some unicode for backwards compatibility)
87 self._encoder = self._decoder = _NullCoder()
88 self.string_type = bytes
89 self.buffer_type = BytesIO
90 self.crlf = b'\r\n'
91 if PY3:
92 self.allowed_string_types = (bytes, str)
93 self.linesep = os.linesep.encode('ascii')
94 def write_to_stdout(b):
95 try:
96 return sys.stdout.buffer.write(b)
97 except AttributeError:
98 # If stdout has been replaced, it may not have .buffer
99 return sys.stdout.write(b.decode('ascii', 'replace'))
100 self.write_to_stdout = write_to_stdout
101 else:
102 self.allowed_string_types = (basestring,) # analysis:ignore
103 self.linesep = os.linesep
104 self.write_to_stdout = sys.stdout.write
105 else:
106 # unicode mode
107 self._encoder = codecs.getincrementalencoder(encoding)(codec_errors)
108 self._decoder = codecs.getincrementaldecoder(encoding)(codec_errors)
109 self.string_type = text_type
110 self.buffer_type = StringIO
111 self.crlf = u'\r\n'
112 self.allowed_string_types = (text_type, )
113 if PY3:
114 self.linesep = os.linesep
115 else:
116 self.linesep = os.linesep.decode('ascii')
117 # This can handle unicode in both Python 2 and 3
118 self.write_to_stdout = sys.stdout.write
119 # storage for async transport
120 self.async_pw_transport = None
121 # This is the read buffer. See maxread.
122 self._buffer = self.buffer_type()
123 # The buffer may be trimmed for efficiency reasons. This is the
124 # untrimmed buffer, used to create the before attribute.
125 self._before = self.buffer_type()
127 def _log(self, s, direction):
128 if self.logfile is not None:
129 self.logfile.write(s)
130 self.logfile.flush()
131 second_log = self.logfile_send if (direction=='send') else self.logfile_read
132 if second_log is not None:
133 second_log.write(s)
134 second_log.flush()
136 # For backwards compatibility, in bytes mode (when encoding is None)
137 # unicode is accepted for send and expect. Unicode mode is strictly unicode
138 # only.
139 def _coerce_expect_string(self, s):
140 if self.encoding is None and not isinstance(s, bytes):
141 return s.encode('ascii')
142 return s
144 # In bytes mode, regex patterns should also be of bytes type
145 def _coerce_expect_re(self, r):
146 p = r.pattern
147 if self.encoding is None and not isinstance(p, bytes):
148 return re.compile(p.encode('utf-8'))
149 # And vice-versa
150 elif self.encoding is not None and isinstance(p, bytes):
151 return re.compile(p.decode('utf-8'))
152 return r
154 def _coerce_send_string(self, s):
155 if self.encoding is None and not isinstance(s, bytes):
156 return s.encode('utf-8')
157 return s
159 def _get_buffer(self):
160 return self._buffer.getvalue()
162 def _set_buffer(self, value):
163 self._buffer = self.buffer_type()
164 self._buffer.write(value)
166 # This property is provided for backwards compatibility (self.buffer used
167 # to be a string/bytes object)
168 buffer = property(_get_buffer, _set_buffer)
170 def read_nonblocking(self, size=1, timeout=None):
171 """This reads data from the file descriptor.
173 This is a simple implementation suitable for a regular file. Subclasses using ptys or pipes should override it.
175 The timeout parameter is ignored.
176 """
178 try:
179 s = os.read(self.child_fd, size)
180 except OSError as err:
181 if err.args[0] == errno.EIO:
182 # Linux-style EOF
183 self.flag_eof = True
184 raise EOF('End Of File (EOF). Exception style platform.')
185 raise
186 if s == b'':
187 # BSD-style EOF
188 self.flag_eof = True
189 raise EOF('End Of File (EOF). Empty string style platform.')
191 s = self._decoder.decode(s, final=False)
192 self._log(s, 'read')
193 return s
195 def _pattern_type_err(self, pattern):
196 raise TypeError('got {badtype} ({badobj!r}) as pattern, must be one'
197 ' of: {goodtypes}, pexpect.EOF, pexpect.TIMEOUT'\
198 .format(badtype=type(pattern),
199 badobj=pattern,
200 goodtypes=', '.join([str(ast)\
201 for ast in self.allowed_string_types])
202 )
203 )
205 def compile_pattern_list(self, patterns):
206 '''This compiles a pattern-string or a list of pattern-strings.
207 Patterns must be a StringType, EOF, TIMEOUT, SRE_Pattern, or a list of
208 those. Patterns may also be None which results in an empty list (you
209 might do this if waiting for an EOF or TIMEOUT condition without
210 expecting any pattern).
212 This is used by expect() when calling expect_list(). Thus expect() is
213 nothing more than::
215 cpl = self.compile_pattern_list(pl)
216 return self.expect_list(cpl, timeout)
218 If you are using expect() within a loop it may be more
219 efficient to compile the patterns first and then call expect_list().
220 This avoid calls in a loop to compile_pattern_list()::
222 cpl = self.compile_pattern_list(my_pattern)
223 while some_condition:
224 ...
225 i = self.expect_list(cpl, timeout)
226 ...
227 '''
229 if patterns is None:
230 return []
231 if not isinstance(patterns, list):
232 patterns = [patterns]
234 # Allow dot to match \n
235 compile_flags = re.DOTALL
236 if self.ignorecase:
237 compile_flags = compile_flags | re.IGNORECASE
238 compiled_pattern_list = []
239 for idx, p in enumerate(patterns):
240 if isinstance(p, self.allowed_string_types):
241 p = self._coerce_expect_string(p)
242 compiled_pattern_list.append(re.compile(p, compile_flags))
243 elif p is EOF:
244 compiled_pattern_list.append(EOF)
245 elif p is TIMEOUT:
246 compiled_pattern_list.append(TIMEOUT)
247 elif isinstance(p, type(re.compile(''))):
248 p = self._coerce_expect_re(p)
249 compiled_pattern_list.append(p)
250 else:
251 self._pattern_type_err(p)
252 return compiled_pattern_list
254 def expect(self, pattern, timeout=-1, searchwindowsize=-1, async_=False, **kw):
255 '''This seeks through the stream until a pattern is matched. The
256 pattern is overloaded and may take several types. The pattern can be a
257 StringType, EOF, a compiled re, or a list of any of those types.
258 Strings will be compiled to re types. This returns the index into the
259 pattern list. If the pattern was not a list this returns index 0 on a
260 successful match. This may raise exceptions for EOF or TIMEOUT. To
261 avoid the EOF or TIMEOUT exceptions add EOF or TIMEOUT to the pattern
262 list. That will cause expect to match an EOF or TIMEOUT condition
263 instead of raising an exception.
265 If you pass a list of patterns and more than one matches, the first
266 match in the stream is chosen. If more than one pattern matches at that
267 point, the leftmost in the pattern list is chosen. For example::
269 # the input is 'foobar'
270 index = p.expect(['bar', 'foo', 'foobar'])
271 # returns 1('foo') even though 'foobar' is a "better" match
273 Please note, however, that buffering can affect this behavior, since
274 input arrives in unpredictable chunks. For example::
276 # the input is 'foobar'
277 index = p.expect(['foobar', 'foo'])
278 # returns 0('foobar') if all input is available at once,
279 # but returns 1('foo') if parts of the final 'bar' arrive late
281 When a match is found for the given pattern, the class instance
282 attribute *match* becomes an re.MatchObject result. Should an EOF
283 or TIMEOUT pattern match, then the match attribute will be an instance
284 of that exception class. The pairing before and after class
285 instance attributes are views of the data preceding and following
286 the matching pattern. On general exception, class attribute
287 *before* is all data received up to the exception, while *match* and
288 *after* attributes are value None.
290 When the keyword argument timeout is -1 (default), then TIMEOUT will
291 raise after the default value specified by the class timeout
292 attribute. When None, TIMEOUT will not be raised and may block
293 indefinitely until match.
295 When the keyword argument searchwindowsize is -1 (default), then the
296 value specified by the class maxread attribute is used.
298 A list entry may be EOF or TIMEOUT instead of a string. This will
299 catch these exceptions and return the index of the list entry instead
300 of raising the exception. The attribute 'after' will be set to the
301 exception type. The attribute 'match' will be None. This allows you to
302 write code like this::
304 index = p.expect(['good', 'bad', pexpect.EOF, pexpect.TIMEOUT])
305 if index == 0:
306 do_something()
307 elif index == 1:
308 do_something_else()
309 elif index == 2:
310 do_some_other_thing()
311 elif index == 3:
312 do_something_completely_different()
314 instead of code like this::
316 try:
317 index = p.expect(['good', 'bad'])
318 if index == 0:
319 do_something()
320 elif index == 1:
321 do_something_else()
322 except EOF:
323 do_some_other_thing()
324 except TIMEOUT:
325 do_something_completely_different()
327 These two forms are equivalent. It all depends on what you want. You
328 can also just expect the EOF if you are waiting for all output of a
329 child to finish. For example::
331 p = pexpect.spawn('/bin/ls')
332 p.expect(pexpect.EOF)
333 print p.before
335 If you are trying to optimize for speed then see expect_list().
337 On Python 3.4, or Python 3.3 with asyncio installed, passing
338 ``async_=True`` will make this return an :mod:`asyncio` coroutine,
339 which you can yield from to get the same result that this method would
340 normally give directly. So, inside a coroutine, you can replace this code::
342 index = p.expect(patterns)
344 With this non-blocking form::
346 index = yield from p.expect(patterns, async_=True)
347 '''
348 if 'async' in kw:
349 async_ = kw.pop('async')
350 if kw:
351 raise TypeError("Unknown keyword arguments: {}".format(kw))
353 compiled_pattern_list = self.compile_pattern_list(pattern)
354 return self.expect_list(compiled_pattern_list,
355 timeout, searchwindowsize, async_)
357 def expect_list(self, pattern_list, timeout=-1, searchwindowsize=-1,
358 async_=False, **kw):
359 '''This takes a list of compiled regular expressions and returns the
360 index into the pattern_list that matched the child output. The list may
361 also contain EOF or TIMEOUT(which are not compiled regular
362 expressions). This method is similar to the expect() method except that
363 expect_list() does not recompile the pattern list on every call. This
364 may help if you are trying to optimize for speed, otherwise just use
365 the expect() method. This is called by expect().
368 Like :meth:`expect`, passing ``async_=True`` will make this return an
369 asyncio coroutine.
370 '''
371 if timeout == -1:
372 timeout = self.timeout
373 if 'async' in kw:
374 async_ = kw.pop('async')
375 if kw:
376 raise TypeError("Unknown keyword arguments: {}".format(kw))
378 exp = Expecter(self, searcher_re(pattern_list), searchwindowsize)
379 if async_:
380 from ._async import expect_async
381 return expect_async(exp, timeout)
382 else:
383 return exp.expect_loop(timeout)
385 def expect_exact(self, pattern_list, timeout=-1, searchwindowsize=-1,
386 async_=False, **kw):
388 '''This is similar to expect(), but uses plain string matching instead
389 of compiled regular expressions in 'pattern_list'. The 'pattern_list'
390 may be a string; a list or other sequence of strings; or TIMEOUT and
391 EOF.
393 This call might be faster than expect() for two reasons: string
394 searching is faster than RE matching and it is possible to limit the
395 search to just the end of the input buffer.
397 This method is also useful when you don't want to have to worry about
398 escaping regular expression characters that you want to match.
400 Like :meth:`expect`, passing ``async_=True`` will make this return an
401 asyncio coroutine.
402 '''
403 if timeout == -1:
404 timeout = self.timeout
405 if 'async' in kw:
406 async_ = kw.pop('async')
407 if kw:
408 raise TypeError("Unknown keyword arguments: {}".format(kw))
410 if (isinstance(pattern_list, self.allowed_string_types) or
411 pattern_list in (TIMEOUT, EOF)):
412 pattern_list = [pattern_list]
414 def prepare_pattern(pattern):
415 if pattern in (TIMEOUT, EOF):
416 return pattern
417 if isinstance(pattern, self.allowed_string_types):
418 return self._coerce_expect_string(pattern)
419 self._pattern_type_err(pattern)
421 try:
422 pattern_list = iter(pattern_list)
423 except TypeError:
424 self._pattern_type_err(pattern_list)
425 pattern_list = [prepare_pattern(p) for p in pattern_list]
427 exp = Expecter(self, searcher_string(pattern_list), searchwindowsize)
428 if async_:
429 from ._async import expect_async
430 return expect_async(exp, timeout)
431 else:
432 return exp.expect_loop(timeout)
434 def expect_loop(self, searcher, timeout=-1, searchwindowsize=-1):
435 '''This is the common loop used inside expect. The 'searcher' should be
436 an instance of searcher_re or searcher_string, which describes how and
437 what to search for in the input.
439 See expect() for other arguments, return value and exceptions. '''
441 exp = Expecter(self, searcher, searchwindowsize)
442 return exp.expect_loop(timeout)
444 def read(self, size=-1):
445 '''This reads at most "size" bytes from the file (less if the read hits
446 EOF before obtaining size bytes). If the size argument is negative or
447 omitted, read all data until EOF is reached. The bytes are returned as
448 a string object. An empty string is returned when EOF is encountered
449 immediately. '''
451 if size == 0:
452 return self.string_type()
453 if size < 0:
454 # delimiter default is EOF
455 self.expect(self.delimiter)
456 return self.before
458 # I could have done this more directly by not using expect(), but
459 # I deliberately decided to couple read() to expect() so that
460 # I would catch any bugs early and ensure consistent behavior.
461 # It's a little less efficient, but there is less for me to
462 # worry about if I have to later modify read() or expect().
463 # Note, it's OK if size==-1 in the regex. That just means it
464 # will never match anything in which case we stop only on EOF.
465 cre = re.compile(self._coerce_expect_string('.{%d}' % size), re.DOTALL)
466 # delimiter default is EOF
467 index = self.expect([cre, self.delimiter])
468 if index == 0:
469 ### FIXME self.before should be ''. Should I assert this?
470 return self.after
471 return self.before
473 def readline(self, size=-1):
474 '''This reads and returns one entire line. The newline at the end of
475 line is returned as part of the string, unless the file ends without a
476 newline. An empty string is returned if EOF is encountered immediately.
477 This looks for a newline as a CR/LF pair (\\r\\n) even on UNIX because
478 this is what the pseudotty device returns. So contrary to what you may
479 expect you will receive newlines as \\r\\n.
481 If the size argument is 0 then an empty string is returned. In all
482 other cases the size argument is ignored, which is not standard
483 behavior for a file-like object. '''
485 if size == 0:
486 return self.string_type()
487 # delimiter default is EOF
488 index = self.expect([self.crlf, self.delimiter])
489 if index == 0:
490 return self.before + self.crlf
491 else:
492 return self.before
494 def __iter__(self):
495 '''This is to support iterators over a file-like object.
496 '''
497 return iter(self.readline, self.string_type())
499 def readlines(self, sizehint=-1):
500 '''This reads until EOF using readline() and returns a list containing
501 the lines thus read. The optional 'sizehint' argument is ignored.
502 Remember, because this reads until EOF that means the child
503 process should have closed its stdout. If you run this method on
504 a child that is still running with its stdout open then this
505 method will block until it timesout.'''
507 lines = []
508 while True:
509 line = self.readline()
510 if not line:
511 break
512 lines.append(line)
513 return lines
515 def fileno(self):
516 '''Expose file descriptor for a file-like interface
517 '''
518 return self.child_fd
520 def flush(self):
521 '''This does nothing. It is here to support the interface for a
522 File-like object. '''
523 pass
525 def isatty(self):
526 """Overridden in subclass using tty"""
527 return False
529 # For 'with spawn(...) as child:'
530 def __enter__(self):
531 return self
533 def __exit__(self, etype, evalue, tb):
534 # We rely on subclasses to implement close(). If they don't, it's not
535 # clear what a context manager should do.
536 self.close()