Appuim并发多进程基于pytest测试框架进行兼容性测试

lsekfe

　前言
　　在实际工作中，如果要用appium实现多设备的兼容性测试，大家想到的也许是“多线程”，但由于python中GIL的影响，多线程并不能做到"多机并行"，这时候可以考虑使用多进程的方式。
　　为什么基于pytest
　　我们知道，pytest中的conftest.py可以定义不同的fixture，测试用例方法可以调用这些fixture，来实现数据共享。以前的框架的思路是：Common目录下的base_driver.py定义生成driver的方法-->conftest.py中调用前者生成driver-->TestCases下的测试用例调用fixture，来实现driver共享 。但是现在不同了，我们有多个设备，这些设备的信息如果只是单纯的写在yml中，我们并行去取的时候似乎也不方便，那可以写在哪里？conftest.py似乎也不是写设备信息的好地方，最后只剩下了main.py，而且将main.py作为多进程的入口再合适不过了。
　　但问题又来了，如果我们想启动多个appium服务，需要考虑以下几点：
　　1、appium通过什么方式启动？
　　2、设备信息如何传递给base_driver方法来生成driver
　　第一点很明确，客户端启动appium server的方式似乎有点不合时宜了，如果你要同时测5个手机，难道要一个个启动客户端吗？最好的方式是启动命令行，因为命令行启动更方便更快。
　　再说第二点前，先整理一下思路：main.py定义多个设备信息-->base_driver方法调用，生成多个driver-->TestCases下的测试用例调用fixture，但是设备信息怎么传递给base_driver方法呢？这时候pytest中的pytestconfig就派上用场了。
　　使用pytestconfig
　　内置的pytestconfig可以通过命令行参数、选项、配置文件、插件、运行目录等方式来控制pytest。pytestconfig是request.config的快捷方式，它在pytest文档里有时候被称为"pytest配置对象"。
　　要理解pytestconfig是如何工作的，可以查看如何添加一个自定义的命令行选项，然后在测试用例中读取该选项。你可以直接从pytestconfig里读取自定义的命令行选项，但是，为了让pytest能够解析它，还需要使用hook函数pytest_addoption。
　　下面使用pytest的hook函数pytest_addoption添加几个命令行选项：

1. pytestconfig/conftest.py
   
2. 　　def pytest_addoption(parser):
   
3. 　　    parser.addoption("--myopt", action="store_true", help="some boolean option")
   
4. 　　    parser.addoption("--foo", action="store", default="bar", help="foo: bar or baz")

复制代码

接下来就可以在测试用例中使用这些选项了：

1. pytest/test_config.py
   
2. 　　import pytest
   
3. 　　def test_option(pytestconfig):
   
4. 　　    print("'foo' set to:", pytestconfig.getoption('foo'))
   
5. 　　    print("'myopt' set to:", pytestconfig.getoption('myopt'))

复制代码

让我们看看它是如何工作的：

1. E:\virtual_workshop\pytest-demo\test_demo7\pytestconfig>pytest -s -q test_config.py::test_config
   
2. 　　'foo' set to: bar
   
3. 　　'myopt' set to: False
   
4. 　　.
   
5. 　　1 passed in 0.02s
   
6. 　　E:\virtual_workshop\pytest-demo\test_demo7\pytestconfig>pytest -s -q --myopt test_config.py::test_config
   
7. 　　'foo' set to: bar
   
8. 　　'myopt' set to: True
   
9. 　　.
   
10. 　　1 passed in 0.01s
    
11. 　　E:\virtual_workshop\pytest-demo\test_demo7\pytestconfig>pytest -s -q --myopt --foo baz test_config.py::test_config
    
12. 　　'foo' set to: baz
    
13. 　　'myopt' set to: True
    
14. 　　.
    
15. 　　1 passed in 0.01s

复制代码

因为pytestconfig是一个fixture，所以它也可以被其他的fixture使用。如果你喜欢，也可以为这些选项创建fixture。

1. @pytest.fixture()
   
2. 　　def foo(pytestconfig):
   
3. 　　    return pytestconfig.option.foo
   
4. 　　   
   
5. 　　@pytest.fixture()
   
6. 　　def myopt(pytestconfig):
   
7. 　　    return pytestconfig.option.myopt
   
8. 　　   
   
9. 　　def test_fixtures_for_options(foo, myopt):
   
10. 　　    print("'foo' set to: ", foo)
    
11. 　　    print("'myopt' set to: ", myopt)

复制代码

具体实现
　　定义main.py
　　既然可以使用pytest命令行参数了，那只需要在pytest.main中加上参数--cmdopt即可，main.py类似这样：

1. import pytest, os
   
2. 　　from multiprocessing import Pool
   
3. 　　device_infos = [{"platform_version": "5.1.1", "server_port": 4723, "device_port": 62001, "system_port": 8200},
   
4. 　　                {"platform_version": "7.1.2", "server_port": 4725, "device_port": 62025, "system_port": 8201}]
   
5. 　　def run_parallel(device_info):
   
6. 　　    pytest.main([f"--cmdopt={device_info}",
   
7. 　　                 "--alluredir", "Reports"])
   
8. 　　    os.system("allure generate Reports -o Reports/html --clean")
   
9. 　　if __name__ == "__main__":
   
10. 　　    with Pool(2) as pool:
    
11. 　　        pool.map(run_parallel, device_infos)
    
12. 　　        pool.close()
    
13. 　　        pool.join()

复制代码

为什么设备信息我只写了四个？platform_version、server_port、device_port、system_port。其他的类似于appPackage、appActivity、platformName等去哪了？当然你也可以写在这儿，其他的应该都是多个设备相同的，我写在yml的配置信息中了
　　·值得注意的是，这里的server_port多个设备不能重复，这是appium server启动的端口号，如果多个设备server_port都重复，那只能启动一个服务了，所以要不同
　　· system_port又是什么？这个是为了防止"互争互抢"现象的发生。多进程多设备并行时，如果多个设备同时使用同一个appium remote port（如8200）。对多个设备而言，它们并不知道相互使用同一port，因此就会出现多个设备发出的Request和接收的Action衔接不上而造成的测试混乱，可能会出现"Original error：Could not proxy command to remote server"的报错
　　定义Caps下的caps.yml
　　这里基本上定义的是多设备相同的desired_caps的公共部分：

1. platformName: Android
   
2. 　　appPackage: com.xxzb.fenwoo
   
3. 　　appActivity: com.xxzb.fenwoo.activity.addition.WelcomeActivity
   
4. 　　newCommonTimeout: 500
   
5. 　　noReset: False

复制代码

定义Common下的base_driver.py
　　这里有几点需要注意下：
　　·多进程在调用BaseDriver类的base_driver方法时，实例化时应该先通过命令行的方式启动appium server，设想一下，如果启动appium server放在get_base_driver中，会出现什么样的场景？conftest中每调用一次get_base_driver方法，就会打开一个cmd窗口，试图去启动appium server
　　· yaml.load方法注意新的写法，加上参数 Loader=yaml.FullLoader，这样据说更安全

1. from appium import webdriver
   
2. 　　from .conf_dir import caps_dir
   
3. 　　import yaml
   
4. 　　import os
   
5. 　　class BaseDriver:
   
6. 　　    def __init__(self, device_info):
   
7. 　　        self.device_info = device_info
   
8. 　　        cmd = "start appium -p {0} -bp {1} -U 127.0.0.1:{2}".format(self.device_info["server_port"], self.device_info["server_port"] + 1, self.device_info["device_port"])
   
9. 　　        os.system(cmd)
   
10. 　　    def base_driver(self, automationName="appium"):
    
11. 　　        fs = open(f"{caps_dir}//caps.yml")
    
12. 　　        #平台名称、包名、Activity名称、超时时间、是否重置、server_ip、
    
13. 　　        desired_caps = yaml.load(fs, Loader=yaml.FullLoader)
    
14. 　　        #版本信息
    
15. 　　        desired_caps["platform_version"] = self.device_info["platform_version"]
    
16. 　　        #设备名称
    
17. 　　        desired_caps["deviceName"] = f"127.0.0.1:{self.device_info['device_port']}"
    
18. 　　        #系统端口号
    
19. 　　        desired_caps["systemPort"] = self.device_info["system_port"]
    
20. 　　        if automationName != "appium":
    
21. 　　            desired_caps["automationName"] = automationName
    
22. 　　        driver = webdriver.Remote(f"http://127.0.0.1:{self.device_info['server_port']}/wd/hub", desired_capabilities=desired_caps)
    
23. 　　        return driver

复制代码

定义conftest.py
　　关键点是pytest_addoption和request.config.getoption这两个函数的使用，一个添加命令行，一个解析命令行，但仍有需要注意的：
　　eval(cmdopt)：之所以使用eval将cmdopt转为字典，是因为cmdopt本身是字符串，类似这样的："{'platform_version': '7.1.2', 'server_port': 4725, 'device_port': 62025, 'system_port': 8201}"，这样取值多不方便。
　　此外，还需要解决一个问题，如果有多个fixture，必须保证第一个测试用例用到的fixture实现BaseDriver的实例化，并且将这一实例化的结果base_driver作为全局变量，供所有的fixture共用，否则就会出现启动多个cmd窗口，启动多个appium server的问题。

1. from common.base_driver import BaseDriver
   
2. 　　import pytest
   
3. 　　driver = None
   
4. 　　def pytest_addoption(parser):
   
5. 　　    parser.addoption("--cmdopt", action="store", default="device_info", help=None)
   
6. 　　@pytest.fixture
   
7. 　　def cmdopt(pytestconfig):
   
8. 　　    #两种写法
   
9. 　　    return pytestconfig.getoption("--cmdopt")
   
10. 　　    #return pytestconfig.option.cmdopt
    
11. 　　#定义公共的fixture
    
12. 　　@pytest.fixture
    
13. 　　def common_driver(cmdopt):
    
14. 　　    global driver
    
15. 　　    base_driver = BaseDriver(eval(cmdopt))
    
16. 　　    driver = base_driver.base_driver()
    
17. 　　    yield driver
    
18. 　　    driver.close_app()
    
19. 　　    driver.quit()

复制代码

因为pytestconfig是request.config的快捷方式，所以cmdopt也可以写作

1. 　@pytest.fixture
   
2. 　　def cmdopt(request):
   
3. 　　    return request.config.getoption("--cmdopt")

复制代码

多进程运行
　　运行main.py，展示多进程运行的截图：

遗留问题
　　多进程兼容性测试也会带来一些问题：
　　·测试报告如何更好的区分多台设备。
　　· 对于分辨率不同的机型，要保证一些操作方法的健壮性和稳定性。如A手机屏幕大，确定按钮就在屏幕可见位置，B手机屏幕小，需要多次滑动才能看到按钮，这就要求定义方法时足够健壮。
　　· 业务逻辑问题。如果并行的去操作（调用同一个接口），会不会有业务逻辑上的限制，比如要抢一个免单券，一天同一个ip，同一个设备只能抢一件，这时候应该只会有一个成功，另一个无疑会失败。这就需要要么调整限制，要么调整方法。