Appium + mitmProxy 实现APP接口稳定性测试

随着 App 用户量的不断增长，任何小的问题都可能放大成严重的线上事故，为了避免对App造成损害的任何可能性，我们必须从各个方面去思考 App 的稳定性建设，尽可能减少任何潜在的威胁。

1.背景介绍

为了保障 App 的稳定性，我们现在有 XMoney 智能遍历测试（崩溃、界面错乱、加载异常等）、UI 自动化（崩溃和业务逻辑验证）、Top1000 小程序遍历（崩溃和业务逻辑报错）、接口稳定性建设（崩溃和业务逻辑验证）。今天要给大家介绍的是接口稳定性建设，就是在后端返回数据如果不可靠的情况下，App 是否依然可以稳定运行。

2.实现思路

方案一

Mock有专门的Mock平台，崩溃有专门的崩溃监控平台，我们只需要把Mock数据下发到手机，如果有崩溃，会被崩溃平台收集到，崩溃会由值班同学分发给研发。

将接口代理到Mock平台，Mock平台根据response 生成批量的Mock数据
通过Appium调用App到制定页面，触发对应请求，请求到了Mock平台后，Mock平台随机下发Mock数据
重复打开指定页面，直到遍历完Mock数据
崩溃平台进行崩溃分发（人工）

优点：

短时间内可以覆盖大量的mock数据。

缺点：

mock数据不可控。

方案二

方案一有太多的随机性，为了让下发的数据和下发数据后的结果能够对应，我们为了控制下发的数据，我们在方案一的基础上自建Mock server，通过Appium脚本和Mock server的交互，控制下发数据。

经调研，Mock server 比较大众的方案就是采用[mitmproxy](https://github.com/mitmproxy/mitmproxy), [mitmproxy](https://github.com/mitmproxy/mitmproxy)可实现 python 脚本的注入，通过 python 脚本可以拦截请求的 request 和 response 数据，然后篡改数据后返回。因为 mitmproxy 是一个shell工具，无法直接和python脚本交互，所以python脚本和mitproxy的交互方式采用的是.ini的形式，互相读取 .ini 文件中的内容来实现数据交互。

优点：

mock数据可控制，可以校验预期结果。

缺点：

需要准备mock数据， mock数据要经常维护。

2种方案我们都有应用，方案一较为简单，在集成测试阶段执行，我们不再展开介绍。方案二稍微麻烦一些，回归测试阶段执行。以下将详细介绍下方案二如何实现。

3.代码设计

方案流程

代码示例

测试Case

通过Mock更新数据，测试小程序升级的逻辑 (因涉及代码较多，大家可通过完整系统查看)

测试小程序的同步升级和异步升级逻辑，修改测试配置后，mitmproxy会监控对应的接口请求，发现存在要修改数据的请求，则进行response修改后返回给手机端。手机端验证是否升级成功。

步骤：

修改测试配置（通过ini文件和mitmproxy实现联动， mitmproxy会在每次触发接口请求时检测当前的测试配置文件）
打开手机app
打开某个小程序

验证是否升级成功（示例为把官方小程序Demo升级成为携程小程序）

""" @File : test_update.py @Author : YangTongGang @Desc : 小程序升级（同步更新、异步更新） """

class TestUpdate(BaseTestCases): """检测小程序同步更新、异步更新已适配(iOS & Android) """ def setUp(self) -> None: # super(TestUpdate, self).setUp() self.business.del_sf_app('智能小程序') if self.is_android: self.business.del_sf_mine_app('智能小程序')

def __update_action(self, is_sync=True):# 把CTS升级成携程, 并修改max age = 0# 控制mitmproxy进行不同的测试if is_sync:test_type = TestType.TestMaxAgeZeroelse:test_type = TestType.TestSaveResponseself.changeTestType(test_type)protocol = CaseManager.official_demo_case().protocolself.business.open_swan(protocol)if is_sync:test_type = TestType.TestSyncUpdateelse:test_type = TestType.TestAsyncUpdateself.changeTestType(test_type)def test_max_age_force_update(self):"""测试小程序同步更新"""self.__update_action()self.business.open_app()protocol = CaseManager.official_demo_case().protocolif self.business.open_swan(protocol):self.assert_validate('汽车票')

Mitmproxy 启动

通过shell脚本启动mitmproxy并加载 http_handle.py文件，加载的python文件通过重写response或者request来达到监控每次请求的request和response的能力。

每次收到测试任务时，测试任务都会有相应的配置，如果发现需要启动代理服务，则会自动调用如下脚本，启动代理，然后手动配置测试机代理到服务地址，或者把任务打到固定的测试机上。

#!/usr/bin/env bash
: '
@File   : start_ui_mock.sh
@Author : YangTongGang
@Desc   : 启动UI Mock服务
'script_file=/CaseTester/HTTPRouter/http_router.pypath=$0
current_path=${path%/*}
current_path=${current_path%/*}
# 启动mock 数据服务
file=/SHELL/start_mock.sh
file_path=${current_path}${file}
chmod +x "${file_path}"
sh "${file_path}" ${script_file}#!/usr/bin/env bash
: '
@File   : start_mock.sh
@Author : YangTongGang
@Desc   : 启动接口Mock服务
'path=$0
file_path=$1
current_path=${path%/*}
current_path=${current_path%/*}
file_path=${current_path}'/..'${file_path}
echo "${file_path}"local_ip=$(ifconfig -a|grep inet|grep -v 127.0.0.1|grep -v inet6|awk '{print $2}'|tr -d "addr:"|head -n 1)
echo '=========== 代理服务地址 ============'
echo 'Script:' "${file_path##*/}"
echo ''
echo "${local_ip}":8088
echo ''
echo '==================================='# mitmdump -s "${file_path}" -p 8088  --flow-detail 0
status=$(mitmdump -s ${file_path} -p 8088  --flow-detail 0)
if [$? == 0]
thenexit 0
fi

监控response

注册对应的接口处理模块，遇到对应的接口时执行对应的测试。

"""
@File   : http_router.py
@Author : YangTongGang
@Desc   : 请求中转站
"""# 注册handle
HANDLES = [# Core更新请求UpdateCoreHandle,# APS拉包请求PkgAPSHandle,# PMS拉包请求PkgPMSHandle,# Update接口UpdateHandle,# 我的页面， 下拉二楼的静默更新个数GetPkgListHandle
]def response(flow):"""接口response"""url = flow.request.urlpath = urlparse.urlsplit(url)['path']proxy_mock_urls = []for handle in HANDLES:proxy_mock_urls.append(handle.identifier)test_type = get_proxy_test_type()if test_type == TestType.TestNone:returnif path not in proxy_mock_urls:returnresponse_dic = json.loads(flow.response.content)for handle in HANDLES:if handle.identifier in path:handler = handle(response_dic, test_type)response_dic = handler.package_dicflow.response.content = bytes(json.dumps(response_dic), encoding='utf8')return