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")
|
