forked from palantir/python-language-server
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_language_server.py
More file actions
104 lines (74 loc) · 2.59 KB
/
Copy pathtest_language_server.py
File metadata and controls
104 lines (74 loc) · 2.59 KB
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
# Copyright 2017 Palantir Technologies, Inc.
import json
import os
from threading import Thread
import jsonrpc
import pytest
from pyls.server import JSONRPCServer
from pyls.language_server import start_io_lang_server
from pyls.python_ls import PythonLanguageServer
class JSONRPCClient(JSONRPCServer):
""" This is a weird way of testing.. but we're going to have two JSONRPCServers
talking to each other. One pretending to be a 'VSCode'-like client, the other is
our language server """
pass
@pytest.fixture
def client_server():
""" A fixture to setup a client/server """
# Client to Server pipe
csr, csw = os.pipe()
# Server to client pipe
scr, scw = os.pipe()
server = Thread(target=start_io_lang_server, args=(
os.fdopen(csr, 'rb'), os.fdopen(scw, 'wb'), PythonLanguageServer
))
server.daemon = True
server.start()
client = JSONRPCClient(os.fdopen(scr, 'rb'), os.fdopen(csw, 'wb'))
yield client, server
try:
client.call('shutdown')
except:
pass
def test_initialize(client_server):
client, server = client_server
client.call('initialize', {
'processId': 1234,
'rootPath': os.path.dirname(__file__),
'initializationOptions': {}
})
response = _get_response(client)
assert 'capabilities' in response['result']
def test_file_closed(client_server):
client, server = client_server
client.rfile.close()
with pytest.raises(Exception):
_get_response(client)
def test_missing_message(client_server):
client, server = client_server
client.call('unknown_method')
response = _get_response(client)
assert response['error']['code'] == -32601 # Method not implemented error
def test_linting(client_server):
client, server = client_server
# Initialize
client.call('initialize', {
'processId': 1234,
'rootPath': os.path.dirname(__file__),
'initializationOptions': {}
})
response = _get_response(client)
assert 'capabilities' in response['result']
# didOpen
client.call('textDocument/didOpen', {
'textDocument': {'uri': 'file:///test', 'text': 'import sys'}
})
response = _get_notification(client)
assert response['method'] == 'textDocument/publishDiagnostics'
assert len(response['params']['diagnostics']) > 0
def _get_notification(client):
request = jsonrpc.jsonrpc.JSONRPCRequest.from_json(client._read_message().decode('utf-8'))
assert request.is_notification
return request.data
def _get_response(client):
return json.loads(client._read_message().decode('utf-8'))