微信号:gh_be6ab0a8dfb7

介绍:国内首个移动测试交流社区,最专业的 Appium 交流社区.专注于移动互联网测试和 Web 相关测试技术研究.我们的理念:Coding Share Show Cool

硬核 | Tencent FAutoTest 源码解读

2019-06-14 07:58 孙圣翔

导读:最近微信小程序的 x5 内核貌似都关闭 debug 属性了,导致测试微信小程序有影响,也引发 TesterHome 社区多位技术专家的讨论[1]。本文作者分享了对 FAT 的硬核源码解读笔记,供耐心参考。文末有福利

01 前言

FAutoTest[2] 简称 FAT,主要用来解决微信内的 UI 的自动化测试问题,包括微信内的 H5 页面和小程序。不过这个项目已经 8 个月没更新了,关注度越来越低,笔者看到这种情况,实在是痛心。

因为这个项目的思路想法都还很不错的,直接通过 chrome dev tools 与小程序交互,而没有使用 chromedriver 这种中间层,稳定性应该更高一些。

这篇文章是我看这个项目的源码学习到的,希望能给大家带来帮助,希望腾讯官网也能够再次重视起来这个项目。

为了防止写文章的时候,突然又有提交了,或者下掉了,我先 Fork 一份代码到了这里 [3]

02 架构

下面直接从官方项目 README 中摘抄

代码结构设计

  1. 整体采用分层设计,API 设计方式参考 WebDriver

  2. 整体框架是一个同步阻塞的模型:在一个线程中循环的执行 receive 方法,等待收到 response,发送消息后,阻塞,只有当 receive 方法获得消息时,才会解除阻塞,发送下一条消息,具备超时异常处理机制

  3. 框架内打包了 Python 版本的 UIAutomator,方便在安卓 Native 页面进行操作

H5 页面/小程序 UI 自动化执行流程

03 使用

在 sample 目录下提供了 3 个使用的例子。考虑到长得都差不多,我们只看 H5Demo.py这个文件。

 
           
  1. from fastAutoTest.core.h5.h5Engine import H5Driver


  2. # http://h5.baike.qq.com/mobile/enter.html 从微信进入此链接,首屏加载完后执行脚本

  3. if __name__ == '__main__':

  4. h5Driver = H5Driver()

  5. h5Driver.initDriver()

  6. h5Driver.clickElementByXpath('/html/body/div[1]/div/div[3]/p')

  7. h5Driver.close()

先看 h5Driver=H5Driver() 这行的实现,需要打开 fastAutoTest/core/h5/h5Engine.py这个文件。

为了方便理解,很多代码我没有贴出来,也有些地方我稍微改了一点。
 
           
  1. # File: fastAutoTest/core/h5/h5Engine.py


  2. class H5Driver():

  3. def __init__(self, device=None):

  4. self.d = uiautomator.Device(device)

  5. self._pageOperator = H5PageOperator()

  6. self._networkHandler = ShortLiveWebSocket(self._webSocketDataTransfer, self._executor, self)

  7. # .... 没贴的代码 ....


  8. def clickElementByXpath(self, xpath):

  9. self.scrollToElementByXpath(xpath)

  10. sendStr = self._pageOperator.getElementRect(xpath)

  11. self._networkHandler.send(sendStr)


  12. x = self._getRelativeDirectionValue("x")

  13. y = self._getRelativeDirectionValue("y")


  14. self.logger.debug('clickElementByXpath --> x:' + str(x) + ' y:' + str(y))

  15. clickCommand = self._pageOperator.clickElementByXpath(x, y)

  16. return self._networkHandler.send(clickCommand)


这里的这个 self._pageOperator对象由 H5PageOperator()实例生成。先看刚才这段代码的最后两行

 
           
  1. clickCommand = self._pageOperator.clickElementByXpath(x, y)

  2. return self._networkHandler.send(clickCommand)

这个 _pageOperator实际上就是用来生成 clickCommand这个命令用的。继续向下追踪 clickElementByXpath的方式实现。

文件 fastAutoTest/core/h5/h5PageOperator.py

 
           
  1. from fastAutoTest.core.common.command.commandProcessor import CommandProcessor

  2. from fastAutoTest.core.h5 import h5UserAPI


  3. class H5PageOperator():

  4. processor = CommandProcessor('h5')


  5. def clickElementByXpath(self, x, y, duration=50, tapCount=1):

  6. params = {"x": x, "y": y, "duration": duration, "tapCount": tapCount}

  7. return self.processor.doCommandWithoutElement(h5UserAPI.ActionType.CLICK, **params)

其中的 h5UserAPI.ActionType.CLICK对应字符串 click

我们再看 CommandProcessor的实现。路径 fastAutoTest/core/common/command/commandProcessor.py

 
           
  1. import string


  2. from jsonConcat import JsonConcat

  3. from fastAutoTest.core.h5 import h5CommandManager, h5UserAPI

  4. from fastAutoTest.core.wx import wxCommandManager, wxUserAPI


  5. class CommandProcessor(object):

  6. def __init__(self, managerType):

  7. if managerType == 'h5':

  8. self.manager = h5CommandManager.H5CommandManager()

  9. self.userAPI = h5UserAPI

  10. self.concat = JsonConcat('h5')

  11. else:

  12. # ...


  13. def doCommandWithoutElement(self, actionType, **kwargs):

  14. return self.concat.concat(actionType, **kwargs)

这里有个 JsonContact 的实现需要查看下 路径 fastAutoTest/core/common/command/jsonConcat.py

 
           
  1. # file: fastAutoTest/core/common/command/jsonConcat.py


  2. from fastAutoTest.core.h5 import h5CommandManager

  3. from fastAutoTest.core.wx import wxCommandManager


  4. class JsonConcat():

  5. def __init__(self, managerType):

  6. if managerType == "h5":

  7. self.manager = h5CommandManager.H5CommandManager()

  8. else:

  9. self.manager = wxCommandManager.WxCommandManager()


  10. def concat(self, action_type, **params):

  11. method = self.manager.getMethod(action_type, None) # return: Input.synthesizeTapGesture

  12. if len(params) != 0:

  13. paramsTemplate = self.manager.getParams(method)

  14. paramsCat = string.Template(paramsTemplate)

  15. paramsResult = paramsCat.substitute(**params)

  16. paramsResult = json.loads(paramsResult)

  17. else:

  18. # 有 getDocument 这些不需要参数的情况

  19. paramsResult = "{}"

  20. result = dict()

  21. result['method'] = method

  22. result['params'] = paramsResult

  23. jsonResult = json.dumps(result)

  24. return jsonResult

这里的 contat 函数返回的是一个 json 字符串,我们用下面这些代码测试下

 
           
  1. from fastAutoTest.core.common.command.jsonConcat import JsonConcat

  2. JsonConcat("h5").concat("click", x=20, y=30, duration=500, tapCount=1)

  3. # output: {"params": {"y": 30, "x": 20, "duration": 500, "tapCount": 1}, "method": "Input.synthesizeTapGesture"}

文件 fastAutoTest/core/h5/h5CommandManager.py有各种各样的定义。

看到这里我也感觉代码真的有点绕了,去掉这些绕绕,我们之前提到的代码 clickCommand=self._pageOperator.clickElementByXpath(x,y) 等价于

 
           
  1. # self._pageOperator.clickElementByXpath(x, y)


  2. clickCommand = json.dumps({

  3. "method": "Input.synthesizeTapGesture",

  4. "params": {"x": 10, "y": 15, "duration": 500, "tapCount": 1},

  5. })

获取到命令之后,就是通过下面的代码 returnself._networkHandler.send(clickCommand)发送给手机的 websocket 进程了。

 
           
  1. from fastAutoTest.utils.singlethreadexecutor import SingleThreadExecutor

  2. from fastAutoTest.core.common.network.websocketdatatransfer import WebSocketDataTransfer

  3. from fastAutoTest.core.common.network.shortLiveWebSocket import ShortLiveWebSocket


  4. class H5Driver():

  5. def __init__(self):

  6. # ...

  7. self._executor = SingleThreadExecutor()

  8. self._webSocketDataTransfer = WebSocketDataTransfer(url=url)

  9. self._networkHandler = ShortLiveWebSocket(self._webSocketDataTransfer, self._executor, self)

跟 websocket 通讯的代码还真少。那个 SingleThreadExecutor的作用就是让函数可以顺序调用,感觉用处不大。

WebSocketDataTransfer 则主要负责 websocket 的通信。

我们只直接来看 self._networkHandler.send(clickCommand)的实现

 
           
  1. File: fastAutoTest/core/common/network/shortLiveWebSocket.py


  2. class ShortLiveWebSocket():

  3. # ... 此处省略无数代码 ...

  4. def __init__(self, ...., driver):

  5. self.driver = driver # H5Driver 实例

  6. # ...


  7. def send(self, data, timeout=60):

  8. self._waitForConnectOrThrow()


  9. # 微信小程序和 QQ 公众号都需要切换页面

  10. if self.driver.getDriverType() == WXDRIVER or self.driver.getDriverType == QQDRIVER:

  11. # 只有点击事件才会导致页面的切换

  12. if 'x=Math.round((left+right)/2)' in data:

  13. time.sleep(WAIT_REFLESH_1_SECOND)

  14. if self.driver.needSwitchNextPage():

  15. self.driver.switchToNextPage()


  16. # 增加 id 字段

  17. jdata = json.loads(data)

  18. jdata["id"] = self._id.getAndIncrement()

  19. data = json.dumps(data)


  20. self._currentRequest = _NetWorkRequset(self._id.getAndIncrement(), data)

  21. currentRequestToJsonStr = self._currentRequest.toSendJsonString()

  22. self.logger.debug(' ---> ' + currentRequestToJsonStr)


  23. # scroll 操作需要滑动到位置才会有返回,如果是 scroll 操作则等待,防止超时退出

  24. if 'synthesizeScrollGesture' in data:

  25. self._webSocketDataTransfer.send(currentRequestToJsonStr)

  26. self._retryEvent.wait(WAIT_REFLESH_40_SECOND)

  27. else:

  28. for num in range(0, SEND_DATA_ATTEMPT_TOTAL_NUM): # 这个 Num 的值是 7

  29. self.logger.debug(" ---> attempt num: " + str(num))


  30. if num != 3 and num != 5:

  31. self._webSocketDataTransfer.send(currentRequestToJsonStr)

  32. self._retryEvent.wait(3)

  33. else:

  34. self.driver.switchToNextPage()

  35. time.sleep(WAIT_REFLESH_2_SECOND)

  36. self.logger.debug('switch when request: ' + currentRequestToJsonStr)

  37. self._webSocketDataTransfer.send(currentRequestToJsonStr)

  38. self._retryEvent.wait(WAIT_REFLESH_2_SECOND)


  39. if self._readWriteSyncEvent.isSet():

  40. break


  41. self._readWriteSyncEvent.wait(timeout=timeout)

  42. self._readWriteSyncEvent.clear()

  43. self._retryEvent.clear()


  44. self._checkReturnOrThrow(self._currentRequest)


  45. return self._currentRequest

这里可以频繁的看到 switchToNextPage这个方法,其实这个方法将 websocket 断了,然后在重连。

 
           
  1. def switchToNextPage(self):

  2. """

  3. 把之前缓存的 html 置为 none

  4. 再重新连接 websocket

  5. """

  6. self.logger.debug('')

  7. self.html = None

  8. self._networkHandler.disconnect()

  9. self._networkHandler.connect()

04 总结

FAT 的代码逻辑太多,这里仅仅分析九牛一毛。接下来我应该会写一篇文章,用简洁一些的代码,讲述 FAT 的原理。花了很多时间去看,感觉还是挺有收获的。

References

[1] 原文链接: https://testerhome.com/topics/19446
[2] FAutoTest: https://github.com/Tencent/FAutoTest
[3] 源码解读: https://github.com/codeskyblue/FAutoTest

05 活动推荐

关于微信小程序的测试,TesterHome 联合腾讯 WeTest 邀请到腾讯技术架构部高级测试工程师谢锦辉分享《微信小程序质量体系议题,腾讯互娱品质管理部高级工程师童立舟分享《小程序全链路性能测试》议题。

另,本文作者孙圣翔,ATX 开源工具作者,阿里巴巴手淘测试开发技术专家,也将分享《ATX 在手淘自动化测试的实践》议题,期待在 MTSC2019 大会上现场交流。

MTSC2019 测试开发大会 | 倒计时

向顶级测试技术专家和质量大咖学习,全面提升软件质量,尽在 MTSC2019!

MTSC2019 第五届中国移动互联网测试开发大会将于 2019 年 6 月 28-29 日在北京国际会议中心举行,60+ 来自 Google、BAT、TMD 等一线互联网企业的测试大咖分享软件质量保障最佳实践,涵盖移动自动化测试、服务端测试、质量保障 QA、高新测试技术(AI+、大数据测试、IOT 测试),游戏测试,工程效率提升等专题。

MTSC2019 大会日程发布,门票即将售罄,预购从速!

大会官网http://2019.test-china.org/
报名地址
https://www.bagevent.com/event/2202999?bag_track=YT

TesterHome 福利时间

转发文章到朋友圈,抽奖赠送 TesterHome 定制版精美礼物(卫衣/图书等)、MTSC2019 测试开发大会门票或优惠折扣,以及其他福利等!


  • Step1: 转发本文到朋友圈,扫描二维码加小助手微信;

  • Step2: 回复「福利」入群抽奖;

  • SteP3: 回复「组团」入群,以团购优惠价购买 MTSC2019 门票;



P.S. 对于 MTSC 往届参会者,可享受 7.5 折福利,请进群咨询;


扫描二维码,或点击“阅读原文,抢购门票!

 
TesterHome 更多文章 利器 | 无线驱动的 Android 专项测试方案 Soloπ(开篇) MTSC2019 大会日程确定,60+ 议题揭秘软件测试技术进阶新风向! 为何软件质量如此重要? “史上最稳定双11”背后的全链路验收实践 MTSC2019 大会日程 V1.0 版公布,投票你最关注的议题领取福利!
猜您喜欢 #TW好文集锦# 精益创业和敏捷 文/施韵涛 Android 内存泄漏探讨——阿里百川 当我们在谈论bug时我们谈论的其实是 百度SSP单页式应用性能优化实践 程序员找女朋友的技术攻略(需求分析篇)