自动化桩的落地实现

自动化桩的落地实现

Administrator 25 2024-09-02

一、背景

  在日常项目中，经常会遇到被测服务需要依赖三方服务才能正常处理业务，但三方服务并没有提供测试环境，导致我们无法在测试环境中去正常测试。最初我们提出“桩”的概念，可以理解为一个假想的服务，去替代依赖服务，模拟与被测服务的场景交互。随着测试深度要求的提升，为满足异常场景的全面验证需求，我们进一步将测试方案演进为“自动化桩”。通过将异常测试用例转化为可编程测试脚本，构建起完整的异常响应机制。在此总结下自动化桩落地过程中遇到的痛点问题和设计实现过程。

二、存在的痛点

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

三、自动化桩的价值

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

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

四、自动化桩的实现

1）落地demo

  在我司业务知识库项目中，多人文档编辑时，有一个痛点，B用户编辑的内容可能会覆盖A用户编辑的内容，所以开发侧提出编辑锁的方案，在A用户进行文档A编辑后，不允许其他用户编辑同一个文档，B用户请求编辑接口会被拒绝。

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

image.png

2）交互设计

1.drawio (1).png

3）交互流程

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客户端执行完全量用例后下线桩服务。

4）脚本编写

下面是python实现的部分demo

  1. main方法
if __name__ == '__main__':
    # 建立并启动socket服务端，实例化桩
    fileAppStub.start_stub()  

    # 启动被测服务
    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')\
  1. 用例层
    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("被测服务校验文件是否存在，请求桩服务，桩接收消息")
        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'])

五、实现结果

image.png