Appium python+appium 自动化测试-Appium 并发测试之 python 启动 appium 服务

Tin · July 17, 2021 · Last by Tin replied at August 02, 2021 · 3429 hits

来自 APP Android 端自动化测试初学者的笔记,写的不对的地方大家多多指教哦

python 启动 appium 服务,需要使用 subprocess 模块,该模块可以创建新的进程,并且连接到进程的输入、输出、错误等管道信息,并且可以获取进程的返回值。

测试场景

使用 python 启动 2 台 appium 服务,端口配置如下:

  • Appium 服务器端口:4723,bp 端口为 4724
  • Appium 服务器端口:4725,bp 端口为 4726

说明:bp 端口(—bootstrap-port)是 appium 和设备之间通信的端口,如果不指定到时无法操作多台设备运行脚本。

一、启动单个 appium 服务

代码实现

import subprocess
from time import ctime

def appium_start(host, port):
    """启动appium server"""
    bootstrap_port = str(port + 1)
    # /b表示后台运行命令行串口
    cmd = 'start /b appium -a ' + host + ' -p ' + str(port) + ' -bp ' + str(bootstrap_port)

    print('%s at %s' % (cmd, ctime()))
    subprocess.Popen(cmd, shell=True, stdout=open('../appium_log/' + str(port) + '.log', 'a'),
                     stderr=subprocess.STDOUT)

# 测试函数,在实际运行过程中可以注释
if __name__ == '__main__':
    host = '127.0.0.1'
    port = 4723
    appium_start(host, port)

运行成功后显示

启动校验

(一)启动后生成日志文件

(二)启动后我们需要校验服务是否启动成功,校验方法如下:

  1. 首先查看有没有生成对应的 log 文件,查看 log 里面的内容。
  2. 使用如下命令来查看,cmd 输入:
netstat -ano |findstr 端口号

netstat 命令解释

netstat 命令是一个监控 TCP/IP 网络的非常有用的工具,它可以显示路由表、实际的网络连接以及每一个网络接口设备的状态信息。输入 netstat -ano 回车.可以查看本机开放的全部端口;输入命令 netstat -h 可以查看全部参数含义。

关闭 Appium 服务

关闭进程有 2 种方式,具体如下:

1.通过 netstat 命令找到对应的 Appium 进程 pid 然后可以在系统任务管理器去关闭进程;win7 系统任务管理器 PID 显示

打开电脑的任务管理器 —> 详细信息,找到对应 pid 后,关闭进程即可

  1. 使用如下命令来关闭:
taskkill -f -pid appium进程pid
注意pid即通过上面netstat命令查看的状态信息如上图

关闭成功后如图所示:

对以上代码封装成类

# 启动单个appium服务
import subprocess
from time import ctime

class MultiAppium:

    def appium_start(self, host, port):
        """启动appium server"""
        bootstrap_port = str(port + 1)
        # /b表示后台运行命令行串口
        cmd = 'start /b appium -a ' + host + ' -p ' + str(port) + ' -bp ' + str(bootstrap_port)

        print('%s at %s' % (cmd, ctime()))
        subprocess.Popen(cmd, shell=True, stdout=open('./appium_log/' + str(port) + '.log', 'a'),
                         stderr=subprocess.STDOUT)

# 测试函数,在实际运行过程中可以注释
if __name__ == '__main__':
    host = '127.0.0.1'
    port = 4723
    appium_start1 = MultiAppium()
    appium_start1.appium_start(host, port)

二、启动多个 appium 服务

只需要在执行环境使用循环调用即可。

代码实现

import subprocess
from time import ctime

def appium_start(host, port):
    """启动appium server"""
    bootstrap_port = str(port + 1)
    # /b表示后台运行命令行串口
    cmd = 'start /b appium -a ' + host + ' -p ' + str(port) + ' -bp ' + str(bootstrap_port)

    print('%s at %s' % (cmd, ctime()))
    subprocess.Popen(cmd, shell=True, stdout=open('../appium_log/' + str(port) + '.log', 'a'),
                     stderr=subprocess.STDOUT)

# 测试函数,在实际运行过程中可以注释
if __name__ == '__main__':
    host = '127.0.0.1'
    for i in range(2):
        port = 4723 + 2 * i
        appium_start(host, port)

启动成功后显示:

用以上方法启动两个 appium 服务,由于显示启动时间较快,看不出区别,实际两个 appium 服务不是同时启动的

会生成两个日志文件:

对以上代码封装成类

# 启动多个appium服务
import subprocess
from time import ctime

class MultiAppium:

    def appium_start(self, host, port):
        """启动appium server"""
        bootstrap_port = str(port + 1)
        # /b表示后台运行命令行串口
        cmd = 'start /b appium -a ' + host + ' -p ' + str(port) + ' -bp ' + str(bootstrap_port)

        print('%s at %s' % (cmd, ctime()))
        subprocess.Popen(cmd, shell=True, stdout=open('./appium_log/' + str(port) + '.log', 'a'),
                         stderr=subprocess.STDOUT)

# 测试函数,在实际运行过程中可以注释
if __name__ == '__main__':
    # 启动多个appium服务
      for i in range(2):
        port = 4723 + 2 * i
        appium_start(host, port)

启动多个 appium 服务和单个 appium 服务区别在于运行时传入的 port 数量

三、多进程并发启动 appium 服务

python 多进程并发启动 appium 服务需要导入 multiprocessing 多进程模块

代码实现

# 多进程并发启动多个appium服务
import multiprocessing
import subprocess
from time import ctime

def appium_start_sync(host, port):
    """启动appium server"""
    bootstrap_port = str(port + 1)
    # /b表示后台运行命令行串口
    cmd = 'start /b appium -a ' + host + ' -p ' + str(port) + ' -bp ' + str(bootstrap_port)

    print('%s at %s' % (cmd, ctime()))
    subprocess.Popen(cmd, shell=True, stdout=open('./appium_log/' + str(port) + '.log', 'a'),
                     stderr=subprocess.STDOUT)

# 构建进程组
appium_process = []

# 加载appium进程
for i in range(2):
    host = '127.0.0.1'
    port = 4723 + 2 * i
    appium_sync = multiprocessing.Process(target=appium_start_sync, args=(host, port))
    appium_process.append(appium_sync)

# 测试函数,在实际运行过程中可以注释
if __name__ == '__main__':
    # 并发启动appium服务
    for appium in appium_process:
        appium.start()
    for appium in appium_process:
        appium.join()

结果同上,对以上代码封装成类:

# 多进程并发启动多个appium服务
import multiprocessing
import subprocess
from time import ctime

class AppiumStartSync:
    """启动appium server"""
    def appium_start_sync(self, host, port):
        bootstrap_port = str(port + 1)
        # /b表示后台运行命令行串口
        cmd = 'start /b appium -a ' + host + ' -p ' + str(port) + ' -bp ' + str(bootstrap_port)
        print('%s at %s' % (cmd, ctime()))
        subprocess.Popen(cmd, shell=True, stdout=open('./appium_log/' + str(port) + '.log', 'a'),
                         stderr=subprocess.STDOUT)

    # 构建进程组
    appium_process = []

    # 加载appium进程
    for i in range(2):
        host = '127.0.0.1'
        port = 4723 + 2 * i
        appium_sync = multiprocessing.Process(target=appium_start_sync, args=(host, port))
        appium_process.append(appium_sync)

# 测试函数,在实际运行过程中可以注释
if __name__ == '__main__':
    appium_start = AppiumStartSync()
    # 并发启动appium服务
    for appium in appium_start.appium_process:
        appium.start()
    for appium in appium_start.appium_process:
        appium.join()
共收到 4 条回复 时间 点赞

思路挺清晰的,不错。

根据自己的经验,提两个小建议:
1、既然用了 shell=True,可以考虑用 shell 命令,而非 cmd 命令,进而兼容 mac 系统?
2、对于端口的使用,python 有一个库叫做 freeport ,可以获取特定范围内空闲的端口号,以及判定指定端口号是否已在使用。可以考虑通过这个来增加端口号检测或自动分配?

陈恒捷 回复

对于端口这个我这样写不知道是否可以,这是我前面自动化里面的一个判断端口是否可用功能,还有实现了释放端口的功能

def check_port(host: str = '127.0.0.1', port: int = 4723):
    """
    function: 检测端口是否被占用,如果sk.connect连接成功, 表示端口已经被占用,如果连接失败,则表示端口未被占用
    :param host: 主机地址:'127.0.0.1'
    :param port: 端口: 4723
    :return: 返回端口是否可用
    """
    sk = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        sk.connect((host, port))
        sk.shutdown(2)
    except OSError:
        return True
    else:
        return False
def release_port(port: int):
    """
    :param port: 需要释放的端口
    :return: 返回True
    """
    cmd = "netstat -ano| findstr {}".format(port)
    result = exec_cmd(cmd)
    if "LISTENING" and str(port) in result:
        pid = result.strip().split(" ")[-1]
        cmd = "taskkill -f -pid {}".format(pid)
        exec_cmd(cmd)
        return True
    else:
        return True

不过都是基于 Windows cmd 命令
这是我前面基于 pytest 封装一个简单多设备运行
gitee 地址:https://gitee.com/King15800/pyAppium

king 回复

第一个端口是否已被占用的,不大明白为啥要有 OSError 异常才算有占用?

第二个释放端口的,直接 kill 有点太粗暴了。。。相比强制释放端口,使用邻近的其他可用端口会更合适。如果必须固定端口号,直接抛异常给用户,让用户操作更好。

至于所有都基于 windows cmd 命令,个人不是很推荐。这样会导致无法跨平台使用,以后搬上 jenkins 也必须是 windows 机器上的 jenkins 。除非确定未来都没有跨平台需要,否则用 shell 会更好。

Tin #4 · August 02, 2021 Author
陈恒捷 回复

感谢大佬指点

需要 Sign In 后方可回复, 如果你还没有账号请点击这里 Sign Up