Text editing cleanup/testability (#577)

* Initial cleanup

Move response parsing to protocol
Extract non-sublime-tied logic from commands
Remove debugging status/logging when applying doc edits.
Simplify sorting of document edits (assuming stable sorting)

* Fix typing issues

* Move commands out of core, add tests

* Fix lint issue

* Name tests correctly and fix data

boot.py

@@ -4,7 +4,7 @@
 from .plugin.core.panels import *
 from .plugin.core.registry import LspRestartClientCommand
 from .plugin.core.documents import *
-from .plugin.core.edit import *
+from .plugin.edit import *
 from .plugin.completion import *
 from .plugin.diagnostics import *
 from .plugin.configuration import *

plugin/core/edit.py

@@ -1,118 +1,48 @@
-import os
-import sublime
-import sublime_plugin
-from timeit import default_timer as timer
-
+from .url import uri_to_filename
 try:
-    from typing import List, Dict, Optional, Any
-    assert List and Dict and Optional and Any
+    from typing import List, Dict, Optional, Any, Iterable, Tuple
+    TextEdit = Tuple[Tuple[int, int], Tuple[int, int], str]
+    assert List and Dict and Optional and Any and Iterable and Tuple
 except ImportError:
     pass
 
-from .url import uri_to_filename
-from .protocol import Range
-from .logging import debug
-from .workspace import get_project_path
-from .views import range_to_region
-
-
-class LspApplyWorkspaceEditCommand(sublime_plugin.WindowCommand):
-    def run(self, changes=None, document_changes=None):
-        documents_changed = 0
-        if changes:
-            for uri, file_changes in changes.items():
-                path = uri_to_filename(uri)
-                self.open_and_apply_edits(path, file_changes)
-                documents_changed += 1
-        elif document_changes:
-            for document_change in document_changes:
-                uri = document_change.get('textDocument').get('uri')
-                path = uri_to_filename(uri)
-                self.open_and_apply_edits(path, document_change.get('edits'))
-                documents_changed += 1
 
-        if documents_changed > 0:
-            message = 'Applied changes to {} documents'.format(documents_changed)
-            self.window.status_message(message)
-        else:
-            self.window.status_message('No changes to apply to workspace')
+def parse_workspace_edit(workspace_edit: 'Dict[str, Any]') -> 'Dict[str, List[TextEdit]]':
+    changes = {}  # type: Dict[str, List[TextEdit]]
+    if 'changes' in workspace_edit:
+        for uri, file_changes in workspace_edit.get('changes', {}).items():
+            changes[uri_to_filename(uri)] = list(parse_text_edit(change) for change in file_changes)
+    if 'documentChanges' in workspace_edit:
+        for document_change in workspace_edit.get('documentChanges', []):
+            uri = document_change.get('textDocument').get('uri')
+            changes[uri_to_filename(uri)] = list(parse_text_edit(change) for change in document_change.get('edits'))
+    return changes
 
-    def open_and_apply_edits(self, path, file_changes):
-        view = self.window.open_file(path)
-        if view:
-            if view.is_loading():
-                # TODO: wait for event instead.
-                sublime.set_timeout_async(
-                    lambda: view.run_command('lsp_apply_document_edit', {'changes': file_changes}),
-                    500
-                )
-            else:
-                view.run_command('lsp_apply_document_edit',
-                                 {'changes': file_changes,
-                                  'show_status': False})
-        else:
-            debug('view not found to apply', path, file_changes)
 
+def parse_range(range: 'Dict[str, int]') -> 'Tuple[int, int]':
+    return range['line'], range['character']
 
-class LspApplyDocumentEditCommand(sublime_plugin.TextCommand):
-    def run(self, edit, changes: 'Optional[List[dict]]' = None, show_status=True):
-        # Apply the changes in reverse, so that we don't invalidate the range
-        # of any change that we haven't applied yet.
-        start = timer()
-        changes2 = changes or []  # New variable of type List[dict]
-        indices = self.changes_order(changes2)
-        for index in indices:
-            change = changes2[index]
-            self.apply_change(self.create_region(change), change.get('newText'), edit)
-        elapsed = timer() - start
 
-        if show_status:
-            window = self.view.window()
-            if window:
-                base_dir = get_project_path(window)
-                file_path = self.view.file_name()
-                relative_file_path = os.path.relpath(file_path, base_dir) if base_dir else file_path
-                message = 'Applied {} change(s) to {} in {:.1f} ms'.format(
-                    len(indices), relative_file_path, elapsed * 1000)
-                window.status_message(message)
-                debug(message)
+def parse_text_edit(text_edit: 'Dict[str, Any]') -> 'TextEdit':
+    return (
+        parse_range(text_edit['range']['start']),
+        parse_range(text_edit['range']['end']),
+        text_edit.get('newText', '')
+    )
 
-    def changes_order(self, changes: 'List[dict]') -> 'List[int]':
-        # Changes look like this:
-        # [
-        #   {
-        #       'newText': str,
-        #       'range': {
-        #            'start': {'line': int, 'character': int},
-        #            'end': {'line': int, 'character': int}
-        #       }
-        #   }
-        # ]
 
-        def get_start_position(index: int):
-            change = changes[index]  # type: Any
-            start = change.get('range').get('start')
-            line = start.get('line')
-            character = start.get('character')
-            return (line, character, index)
+def sort_by_application_order(changes: 'Iterable[TextEdit]') -> 'List[TextEdit]':
 
-        # The spec reads:
-        # > However, it is possible that multiple edits have the same start position: multiple
-        # > inserts, or any number of inserts followed by a single remove or replace edit. If
-        # > multiple inserts have the same position, the order in the array defines the order in
-        # > which the inserted strings appear in the resulting text.
-        # So we sort by start position. But if multiple text edits start at the same position,
-        # we use the index in the array as the key.
-        return sorted(range(len(changes)), key=get_start_position, reverse=True)
+    def get_start_position(pair: 'Tuple[int, TextEdit]'):
+        index, change = pair
+        return change[0][0], change[0][1], index
 
-    def create_region(self, change):
-        return range_to_region(Range.from_lsp(change['range']), self.view)
+    # The spec reads:
+    # > However, it is possible that multiple edits have the same start position: multiple
+    # > inserts, or any number of inserts followed by a single remove or replace edit. If
+    # > multiple inserts have the same position, the order in the array defines the order in
+    # > which the inserted strings appear in the resulting text.
+    # So we sort by start position. But if multiple text edits start at the same position,
+    # we use the index in the array as the key.
 
-    def apply_change(self, region, newText, edit):
-        if region.empty():
-            self.view.insert(edit, region.a, newText)
-        else:
-            if len(newText) > 0:
-                self.view.replace(edit, region, newText)
-            else:
-                self.view.erase(edit, region)
+    return list(map(lambda pair: pair[1], sorted(enumerate(changes), key=get_start_position, reverse=True)))

plugin/core/test_edit.py

@@ -0,0 +1,71 @@
+import unittest
+from .edit import sort_by_application_order, parse_workspace_edit, parse_text_edit
+from .test_protocol import LSP_RANGE
+
+try:
+    from typing import List, Dict, Optional, Any, Iterable, Tuple
+    from .edit import TextEdit
+    assert List and Dict and Optional and Any and Iterable and Tuple and TextEdit
+except ImportError:
+    pass
+
+LSP_TEXT_EDIT = dict(newText='newText', range=LSP_RANGE)
+
+LSP_EDIT_CHANGES = {
+    'changes': {
+        'file:///file.py': [LSP_TEXT_EDIT]
+    }
+}
+
+LSP_EDIT_DOCUMENT_CHANGES = {
+    'documentChanges': [{
+        'textDocument': {'uri': 'file:///file.py'},
+        'edits': [LSP_TEXT_EDIT]
+    }]
+}
+
+
+class TextEditTests(unittest.TestCase):
+
+    def test_parse_from_lsp(self):
+        (start, end, newText) = parse_text_edit(LSP_TEXT_EDIT)
+        self.assertEqual(newText, 'newText')
+        self.assertEqual(start[0], 10)
+        self.assertEqual(start[1], 4)
+        self.assertEqual(end[0], 11)
+        self.assertEqual(end[1], 3)
+
+
+class WorkspaceEditTests(unittest.TestCase):
+
+    def test_parse_no_changes_from_lsp(self):
+        edit = parse_workspace_edit(dict())
+        self.assertEqual(len(edit), 0)
+
+    def test_parse_changes_from_lsp(self):
+        edit = parse_workspace_edit(LSP_EDIT_CHANGES)
+        self.assertEqual(len(edit), 1)
+        self.assertEqual(len(edit['/file.py']), 1)
+
+    def test_parse_document_changes_from_lsp(self):
+        edit = parse_workspace_edit(LSP_EDIT_DOCUMENT_CHANGES)
+        self.assertEqual(len(edit), 1)
+        self.assertEqual(len(edit['/file.py']), 1)
+
+
+class SortByApplicationOrderTests(unittest.TestCase):
+
+    def test_empty_sort(self):
+        self.assertEqual(sort_by_application_order([]), [])
+
+    def test_sorts_backwards(self):
+        edits = [
+            ((0, 0), (0, 0), 'b'),
+            ((0, 0), (0, 0), 'a'),
+            ((0, 2), (0, 2), 'c')
+        ]
+        # expect 'c' (higher start), 'a' now reverse order before 'b'
+        sorted = sort_by_application_order(edits)
+        self.assertEqual(sorted[0][2], 'c')
+        self.assertEqual(sorted[1][2], 'a')
+        self.assertEqual(sorted[2][2], 'b')

plugin/core/windows.py

@@ -3,6 +3,7 @@
 from .logging import debug, server_log
 from .types import ClientStates, ClientConfig, WindowLike, ViewLike, LanguageConfig, config_supports_syntax
 from .protocol import Notification, Response
+from .edit import parse_workspace_edit
 from .sessions import Session
 from .url import filename_to_uri
 from .workspace import get_project_path
@@ -419,8 +420,8 @@ def _end_old_sessions(self):
 
     def _apply_workspace_edit(self, params: 'Dict[str, Any]', client: Client, request_id: int) -> None:
         edit = params.get('edit', dict())
-        self._window.run_command('lsp_apply_workspace_edit', {'changes': edit.get('changes'),
-                                                              'document_changes': edit.get('documentChanges')})
+        changes = parse_workspace_edit(edit)
+        self._window.run_command('lsp_apply_workspace_edit', {'changes': changes})
         # TODO: We should ideally wait for all changes to have been applied.
         # This however seems overly complicated, because we have to bring along a string representation of the
         # client through the sublime-command invocations (as well as the request ID, but that is easy), and then

plugin/edit.py

@@ -0,0 +1,61 @@
+import sublime
+import sublime_plugin
+from .core.edit import sort_by_application_order
+try:
+    from typing import List, Dict, Optional, Any, Iterable, Tuple
+    from .core.edit import TextEdit
+    assert List and Dict and Optional and Any and Iterable and Tuple and TextEdit
+except ImportError:
+    pass
+from .core.logging import debug
+
+
+class LspApplyWorkspaceEditCommand(sublime_plugin.WindowCommand):
+    def run(self, changes: 'Optional[Dict[str, List[TextEdit]]]'=None):
+        documents_changed = 0
+        if changes:
+            for path, document_changes in changes.items():
+                self.open_and_apply_edits(path, document_changes)
+                documents_changed += 1
+
+        if documents_changed > 0:
+            message = 'Applied changes to {} documents'.format(documents_changed)
+            self.window.status_message(message)
+        else:
+            self.window.status_message('No changes to apply to workspace')
+
+    def open_and_apply_edits(self, path, file_changes):
+        view = self.window.open_file(path)
+        if view:
+            if view.is_loading():
+                # TODO: wait for event instead.
+                sublime.set_timeout_async(
+                    lambda: view.run_command('lsp_apply_document_edit', {'changes': file_changes}),
+                    500
+                )
+            else:
+                view.run_command('lsp_apply_document_edit',
+                                 {'changes': file_changes,
+                                  'show_status': False})
+        else:
+            debug('view not found to apply', path, file_changes)
+
+
+class LspApplyDocumentEditCommand(sublime_plugin.TextCommand):
+    def run(self, edit, changes: 'Optional[List[TextEdit]]'=None):
+        # Apply the changes in reverse, so that we don't invalidate the range
+        # of any change that we haven't applied yet.
+        if changes:
+            for change in sort_by_application_order(changes):
+                start, end, newText = change
+                region = sublime.Region(self.view.text_point(*start), self.view.text_point(*end))
+                self.apply_change(region, newText, edit)
+
+    def apply_change(self, region: 'sublime.Region', newText: str, edit):
+        if region.empty():
+            self.view.insert(edit, region.a, newText)
+        else:
+            if len(newText) > 0:
+                self.view.replace(edit, region, newText)
+            else:
+                self.view.erase(edit, region)

plugin/formatting.py

@@ -1,4 +1,5 @@
 from .core.protocol import Request
+from .core.edit import parse_text_edit
 from .core.registry import client_for_view, LspTextCommand
 from .core.types import ViewLike
 from .core.url import filename_to_uri
@@ -18,6 +19,11 @@ def options_for_view(view: ViewLike) -> 'Dict[str, Any]':
     }
 
 
+def apply_response_to_view(response, view):
+    edits = list(parse_text_edit(change) for change in response) if response else []
+    view.run_command('lsp_apply_document_edit', {'changes': edits})
+
+
 class LspFormatDocumentCommand(LspTextCommand):
     def __init__(self, view):
         super().__init__(view)
@@ -28,7 +34,6 @@ def is_enabled(self, event=None):
     def run(self, edit):
         client = client_for_view(self.view)
         if client:
-            pos = self.view.sel()[0].begin()
             params = {
                 "textDocument": {
                     "uri": filename_to_uri(self.view.file_name())
@@ -37,11 +42,7 @@ def run(self, edit):
             }
             request = Request.formatting(params)
             client.send_request(
-                request, lambda response: self.handle_response(response, pos))
-
-    def handle_response(self, response, pos):
-        self.view.run_command('lsp_apply_document_edit',
-                              {'changes': response})
+                request, lambda response: apply_response_to_view(response, self.view))
 
 
 class LspFormatDocumentRangeCommand(LspTextCommand):
@@ -68,5 +69,4 @@ def run(self, _):
                 "options": options_for_view(self.view)
             }
             client.send_request(Request.rangeFormatting(params),
-                                lambda response: self.view.run_command('lsp_apply_document_edit',
-                                                                       {'changes': response}))
+                                lambda response: apply_response_to_view(response, self.view))

plugin/rename.py

@@ -1,6 +1,7 @@
 import sublime_plugin
 from .core.registry import client_for_view, LspTextCommand
 from .core.protocol import Request
+from .core.edit import parse_workspace_edit
 from .core.documents import get_document_position, get_position, is_at_word
 try:
     from typing import List, Dict, Optional
@@ -64,9 +65,9 @@ def request_rename(self, params, new_name) -> None:
 
     def handle_response(self, response: 'Optional[Dict]') -> None:
         if response:
+            changes = parse_workspace_edit(response)
             self.view.window().run_command('lsp_apply_workspace_edit',
-                                           {'changes': response.get('changes'),
-                                            'document_changes': response.get('documentChanges')})
+                                           {'changes': changes})
         else:
             self.view.window().status_message('No rename edits returned')