一种基于闭包函数实现自动化框架断言组件的设计实践 | 京东物流技术团队
1 背景
目前测试组同学基本具备自动化脚本编写能力，为了提高效率，如何灵活运用这些维护的脚本去替代部分手工的重复工作？为了达到测试过程中更多的去使用自动化方式，如何能够保证通过脚本覆盖更多的校验点，提高自动化测试的精度和力度？那么一定是不断的丰富断言，符合预期场景。紧接着棘手的问题就是，在前人维护的脚本不清楚如果在方法内部修改？担心修改原来逻辑影响正向流程运行？一个断言方法希望应用到更多的用例中？本文意在介绍通过闭包函数，实现自动化框架中断言组件的设计实践。
2设计方法
2.1 设计思路
随着脚本维护量不断增大，维护的人越来越多，即要增加断言场景又要保证每天持续集成运行原有用例的成功率。我们理想的断言组件，一定是在不改变原来用例结构和调用方式基础之上，对前人写的代码零侵入，通过装饰器增加更多场景断言，并且做到复用断言组件到更多的测试用例上。
2.2 原理解读
2.2.1 闭包函数解读
名词解释：
闭包函数是函数的嵌套，函数内还有函数，即外层函数嵌套一个内层函数，在外层函数定义局部变量，在内层函数通过nonlocal引用，并实现指定功能，比如计数，最后外层函数return内层函数。
主要作用：
可以变相实现私有变量的功能，即用内层函数访问外层函数内的变量，并让外层函数内的变量常驻内存。
实现原理：
闭包函数之所以可以实现让外层函数内的变量常驻内存，关键就是其定义了个内层函数，并通过内层函数访问外层函数的变量，并最后由外层函数将内层函数返回出去并赋值给另外一个变量。此时因为内层函数被赋值给一个变量，其内存空间不会被释放，而内层函数又在其函数体内引用了外层函数的变量，导致该变量的内存也不会被回收。一般情况下，当一个函数运行完毕后，其内存空间即被回收释放，下次再调用该函数的时候，会重新完整运行一次被调用函数，但闭包函数主要是利用Python的内存回收机制，实现了闭包的效果。
2.2.2 装饰器解读
名词解释：
装饰器自身是一个返回可调用对象的可调用对象，本质是一个闭包函数。
结构特点：
装饰器也是函数的嵌套结构，可能还会存在三层嵌套，外层函数就是装饰器函数，接受的参数是一个函数，一般是传入被装饰函数；内层函数实现具体的装饰器功能，比如日志记录、登录鉴权、逻辑校验等，内层函数return一次传入的函数调用，外层函数return内层函数；如果是多层嵌套，最内层是实现具体装饰器功能的函数，并负责调用一次传入的函数，最外一层函数return第二层函数，依次类推，不过一般最多就是三层函数嵌套。
3 解决方案
3.1 现有用例
def test_enquiry_bill_for_two_driver_quote_price(params):
"""
终端来源两个司机同时报价再修改其一报价
Args:
params:测试用例数据
Returns:测试用例实际返回结果
"""
# 询价接单
enquiry_code = jsf_receive_enquiry_bill(**params['expect'][0]).get("data")
params['actual'].append({"enquiryCode": enquiry_code})
# 获取单趟任务
transit_job_code = get_transit_job_code(enquiry_code=enquiry_code).get('transit_job_code')
# 司机报名,报价
params['expect'][1].update({"transitJobCode": transit_job_code})
jsf_apply_transit_job_by_param(**params['expect'][1])
# 第二位司机报名，报价
params['expect'][2].update({"transitJobCode": transit_job_code})
jsf_apply_transit_job_by_param(**params['expect'][2])
# 第二位司机修改报价
params['expect'][2].update({"quotePrice": 100})
actual = jsf_apply_transit_job_by_param(**params['expect'][2])
params['actual'].append(actual)
assert actual.get('code') == 1
assert actual.get('message') == '重新报价成功'
log.info(f'验证预期结果为 {actual.get("data")} 通过')
return params3.2 断言组件设计
单一业务节点校验组件：
如上对询价单报价场景，现有测试用例完全可以单独运行，目前只有简单的返回值断言，缺少很多关键节点校验。比如，步骤一询价接单是否落库成功，步骤二单趟任务是否创建成功；步骤三司机报价后的单趟价格，步骤四司机再次提交报价，调用接口后的价格是否修改成功，我们为了不影响原来用例执行，对原代码做到零侵入，且自动实现断言异常捕获，可以通过增加一个断言组件完成。
def validation(func):
@wraps(func)
def wrapper(*args, **kwargs):
try:
# 执行函数
data=func(*args, **kwargs)
actual_enquiry=hash_db.query_enquiry_bill(data['actual']['enquiryCode'])
actual_transit=hash_db.query_transit_job_bill(data['expect'][1]['transitJobCode'])
assert data.get("expect")[2]['quotePrice'] == actual_transit['quote_price']
except Exception as ex:
log.exception(ex)
return wrapper公共校验组件：
如上实现了通过一个装饰器去完成断言，但有些同学认为，以上断言方法又不能适用于其他用例，为什么还要额外重写一个函数呢？其实这种方式，更多的会应用到公共组件，比如以下通过装饰器完成用例返回值与对应数据库的断言场景。
def validation_db(sql,**kwargs):
def validation(func):
@wraps(func)
def wrapper(*args, **kwargs):
try:
counts, results = tms_mysql.execute_query(sql)
if counts:
# 根据获取数据开始断言
for key_res, value_res in results[0].items():
for key_arg, value_arg in kwargs.items():
if field_change(key_res, change_type='to_arg') == key_arg:
log.info(f'断言{key_arg}字段,预期值是{value_res}，实际值是{value_arg}')
assert value_res == value_arg
else:
return counts
except Exception as ex:
log.exception(ex)
return wrapper
return validation3.3 改造用例
单一装饰器组件
如下所示，用例test_enquiry_bill_for_two_driver_quote_price内部代码依旧不变，仅是在方法上，加上
@validation，目前在执行原有用例时，增加校验过程数据，比如第一次提交报价的值，更改后提交数据的变化，增加现有自动化测试用例的可靠性。
@validation
def test_enquiry_bill_for_two_driver_quote_price(params):
"""
终端来源两个司机同时报价再修改其一报价
Args:
params:测试用例数据
Returns:测试用例实际返回结果
"""
# 询价接单
enquiry_code = jsf_receive_enquiry_bill(**params['expect'][0]).get("data")
params['actual'].append({"enquiryCode": enquiry_code})
# 获取单趟任务
transit_job_code = get_transit_job_code(enquiry_code=enquiry_code).get('transit_job_code')
# 司机报名,报价
params['expect'][1].update({"transitJobCode": transit_job_code})
jsf_apply_transit_job_by_param(**params['expect'][1])
# 第二位司机报名，报价
params['expect'][2].update({"transitJobCode": transit_job_code})
jsf_apply_transit_job_by_param(**params['expect'][2])
# 第二位司机修改报价
params['expect'][2].update({"quotePrice": 100})
actual = jsf_apply_transit_job_by_param(**params['expect'][2])
params['actual'].append(actual)
assert actual.get('code') == 1
assert actual.get('message') == '重新报价成功'
log.info(f'验证预期结果为 {actual.get("data")} 通过')
return params多个装饰器嵌套
如下是多个组件嵌套使用方式，及执行顺序解读
@dec1
@dec2
@dec3
def func():
pass此时：可以对某个被装饰函数，增加多个功能
装饰器生效顺序，从上到下，即dec1＞dec2＞dec3
在第一步改造后，仅是增加了对核心字段的过程数据校验，有的同学希望用例更加准确，不用再切换去看数据库，直接将所有返回值字段，与库里进行预期比较。
如下所示，同样在原有用例上增加多个装饰器，即多个断言组件，按顺序依次断言。下面是，增加定义的单个用例的私有断言@validation和数据库公共断言@validation_db
增加后不会影响原来测试流程执行，大家也可以按照需求，在断言组件内声明，断言异常是否中断。
@validation
@validation_db(enquiry_sql)
def test_enquiry_bill_for_two_driver_quote_price(params):
"""
终端来源两个司机同时报价再修改其一报价
Args:
params:测试用例数据
Returns:测试用例实际返回结果
"""
# 询价接单
enquiry_code = jsf_receive_enquiry_bill(**params['expect'][0]).get("data")
params['actual'].append({"enquiryCode": enquiry_code})
# 获取单趟任务
transit_job_code = get_transit_job_code(enquiry_code=enquiry_code).get('transit_job_code')
# 司机报名,报价
params['expect'][1].update({"transitJobCode": transit_job_code})
jsf_apply_transit_job_by_param(**params['expect'][1])
# 第二位司机报名，报价
params['expect'][2].update({"transitJobCode": transit_job_code})
jsf_apply_transit_job_by_param(**params['expect'][2])
# 第二位司机修改报价
params['expect'][2].update({"quotePrice": 100})
actual = jsf_apply_transit_job_by_param(**params['expect'][2])
params['actual'].append(actual)
assert actual.get('code') == 1
assert actual.get('message') == '重新报价成功'
log.info(f'验证预期结果为 {actual.get("data")} 通过')
return params4 总结
以上实践案例，是基于运力测试团队现有的自动化维护情况，前期脚本已大量堆砌但缺少断言，现阶段测试流程没有变化，但为了增加自动化脚本的测试力度需要批量增加断言。是否利用装饰器来实现断言，一定要取决于团队中维护用例的情况，如果当前用例从头到尾都是你一个人维护，里面的场景也没办法给其他人公用，那么大可不必！不过学习好装饰器后，在代码编写过程中希望一处实现多处复用，也可以通过装饰器方式去提升代码可读性和可维护性。
作者：京东物流 刘红妍
来源：京东云开发者社区 自猿其说Tech 转载请注明来源
