Run part of Python unit tests on remote Linux machine from a Windows host

Question

My situation goes like this:

• I have a windows-based server program and a linux-based client.
• I have many tests for the linux clients which run and are required to run on local linux machines
• I need to run some code from windows server machine which will send some messages to the linux clients. Then, a test shall be executed on linux client machines which verifies the effect of those messages

So a typical test case would look like this, running on windows host:

test_example_message(self):
    # has to be executed locally on windows server
    send_message(EXAMPLE, hosts)
    # has to be executed locally on linux clients
    for host in hosts:
        verify_message_effect(EXAMPLE, host)

I have found that pytest-xdist is somehow able to do that.

I there any good tutorial or code example on how to use it?

Answer 1

def execute_tc(config):
"""Execute test code in parallel"""
    # create shared results dictionary
    manager = multiprocessing.Manager()
    result_dict = manager.dict({})
    # create processes
    processes = []
    for func, platform, args in config:
        function = getattr(test_functions, func)
        worker = multiprocessing.Process(target=function, args=[result_dict, platform, args])
        worker.daemon = True
        worker.start()
        processes.append(worker)

    for process in processes:
        process.join()

    return result_dict

I think you have change the execute test case method to execute particular test cases.

Answer 2

My final design utilizes ssh & multiprocessing instead of xdist (in the execute_tc() function):

import multiprocessing
import test_functions

def test_example_message(self):
"""Example test case"""
    # to get IPs, usernames, passwords, other required data
    config = get_test_config('example_message')
    # will take care of threading and executing parts
    result_dict = execute_tc(config)
    # to fail or not to fail. take care of proper reporting
    process_results(result_dict)

def execute_tc(config):
"""Execute test code in parallel"""
    # create shared results dictionary
    manager = multiprocessing.Manager()
    result_dict = manager.dict({})
    # create processes
    processes = []
    for func, platform, args in config:
        function = getattr(test_functions, func)
        worker = multiprocessing.Process(target=function, args=[result_dict, platform, args])
        worker.daemon = True
        worker.start()
        processes.append(worker)

    for process in processes:
        process.join()

    return result_dict