Mercurial > vim
view src/testdir/test_channel_lsp.py @ 34686:83875247fbc0 v9.1.0224
patch 9.1.0224: cursor may move too many lines over "right" & "below" virt text
Commit: https://github.com/vim/vim/commit/515f734e687f28f7199b2a8042197624d9f3ec15
Author: Dylan Thacker-Smith <dylan.ah.smith@gmail.com>
Date: Thu Mar 28 12:01:14 2024 +0100
patch 9.1.0224: cursor may move too many lines over "right" & "below" virt text
Problem: If a line has "right" & "below" virtual text properties,
where the "below" property may be stored first due to lack of
ordering between them, then the line height is calculated to
be 1 more and causes the cursor to far over the line.
Solution: Remove some unnecessary setting of a
`next_right_goes_below = TRUE` flag for "below" and "above"
text properties. (Dylan Thacker-Smith)
I modified a regression test I recently added to cover this case,
leveraging the fact that "after", "right" & "below" text properties are
being stored in the reverse of the order they are added in. The
previous version of this regression test was crafted to workaround this
issue so it can be addressed by this separate patch.
closes: #14317
Signed-off-by: Dylan Thacker-Smith <dylan.ah.smith@gmail.com>
Signed-off-by: Christian Brabandt <cb@256bit.org>
author | Christian Brabandt <cb@256bit.org> |
---|---|
date | Thu, 28 Mar 2024 12:15:03 +0100 |
parents | 41e2414d2886 |
children |
line wrap: on
line source
#!/usr/bin/env python # # Server that will accept connections from a Vim channel. # Used by test_channel.vim to test LSP functionality. # # This requires Python 2.6 or later. from __future__ import print_function import json import socket import sys import time import threading try: # Python 3 import socketserver except ImportError: # Python 2 import SocketServer as socketserver class ThreadedTCPRequestHandler(socketserver.BaseRequestHandler): def setup(self): self.request.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) def debuglog(self, msg): if self.debug: with open("Xlspserver.log", "a") as myfile: myfile.write(msg) def send_lsp_req(self, msgid, method, params): v = {'jsonrpc': '2.0', 'id': msgid, 'method': method} if len(params) != 0: v['params'] = params s = json.dumps(v) req = "Content-Length: " + str(len(s)) + "\r\n" req += "Content-Type: application/vscode-jsonrpc; charset=utf-8\r\n" req += "\r\n" req += s if self.debug: self.debuglog("SEND: ({0} bytes) '{1}'\n".format(len(req), req)) self.request.sendall(req.encode('utf-8')) def send_lsp_resp(self, msgid, resp_dict): v = {'jsonrpc': '2.0', 'result': resp_dict} if msgid != -1: v['id'] = msgid s = json.dumps(v) resp = "Content-Length: " + str(len(s)) + "\r\n" resp += "Content-Type: application/vscode-jsonrpc; charset=utf-8\r\n" resp += "\r\n" resp += s if self.debug: self.debuglog("SEND: ({0} bytes) '{1}'\n".format(len(resp), resp)) self.request.sendall(resp.encode('utf-8')) def send_wrong_payload(self): v = 'wrong-payload' s = json.dumps(v) resp = "Content-Length: " + str(len(s)) + "\r\n" resp += "Content-Type: application/vscode-jsonrpc; charset=utf-8\r\n" resp += "\r\n" resp += s self.request.sendall(resp.encode('utf-8')) def send_empty_header(self, msgid, resp_dict): v = {'jsonrpc': '2.0', 'id': msgid, 'result': resp_dict} s = json.dumps(v) resp = "\r\n" resp += s self.request.sendall(resp.encode('utf-8')) def send_empty_payload(self): resp = "Content-Length: 0\r\n" resp += "Content-Type: application/vscode-jsonrpc; charset=utf-8\r\n" resp += "\r\n" self.request.sendall(resp.encode('utf-8')) def send_extra_hdr_fields(self, msgid, resp_dict): # test for sending extra fields in the http header v = {'jsonrpc': '2.0', 'id': msgid, 'result': resp_dict} s = json.dumps(v) resp = "Host: abc.vim.org\r\n" resp += "User-Agent: Python\r\n" resp += "Accept-Language: en-US,en\r\n" resp += "Content-Type: application/vscode-jsonrpc; charset=utf-8\r\n" resp += "Content-Length: " + str(len(s)) + "\r\n" resp += "\r\n" resp += s self.request.sendall(resp.encode('utf-8')) def send_delayed_payload(self, msgid, resp_dict): # test for sending the hdr first and then after some delay, send the # payload v = {'jsonrpc': '2.0', 'id': msgid, 'result': resp_dict} s = json.dumps(v) resp = "Content-Length: " + str(len(s)) + "\r\n" resp += "\r\n" self.request.sendall(resp.encode('utf-8')) time.sleep(0.05) resp = s self.request.sendall(resp.encode('utf-8')) def send_hdr_without_len(self, msgid, resp_dict): # test for sending the http header without length v = {'jsonrpc': '2.0', 'id': msgid, 'result': resp_dict} s = json.dumps(v) resp = "Content-Type: application/vscode-jsonrpc; charset=utf-8\r\n" resp += "\r\n" resp += s self.request.sendall(resp.encode('utf-8')) def send_hdr_with_wrong_len(self, msgid, resp_dict): # test for sending the http header with wrong length v = {'jsonrpc': '2.0', 'id': msgid, 'result': resp_dict} s = json.dumps(v) resp = "Content-Length: 1000\r\n" resp += "\r\n" resp += s self.request.sendall(resp.encode('utf-8')) def send_hdr_with_negative_len(self, msgid, resp_dict): # test for sending the http header with negative length v = {'jsonrpc': '2.0', 'id': msgid, 'result': resp_dict} s = json.dumps(v) resp = "Content-Length: -1\r\n" resp += "\r\n" resp += s self.request.sendall(resp.encode('utf-8')) def do_ping(self, payload): time.sleep(0.2) self.send_lsp_resp(payload['id'], 'alive') def do_echo(self, payload): self.send_lsp_resp(-1, payload) def do_simple_rpc(self, payload): # test for a simple RPC request self.send_lsp_resp(payload['id'], 'simple-rpc') def do_rpc_with_notif(self, payload): # test for sending a notification before replying to a request message self.send_lsp_resp(-1, 'rpc-with-notif-notif') # sleep for some time to make sure the notification is delivered time.sleep(0.2) self.send_lsp_resp(payload['id'], 'rpc-with-notif-resp') def do_wrong_payload(self, payload): # test for sending a non dict payload self.send_wrong_payload() time.sleep(0.2) self.send_lsp_resp(-1, 'wrong-payload') def do_large_payload(self, payload): # test for sending a large (> 64K) payload self.send_lsp_resp(payload['id'], payload) def do_rpc_resp_incorrect_id(self, payload): self.send_lsp_resp(-1, 'rpc-resp-incorrect-id-1') self.send_lsp_resp(-1, 'rpc-resp-incorrect-id-2') self.send_lsp_resp(1, 'rpc-resp-incorrect-id-3') time.sleep(0.2) self.send_lsp_resp(payload['id'], 'rpc-resp-incorrect-id-4') def do_simple_notif(self, payload): # notification message test self.send_lsp_resp(-1, 'simple-notif') def do_multi_notif(self, payload): # send multiple notifications self.send_lsp_resp(-1, 'multi-notif1') self.send_lsp_resp(-1, 'multi-notif2') def do_msg_with_id(self, payload): self.send_lsp_resp(payload['id'], 'msg-with-id') def do_msg_specific_cb(self, payload): self.send_lsp_resp(payload['id'], 'msg-specific-cb') def do_server_req(self, payload): self.send_lsp_resp(201, {'method': 'checkhealth', 'params': {'a': 20}}) def do_extra_hdr_fields(self, payload): self.send_extra_hdr_fields(payload['id'], 'extra-hdr-fields') def do_delayed_payload(self, payload): self.send_delayed_payload(payload['id'], 'delayed-payload') def do_hdr_without_len(self, payload): self.send_hdr_without_len(payload['id'], 'hdr-without-len') def do_hdr_with_wrong_len(self, payload): self.send_hdr_with_wrong_len(payload['id'], 'hdr-with-wrong-len') def do_hdr_with_negative_len(self, payload): self.send_hdr_with_negative_len(payload['id'], 'hdr-with-negative-len') def do_empty_header(self, payload): self.send_empty_header(payload['id'], 'empty-header') def do_empty_payload(self, payload): self.send_empty_payload() def do_server_req_in_middle(self, payload): # Send a notification message to the client in the middle of processing # a request message from the client self.send_lsp_req(-1, 'server-req-in-middle', {'text': 'server-notif'}) # Send a request message to the client in the middle of processing a # request message from the client. self.send_lsp_req(payload['id'], 'server-req-in-middle', {'text': 'server-req'}) def do_server_req_in_middle_resp(self, payload): # After receiving a response from the client send the response to the # client request. self.send_lsp_resp(payload['id'], {'text': 'server-resp'}) def process_msg(self, msg): try: decoded = json.loads(msg) if 'method' in decoded: test_map = { 'ping': self.do_ping, 'echo': self.do_echo, 'simple-rpc': self.do_simple_rpc, 'rpc-with-notif': self.do_rpc_with_notif, 'wrong-payload': self.do_wrong_payload, 'large-payload': self.do_large_payload, 'rpc-resp-incorrect-id': self.do_rpc_resp_incorrect_id, 'simple-notif': self.do_simple_notif, 'multi-notif': self.do_multi_notif, 'msg-with-id': self.do_msg_with_id, 'msg-specific-cb': self.do_msg_specific_cb, 'server-req': self.do_server_req, 'extra-hdr-fields': self.do_extra_hdr_fields, 'delayed-payload': self.do_delayed_payload, 'hdr-without-len': self.do_hdr_without_len, 'hdr-with-wrong-len': self.do_hdr_with_wrong_len, 'hdr-with-negative-len': self.do_hdr_with_negative_len, 'empty-header': self.do_empty_header, 'empty-payload': self.do_empty_payload, 'server-req-in-middle': self.do_server_req_in_middle, 'server-req-in-middle-resp': self.do_server_req_in_middle_resp, } if decoded['method'] in test_map: test_map[decoded['method']](decoded) else: self.debuglog("Error: Unsupported method - " + decoded['method'] + "\n") else: self.debuglog("Error: 'method' field is not found\n") except ValueError: self.debuglog("Error: json decoding failed\n") def process_msgs(self, msgbuf): while True: sidx = msgbuf.find('Content-Length: ') if sidx == -1: # partial message received return msgbuf sidx += 16 eidx = msgbuf.find('\r\n') if eidx == -1: # partial message received return msgbuf msglen = int(msgbuf[sidx:eidx]) hdrend = msgbuf.find('\r\n\r\n') if hdrend == -1: # partial message received return msgbuf if msglen > len(msgbuf[hdrend + 4:]): if self.debug: self.debuglog("Partial message ({0} bytes)\n".format(len(msgbuf))) # partial message received return msgbuf if self.debug: self.debuglog("Complete message ({0} bytes) received\n".format(msglen)) # Remove the header msgbuf = msgbuf[hdrend + 4:] payload = msgbuf[:msglen] self.process_msg(payload) # Remove the processed message msgbuf = msgbuf[msglen:] def handle(self): self.debug = False self.debuglog("=== socket opened ===\n") msgbuf = '' while True: try: received = self.request.recv(4096).decode('utf-8') except socket.error: self.debuglog("=== socket error ===\n") break except IOError: self.debuglog("=== socket closed ===\n") break if received == '': self.debuglog("=== socket closed ===\n") break # Write the received lines into the file for debugging if self.debug: self.debuglog("RECV: ({0} bytes) '{1}'\n".format(len(received), received)) # Can receive more than one line in a response or a partial line. # Accumulate all the received characters and process one line at # a time. msgbuf += received msgbuf = self.process_msgs(msgbuf) class ThreadedTCPServer(socketserver.ThreadingMixIn, socketserver.TCPServer): pass def writePortInFile(port): # Write the port number in Xportnr, so that the test knows it. f = open("Xportnr", "w") f.write("{0}".format(port)) f.close() def main(host, port, server_class=ThreadedTCPServer): # Wait half a second before opening the port to test waittime in ch_open(). # We do want to get the port number, get that first. We cannot open the # socket, guess a port is free. if len(sys.argv) >= 2 and sys.argv[1] == 'delay': port = 13684 writePortInFile(port) time.sleep(0.5) addrs = socket.getaddrinfo(host, port, 0, 0, socket.IPPROTO_TCP) # Each addr is a (family, type, proto, canonname, sockaddr) tuple sockaddr = addrs[0][4] server_class.address_family = addrs[0][0] server = server_class(sockaddr[0:2], ThreadedTCPRequestHandler) ip, port = server.server_address[0:2] # Start a thread with the server. That thread will then start a new thread # for each connection. server_thread = threading.Thread(target=server.serve_forever) server_thread.start() writePortInFile(port) # Main thread terminates, but the server continues running # until server.shutdown() is called. try: while server_thread.is_alive(): server_thread.join(1) except (KeyboardInterrupt, SystemExit): server.shutdown() if __name__ == "__main__": main("localhost", 0)