intermediate changes

ref:cde9a383711a11544ce7e107a78147fb96cc4029

1 files changed, 72 insertions, 0 deletions

diff --git a/contrib/python/prompt-toolkit/py3/prompt_toolkit/input/posix_pipe.py b/contrib/python/prompt-toolkit/py3/prompt_toolkit/input/posix_pipe.py
new file mode 100644
index 0000000000..de47c64933
--- /dev/null
+++ b/contrib/python/prompt-toolkit/py3/prompt_toolkit/input/posix_pipe.py
@@ -0,0 +1,72 @@
+import os
+from typing import ContextManager, TextIO, cast
+
+from ..utils import DummyContext
+from .base import PipeInput
+from .vt100 import Vt100Input
+
+__all__ = [
+    "PosixPipeInput",
+]
+
+
+class PosixPipeInput(Vt100Input, PipeInput):
+    """
+    Input that is send through a pipe.
+    This is useful if we want to send the input programmatically into the
+    application. Mostly useful for unit testing.
+
+    Usage::
+
+        input = PosixPipeInput()
+        input.send_text('inputdata')
+    """
+
+    _id = 0
+
+    def __init__(self, text: str = "") -> None:
+        self._r, self._w = os.pipe()
+
+        class Stdin:
+            encoding = "utf-8"
+
+            def isatty(stdin) -> bool:
+                return True
+
+            def fileno(stdin) -> int:
+                return self._r
+
+        super().__init__(cast(TextIO, Stdin()))
+        self.send_text(text)
+
+        # Identifier for every PipeInput for the hash.
+        self.__class__._id += 1
+        self._id = self.__class__._id
+
+    def send_bytes(self, data: bytes) -> None:
+        os.write(self._w, data)
+
+    def send_text(self, data: str) -> None:
+        "Send text to the input."
+        os.write(self._w, data.encode("utf-8"))
+
+    def raw_mode(self) -> ContextManager[None]:
+        return DummyContext()
+
+    def cooked_mode(self) -> ContextManager[None]:
+        return DummyContext()
+
+    def close(self) -> None:
+        "Close pipe fds."
+        os.close(self._r)
+        os.close(self._w)
+
+        # We should assign `None` to 'self._r` and 'self._w',
+        # The event loop still needs to know the the fileno for this input in order
+        # to properly remove it from the selectors.
+
+    def typeahead_hash(self) -> str:
+        """
+        This needs to be unique for every `PipeInput`.
+        """
+        return "pipe-input-%s" % (self._id,)