Test against an external processes with dependencies, with pytest-xprocess

conftest.py

import sys

import py
import pytest
from xprocess import ProcessStarter

from servers import ItalicTCPHandler, EchoTCPHandler

python_executable_full_path = sys.executable
python_server_script_full_path = py.path.local(__file__).dirpath("servers.py")


def start_server(xprocess, server_class):
    class Starter(ProcessStarter):
        pattern = 'completed'
        args = [python_executable_full_path, python_server_script_full_path, server_class.__name__]

    xprocess.ensure(server_class.__name__, Starter)


def stop_server(xprocess, server_class):
    xprocess.getinfo(server_class.__name__).terminate()


@pytest.fixture(autouse=True, scope='session')
def start_italic_server(xprocess):
    start_server(xprocess, ItalicTCPHandler)
    yield
    stop_server(xprocess, ItalicTCPHandler)


@pytest.fixture(autouse=True, scope='session')
def start_echo_server(xprocess, start_italic_server):
    start_server(xprocess, EchoTCPHandler)
    yield
    stop_server(xprocess, EchoTCPHandler)

servers.py

import socket
import socketserver
import sys
import time


class EchoTCPHandler(socketserver.StreamRequestHandler):
    host, port = 'localhost', 6667

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

    @staticmethod
    def _query_server(handler_class, data):
        buffer_size = 4096
        with socket.socket() as sock:
            sock.connect((handler_class.host, handler_class.port))
            sock.sendall('{}\n'.format(data).encode('UTF-8'))
            data = sock.recv(buffer_size)
            return data


class ItalicTCPHandler(socketserver.StreamRequestHandler):
    host, port = 'localhost', 6668

    def handle(self):
        data = self.rfile.readline().strip().decode('UTF-8')
        print('[ITALIC SERVER] Received data: {}'.format(data))
        self.wfile.write('<i>{}</i>'.format(data).encode('UTF-8'))


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


if __name__ == '__main__':
    # Obviously not safe, only for simplicity purpose
    server_class = eval(sys.argv[1])

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

from servers import EchoTCPHandler


def test_echo_server():
    _test_with_string('hello')


def _test_with_string(s):
    buffer_size = 4096

    sock = socket.socket()
    sock.connect((EchoTCPHandler.host, EchoTCPHandler.port))
    sock.sendall('{}\n'.format(s).encode('UTF-8'))
    c = sock.recv(buffer_size)
    assert c == 'Echo: <i>{}</i>'.format(s).encode('UTF-8')

test_echo_server_2.py

import socket

from servers import EchoTCPHandler


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


def _test_with_string(s):
    buffer_size = 4096

    sock = socket.socket()
    sock.connect((EchoTCPHandler.host, EchoTCPHandler.port))
    sock.sendall('{}\n'.format(s).encode('UTF-8'))
    c = sock.recv(buffer_size)
    assert c == 'Echo: <i>{}</i>'.format(s).encode('UTF-8')

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