我的情况是这样的：
因此，一个典型的testing用例将如下所示，在Windows主机上运行：
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) 我发现pytest-xdist在某种程度上能够做到这一点。
我有什么好的教程或代码示例如何使用它？
我的最终设计使用ssh＆multiprocessing而不是xdist（在execute_tc（）函数中）：
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
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
我认为你已经改变了执行测试用例方法来执行特定的测试用例。