klammertext/tst/editor/ls_test.py

#!/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()