自动化桩的落地实现-Go-Tester

前言

之前在公司接触的项目，会遇到被测服务需要依赖三方服务才能正常处理业务，但三方服务却无法在测试环境接入的场景。一开始我们提出桩的概念，可以理解为假想的服务，去替代依赖服务，模拟与被测服务的场景交互。但后面需要考虑校验异常场景，又实现了自动化桩，也就是将校验依赖服务异常测试用例转化成自动化脚本。在这里总结下自动化桩遇到的痛点问题和设计实现过程。

存在的痛点

三方服务异常的处理行为覆盖不到。开发一般会提供初级桩，只能保证通信过程，固定返回200的处理结果。但只能测试正常而且单一的交互场景，无法对满足交互异常情况的测试。
被测服务请求三方服务的协议正确性难校验。如果三方服务可以接入测试，按之前只能靠抓包工具拦截真实服务，获取被测服务请求三方服务的接口，再手动对齐接口文档。但无法接入的话，仅靠初级桩是无法校验到协议正确性。
异常场景无法落地成自动化。对于三方服务交互异常的场景如返回超时、返回错误的状态码、三方服务器突然宕机等。可以通过抓包工具去手工测试模拟接口拦截、数据的篡改、kill -9使服务下线。但这些异常场景的测试用例无法转换成自动化脚本。
线上测试的成本过高。因为无法接入真实的三方服务测试，所以没法保障被测服务上线以后能和三方服务正常通信，还需要额外的线上调试成本，还可能影响线上的真实数据。

自动化桩的价值

基于以上存在的痛点问题，我在初级桩的概念上，引入了高级桩，也就是自动化桩。自动化桩能够实现以下的能力：

能够在脚本上篡改桩的返回。包括状态码、返回体内容、返回时间，可以覆盖完整的交互异常场景。
能校验被测服务请求三方服务的协议正确性。可以提供请求依赖服务协议的断言方法。
能将测试用例转化为自动化脚本，实现自动化的效果。

自动化桩的实现

落地背景

多人文档编辑时，有一个痛点，B用户编辑的内容可能会覆盖A用户编辑的内容，所以提出编辑锁的方案，A用户进行文档A编辑后，不允许其他用户编辑同一个文档，B用户请求编辑接口会被拒绝。

被测服务（docApp）需要先通过依赖服务（fileApp）校验文档的状态。
自动化桩替代了依赖服务（fileApp），与被测服务进行交互。

交互设计

交互流程

step1：在python客户端实现socket服务端，提供通信方式

step2：在桩模块下，实现socket客户端，和python客户端的socket服务建立通信，明确channel并实例化。

step3：根据桩启动配置所描述的端口在桩模块基于flask框架启动http的桩，满足任意接口和任意请求方式。

step4：启动被测服务，满足依赖关系。

step5：测试用例开始执行，使用桩模块提供的异步http请求方法调用被测服务的接口。

step6：被测服务在入参协议校验完成后开始请求依赖服务获取业务的处理结果，被测服务调用依赖服务的接口（实际上就是请求桩）。

step7：桩服务会通过channel发送消息，将从被测服务接到的请求打包起来发送到python客户端的socket服务。

step8：python客户端会通过拉取消息的方式接收消息。

step9：python客户端需要定义桩的结果返回，定义完成通过socket提供的send进行消息的回调。

step10：桩接收到回调消息后包装成回调消息直接进行被测服务的消息回复。

step11：被测服务通过接口返回返回业务的处理结果。

step12：python客户端能正常完成用例的流转过程。

step13：python客户端执行完全量用例后下线桩服务。

编写脚本

下面是python实现的部分demo

1.main方法

if __name__ == '__main__':
    # 建立并启动socket服务端，实例化了整个桩
    fileAppStub.start_stub()  # 建立并启动socket服务端，实例化了整个桩

    # 启动被测服务
    t = threading.Thread(target = start_server)
    t.start()

    # 启动桩的校验
    check_server = {'http_stub', 'socket_server'}
    check_server_start(check_server)

    # 执行用例
    testsuite = unittest.defaultTestLoader.discover(
        start_dir = DIR + '/testCase',
        pattern = 'test*.py'
    )
    run(testsuite)

    # 桩下线
    fileAppStub.shutdown_stub()

    # 停止被测服务
    os.chdir(DIR + '/app')
    os.system('python appStop.py')\

2.用例层

    def testCase01_demo(self):
        """主流程"""
        file_id = '121'
        user_id = '8771'

        info_log("前置：清空文件协作用户")
        self.apiBusiness.clear_file_members(file_id)

        info_log("请求被测服务接口")
        data = {
            "file_id": file_id,
            "status": "edit"
        }
        headers = {
            "Cookie": f"user_id={user_id}",
            "Content-Type": "application/json"
        }

        hc = HttpCommon()
        hc.thread_run_requests(method="POST", url=self.host + "/edit", json=data, headers=headers)

        info_log("被测服务校验文件是否存在，请求file服务(桩)，桩接收消息")
        receive_msg = fileAppStub.receive_msg()
        info_log("校验fileApp桩收到的消息")
        receive_body = receive_msg['body']
        self.assertEqual(1, len(receive_body.keys()))  # 校验请求体没有其他字段
        self.assertEqual(file_id, receive_body['file_id'])  # 校验file_id的值和对外接口输入的一致
        self.assertEqual('file', receive_msg['path'])  # 校验接口请求的路由信息是否正确
        self.assertEqual('GET', receive_msg['method'])  # 校验http的请求方式是否正确

        info_log("fileApp桩服务回复正常消息")
        send_data = {
            "body": {"msg": "success"},
            "status_code": 200
        }
        # 将消息包发送给桩
        fileAppStub.send_msg(send_data)

        self.assertEqual(200, hc.status_code)
        self.assertEqual('edit success', json.loads(hc.res_text)['msg'])

实现结果

自动化桩.png

目录CONTENT

自动化桩的落地实现

前言