一般来说，在接口测试完成以后，就可以开始进行服务故障演练，如果依赖服务特别多，演练的成本就相对来说比较高，甚至到时候进行回归测试也会很繁琐。如果可以实现一个自动化桩，能直接通过自动化脚本实现这样的一个场景模拟，那就能直接给当前团队提高测试效率。本文主要介绍自动化桩的设计思路，并围绕一个文档编辑场景实现一个案例 demo。
自动化桩是什么？
应用于自动化框架，可以通过脚本调用的方式模拟依赖服务的请求和返回行为，并且可以把结果同步到用例自动化case当中；同时兼容桩和 mock 的能力，并且能进行自动化脚本的转化，还可以减轻开发的可测试性保障成本。
自动化桩有什么价值？
自动化桩的能力：
能够在脚本上篡改桩的返回。包括状态码、返回体内容、返回时间，可以覆盖完整的交互异常场景。
能校验被测服务请求三方服务的协议正确性。可以提供请求依赖服务协议的断言方法。
能将测试用例转化为自动化脚本，实现自动化的效果。
自动化桩能解决的痛点问题：
三方服务异常的处理行为覆盖不到。开发一般会提供初级桩，只能保证通信过程，固定返回200的处理结果。但只能测试正常而且单一的交互场景，无法对满足交互异常情况的测试。
被测服务请求三方服务的协议正确性难校验。如果三方服务可以接入测试，按之前只能靠抓包工具拦截真实服务，获取被测服务请求三方服务的接口，再手动对齐接口文档。但无法接入的话，仅靠初级桩是无法校验到协议正确性。
异常场景测试无法自动化。对于三方服务交互异常的场景如返回超时、返回错误的状态码、三方服务器突然宕机等。可以通过抓包工具去手工测试模拟接口拦截、数据的篡改、kill -9使服务下线。但这些异常场景的测试用例无法转换成自动化脚本。
线上测试的成本过高。因为无法接入真实的三方服务测试，所以没法保障被测服务上线以后能和三方服务正常通信，还需要额外的线上调试成本，还可能影响线上的真实数据。
自动化桩的设计思路
设计思路：
实现基础桩，手动模拟交互场景
实现 autoClient 和桩的通信
定义交互协议
如何实现桩和 autoClient 的多线程交互和协作
实现桩实例管理
超时处理器
入参协议校验、异常捕捉的场景覆盖
完整落地
自动生成无冲突的端口号来避免端口冲突
自动化桩的交互流程：
（蓝色框指的是框里的逻辑实现都是在 flask 同一 route 下完成）
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 客户端执行完全量用例后下线桩服务。
自动化桩的落地实现
背景：
多人文档编辑时，有一个痛点，B用户编辑的内容可能会覆盖A用户编辑的内容，所以提出编辑锁的方案，A用户进行文档A编辑后，不允许其他用户编辑同一个文档，B用户请求编辑接口会被拒绝。
测试用例：
testCase01_major：
正向用例，请求 docteamApp 接口，文档不存在时进行文档编辑。
操作步骤：
调用接口，请求参数....
fileApp 接收到 docteamApp 的请求后返回200的处理结果。
期望：
docteamApp 返回200的状态，文档编辑成功。
fileApp 收到 docteamApp 的请求协议为 {...}。
testCase02：
请求文档编辑接口，fileapp 返回文档不存在的情况，返回403状态码。
期望值：文档编辑接口返回 403状态码，返回体 {XXXXXX}。
testCase03：
请求文档编辑接口，fileapp 返回超时。
期望值：文档编辑接口返回 500状态码，返回体 {XXXXXX}。
testCase04：
请求文档编辑接口，fileapp 异常下线。
期望值：文档编辑接口返回 503状态码，返回体 {XXXXXX}。
桩实例回收方法实现：
在 flask 的 route 中增加一个接口，如果该接口被调用就进行实例回收。
if path == 'shutdown':
os.kill(os.getpid(), signal.SIGINT)
return 'Server shutting down...', 200在用例执行完成后可以调用桩对象的下线方法完成实例回收。
def stub_stop(self):
requests.post(url=f"http://127.0.0.1:{self.stub_port}/shutdown", timeout=3)异步 http 请求实现，使接口请求和返回异步开来，能够在 http 结果返回前控制桩消息的收发：
import threading
import requests
class HttpCommon:
def __init__(self):
self.res = None
self.status_code = None
self.res_text = None
def __target_http_func(self, **kwargs):
self.res = requests.request(**kwargs)
self.status_code = self.res.status_code
self.res_text = self.res.text
def http_requests(self, **kwargs):
"""支持http协议请求时的多线程等待"""
"""Constructs and sends a :class:`Request <Request>`.
支持http协议请求时的多线程等待
"""
t = threading.Thread(target=self.__target_http_func, kwargs=kwargs)
t.start()基于 flask 框架实现 http 协议的桩：
from flask import Flask, request, jsonify
import socket
from werkzeug.serving import make_server
import json
import os
import signal
class HttpStub:
def __init__(self, stub_port, socket_server_port):
self.channel = None
self.stub_port = stub_port
self.socket_server_port = socket_server_port
self.app = Flask(__name__)
self.app.route('/<path:path>', methods=['GET', 'POST', 'PUT', 'DELETE', 'PATCH'])(self.common_route)
def socket_client(self):
self.channel = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) # 套接字：1 ==》 udp传输协议
def common_route(self, path):
if path == 'shutdown':
os.kill(os.getpid(), signal.SIGINT)
return 'Server shutting down...', 200
msg = {
'path': path,
'method': request.method,
'headers': dict(request.headers),
'cookies': dict(request.cookies),
'params': dict(request.args) or None,
'data': request.data.decode('utf-8') or None
}
server_address = ('127.0.0.1', int(self.socket_server_port))
self.channel.sendto(json.dumps(msg).encode('utf-8'), server_address)
data = self.channel.recv(1024).decode('utf-8')
send_data = json.loads(data)
send_response = send_data["body"]
send_status_code = send_data["code"]
return send_response, send_status_code
def server_run(self):
self.socket_client()
server = make_server('127.0.0.1', self.stub_port, self.app)
self.app.debug = True
server.serve_forever()桩的运行步骤：
需要先启动 udp socket 的客户端
flask 启动
flask 多线程运行
定义通用 route 来兼容所有 http 接口协议
在 route 中实现被动接受被测服务的请求，并将请求内容打包成一个消息包再发送到 autoClient
autoClient 可以定义桩的返回结果，通过 socket 发送到 flask 实例，flask 接受到消息解析完成后 return 给调用方
基于面向对象进行开发，以便其他模块变量的复用
文档不存在时的测试用例 demo：
def testCase02_fileApp_res_403(self):
"""请求文档编辑接口，fileapp返回403的状态码 期望值：文档编辑接口返回 500状态码，返回体{XXXXXX}"""
print("请求文档编辑接口，fileapp返回403的状态码 期望值：文档编辑接口返回 500状态码，返回体{XXXXXX}")
headers = {
'Content-Type': 'application/json',
'Cookie': 'user_id=B'
}
data = {
"file_id": "1111",
"status": "edit"
}
hc = HttpCommon()
# 异步请求被测服务，便于后续桩的消息收发
hc.http_requests(url=self.url, method='post', headers=headers, json=data)
data = fileAppStub.receive() # 接收桩打包的从被测服务收到的请求信息
expect = {'file_id': '1111'}
# 验证被测服务请求三方服务协议的正确性
self.ac.json_assert(expect, data['params'])
# 提供给桩的具体数据，也是桩要返回给被测服务的数据
send_data = {
"body": {"msg": "fail", "body_code": 999999},
"code": 403
}
fileAppStub.send(send_data)
time.sleep(2)
self.ac.code_assert(500, hc.status_code)
expect = {
"msg": "SERVER ERROR"
}
self.ac.json_assert(expect, json.loads(hc.res_text))
# 打印被测服务的接口返回结果
print(hc.status_code)
print(hc.res_text)结果输出：
评论区