背景
对于开发人员来说，在实际项目中写单元测试的过程中会发现需要测试的类有很多依赖，这些依赖项又会有依赖，导致在单元测试代码里几乎无法完成构建，尤其是当依赖项尚未构建完成时会导致单元测试无法进行。
为了解决这类问题开发人员引入了Mock的概念。
而对于测试的我们来说，我们在测试被测服务过程中，会遇到被测服务需要依赖三方服务才能正常启动和处理业务的场景，仅仅使用mock工具没办法覆盖全部测试用例。
所以，我们可以要求开发实现一个简易版本的三方服务，确保测试用例的执行，开发可能有以下三种处理方案：
写一个桩（假想的服务），一个空心服务，只能保证通信过程，固定返回200的处理结果
修改被测服务源码，屏蔽掉所有三方服务的交互代码
写死三方服务的处理结果，不进行通信
只提供线上环境来测试
痛点
基于上述这些手段可能存在的问题和痛点：
三方服务异常的处理行为覆盖不到
定义的三方处理数据缺少真实性
没法保障被测服务上线以后能和三方服务正常通信，还需要额外的线上调试成本
开发不愿花更多的人力成本时间成本来保障可测试性
线上测试的成本过高，或可能影响线上的真实数据
价值
基于上述问题背景，引入了自动化桩的概念，在桩的基础上我实现了更高阶的能力，能够实现不仅仅只是对请求做出响应，它还能融合嵌入到接口自动化框架中，不再需要测试人员手动去执行，只需前置在用例层代码定义好三方服务的返回即可。
自动化桩实现的能力如下：
能够接受桩的消息进行校验，相当于验证被测服务请求三方服务协议的正确性！
能在测试用例层直接定义桩的结果返回！
且全程无需开发协助实现，无需开发协助部署。
设计交互图
交互流程
首先修改被测服务的配置，指向依赖服务的地址修改为自动化桩的ip和端口
启动python与桩的通信信道，实现socket服务端
启动socket客户端
启动自动化桩的flask桩实例
启动被测服务
python客户端发送请求给被测服务
被测服务发送请求给桩
桩接收被测服务的请求后发送给python客户端去断言协议正确性
python客户端定义桩的返回体发送给桩
桩发送返回体给被测服务
被测服务收到桩的响应后，发送响应给python客户端
python客户端断言被测服务返回体
实战演练
完整项目背景：
域管平台由人员管理、终端管理、策略中心、运维服务、软件管理、企业服务、日志审计、报表管理、消息中心、平台管理等模块组成。管理员可以通过域管平台，对终端设备进行统一用户认证管理、安全策略集中控制、软件更新与分发等功能。
问题背景：
在运维服务-任务管理模块，下发任务到设备流程中，有一个痛点问题：任务下发到关机设备时，会出现任务无法送达设备并正确执行的问题。所以，提出检测机器是否在线的方案。管理员下发任务到设备时，只有检测到设备在线，才允许下发任务。
此处，我用自动化桩模拟的三方服务就是machineServer
展开服务端的测试工作，需要设计依赖服务异常的测试用例。
请求task接口，machine返回403的状态码 期望值：task接口返回 500状态码，返回体{XXXXXX}
请求task接口，machine返回超时 期望值：task接口返回 500状态码，返回体{XXXXXX}
请求task接口，machine异常下线 期望值：task接口返回 503状态码，返回体{XXXXXX}
代码实现细节
main.py
import time
import unittest
import os
from BeautifulReport import BeautifulReport
from threading import Thread
from client_stub import machineStub
import os
import psutil
DIR = os.path.dirname(os.path.abspath(__machine__))
ENVIRON = 'Online' # 'Online' -> 线上环境， 'Offline' -> 测试环境
if __name__ == '__main__':
# 杀python进程
current_process = os.getpid()
for process in psutil.process_iter():
try:
if "python" in process.name().lower() and process.pid != current_process:
process.terminate()
except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
pass
# 启动桩
machineStub.stub_start()
def start_server():
os.chdir(DIR + '/app')
os.system('python taskServer.py') # ./XXX.sh
t = Thread(target=start_server)
t.start()
time.sleep(5)
run_pattern = 'all' # all 全量测试用例执行 smoking 冒烟测试执行 指定执行文件
if run_pattern == 'all':
pattern = 'test_*.py'
elif run_pattern == 'smoking':
pattern = 'test_major*.py'
else:
pattern = run_pattern + '.py'
suite = unittest.TestLoader().discover(DIR + '\\testCase', pattern=pattern)
result = BeautifulReport(suite)
result.report(machinename="report.html", description='测试报告', report_dir=DIR)
machineStub.shutdown_stub()
用例层
import time
import unittest
import requests
import json
from client_stub import machineStub
from httpStubFramework.httpCommon import HttpCommon
from common.logsMethod import step
from common.logsMethod import class_case_log
from common.checkOutput import CheckOutput
@class_case_log
class Pro(unittest.TestCase):
def setUp(self) -> None:
url = 'http://127.0.0.1:8530/clear'
headers = {
'Content-Type': 'application/json'
}
requests.delete(url=url, headers=headers, json={'machine_id': '1111'})
def testCase01_major(self):
"""请求task接口，machine返回200的状态码 期望值：task接口返回 200状态码，返回体{XXXXXX}"""
url = 'http://127.0.0.1:8530/publish'
headers = {
'Content-Type': 'application/json',
'Cookie': 'user_id=B'
}
data = {
"machine_id": "1111",
"status": "publish"
}
step("实例化桩Flask服务")
hc = HttpCommon()
hc.http_requests(url=url, method='post', headers=headers, json=data)
# 桩的socket接收消息
machine_recv = machineStub.receive()
print("machine_recv:")
print(machine_recv)
step("校验被测服务请求stub的协议正确性")
self.assertEqual(machine_recv['headers'], {'Host': '127.0.0.1:8771', 'User-Agent': 'python-requests/2.27.1',
'Accept-Encoding': 'gzip, deflate', 'Accept': '*/*',
'Connection': 'keep-alive'})
self.assertEqual(machine_recv['body'], {"machine_id": "1111"})
self.assertEqual(machine_recv['method'], "GET")
step("用例层自定义桩的异常情况和信息")
major_data = {
"body": {"msg": "publish success"},
"code": 200
}
step("桩收到异常信息后，发送异常信息给被测服务")
# 桩的socket发送消息
machineStub.send(major_data)
time.sleep(1)
step("用例层预期值校验：被测服务发送用例层的预期结果是200")
expect = {"msg": "publish success"}
hc.res_json = json.loads(hc.res_text)
CheckOutput().output_check(expect, hc.res_json)
self.assertEqual(200, hc.status_code)
def testCase02_machineApp_res_403(self):
"""请求task接口，machine返回403的状态码 期望值：task接口返回 500状态码，返回体{XXXXXX}"""
url = 'http://127.0.0.1:8530/publish'
headers = {
'Content-Type': 'application/json',
'Cookie': 'user_id=B'
}
data = {
"machine_id": "1111",
"status": "publish"
}
step("实例化桩Flask服务")
hc = HttpCommon()
hc.http_requests(url=url, method='post', headers=headers, json=data)
# 桩的socket接收消息
machine_recv = machineStub.receive()
step("校验被测服务请求stub的协议正确性")
self.assertEqual(machine_recv['headers'], {'Host': '127.0.0.1:8771', 'User-Agent': 'python-requests/2.27.1',
'Accept-Encoding': 'gzip, deflate', 'Accept': '*/*',
'Connection': 'keep-alive'})
self.assertEqual(machine_recv['body'], {"machine_id": "1111"})
self.assertEqual(machine_recv['method'], "GET")
step("用例层自定义桩的异常情况和信息")
fail_data = {
"body": {"msg": "fail", "body_code": 999999},
"code": 403
}
step("桩收到异常信息后，发送异常信息给被测服务")
# 桩的socket发送消息
machineStub.send(fail_data)
time.sleep(1)
step("用例层预期值校验：被测服务发送用例层的预期结果是500")
expect = {"msg": "SERVER ERROR"}
hc.res_json = json.loads(hc.res_text)
CheckOutput().output_check(expect, hc.res_json)
self.assertEqual(500, hc.status_code)