1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179
|
# Copyright 2017-2020 Palantir Technologies, Inc.
# Copyright 2021- Python Language Server Contributors.
import os
from io import StringIO
from unittest.mock import MagicMock
import pytest
from pylsp_jsonrpc.dispatchers import MethodDispatcher
from pylsp_jsonrpc.endpoint import Endpoint
from pylsp_jsonrpc.exceptions import JsonRpcException
from pylsp import uris
from pylsp.config.config import Config
from pylsp.python_lsp import PythonLSPServer
from pylsp.workspace import Document, Workspace
from test.test_utils import CALL_TIMEOUT_IN_SECONDS, ClientServerPair
DOC_URI = uris.from_fs_path(__file__)
DOC = """import sys
def main():
print sys.stdin.read()
"""
class FakeEditorMethodsMixin:
"""
Represents the methods to be added to a dispatcher class when faking an editor.
"""
def m_window__work_done_progress__create(self, *_args, **_kwargs):
"""
Fake editor method `window/workDoneProgress/create`.
related spec:
https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#window_workDoneProgress_create
"""
return None
class FakePythonLSPServer(FakeEditorMethodsMixin, PythonLSPServer):
pass
class FakeEndpoint(Endpoint):
"""
Fake Endpoint representing the editor / LSP client.
The `dispatcher` dict will be used to synchronously calculate the responses
for calls to `.request` and resolve the futures with the value or errors.
Fake methods in the `dispatcher` should raise `JsonRpcException` for any
error.
"""
def request(self, method, params=None):
request_future = super().request(method, params)
try:
request_future.set_result(self._dispatcher[method](params))
except JsonRpcException as e:
request_future.set_exception(e)
return request_future
@pytest.fixture
def pylsp(tmpdir):
"""Return an initialized python LS"""
ls = FakePythonLSPServer(StringIO, StringIO, endpoint_cls=FakeEndpoint)
ls.m_initialize(
processId=1, rootUri=uris.from_fs_path(str(tmpdir)), initializationOptions={}
)
return ls
@pytest.fixture
def pylsp_w_workspace_folders(tmpdir):
"""Return an initialized python LS"""
ls = FakePythonLSPServer(StringIO, StringIO, endpoint_cls=FakeEndpoint)
folder1 = tmpdir.mkdir("folder1")
folder2 = tmpdir.mkdir("folder2")
ls.m_initialize(
processId=1,
rootUri=uris.from_fs_path(str(folder1)),
initializationOptions={},
workspaceFolders=[
{"uri": uris.from_fs_path(str(folder1)), "name": "folder1"},
{"uri": uris.from_fs_path(str(folder2)), "name": "folder2"},
],
)
workspace_folders = [folder1, folder2]
return (ls, workspace_folders)
@pytest.fixture()
def consumer():
return MagicMock()
@pytest.fixture()
def endpoint(consumer):
class Dispatcher(FakeEditorMethodsMixin, MethodDispatcher):
pass
return FakeEndpoint(Dispatcher(), consumer, id_generator=lambda: "id")
@pytest.fixture
def workspace(tmpdir, endpoint) -> None:
"""Return a workspace."""
ws = Workspace(uris.from_fs_path(str(tmpdir)), endpoint)
ws._config = Config(ws.root_uri, {}, 0, {})
yield ws
ws.close()
@pytest.fixture
def workspace_other_root_path(tmpdir, endpoint):
"""Return a workspace with a root_path other than tmpdir."""
ws_path = str(tmpdir.mkdir("test123").mkdir("test456"))
ws = Workspace(uris.from_fs_path(ws_path), endpoint)
ws._config = Config(ws.root_uri, {}, 0, {})
return ws
@pytest.fixture
def config(workspace):
"""Return a config object."""
cfg = Config(workspace.root_uri, {}, 0, {})
cfg._plugin_settings = {
"plugins": {"pylint": {"enabled": False, "args": [], "executable": None}}
}
return cfg
@pytest.fixture
def doc(workspace):
return Document(DOC_URI, workspace, DOC)
@pytest.fixture
def temp_workspace_factory(workspace):
"""
Returns a function that creates a temporary workspace from the files dict.
The dict is in the format {"file_name": "file_contents"}
"""
def fn(files):
def create_file(name, content):
fn = os.path.join(workspace.root_path, name)
with open(fn, "w", encoding="utf-8") as f:
f.write(content)
workspace.put_document(uris.from_fs_path(fn), content)
for name, content in files.items():
create_file(name, content)
return workspace
return fn
@pytest.fixture
def client_server_pair() -> None:
"""A fixture that sets up a client/server pair and shuts down the server"""
client_server_pair_obj = ClientServerPair()
yield (client_server_pair_obj.client, client_server_pair_obj.server)
shutdown_response = client_server_pair_obj.client._endpoint.request(
"shutdown"
).result(timeout=CALL_TIMEOUT_IN_SECONDS)
assert shutdown_response is None
client_server_pair_obj.client._endpoint.notify("exit")
|