win系统airtest+pytest-xdist服务器分布式运行。

1.准备至少两台服务器，集群全部是局域网，（启动脚本的时候可以使用外网ip）。

2.输出的报告地址，需要把文件夹设置成共享文件夹，（连接的时候使用内外ip）。

启动脚本文件

import os, datetime, sys# sys包启动py文件的时候可以给文件入参运行

HOST1 = "服务器1"

HOST2 = "服务器2"

MASTER_HOST = r"\\127.0.0.1"#主服务器，需要换成内外ip，不建议使用127.0.0.1

MASTER_HOST_report_PATH=fr"{MASTER_HOST}\output\report"

date = datetime.datetime.now().strftime("%Y_%m_%d")

dates = datetime.datetime.now().strftime("%Y_%m_%d_%H_%M_%S")

BASE_PATH = os.path.dirname(__file__)

filename = dates

param = sys.argv[1]

if param == "polis" or param == "loudliveroom" or param == "fittingroom":

    os.mkdir(fr"{MASTER_HOST_report_PATH}\{param}\{filename}")

    os.system(

        fr'pytest -sv  -d --tx socket={HOST1}  --rsyncdir {BASE_PATH}\casetests  -m "polis" --alluredir={MASTER_HOST_report_PATH}\{param}\{filename}\allure')#输出的报告路径需要主服务器地址加共享的文件夹路径（否则报告不会收集起来）

    os.system(

        fr"allure generate {MASTER_HOST_report_PATH}\{param}\{filename}\allure -o {MASTER_HOST_report_PATH}\{param}\{filename}\report --clean")

elif param == "loudliveroom,fitting" or param == "loudliveroom,polis" or param == "fittingroom,polis" or param == "loudliveroom,fittingroom,polis":

    os.mkdir(fr"{MASTER_HOST_report_PATH}\all_report\{filename}")

    os.system(

        fr'pytest -sv  -d --tx socket={HOST1}   --rsyncdir {BASE_PATH}\casetests  -m "polis" --alluredir={MASTER_HOST_report_PATH}\all_report\{filename}\allure')

    os.system(

        fr"allure generate {MASTER_HOST_report_PATH}\all_report\{filename}\allure -o {MASTER_HOST_report_PATH}\all_report\{filename}\report --clean")

else:

    print("脚本启动错误")

3.最重要文件：需要用sockserver.py在各个服务器先执行包括主服务器和子服务器。文件放在项目根目录下。

#! /usr/bin/env python

# -*- coding: utf-8 -*-

"""

    start socket based minimal readline exec server

    it can exeuted in 2 modes of operation

    1. as normal script, that listens for new connections

    2. via existing_gateway.remote_exec (as imported module)

"""

# this part of the program only executes on the server side

#

import os

import sys

progname = "socket_readline_exec_server-1.2"

def get_fcntl():

    try:

        import fcntl

    except ImportError:

        fcntl = None

    return fcntl

fcntl = get_fcntl()

debug = 0

if debug:  # and not os.isatty(sys.stdin.fileno())

    f = open("/tmp/execnet-socket-pyout.log", "w")

    old = sys.stdout, sys.stderr

    sys.stdout = sys.stderr = f

def print_(*args):

    print(" ".join(str(arg) for arg in args))

if sys.version_info > (3, 0):

    exec(

        """def exec_(source, locs):

    exec(source, locs)"""

    )

else:

    exec(

        """def exec_(source, locs):

    exec source in locs"""

    )

def exec_from_one_connection(serversock):

    print_(progname, "Entering Accept loop", serversock.getsockname())

    clientsock, address = serversock.accept()

    print_(progname, "got new connection from %s %s" % address)

    clientfile = clientsock.makefile("rb")

    print_("reading line")

    # rstrip so that we can use \r\n for telnet testing

    source = clientfile.readline().rstrip()

    clientfile.close()

    g = {"clientsock": clientsock, "address": address, "execmodel": execmodel}

    source = eval(source)

    if source:

        co = compile(source + "\n", "<socket server>", "exec")

        print_(progname, "compiled source, executing")

        try:

            exec_(co, g)  # noqa

        finally:

            print_(progname, "finished executing code")

            # background thread might hold a reference to this (!?)

            # clientsock.close()

def bind_and_listen(hostport, execmodel):

    socket = execmodel.socket

    if isinstance(hostport, str):

        host, port = hostport.split(":")

        hostport = (host, int(port))

    serversock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    # set close-on-exec

    if hasattr(fcntl, "FD_CLOEXEC"):

        old = fcntl.fcntl(serversock.fileno(), fcntl.F_GETFD)

        fcntl.fcntl(serversock.fileno(), fcntl.F_SETFD, old | fcntl.FD_CLOEXEC)

    # allow the address to be re-used in a reasonable amount of time

    if os.name == "posix" and sys.platform != "cygwin":

        serversock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

    serversock.bind(hostport)

    serversock.listen(5)

    return serversock

def startserver(serversock, loop=False):

    execute_path = os.getcwd()

    try:

        while 1:

            try:

                exec_from_one_connection(serversock)

            except (KeyboardInterrupt, SystemExit):

                raise

            except:

                if debug:

                    import traceback

                    traceback.print_exc()

                else:

                    excinfo = sys.exc_info()

                    print_("got exception", excinfo[1])

            os.chdir(execute_path)

            if not loop:

                break

    finally:

        print_("leaving socketserver execloop")

        serversock.shutdown(2)

if __name__ == "__main__":

    import sys

    if len(sys.argv) > 1:

        hostport = sys.argv[1]

    else:

        hostport = ":8888"

    from execnet.gateway_base import get_execmodel

    execmodel = get_execmodel("thread")

    serversock = bind_and_listen(hostport, execmodel)

    startserver(serversock, loop=True)

elif __name__ == "__channelexec__":

    chan = globals()["channel"]

    execmodel = chan.gateway.execmodel

    bindname = chan.receive()

    sock = bind_and_listen(bindname, execmodel)

    port = sock.getsockname()

    chan.send(port)

    startserver(sock)

4.如果项目需要截图和用到图片识别的时候，记得把图片读取路径和存放路径放在（pyexecnetcache）文件路径下。

巴特西

win系统airtest+pytest-xdist服务器分布式运行。

最新文章

热门文章