summaryrefslogtreecommitdiffstats
path: root/src/testdir/test_channel_unix.py
diff options
context:
space:
mode:
authorLemonBoy <thatlemon@gmail.com>2022-04-04 15:46:58 +0100
committerBram Moolenaar <Bram@vim.org>2022-04-04 15:46:58 +0100
commitcc766a85f460ebb7f8c915508447548b5f5b99bc (patch)
tree063d92f755b260af22cd1d40bd95379b7339e9d2 /src/testdir/test_channel_unix.py
parent4829c1c9e9095a3303caec9af7d02f6547f6df0e (diff)
patch 8.2.4684: cannot open a channel on a Unix domain socketv8.2.4684
Problem: Cannot open a channel on a Unix domain socket. Solution: Add Unix domain socket support. (closes #10062)
Diffstat (limited to 'src/testdir/test_channel_unix.py')
-rw-r--r--src/testdir/test_channel_unix.py50
1 files changed, 50 insertions, 0 deletions
diff --git a/src/testdir/test_channel_unix.py b/src/testdir/test_channel_unix.py
new file mode 100644
index 0000000000..4836e26ab8
--- /dev/null
+++ b/src/testdir/test_channel_unix.py
@@ -0,0 +1,50 @@
+#!/usr/bin/env python
+#
+# Server that will accept connections from a Vim channel.
+# Used by test_channel.vim.
+#
+# This requires Python 2.6 or later.
+
+from __future__ import print_function
+from test_channel import ThreadedTCPServer, ThreadedTCPRequestHandler, \
+ writePortInFile
+import socket
+import threading
+import os
+
+try:
+ FileNotFoundError
+except NameError:
+ # Python 2
+ FileNotFoundError = (IOError, OSError)
+
+class ThreadedUnixServer(ThreadedTCPServer):
+ address_family = socket.AF_UNIX
+
+def main(path):
+ server = ThreadedUnixServer(path, ThreadedTCPRequestHandler)
+
+ # 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()
+
+ # Signal the test harness we're ready, the port value has no meaning.
+ writePortInFile(1234)
+
+ print("Listening on {0}".format(server.server_address))
+
+ # 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__":
+ try:
+ os.remove("Xtestsocket")
+ except FileNotFoundError:
+ pass
+ main("Xtestsocket")