最近一直在忙着点点点，好长时间没更新新的文章了，今天写一下最近做功能测试过程中的一个思路。
需求背景：
有一系列的任务调2-3个外部接口获取数据后，入库到mysql数据库里面，然后会对外提供接口返回清洗后的数据。需要对这整个过程进行验证。
这中间可能会涉及到的点有：
1、外部接口的数据分别入库到mysql里面的数据是否正确，包括字段取值映射关系，数据总记录数等等。
2、数据源数据更新时，通过监听kafka消息及时更新mysql中的数据
3、测试一下对外提供的接口和kafka消息等。
测试过程中遇到的问题：
1、其实这些东西测起来不是很难，只是字段比较多 ，需要耗费一定的时间。加上最近的需求又有点多，以后类似这样的需求还有不少，再加上是新接手这块的需求，对一些字段映射关系啥的不是很清楚，对上游数据的改动和来源不熟悉，造数据覆盖不同场景需要耗费一定的时间
2、没开始测之前，以为接口对外输出应该比较好验证，结果等到我测的时候才发现，比我想象中稍微要麻烦一点点，本以为字段都是平铺返回的，这样我顶多处理一下字段映射关系，结果发现接口返回的时候还对不同的属性进行了分组，这样就导致到时候写代码的时候又会变得复杂很多。
接下来分享一下最后接口验证这块的一个写脚本的思路：
1、将mysql中的数据查出来，然后调对应的接口
2、按照接口返回的格式定义一套模板，将数据库里面的字段名和接口的字段名之间做一个映射关系转换
3、定义一个方法，传入mysql中的数据，替换调模板中的变量，然后按照接口的格式进行返回
4、用deepdiff库去对比从库中查出来的按照模板格式化后的数据和接口返回的数据进行对比。
注意：如果接口涉及到批量查询的时候，返回的大概率是一个对象的list，用deepdiff对比的时候，要注意列表中元素的顺序，最好自己将两边数据顺序都处理成一致的，避免插件对比的时候结果不符合预期。
下面附上替换模板中变量的参考代码，其中${xxx}格式表示xxx为变量名：
import pprint
# 模板字典，包含嵌套字段
template = {
"userName": "${user_name}",
"userAge": "${user_age1}",
"userInfo": {
"phone": "${user_phone}",
"email": "${user_email}"
},
"userInfoExt": {
"wxName": "${wx_name}"
}
}
# 实际数据字典
data = {
"user_name": "小博",
"wx_name": "小博测试成长之路",
"user_age": 30,
"user_age1": 1,
"user_phone": "1234567890",
"user_email": "example@example.com",
"additional_field_1": "Additional Data 1",
"additional_field_2": "Additional Data 2"
}
def replace_variable(template, data):
if isinstance(template, dict):
result = {}
for key, value in template.items():
if isinstance(value, str):
modified_value = ""
i = 0
while i < len(value):
if value[i:i + 2] == "${" and "}" in value[i + 2:]:
start = i
end = value.index("}", i + 2)
field_name = value[start + 2:end] # 提取字段名
if field_name in data:
modified_value += str(data[field_name])
else:
modified_value += value[start:end + 1] # 如果字段不存在，保留占位符不变
i = end + 1
else:
modified_value += value[i]
i += 1
result[key] = modified_value
elif isinstance(value, dict):
# 如果值是字典，递归处理
result[key] = replace_variable(value, data)
else:
result[key] = value
return result
else:
return template
# 调用递归函数进行替换
result = replace_variable(template, data)
# 打印映射后的结果
pprint.pprint(result)最后，考虑到接口取数据库的字段可能不止是字段名映射，可能还涉及到映射关系的转换或者计算之类，可以在定义一个函数去解析模板中符合某种格式的自定义函数，将函数返回值替换模板中的数据，下面代码仅供参考，格式$.xxx()为自定义函数，其中xxx为函数名。
import re
from pprint import pprint
template = {'userAge': '$.int(1)',
'userAge1': 'a$.int(1)',
'userInfo': {'email': 'example@example.com', 'phone': '1234567890'},
'userInfoExt': {'wxName': '小博测试成长之路'},
'userName': '小博',
"sex": "$.custom_func(1)"}
def custom_func(value):
return {"1": "男", "2": "女"}.get(value)
def custom_function(value):
try:
# 使用正则表达式匹配函数调用
match = re.match(r'(.*?)\$\.(.*?)\((.*?)\)(.*)', value)
if match:
text_before = match.group(1) if match.group(1) else ''
function_name = match.group(2)
function_args = match.group(3)
text_after = match.group(4) if match.group(4) else ''
# 检查函数名是否是内置函数
if function_name not in globals() or not callable(globals()[function_name]):
# 如果是内置函数，使用 eval 调用自定义函数
result = eval(f'{function_name}({function_args})')
else:
# 调用自定义函数
result = globals()[function_name](function_args)
if text_before == "" and text_after == "":
return result
return f'{text_before}{result}{text_after}'
return value
except Exception:
return value
def replace_funcs(template):
if isinstance(template, dict):
result = {}
for key, value in template.items():
result[key] = replace_funcs(value)
return result
elif isinstance(template, str):
return custom_function(template)
else:
return template
parsed_template = replace_funcs(template)
pprint(parsed_template)提供另一个demo：
import re
from pprint import pprint
class SMTools:
@staticmethod
def custom_func1(value):
return f"Custom Function 2 Result for {value}"
@staticmethod
def replace_funcs(template):
if isinstance(template, dict):
for key, value in template.items():
if isinstance(value, str):
# 使用正则表达式匹配函数调用
match = re.match(r'(.*?)\$\.(\w+)\((.*?)\)(.*)', value)
if match:
text_before = match.group(1) if match.group(1) else ''
function_name = match.group(2)
function_args = match.group(3)
text_after = match.group(4) if match.group(4) else ''
try:
if hasattr(SMTools, function_name) and callable(getattr(SMTools, function_name)):
# 检查是否是SMTools类中的自定义函数并调用它
func = getattr(SMTools, function_name)
result = func(function_args)
value = f"{text_before}{result}{text_after}"
else:
# 处理非自定义函数
result = eval(f'{function_name}({function_args})')
if text_before == "" and text_after == "":
value = result
else:
value = f"{text_before}{result}{text_after}"
except Exception as e:
# 处理异常情况
value = f"{value} Error: {str(e)} "
template[key] = value
elif isinstance(value, dict):
# 递归处理嵌套字典
template[key] = SMTools.replace_funcs(value)
return template # 返回处理后的template
# 示例用法
template = {'userAge': '$.int(1)',
'userAge1': 'a$.int(1)',
'userInfo': {'email': 'example@example.com', 'phone': '1234567890'},
'userInfoExt': {'wxName': '小博测试成长之路'},
'userName': '小博',
"sex": "$.custom_func1(1)"}
SMTools.replace_funcs(template)
pprint(template)此处的代码还不够完善，可能后续运行过程中模板转换可能会有不满足预期的情况，到时候调整对应的转换函数即可。重要的是解决问题的一个思路。
#注意：
以上代码仅供参考，要根据自己写的模板进行联调，把一些异常情况考虑进去，比如变量中嵌套使用的场景等。
End
如果觉得文章对你有帮助的话，欢迎点赞转发~
如果还没有进粉丝交流群的小伙伴，赶快添加好友(xiaobotester)邀请你们进群喔。👇👇👇关注公众号，测试干货及时送达
为感谢所有粉丝的关注和长期以来的支持，特在公众号后台为大家准备了一些资料，可在后台回复【百宝箱/关键字】获取相关链接，资料会持续更新，有需要的可以先私聊作者补充文档。
往期精彩回顾