Test against an external process with pytest-xprocess

echo_server.py

import socketserver
import sys
import time


class EchoTCPHandler(socketserver.StreamRequestHandler):
    def handle(self):
        data = self.rfile.readline().strip().decode('UTF-8')
        print('[ECHO SERVER] Received data: {}'.format(data))
        self.wfile.write('Echo: {}'.format(data).encode('UTF-8'))


def print_err(s):
    sys.stderr.write(s)
    sys.stderr.flush()


if __name__ == '__main__':

    default_port = 9998
    host, port = 'localhost', int(sys.argv[1]) if len(sys.argv) > 1 else default_port

    server = socketserver.TCPServer((host, port), EchoTCPHandler)
    print_err('Server starting...')

    startup_time = 3
    time.sleep(startup_time)

    print_err('Server startup completed!')
    server.serve_forever()

test_echo_server.py

import socket
import sys

import py
import pytest
from xprocess import ProcessStarter

server_name = 'echo-server'
hostname, port = 'localhost', 6777


@pytest.fixture(autouse=True)
def start_server(xprocess):
    python_executable_full_path = sys.executable
    python_server_script_full_path = py.path.local(__file__).dirpath("echo_server.py")

    class Starter(ProcessStarter):
        pattern = 'completed'
        args = [python_executable_full_path, python_server_script_full_path, str(port)]

    xprocess.ensure(server_name, Starter)

    yield
    
    xprocess.getinfo(server_name).terminate()


def test_echo_server():
    buffer_size = 4096

    sock = socket.socket()
    sock.connect((hostname, port))
    sock.sendall('hello\n'.encode('utf8'))
    c = sock.recv(buffer_size)
    assert c == 'Echo: hello'.encode('utf8')

test_echo_server_two_tests.py

# Imports, fixture not repeated here

def test_echo_server():
    _test_with_string('hello')


def test_echo_server_2():
    _test_with_string('hello world!')


def _test_with_string(s):
    buffer_size = 4096

    sock = socket.socket()
    sock.connect((hostname, port))
    sock.sendall('{}\n'.format(s).encode('utf8'))
    c = sock.recv(buffer_size)
    assert c == 'Echo: {}'.format(s).encode('utf8')

The code explanations can be found here: https://guillaumehumbert.blogspot.sg/2018/01/testing-against-external-processes-with.html