Class Responder

invoke/watchers.py:53–110

A parameterizable object that submits responses to specific patterns. Commonly used to implement password auto-responds for things like ``sudo``. .. versionadded:: 1.0

class Responder(StreamWatcher):
    """
    A parameterizable object that submits responses to specific patterns.

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

    .. versionadded:: 1.0
    """

    def __init__(self, pattern: str, response: str) -> None:
        r"""
        Imprint this `Responder` with necessary parameters.

        :param pattern:
            A raw string (e.g. ``r"\[sudo\] password for .*:"``) which will be
            turned into a regular expression.

        :param response:
            The string to submit to the subprocess' stdin when ``pattern`` is
            detected.
        """
        # TODO: precompile the keys into regex objects
        self.pattern = pattern
        self.response = response
        self.index = 0

    def pattern_matches(
        self, stream: str, pattern: str, index_attr: str
    ) -> Iterable[str]:
        """
        Generic "search for pattern in stream, using index" behavior.

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

        :param str stream: The same data passed to ``submit``.
        :param str pattern: The pattern to search for.
        :param str index_attr: The name of the index attribute to use.
        :returns: An iterable of string matches.

        .. versionadded:: 1.0
        """
        # NOTE: generifies scanning so it can be used to scan for >1 pattern at
        # once, e.g. in FailingResponder.
        # Only look at stream contents we haven't seen yet, to avoid dupes.
        index = getattr(self, index_attr)
        new = stream[index:]
        # Search, across lines if necessary
        matches = re.findall(pattern, new, re.S)
        # Update seek index if we've matched
        if matches:
            setattr(self, index_attr, index + len(new))
        return matches

    def submit(self, stream: str) -> Generator[str, None, None]:
        # Iterate over findall() response in case >1 match occurred.
        for _ in self.pattern_matches(stream, self.pattern, "index"):
            yield self.response