290 lines
13 KiB
Python
290 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""Protocol test for the Klammertext language server (klammertext_ls.py).
|
|
|
|
Spawns the server as an LSP client would (JSON-RPC over stdio) and exercises
|
|
every capability against the editor fixtures:
|
|
|
|
* initialize / initialized handshake and the advertised capabilities
|
|
* didOpen + publishDiagnostics (clean fixture -> no diagnostics;
|
|
unbalanced text -> the expected errors)
|
|
* textDocument/formatting on every indent fixture == its expected file
|
|
* textDocument/rangeFormatting (a sub-range only)
|
|
* textDocument/documentHighlight + definition (the delimiter matcher)
|
|
* workspace/executeCommand klammertext.alignTable -> workspace/applyEdit,
|
|
applied result == every align fixture's expected file
|
|
* shutdown / exit (exit code 0)
|
|
|
|
Usage: ls_test.py SHARED_DIR FIXTURE_DIR
|
|
Prints one PASS/FAIL line per check; exits nonzero on any failure.
|
|
"""
|
|
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
|
|
|
|
class Client:
|
|
"""A minimal scripted LSP client over a server subprocess."""
|
|
|
|
def __init__(self, server_path):
|
|
self.proc = subprocess.Popen(
|
|
[sys.executable, server_path],
|
|
stdin=subprocess.PIPE, stdout=subprocess.PIPE)
|
|
self.next_id = 1
|
|
self.notifications = [] # collected server notifications
|
|
self.server_requests = [] # collected server->client requests
|
|
|
|
def send(self, message):
|
|
body = json.dumps(message).encode('utf-8')
|
|
self.proc.stdin.write(b'Content-Length: %d\r\n\r\n' % len(body) + body)
|
|
self.proc.stdin.flush()
|
|
|
|
def read_message(self):
|
|
length = None
|
|
while True:
|
|
line = self.proc.stdout.readline()
|
|
if not line:
|
|
return None
|
|
line = line.strip()
|
|
if not line:
|
|
break
|
|
if line.lower().startswith(b'content-length:'):
|
|
length = int(line.split(b':', 1)[1])
|
|
return json.loads(self.proc.stdout.read(length).decode('utf-8'))
|
|
|
|
def request(self, method, params):
|
|
"""Send a request and pump messages until its response arrives."""
|
|
msg_id = self.next_id
|
|
self.next_id += 1
|
|
self.send({'jsonrpc': '2.0', 'id': msg_id,
|
|
'method': method, 'params': params})
|
|
while True:
|
|
msg = self.read_message()
|
|
assert msg is not None, 'server closed during ' + method
|
|
if msg.get('id') == msg_id and 'method' not in msg:
|
|
assert 'error' not in msg, 'error response: %r' % msg
|
|
return msg.get('result')
|
|
self._collect(msg)
|
|
|
|
def notify(self, method, params):
|
|
self.send({'jsonrpc': '2.0', 'method': method, 'params': params})
|
|
|
|
def _collect(self, msg):
|
|
if 'method' in msg and msg.get('id') is not None:
|
|
self.server_requests.append(msg)
|
|
# acknowledge server->client requests (e.g. applyEdit)
|
|
self.send({'jsonrpc': '2.0', 'id': msg['id'],
|
|
'result': {'applied': True}})
|
|
elif 'method' in msg:
|
|
self.notifications.append(msg)
|
|
|
|
def wait_notification(self, method):
|
|
"""Pump messages until a notification with METHOD arrives."""
|
|
while True:
|
|
for i, msg in enumerate(self.notifications):
|
|
if msg['method'] == method:
|
|
return self.notifications.pop(i)['params']
|
|
msg = self.read_message()
|
|
assert msg is not None, 'server closed waiting for ' + method
|
|
self._collect(msg)
|
|
|
|
def wait_server_request(self, method):
|
|
while True:
|
|
for i, msg in enumerate(self.server_requests):
|
|
if msg['method'] == method:
|
|
return self.server_requests.pop(i)['params']
|
|
msg = self.read_message()
|
|
assert msg is not None, 'server closed waiting for ' + method
|
|
self._collect(msg)
|
|
|
|
|
|
PASS = 0
|
|
FAIL = 0
|
|
|
|
|
|
def check(name, ok, detail=''):
|
|
global PASS, FAIL
|
|
if ok:
|
|
PASS += 1
|
|
print('PASS %s' % name)
|
|
else:
|
|
FAIL += 1
|
|
print('FAIL %s %s' % (name, detail))
|
|
|
|
|
|
def apply_edits(ls, text, edits):
|
|
"""Apply LSP TextEdits to TEXT (offsets via the server's own helpers)."""
|
|
resolved = [(ls.pos_to_offset(text, e['range']['start']),
|
|
ls.pos_to_offset(text, e['range']['end']),
|
|
e['newText']) for e in edits]
|
|
for a, b, new in sorted(resolved, reverse=True):
|
|
text = text[:a] + new + text[b:]
|
|
return text
|
|
|
|
|
|
def main():
|
|
shared_dir, fixture_dir = sys.argv[1], sys.argv[2]
|
|
sys.path.insert(0, shared_dir)
|
|
import klammertext_ls as ls
|
|
|
|
client = Client(shared_dir + '/klammertext_ls.py')
|
|
uri = 'file:///ls_test.kt'
|
|
|
|
# -- handshake --
|
|
result = client.request('initialize', {'capabilities': {}})
|
|
caps = result.get('capabilities', {})
|
|
check('initialize capabilities',
|
|
caps.get('documentFormattingProvider') is True
|
|
and caps.get('documentHighlightProvider') is True
|
|
and 'klammertext.alignTable'
|
|
in caps.get('executeCommandProvider', {}).get('commands', []))
|
|
client.notify('initialized', {})
|
|
|
|
# -- diagnostics: clean fixture, then unbalanced text --
|
|
text = open(fixture_dir + '/indent_list_expected.kt').read()
|
|
client.notify('textDocument/didOpen',
|
|
{'textDocument': {'uri': uri, 'languageId': 'klammertext',
|
|
'version': 1, 'text': text}})
|
|
diags = client.wait_notification('textDocument/publishDiagnostics')
|
|
check('clean fixture: no diagnostics', diags['diagnostics'] == [],
|
|
repr(diags['diagnostics']))
|
|
|
|
bad = '@i abc\n@ol x ul@\n'
|
|
client.notify('textDocument/didChange',
|
|
{'textDocument': {'uri': uri, 'version': 2},
|
|
'contentChanges': [{'text': bad}]})
|
|
diags = client.wait_notification('textDocument/publishDiagnostics')
|
|
msgs = [d['message'] for d in diags['diagnostics']]
|
|
check('unbalanced text: diagnostics',
|
|
any('@i' in m and 'never closed' in m for m in msgs)
|
|
and any('ul@' in m for m in msgs), repr(msgs))
|
|
|
|
# -- formatting == every indent fixture's expected file --
|
|
indent_fixtures = ['indent_list', 'indent_document', 'indent_table',
|
|
'indent_untouched', 'indent_defs', 'indent_escapes',
|
|
'indent_named_close']
|
|
for f in indent_fixtures:
|
|
src = open(fixture_dir + '/%s.kt' % f).read()
|
|
exp = open(fixture_dir + '/%s_expected.kt' % f).read()
|
|
client.notify('textDocument/didChange',
|
|
{'textDocument': {'uri': uri, 'version': 3},
|
|
'contentChanges': [{'text': src}]})
|
|
client.wait_notification('textDocument/publishDiagnostics')
|
|
edits = client.request('textDocument/formatting',
|
|
{'textDocument': {'uri': uri},
|
|
'options': {'tabSize': 2,
|
|
'insertSpaces': True}})
|
|
check('formatting %s' % f, apply_edits(ls, src, edits or []) == exp)
|
|
|
|
# -- rangeFormatting: only the requested lines change --
|
|
src = '@ol\nzero\n one\n@\n'
|
|
client.notify('textDocument/didChange',
|
|
{'textDocument': {'uri': uri, 'version': 4},
|
|
'contentChanges': [{'text': src}]})
|
|
client.wait_notification('textDocument/publishDiagnostics')
|
|
edits = client.request('textDocument/rangeFormatting',
|
|
{'textDocument': {'uri': uri},
|
|
'range': {'start': {'line': 2, 'character': 0},
|
|
'end': {'line': 2, 'character': 0}},
|
|
'options': {}})
|
|
check('rangeFormatting touches only its lines',
|
|
apply_edits(ls, src, edits or []) == '@ol\nzero\n one\n@\n')
|
|
|
|
# -- matcher: documentHighlight + definition --
|
|
src = '@i abc @\n'
|
|
client.notify('textDocument/didChange',
|
|
{'textDocument': {'uri': uri, 'version': 5},
|
|
'contentChanges': [{'text': src}]})
|
|
client.wait_notification('textDocument/publishDiagnostics')
|
|
hl = client.request('textDocument/documentHighlight',
|
|
{'textDocument': {'uri': uri},
|
|
'position': {'line': 0, 'character': 0}})
|
|
check('documentHighlight pair', hl is not None and len(hl) == 2, repr(hl))
|
|
defn = client.request('textDocument/definition',
|
|
{'textDocument': {'uri': uri},
|
|
'position': {'line': 0, 'character': 0}})
|
|
check('definition = the matching close',
|
|
defn is not None and defn['range']['start']['character'] == 7,
|
|
repr(defn))
|
|
off = client.request('textDocument/documentHighlight',
|
|
{'textDocument': {'uri': uri},
|
|
'position': {'line': 0, 'character': 4}})
|
|
check('documentHighlight off-delimiter is null', off is None, repr(off))
|
|
|
|
# -- the custom matchInfo request (drives the VS Code decorations) --
|
|
hl2 = client.request('textDocument/documentHighlight',
|
|
{'textDocument': {'uri': uri},
|
|
'position': {'line': 0, 'character': 7}})
|
|
check('documentHighlight from the bare close',
|
|
hl2 is not None and len(hl2) == 2, repr(hl2))
|
|
mi = client.request('klammertext/matchInfo',
|
|
{'textDocument': {'uri': uri},
|
|
'position': {'line': 0, 'character': 7}})
|
|
check('matchInfo from the bare close',
|
|
mi is not None and not mi['mismatch']
|
|
and mi['matchToken']['start']['character'] == 0, repr(mi))
|
|
client.notify('textDocument/didChange',
|
|
{'textDocument': {'uri': uri, 'version': 6},
|
|
'contentChanges': [{'text': '@ol x ul@\n'}]})
|
|
client.wait_notification('textDocument/publishDiagnostics')
|
|
mi = client.request('klammertext/matchInfo',
|
|
{'textDocument': {'uri': uri},
|
|
'position': {'line': 0, 'character': 0}})
|
|
check('matchInfo reports a mismatch',
|
|
mi is not None and mi['mismatch'] and 'ul@' in (mi['message'] or ''),
|
|
repr(mi))
|
|
mi = client.request('klammertext/matchInfo',
|
|
{'textDocument': {'uri': uri},
|
|
'position': {'line': 0, 'character': 4}})
|
|
check('matchInfo off-delimiter is null', mi is None, repr(mi))
|
|
|
|
# -- alignTable via executeCommand -> applyEdit, on every align fixture --
|
|
align_fixtures = ['align_mixed', 'align_empty_cells', 'align_boundary',
|
|
'align_colspan', 'align_escapes']
|
|
for f in align_fixtures:
|
|
src = open(fixture_dir + '/%s.kt' % f).read()
|
|
exp = open(fixture_dir + '/%s_expected.kt' % f).read()
|
|
client.notify('textDocument/didChange',
|
|
{'textDocument': {'uri': uri, 'version': 6},
|
|
'contentChanges': [{'text': src}]})
|
|
client.wait_notification('textDocument/publishDiagnostics')
|
|
pos = ls.offset_to_pos(src, src.index('|'))
|
|
client.request('workspace/executeCommand',
|
|
{'command': 'klammertext.alignTable',
|
|
'arguments': [{'uri': uri, 'position': pos}]})
|
|
if src == exp: # already-aligned fixture: no applyEdit
|
|
continue
|
|
params = client.wait_server_request('workspace/applyEdit')
|
|
edits = params['edit']['changes'][uri]
|
|
check('alignTable %s' % f, apply_edits(ls, src, edits) == exp)
|
|
|
|
# align_too_wide must produce NO applyEdit (limit refusal) — verify via
|
|
# a message arriving with no pending applyEdit request.
|
|
src = open(fixture_dir + '/align_too_wide.kt').read()
|
|
client.notify('textDocument/didChange',
|
|
{'textDocument': {'uri': uri, 'version': 7},
|
|
'contentChanges': [{'text': src}]})
|
|
client.wait_notification('textDocument/publishDiagnostics')
|
|
client.notifications.clear() # drop showMessages from the loop above
|
|
pos = ls.offset_to_pos(src, src.index('|'))
|
|
client.request('workspace/executeCommand',
|
|
{'command': 'klammertext.alignTable',
|
|
'arguments': [{'uri': uri, 'position': pos}]})
|
|
msg = client.wait_notification('window/showMessage')
|
|
check('alignTable respects ROW_MAX (refusal message, no edit)',
|
|
'not aligning' in msg['message'] and not client.server_requests,
|
|
repr(msg))
|
|
|
|
# -- shutdown --
|
|
client.request('shutdown', None)
|
|
client.notify('exit', None)
|
|
code = client.proc.wait(timeout=10)
|
|
check('clean exit', code == 0, 'exit code %d' % code)
|
|
|
|
print('ls_test: %d passed, %d failed' % (PASS, FAIL))
|
|
sys.exit(1 if FAIL else 0)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|