Appium-Pytest多开设备运行自动化用例

2021-09-24  本文已影响0人  Alan_9149

Windows-appium 环境配置

1. 安装node.js

2. 设置缓存
npm config set cache "D:\Program Files\nodejs
3. 设置全局模块存放路径
npm config set prefix "D:\Program Files\nodejs
4. 设置淘宝npm镜像

npm config set registry "https://registry.npm.taobao.org"
npm install -g cnpm --registry=https://registry.npm.taobao.org

5. 安装Android SDK 配置Android sdk环境

6. 安装Appium 运行cmd

7. 安装Appium-doctor

8. 配置系统cmd环境变量

项目目录结构说明

.
├─Common            ## 公共文件
    │ CheckPort.py           ## 检查Appium端口
    │ StartServer.py         ## 启动Appium Server
    │ AppDriver.py           ##  启用Driver Server
│
└─Config                     ##  配置文件
    │  cap.yaml              ## 启动Appium Server参数配置
    │  root_config.py        ## 系统常量配置
│
└─testcase     ## 测试用例
│
└─conftest.py  ## pytest fixture 文件
│
└─main.py  ## 运行用例文件

CheckPort.py

socket

socket通常也称作"套接字"，用于描述IP地址和端口，是一个通信链的句柄，应用程序通常通过"套接字"向网络发出请求或者应答网络请求。socket起源于Unix，而Unix/Linux基本哲学之一就是“一切皆文件”，对于文件用【打开】【读写】【关闭】模式来操作。socket就是该模式的一个实现，socket即是一种特殊的文件，一些socket函数就是对其进行的操作（读/写IO、打开、关闭）

import socket
import os


def check_port(host, port):
    """检测指定的端口是否被占用"""
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)  # 创建socket对象
    try:
        # 连接到address处的套接字。一般，address的格式为元组（hostname,port）
        s.connect((host, port))
        s.shutdown(2)
    except OSError:
        print('port %s is available! ' % port)
        return True
    else:
        print('port %s already be in use !' % port)
        return False


def release_port(port):
    """释放指定的端口"""
    cmd_find = 'netstat -aon | findstr {}'.format(port)  # 查找对应端口的pid
    # 返回命令执行后的结果
    result = os.popen(cmd_find).read()
    if str(port) and 'LISTENING' in result:
        # 获取端口对应的pid进程
        i = result.index('LISTENING')
        start = i + len('LISTENING') + 7
        end = result.index('\n')
        pid = result[start:end]
        cmd_kill = 'taskkill -f -pid %s' % pid  # 关闭被占用端口的pid
        os.popen(cmd_kill)
        print(f'释放进程端口:{port}')
    else:
        print('port %s is available !' % port)


if __name__ == '__main__':
    host = '127.0.0.1'
    port = 4723
    release_port(4723)
    release_port(4725)
    if not check_port(host, port):
        print("端口被占用")
        release_port(port)

StartServer.py

subprocess Popen类
用于在一个新的进程中执行一个子程序;即允许你去创建一个新的进程让其执行另外的程序，并与它进行通信，获取标准的输入、标准输出、标准错误以及返回码等。
主要参数说明：

  • args：用于指定进程的可执行文件及其参数。如果是一个序列类型参数，则序列的第一个元素通常都必须是一个可执行文件的路径。当然也可以使用executeable参数来指定可执行文件的路径。
  • stdin,stdout,stderr：分别表示程序的标准输入、标准输出、标准错误。有效的值可以是PIPE，存在的文件描述符，存在的文件对象或None，如果为None需从父进程继承过来，stdout可以是PIPE，表示对子进程创建一个管道，stderr可以是STDOUT，表示标准错误数据应该从应用程序中捕获并作为标准输出流stdout的文件句柄。
  • shell：如果这个参数被设置为True，程序将通过shell来执行。
  • env：它描述的是子进程的环境变量。如果为None，子进程的环境变量将从父进程继承而来。
import subprocess
import time
from config.root_config import LOG_DIR


def appium_start(host, port, log_name):
    """
    启动appium server
    :param host:
    :param port:
    :param log_name:
    :return:
    """
    # 指定bp端口号
    bootstrap_port = str(port + 1)
    # cmd命令
    cmd = 'start /b appium -a ' + host + ' -p ' + str(port) + ' -bp ' + str(bootstrap_port)
    # 去掉 “/b”，即可以打开cmd弹窗运行
    # cmd = 'start  appium -a ' + host+' -p '+str(port) +' -bp '+ str(bootstrap_port)
    print("%s at %s " % (cmd, time.ctime()))
    # cmd 需要执行的shell命令
    subprocess.Popen(cmd, shell=True, stdout=open(LOG_DIR + '/appium_log/' + f'{log_name}.log', 'w',encoding='utf8'),
                     stderr=subprocess.STDOUT)


if __name__ == '__main__':
    host = '127.0.0.1'
    port = 4723

conftest.py 配置文件

Hook 方法之 pytest_addoption ：
pytest_addoption 可以让用户注册一个自定义的命令行参数，方便用户将数据传递给 pytest；
这个 Hook 方法一般和 内置 fixture pytestconfig 配合使用，pytest_addoption 注册命令行参数，pytestconfig 通过配置对象读取参数的值；

parser.addoption() 参数说明：

  • name：自定义命令行参数的名字，可以是："foo"， "-foo" 或 "--foo"；
  • action：在命令行中遇到此参数时要采取的基本操作类型；
  • default：如果参数不在命令行中，则生成的默认值。
  • help：对参数作用的简要说明；
import time
import pytest
from Common.app_driver import BaseDriver
from Common.check_port import release_port

base_driver = None

# 注册自定义参数cmdopt到配置对象
def pytest_addoption(parser):
    parser.addoption("--cmdopt", action="store", default="device_info", help=None)

# 从配置对象获取cmdopt的值
@pytest.fixture(scope="session")
def cmd_opt(request):
    return request.config.getoption("--cmdopt")

# 全局启动appium driver
@pytest.fixture(scope="session")
def common_driver(cmd_opt):
    cmd_opt = eval(cmd_opt)
    global base_driver
    base_driver = BaseDriver(cmd_opt)
    time.sleep(1)
    driver = base_driver.get_base_driver()
    driver.implicitly_wait(10)
    yield driver
    driver.quit() # 退出driver
    release_port(cmd_opt["server_port"]) # 释放端口

AppDriver.py

import yaml
from appium import webdriver
from Common.check_port import check_port, release_port
from Common.start_server import appium_start
from config.root_config import CONFIG_PATH


class BaseDriver(object):

    def __init__(self, device_info):
        self.device_info = device_info
        self.host = self.device_info["server_host"]
        self.port = int(self.device_info["server_port"])
        if not check_port(self.host, self.port):
            release_port(self.port)
        appium_start(self.host, self.port, self.device_info['title'])
        with open(CONFIG_PATH, 'r') as f:
            self.data = yaml.load(f, Loader=yaml.FullLoader)

    def get_base_driver(self):
        """获取driver"""
        caps = self.data[self.device_info['title']]
        # 启用webdriver服务
        driver = webdriver.Remote(
            command_executor='http://' + self.host + ':' + str(self.port) + '/wd/hub',
            desired_capabilities=caps # 负责启动服务端时的参数设置
        )
        return driver


if __name__ == '__main__':
    device_info = {"title": "Emulator_two", "server_host": "127.0.0.1", "server_port": "4725", }
    base = BaseDriver(device_info)

main.py

import os
import threading
import time
from multiprocessing import Pool
import pytest
from config.root_config import ROOT_DIR

"""多设备启动文件"""
# appium 端口
device_infos = [
    {"title": "Emulator_one", "server_host": "127.0.0.1", "server_port": "4723", },
    {"title": "Emulator_two", "server_host": "127.0.0.1", "server_port": "4725", }
]
detail_report_path = ROOT_DIR + "\\report\\html"
xml_report_path = ROOT_DIR + "\\report\\xml"


def main(device_info):
    pytest.main(["--cmdopt={}".format(device_info),
                 "--alluredir", xml_report_path, "-vs"])
    os.system("allure generate %s -o %s --clean" % (xml_report_path, detail_report_path))


if __name__ == '__main__':
    # 创建进程池
    with Pool(len(device_infos)) as pool:
        # device_infos:要处理的数据列表，main：处理device_infos列表中数据的函数
        pool.map(main, device_infos) # 并行
        pool.close() # 关闭进程池，不再接受新的进程
        pool.join() #  主进程阻塞等待子进程的退出
上一篇 下一篇

猜你喜欢

热点阅读