Test the ability to inherit sockets in multiprocessing on Windows (Python 3.6)

test_multiproc_socket_inherit.py

#!/usr/bin/env python
# -*- coding: utf-8; mode: python; py-indent-offset: 4; indent-tabs-mode: nil -*-
# vim: fileencoding=utf-8 tabstop=4 expandtab shiftwidth=4


"""
Test socket inheritance on Windows with multiprocessing.
"""

import time
import os
import socket
import multiprocessing

my_pid = os.getpid()


def log(msg):
    print("[%s] %s" % (my_pid, msg,))


def run_server():
    parent, child = socket.socketpair(family=socket.AF_INET, type=socket.SOCK_STREAM)
    parent.setblocking(1)
    child.setblocking(1)
    parent.set_inheritable(True)
    child.set_inheritable(True)

    proc_args = (child.fileno(),)
    child_process = multiprocessing.Process(
        target=run_child, args=proc_args)
    child_process.start()
    log("Launched child process")

    time.sleep(1)
    child.close()

    parent.send(b"HELLO\n")

    data_from_child = parent.recv(4)
    log("data from child: %s" % (data_from_child,))
    if data_from_child == b"BYE\n":
        log("Test succeeded.")
    else:
        log("Test failed: got: %s" % (data_from_child,))
    parent.close()
    child_process.join()


def run_child(child_socket_fileno):
    log("In child process with sockets %s" % (child_socket_fileno,))
    family = socket.AF_INET
    stype = socket.SOCK_STREAM
    child_sock = socket.fromfd(child_socket_fileno, family, stype)
    p_data = child_sock.recv(6)
    log("PARENT SAID: %s" % (p_data,))
    child_sock.send(b"BYE\n")
    child_sock.close()
    return


if __name__ == "__main__":
    run_server()