test_ipcs.py

import os
import socket
import time
import atexit

# Fork the process
pid = os.fork()

if pid == 0:
    # This is the child process
    print("Hello from child process!")
    atexit.register(os.unlink, '/tmp/mysocket.sock')
    
    # Create a Unix socket
    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    
    # Bind the socket to a file in the filesystem
    socket_path = '/tmp/mysocket.sock'
    sock.bind(socket_path)
    
    # Start listening for connections
    sock.listen(1)

    # Wait for the parent process to connect
    conn, _ = sock.accept()
    
    # Receive a message from the parent process
    message = conn.recv(1024).decode('utf-8')
    print("Received message from parent process:", message)
    time.sleep(1)
    # Send a message to the parent process
    message = "Hello from child process!"
    conn.sendall(message.encode('utf-8'))
    
    # Close the connection and the socket
    conn.close()
    sock.close()

else:
    # This is the parent process
    print("Hello from parent process!")
    
    # Wait for the child process to create the Unix socket
    while not os.path.exists('/tmp/mysocket.sock'):
        print("Checking for socket...")

    # Connect to the child process's Unix socket
    conn = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    conn.connect('/tmp/mysocket.sock')

    # Send a message to the child process
    message = "Hello from parent process!"
    conn.sendall(message.encode('utf-8'))
    
    # Receive a message from the child process
    message = conn.recv(1024).decode('utf-8')
    print("Received message from child process:", message)
    
    # Close the connection
    conn.close()