FilesExpand file tree

develop

/

fixtures.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

179 lines (131 loc) · 4.91 KB

develop

/

fixtures.py

Copy path

Top

File metadata and controls

• Code
• Blame

179 lines (131 loc) · 4.91 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

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