Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/invoke/watchers.py: 35%

1import re 

2import threading 

3from typing import Generator, Iterable 

4 

5from .exceptions import ResponseNotAccepted 

6 

7 

8class StreamWatcher(threading.local): 

9 """ 

10 A class whose subclasses may act on seen stream data from subprocesses. 

11 

12 Subclasses must exhibit the following API; see `Responder` for a concrete 

13 example. 

14 

15 * ``__init__`` is completely up to each subclass, though as usual, 

16 subclasses *of* subclasses should be careful to make use of `super` where 

17 appropriate. 

18 * `submit` must accept the entire current contents of the stream being 

19 watched, as a string, and may optionally return an iterable of strings 

20 (or act as a generator iterator, i.e. multiple calls to ``yield 

21 <string>``), which will each be written to the subprocess' standard 

22 input. 

23 

24 .. note:: 

25 `StreamWatcher` subclasses exist in part to enable state tracking, such 

26 as detecting when a submitted password didn't work & erroring (or 

27 prompting a user, or etc). Such bookkeeping isn't easily achievable 

28 with simple callback functions. 

29 

30 .. note:: 

31 `StreamWatcher` subclasses `threading.local` so that its instances can 

32 be used to 'watch' both subprocess stdout and stderr in separate 

33 threads. 

34 

35 .. versionadded:: 1.0 

36 """ 

37 

38 def submit(self, stream: str) -> Iterable[str]: 

39 """ 

40 Act on ``stream`` data, potentially returning responses. 

41 

42 :param str stream: 

43 All data read on this stream since the beginning of the session. 

44 

45 :returns: 

46 An iterable of ``str`` (which may be empty). 

47 

48 .. versionadded:: 1.0 

49 """ 

50 raise NotImplementedError 

51 

52 

53class Responder(StreamWatcher): 

54 """ 

55 A parameterizable object that submits responses to specific patterns. 

56 

57 Commonly used to implement password auto-responds for things like ``sudo``. 

58 

59 .. versionadded:: 1.0 

60 """ 

61 

62 def __init__(self, pattern: str, response: str) -> None: 

63 r""" 

64 Imprint this `Responder` with necessary parameters. 

65 

66 :param pattern: 

67 A raw string (e.g. ``r"\[sudo\] password for .*:"``) which will be 

68 turned into a regular expression. 

69 

70 :param response: 

71 The string to submit to the subprocess' stdin when ``pattern`` is 

72 detected. 

73 """ 

74 # TODO: precompile the keys into regex objects 

75 self.pattern = pattern 

76 self.response = response 

77 self.index = 0 

78 

79 def pattern_matches( 

80 self, stream: str, pattern: str, index_attr: str 

81 ) -> Iterable[str]: 

82 """ 

83 Generic "search for pattern in stream, using index" behavior. 

84 

85 Used here and in some subclasses that want to track multiple patterns 

86 concurrently. 

87 

88 :param str stream: The same data passed to ``submit``. 

89 :param str pattern: The pattern to search for. 

90 :param str index_attr: The name of the index attribute to use. 

91 :returns: An iterable of string matches. 

92 

93 .. versionadded:: 1.0 

94 """ 

95 # NOTE: generifies scanning so it can be used to scan for >1 pattern at 

96 # once, e.g. in FailingResponder. 

97 # Only look at stream contents we haven't seen yet, to avoid dupes. 

98 index = getattr(self, index_attr) 

99 new = stream[index:] 

100 # Search, across lines if necessary 

101 matches = re.findall(pattern, new, re.S) 

102 # Update seek index if we've matched 

103 if matches: 

104 setattr(self, index_attr, index + len(new)) 

105 return matches 

106 

107 def submit(self, stream: str) -> Generator[str, None, None]: 

108 # Iterate over findall() response in case >1 match occurred. 

109 for _ in self.pattern_matches(stream, self.pattern, "index"): 

110 yield self.response 

111 

112 

113class FailingResponder(Responder): 

114 """ 

115 Variant of `Responder` which is capable of detecting incorrect responses. 

116 

117 This class adds a ``sentinel`` parameter to ``__init__``, and its 

118 ``submit`` will raise `.ResponseNotAccepted` if it detects that sentinel 

119 value in the stream. 

120 

121 .. versionadded:: 1.0 

122 """ 

123 

124 def __init__(self, pattern: str, response: str, sentinel: str) -> None: 

125 super().__init__(pattern, response) 

126 self.sentinel = sentinel 

127 self.failure_index = 0 

128 self.tried = False 

129 

130 def submit(self, stream: str) -> Generator[str, None, None]: 

131 # Behave like regular Responder initially 

132 response = super().submit(stream) 

133 # Also check stream for our failure sentinel 

134 failed = self.pattern_matches(stream, self.sentinel, "failure_index") 

135 # Error out if we seem to have failed after a previous response. 

136 if self.tried and failed: 

137 err = 'Auto-response to r"{}" failed with {!r}!'.format( 

138 self.pattern, self.sentinel 

139 ) 

140 raise ResponseNotAccepted(err) 

141 # Once we see that we had a response, take note 

142 if response: 

143 self.tried = True 

144 # Again, behave regularly by default. 

145 return response