src/testdir/test_channel_lsp.py - android_external_vim - Gitiles

 #!/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 send_lsp_msg(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/vim-jsonrpc; charset=utf-8\r\n"
         resp += "\r\n"
         resp += s
         if self.debug:
             with open("Xlspdebug.log", "a") as myfile:
                 myfile.write("\n=> send\n" + 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/vim-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/vim-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/vim-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_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/vim-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_msg(payload['id'], 'alive')

     def do_echo(self, payload):
         self.send_lsp_msg(-1, payload)

     def do_simple_rpc(self, payload):
         # test for a simple RPC request
         self.send_lsp_msg(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_msg(-1, 'rpc-with-notif-notif')
         # sleep for some time to make sure the notification is delivered
         time.sleep(0.2)
         self.send_lsp_msg(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_msg(-1, 'wrong-payload')

     def do_rpc_resp_incorrect_id(self, payload):
         self.send_lsp_msg(-1, 'rpc-resp-incorrect-id-1')
         self.send_lsp_msg(-1, 'rpc-resp-incorrect-id-2')
         self.send_lsp_msg(1, 'rpc-resp-incorrect-id-3')
         time.sleep(0.2)
         self.send_lsp_msg(payload['id'], 'rpc-resp-incorrect-id-4')

     def do_simple_notif(self, payload):
         # notification message test
         self.send_lsp_msg(-1, 'simple-notif')

     def do_multi_notif(self, payload):
         # send multiple notifications
         self.send_lsp_msg(-1, 'multi-notif1')
         self.send_lsp_msg(-1, 'multi-notif2')

     def do_msg_with_id(self, payload):
         self.send_lsp_msg(payload['id'], 'msg-with-id')

     def do_msg_specific_cb(self, payload):
         self.send_lsp_msg(payload['id'], 'msg-specifc-cb')

     def do_server_req(self, payload):
         self.send_lsp_msg(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_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 process_msg(self, msg):
         try:
             decoded = json.loads(msg)
             print("Decoded:")
             print(str(decoded))
             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,
                         '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-specifc-cb': self.do_msg_specific_cb,
                         'server-req': self.do_server_req,
                         'extra-hdr-fields': self.do_extra_hdr_fields,
                         '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
                         }
                 if decoded['method'] in test_map:
                     test_map[decoded['method']](decoded)
                 else:
                     print("Error: Unsupported method: " + decoded['method'])
             else:
                 print("Error: 'method' field is not found")

         except ValueError:
             print("json decoding failed")

     def process_msgs(self, msgbuf):
         while True:
             sidx = msgbuf.find('Content-Length: ')
             if sidx == -1:
                 return msgbuf
             sidx += 16
             eidx = msgbuf.find('\r\n')
             if eidx == -1:
                 return msgbuf
             msglen = int(msgbuf[sidx:eidx])

             hdrend = msgbuf.find('\r\n\r\n')
             if hdrend == -1:
                 return msgbuf

             # Remove the header
             msgbuf = msgbuf[hdrend + 4:]
             payload = msgbuf[:msglen]

             self.process_msg(payload)

             # Remove the processed message
             msgbuf = msgbuf[msglen:]

     def handle(self):
         print("=== socket opened ===")
         self.debug = False
         msgbuf = ''
         while True:
             try:
                 received = self.request.recv(4096).decode('utf-8')
             except socket.error:
                 print("=== socket error ===")
                 break
             except IOError:
                 print("=== socket closed ===")
                 break
             if received == '':
                 print("=== socket closed ===")
                 break
             print("\nReceived:\n{0}".format(received))

             # Write the received lines into the file for debugging
             if self.debug:
                 with open("Xlspdebug.log", "a") as myfile:
                     myfile.write("\n<= recv\n" + 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)

         print("Wait for it...")
         time.sleep(0.5)

     server = server_class((host, port), 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)

     print("Listening on port {0}".format(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)
	#!/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 send_lsp_msg(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/vim-jsonrpc; charset=utf-8\r\n"
	resp += "\r\n"
	resp += s
	if self.debug:
	with open("Xlspdebug.log", "a") as myfile:
	myfile.write("\n=> send\n" + 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/vim-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/vim-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/vim-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_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/vim-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_msg(payload['id'], 'alive')

	def do_echo(self, payload):
	self.send_lsp_msg(-1, payload)

	def do_simple_rpc(self, payload):
	# test for a simple RPC request
	self.send_lsp_msg(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_msg(-1, 'rpc-with-notif-notif')
	# sleep for some time to make sure the notification is delivered
	time.sleep(0.2)
	self.send_lsp_msg(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_msg(-1, 'wrong-payload')

	def do_rpc_resp_incorrect_id(self, payload):
	self.send_lsp_msg(-1, 'rpc-resp-incorrect-id-1')
	self.send_lsp_msg(-1, 'rpc-resp-incorrect-id-2')
	self.send_lsp_msg(1, 'rpc-resp-incorrect-id-3')
	time.sleep(0.2)
	self.send_lsp_msg(payload['id'], 'rpc-resp-incorrect-id-4')

	def do_simple_notif(self, payload):
	# notification message test
	self.send_lsp_msg(-1, 'simple-notif')

	def do_multi_notif(self, payload):
	# send multiple notifications
	self.send_lsp_msg(-1, 'multi-notif1')
	self.send_lsp_msg(-1, 'multi-notif2')

	def do_msg_with_id(self, payload):
	self.send_lsp_msg(payload['id'], 'msg-with-id')

	def do_msg_specific_cb(self, payload):
	self.send_lsp_msg(payload['id'], 'msg-specifc-cb')

	def do_server_req(self, payload):
	self.send_lsp_msg(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_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 process_msg(self, msg):
	try:
	decoded = json.loads(msg)
	print("Decoded:")
	print(str(decoded))
	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,
	'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-specifc-cb': self.do_msg_specific_cb,
	'server-req': self.do_server_req,
	'extra-hdr-fields': self.do_extra_hdr_fields,
	'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
	}
	if decoded['method'] in test_map:
	test_map[decoded['method']](decoded)
	else:
	print("Error: Unsupported method: " + decoded['method'])
	else:
	print("Error: 'method' field is not found")

	except ValueError:
	print("json decoding failed")

	def process_msgs(self, msgbuf):
	while True:
	sidx = msgbuf.find('Content-Length: ')
	if sidx == -1:
	return msgbuf
	sidx += 16
	eidx = msgbuf.find('\r\n')
	if eidx == -1:
	return msgbuf
	msglen = int(msgbuf[sidx:eidx])

	hdrend = msgbuf.find('\r\n\r\n')
	if hdrend == -1:
	return msgbuf

	# Remove the header
	msgbuf = msgbuf[hdrend + 4:]
	payload = msgbuf[:msglen]

	self.process_msg(payload)

	# Remove the processed message
	msgbuf = msgbuf[msglen:]

	def handle(self):
	print("=== socket opened ===")
	self.debug = False
	msgbuf = ''
	while True:
	try:
	received = self.request.recv(4096).decode('utf-8')
	except socket.error:
	print("=== socket error ===")
	break
	except IOError:
	print("=== socket closed ===")
	break
	if received == '':
	print("=== socket closed ===")
	break
	print("\nReceived:\n{0}".format(received))

	# Write the received lines into the file for debugging
	if self.debug:
	with open("Xlspdebug.log", "a") as myfile:
	myfile.write("\n<= recv\n" + 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)

	print("Wait for it...")
	time.sleep(0.5)

	server = server_class((host, port), 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)

	print("Listening on port {0}".format(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)