python-lsp-server/test/test_utils.py at develop · StarkZhidian/python-lsp-server · GitHub

History

264 lines (205 loc) · 6.91 KB

Raw

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

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

import multiprocessing

import os

import sys

import time

from threading import Thread

from typing import Any, Dict, List

from unittest import mock

from docstring_to_markdown import UnknownFormatError

from flaky import flaky

from pylsp import _utils

from pylsp.lsp import NotebookCellKind

from pylsp.python_lsp import PythonLSPServer, start_io_lang_server

CALL_TIMEOUT_IN_SECONDS = 30

def send_notebook_did_open(client, cells: List[str]):

"""

Sends a notebookDocument/didOpen notification with the given python cells.

The notebook has the uri "notebook_uri" and the cells have the uris

"cell_1_uri", "cell_2_uri", etc.

"""

client._endpoint.notify(

"notebookDocument/didOpen", notebook_with_python_cells(cells)

)

def notebook_with_python_cells(cells: List[str]):

"""

Create a notebook document with the given python cells.

The notebook has the uri "notebook_uri" and the cells have the uris

"cell_1_uri", "cell_2_uri", etc.

"""

return {

"notebookDocument": {

"uri": "notebook_uri",

"notebookType": "jupyter-notebook",

"cells": [

{

"kind": NotebookCellKind.Code,

"document": f"cell_{i+1}_uri",

}

for i in range(len(cells))

"cellTextDocuments": [

{

"uri": f"cell_{i+1}_uri",

"languageId": "python",

"text": cell,

}

for i, cell in enumerate(cells)

}

def send_initialize_request(client, initialization_options: Dict[str, Any] = None):

return client._endpoint.request(

"initialize",

{

"processId": 1234,

"rootPath": os.path.dirname(__file__),

"initializationOptions": initialization_options,

).result(timeout=CALL_TIMEOUT_IN_SECONDS)

def start(obj):

obj.start()

class ClientServerPair:

"""

A class to setup a client/server pair.

args:

start_server_in_process: if True, the server will be started in a process.

check_parent_process: if True, the server_process will check if the parent process is alive.

"""

def __init__(self, start_server_in_process=False, check_parent_process=False):

# Client to Server pipe

csr, csw = os.pipe()

# Server to client pipe

scr, scw = os.pipe()

if start_server_in_process:

ParallelKind = self._get_parallel_kind()

self.server_process = ParallelKind(

target=start_io_lang_server,

args=(

os.fdopen(csr, "rb"),

os.fdopen(scw, "wb"),

check_parent_process,

PythonLSPServer,

)

self.server_process.start()

else:

self.server = PythonLSPServer(os.fdopen(csr, "rb"), os.fdopen(scw, "wb"))

self.server_thread = Thread(target=start, args=[self.server])

self.server_thread.start()

self.client = PythonLSPServer(os.fdopen(scr, "rb"), os.fdopen(csw, "wb"))

self.client_thread = Thread(target=start, args=[self.client])

self.client_thread.start()

def _get_parallel_kind(self):

if os.name == "nt":

return Thread

if sys.version_info[:2] >= (3, 8):

return multiprocessing.get_context("fork").Process

return multiprocessing.Process

@flaky(max_runs=6, min_passes=1)

def test_debounce():

interval = 0.1

obj = mock.Mock()

@_utils.debounce(0.1)

def call_m():

obj()

assert not obj.mock_calls

call_m()

assert not obj.mock_calls

time.sleep(interval * 2)

assert len(obj.mock_calls) == 1

call_m()

time.sleep(interval * 2)

assert len(obj.mock_calls) == 2

@flaky(max_runs=6, min_passes=1)

def test_debounce_keyed_by():

interval = 0.1

obj = mock.Mock()

@_utils.debounce(0.1, keyed_by="key")

def call_m(key):

obj(key)

assert not obj.mock_calls

call_m(1)

call_m(2)

call_m(3)

assert not obj.mock_calls

time.sleep(interval * 2)

obj.assert_has_calls(

[

mock.call(1),

mock.call(2),

mock.call(3),

any_order=True,

)

assert len(obj.mock_calls) == 3

call_m(1)

time.sleep(interval * 2)

assert len(obj.mock_calls) == 4

def test_list_to_string():

assert _utils.list_to_string("string") == "string"

assert _utils.list_to_string(["a", "r", "r", "a", "y"]) == "a,r,r,a,y"

def test_find_parents(tmpdir):

subsubdir = tmpdir.ensure_dir("subdir", "subsubdir")

path = subsubdir.ensure("path.py")

test_cfg = tmpdir.ensure("test.cfg")

assert _utils.find_parents(tmpdir.strpath, path.strpath, ["test.cfg"]) == [

test_cfg.strpath

]

def test_merge_dicts():

assert _utils.merge_dicts(

{"a": True, "b": {"x": 123, "y": {"hello": "world"}}},

{"a": False, "b": {"y": [], "z": 987}},

) == {"a": False, "b": {"x": 123, "y": [], "z": 987}}

def test_clip_column():

assert _utils.clip_column(0, [], 0) == 0

assert _utils.clip_column(2, ["123"], 0) == 2

assert _utils.clip_column(3, ["123"], 0) == 3

assert _utils.clip_column(5, ["123"], 0) == 3

assert _utils.clip_column(0, ["\n", "123"], 0) == 0

assert _utils.clip_column(1, ["\n", "123"], 0) == 0

assert _utils.clip_column(2, ["123\n", "123"], 0) == 2

assert _utils.clip_column(3, ["123\n", "123"], 0) == 3

assert _utils.clip_column(4, ["123\n", "123"], 1) == 3

@mock.patch("docstring_to_markdown.convert")

def test_format_docstring_valid_rst_signature(mock_convert):

"""Test that a valid RST docstring includes the function signature."""

docstring = """A function docstring.

Parameters

----------

a : str, something

"""

# Mock the return value to avoid depedency on the real thing

mock_convert.return_value = """A function docstring.

#### Parameters

- `a`: str, something

"""

markdown = _utils.format_docstring(

docstring,

"markdown",

["something(a: str) -> str"],

)["value"]

assert markdown.startswith(

_utils.wrap_signature("something(a: str) -> str"),

)

@mock.patch("docstring_to_markdown.convert", side_effect=UnknownFormatError)

def test_format_docstring_invalid_rst_signature(_):

"""Test that an invalid RST docstring includes the function signature."""

docstring = """A function docstring.

Parameters

----------

a : str, something

"""

markdown = _utils.format_docstring(

docstring,

"markdown",

["something(a: str) -> str"],

)["value"]

assert markdown.startswith(

_utils.wrap_signature("something(a: str) -> str"),

)

python-lsp-server/test/test_utils.py at develop · StarkZhidian/python-lsp-server · GitHub

Expand file tree

Breadcrumbs

test_utils.py

Latest commit

History

Breadcrumbs

test_utils.py

File metadata and controls