Improve search integration with the documentation panel

awsshell/app.py

@@ -18,6 +18,7 @@
 from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
 from prompt_toolkit.history import InMemoryHistory, FileHistory
 from prompt_toolkit.enums import EditingMode
+from prompt_toolkit.enums import DEFAULT_BUFFER
 
 from awsshell.ui import create_default_layout
 from awsshell.config import Config
@@ -431,6 +432,7 @@ def create_application(self, completer, history,
             editing_mode = EditingMode.EMACS
 
         return Application(
+            use_alternate_screen=self._output is not None,
             editing_mode=editing_mode,
             layout=self.create_layout(display_completions_in_columns, toolbar),
             mouse_support=False,
@@ -444,7 +446,7 @@ def create_application(self, completer, history,
         )
 
     def on_input_timeout(self, cli):
-        if not self.show_help:
+        if not self.show_help or cli.current_buffer_name != DEFAULT_BUFFER:
             return
         document = cli.current_buffer.document
         text = document.text

awsshell/keys.py

@@ -11,6 +11,7 @@
 # ANY KIND, either express or implied. See the License for the specific
 # language governing permissions and limitations under the License.
 from prompt_toolkit.key_binding.manager import KeyBindingManager
+from prompt_toolkit.enums import DEFAULT_BUFFER
 from prompt_toolkit.keys import Keys
 
 
@@ -154,7 +155,7 @@ def handle_f9(event):
 
             """
             if event.cli.current_buffer_name == u'clidocs':
-                event.cli.focus(u'DEFAULT_BUFFER')
+                event.cli.focus(DEFAULT_BUFFER)
             else:
                 event.cli.focus(u'clidocs')
 

awsshell/toolbar.py

@@ -11,6 +11,7 @@
 # ANY KIND, either express or implied. See the License for the specific
 # language governing permissions and limitations under the License.
 from pygments.token import Token
+from prompt_toolkit.enums import DEFAULT_BUFFER
 
 
 class Toolbar(object):
@@ -87,7 +88,7 @@ def get_toolbar_items(cli):
             else:
                 show_help_token = Token.Toolbar.Off
                 show_help_cfg = 'OFF'
-            if cli.current_buffer_name == 'DEFAULT_BUFFER':
+            if cli.current_buffer_name == DEFAULT_BUFFER:
                 show_buffer_name = 'cli'
             else:
                 show_buffer_name = 'doc'

awsshell/ui.py

@@ -81,15 +81,13 @@ def create_default_layout(app, message='',
 
     # Create processors list.
     # (DefaultPrompt should always be at the end.)
+    def search_highlighting(name):
+        return ConditionalProcessor(
+            HighlightSearchProcessor(preview_search=PreviousFocus(name)),
+            HasFocus(SEARCH_BUFFER))
+
     input_processors = [
-        ConditionalProcessor(
-            # By default, only highlight search when the search
-            # input has the focus. (Note that this doesn't mean
-            # there is no search: the Vi 'n' binding for instance
-            # still allows to jump to the next match in
-            # navigation mode.)
-            HighlightSearchProcessor(preview_search=Always()),
-            HasFocus(SEARCH_BUFFER)),
+        search_highlighting(DEFAULT_BUFFER),
         HighlightSelectionProcessor(),
         ConditionalProcessor(
             AppendAutoSuggestion(), HasFocus(DEFAULT_BUFFER) & ~IsDone()),
@@ -155,7 +153,7 @@ def separator():
                         lexer=lexer,
                         # Enable preview_search, we want to have immediate
                         # feedback in reverse-i-search mode.
-                        preview_search=Always(),
+                        preview_search=PreviousFocus(DEFAULT_BUFFER),
                         focus_on_click=True,
                     ),
                     get_height=get_height,
@@ -183,6 +181,8 @@ def separator():
                 BufferControl(
                     focus_on_click=True,
                     buffer_name=u'clidocs',
+                    input_processors=[search_highlighting(u'clidocs')],
+                    preview_search=PreviousFocus(u'clidocs'),
                 ),
                 height=LayoutDimension(max=15)),
             filter=HasDocumentation(app) & ~IsDone(),
@@ -233,3 +233,13 @@ def __init__(self, app):
 
     def __call__(self, cli):
         return bool(self._app.current_docs)
+
+
+class PreviousFocus(Filter):
+    def __init__(self, buffer_name):
+        self._buffer_name = buffer_name
+
+    def __call__(self, cli):
+        previous_buffer = cli.buffers.previous(cli)
+        target_buffer = cli.buffers.get(self._buffer_name)
+        return previous_buffer is target_buffer

requirements-test.txt

@@ -6,3 +6,4 @@ configobj==5.0.6
 # Note you need at least pip --version of 6.0 or
 # higher to be able to pick on these version specifiers.
 unittest2==1.1.0; python_version == '2.6'
+pyte==0.6.0

tests/integration/test_ui.py

@@ -0,0 +1,181 @@
+import pyte
+import time
+import unittest
+import threading
+from awsshell import shellcomplete, autocomplete
+from awsshell.app import AWSShell
+from awsshell.docs import DocRetriever
+from prompt_toolkit.keys import Keys
+from prompt_toolkit.input import PipeInput
+from prompt_toolkit.layout.screen import Size
+from prompt_toolkit.terminal.vt100_output import Vt100_Output
+from prompt_toolkit.terminal.vt100_input import ANSI_SEQUENCES
+
+_polly_index = {
+    "children": {
+        "describe-voices": {
+            "children": {},
+            "argument_metadata": {
+                "--language-code": {
+                    "type_name": "string",
+                    "example": "",
+                    "api_name": "LanguageCode",
+                    "required": False,
+                    "minidoc": "LanguageCode minidoc string"
+                }
+            },
+            "arguments": ["--language-code"],
+            "commands": []
+        }
+    },
+    "argument_metadata": {},
+    "arguments": [],
+    "commands": ["describe-voices"]
+}
+
+_index_data = {
+    'aws': {
+        'commands': ['polly'],
+        'children': {'polly': _polly_index},
+        'arguments': [],
+        'argument_metadata': {},
+    }
+}
+
+_doc_db = {b'aws.polly': 'Polly is a service'}
+
+
+class PyteOutput(object):
+
+    def __init__(self, columns=80, lines=24):
+        self._columns = columns
+        self._lines = lines
+        self._screen = pyte.Screen(self._columns, self._lines)
+        self._stream = pyte.ByteStream(self._screen)
+        self.encoding = 'utf-8'
+
+    def write(self, data):
+        self._stream.feed(data)
+
+    def flush(self):
+        pass
+
+    def get_size(self):
+        def _get_size():
+            return Size(columns=self._columns, rows=self._lines)
+        return _get_size
+
+    def display(self):
+        return self._screen.display
+
+
+def _create_shell(ptk_input, ptk_output):
+    io = {
+        'input': ptk_input,
+        'output': ptk_output,
+    }
+    doc_data = DocRetriever(_doc_db)
+    model_completer = autocomplete.AWSCLIModelCompleter(_index_data)
+    completer = shellcomplete.AWSShellCompleter(model_completer)
+    return AWSShell(completer, model_completer, doc_data, **io)
+
+
+_ansi_sequence = dict((key, ansi) for ansi, key in ANSI_SEQUENCES.items())
+
+
+class VirtualShell(object):
+
+    def __init__(self, shell=None):
+        self._ptk_input = PipeInput()
+        self._pyte = PyteOutput()
+        self._ptk_output = Vt100_Output(self._pyte, self._pyte.get_size())
+
+        if shell is None:
+            shell = _create_shell(self._ptk_input, self._ptk_output)
+
+        def _run_shell():
+            shell.run()
+
+        self._thread = threading.Thread(target=_run_shell)
+        self._thread.start()
+
+    def write(self, data):
+        self._ptk_input.send_text(data)
+
+    def press_keys(self, *keys):
+        sequences = [_ansi_sequence[key] for key in keys]
+        self.write(''.join(sequences))
+
+    def display(self):
+        return self._pyte.display()
+
+    def quit(self):
+        # figure out a better way to quit
+        self.press_keys(Keys.F10)
+        self._thread.join()
+        self._ptk_input.close()
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.quit()
+
+
+class ShellTest(unittest.TestCase):
+
+    def setUp(self):
+        self.shell = VirtualShell()
+
+    def tearDown(self):
+        self.shell.quit()
+
+    def write(self, data):
+        self.shell.write(data)
+
+    def press_keys(self, *keys):
+        self.shell.press_keys(*keys)
+
+    def _poll_display(self, timeout=2, interval=0.1):
+        start = time.time()
+        while time.time() <= start + timeout:
+            display = self.shell.display()
+            yield display
+            time.sleep(interval)
+
+    def await_text(self, text):
+        for display in self._poll_display():
+            for line in display:
+                if text in line:
+                    return
+        display = '\n'.join(display)
+        fail_message = '"{text}" not found on screen: \n{display}'
+        self.fail(fail_message.format(text=text, display=display))
+
+    def test_toolbar_appears(self):
+        self.await_text('[F3] Keys')
+
+    def test_input_works(self):
+        self.write('ec2')
+        self.await_text('ec2')
+
+    def test_completion_menu_operation(self):
+        self.write('polly desc')
+        self.await_text('describe-voices')
+
+    def test_completion_menu_argument(self):
+        self.write('polly describe-voices --l')
+        self.await_text('--language-code')
+
+    def test_doc_menu_appears(self):
+        self.write('polly ')
+        self.await_text('Polly is a service')
+
+    def test_doc_menu_is_searchable(self):
+        self.write('polly ')
+        self.await_text('Polly is a service')
+        self.press_keys(Keys.F9)
+        self.write('/')
+        # wait for the input timeout
+        time.sleep(0.6)
+        self.await_text('Polly is a service')