利用自动化平台可以做的那亿点事 ｜得物技术

前语

相信咱们对接口主动化现已不陌生了，这是几乎咱们每个迭代都会投入的工作，但耗费了这么多精力去编写和保护，实际的收益怎么呢？假如收益欠好，是不是说明咱们主动化case的完成办法、运用办法还有改善的地方呢？以下是接入得物接口主动化渠道后的一些实践和主意，欢迎咱们活跃沟通～

浅谈接口主动化

1.1运用场景&能够带来的作用

• 给开发用 – 进步自测功率&提测质量

在接入主动化渠道前，咱们只能本地拉取代码->履行用例，所以履行者也只要测验人员。接入渠道后，经过宣导or共享，开发能够便利的找到需求的用例（用例模块和标题需描述清晰），然后帮助他们造数或自测。

对于一些核心场景，即便事务迭代，通常成果也不会发生太大改变，这一类的场景case假如规划地较为安稳（当然这儿的安稳不是只校验code=200就行），能够共享给开发用于自测，依据开发同学运用后的反馈，他们自测简略了许多，也有帮助他们发现过问题。

别的有一些本迭代内的新增接口，在接口评定完成后，咱们能够提前编写好，依据具体情况决定是先保证接口状况的正常，后续再弥补数据逻辑的校验，还是直接先把case写好。由于很多时候开发自测都仅仅调用本地代码，提测后连接口都调不通，假如提测前能够先进行根本的校验，就能削减冒烟测验被堵塞的概率。

• 给测验用 – 进步测验功率

_冒烟测验：_针对改动点挑出触及的接口case，再加上P0等级case，提测后先履行一遍看看是否正常，假如核心链路反常，堵塞了后续测验，就能够直接打回了。

_验证bug：_有些杂乱场景，测验链路较长，测验数据准备又很困难，很简略呈现bug，而呈现bug也就算了，偏偏改一遍还不必定能改好…这时候主动化的价值就体现了，把这些场景运用主动化完成，验证bug时直接一键履行就能得出成果，大大节省了时刻，同时也安稳了自己濒临暴躁的心情。

_回归测验：_在每次的bvt测验、掩盖率跟进中，有些case可能并不触及本次需求改动范围，场景又比较简略基础，咱们就能够运用主动化去掩盖。履行经过，视具体情况能够简略看一眼或许不再回归。

• 给需求的人用 – 简易的造数东西

虽然咱们现在有了造数渠道，但完成起来有必定的本钱，一些场景可能除了自己没有别的事务方有造数需求，并且场景很简略，只需调个接口，改个数据表就行，那么最快的造数办法就是主动化脚本。现在有了主动化渠道，咱们能够更好地共享给有造数需求的开发、产品、测验。

当然，以上作用的条件是咱们的主动化case比较安稳，不能每次履行都一堆不经过，这样时刻都耗费在排查问题上了，作用会大打折扣，他人也不会再乐意运用。

1.2 什么时刻去写主动化case

通常一部分同学会在用例评定完毕，开发提测之前进行case编写，此刻需求完成主动化的场景现已清晰，根本上触及的接口和收支参都已确认，主动化case的大致框架就形成了。这时候完成主动化，就能够最大化地发挥其价值，在上述触及到的几个场景都能投入运用。假如由于时刻不够或接口没有清晰，能够先梳理好需求完成主动化的场景过程，在提测后一边手动履行用例一边弥补接口参数和校验点。针对等级较低的接口场景，也能够放在版本完毕后再完成，仅仅作用会降低一些。

1.3 主动化保护本钱太高怎么办

咱们保护的case一般有两种，一是自己写的，二是他人写的。自己写的，含着泪也要日常保护。他人写的，由于咱们的编码风格千差万别，在接入主动化渠道前，保护起来几乎困难重重，当咱们为了经过率去推动case更新时，往往这一类的难以推动。现在接入了渠道，根本上一致了case模板，当由于需求变化需求更新时，有时只需求修正收支参和断言即可，必定程度上现已降低了保护本钱。

别的，当case经常报错时，能够看看规划上是否能优化。有些依赖性强的数据，是否能够经过其他手法让这部分数据安稳下来。比如发优惠券的场景，条件需求一张有用的券，那咱们在发券前能够先获取一张有用的券信息，或许在发券前先创立一张券，发完券后假如需求对券信息进行校验，也经过变量的办法。针对单个测验点完成主动化时，能够尽可能地与其他测验点解藕，充分运用前置脚本，经过修正数据表等办法较少依赖。case中也能够设置失利重试次数，削减由于环境不安稳等原因造成的失利。

在主动化渠道上的实践

2.1场景case的编写

举个例子：“得物App新客人群领取优惠券并触发金额胀大，多次触发胀大应该只要一次胀大成功”。

这个case在迭代中进步了测验功率，并且在后续需求改变时，帮助开发自测，处理造数问题并发现了bug。

• 由于事务特性，只要射中实验组的新用户才可领券。那么首要需求创立一个新用户，并增加到ab白名单。然后在领券前先对领券状况、用户身份进行校验；

• 由于后台会装备3套券，初次领券成功后，只会发放其间一套，所以在对领券接口的出参进行根本校验后，还需对券记载进行具体的检查，就需求运用后置脚本，获取到券装备后再对数据表进行核对，需求校验的表包括事务本身的领券记载表和优惠事务侧的账户表；

import json
import requests
from util.db_mysql import DBMySQL
from util.db_redis import DbRedis
def call(env_vars, g_vars, l_vars, sys_funcs, asserts, logger, **kwargs):
    userId = l_vars.get('userId')
    n = int(userId)%4
    dbA = DBMySQL(env_vars.get("db.A"))
    dbB = DBMySQL(env_vars.get("db.B"))
    try:
        sql_1 = "SELECT * FROM table_A WHERE user_id = %s;"%userId
        # 领券后，用户领券状况校验
        user_coupon_info = dbA.select(sql_1)
        logger.info(newbie_res)
        asserts.assertEqual(user_coupon_info[0].get("status"), 1, msg="数据表领券状况为true")
        asserts.assertEqual(user_coupon_info[0].get("type"), 0, msg="当时券类型为0")
        asserts.assertIsEmpty(user_coupon_info[0].get("coupon1"), msg="无财物1")
        asserts.assertIsEmpty(user_coupon_info[0].get("coupon2"), msg="无财物2")
        asserts.assertIsEmpty(user_coupon_info[0].get("coupon4"), msg="无财物4")
        asserts.assertIsNotEmpty(user_coupon_info[0].get("info"), msg="券包信息非空")
        #获取用户分组，确认用户是射中了实验组的
        group = user_coupon_info[0].get("group")
        asserts.assertNotEqual(group, 0, msg="用户射中对照组，无胀大券")
        #获取胀大财物装备
        sql_2 = "SELECT * FROM table_B WHERE id = 50%s and deleted=0"%group
        logger.info("sql_2："+sql_2)
        coupon_config = dbA.select(sql_2)
        logger.info("coupon_config："+coupon_config)
        content = json.loads(coupon_config[0].get("content_info"))
        for i in range(3):
            activityId = content[i]["activityId"]
            l_vars.set('activityId_{}'.format(i+1), activityId)
            # 优惠券表校验
            sql_3 = "SELECT * FROM a_coupon_%s WHERE user_id = %s and activity_id = %s;"%(n,userId,activityId)
            logger.info("sql_3："+sql_3)
            coupon_res = dbB.select(sql_3)
            logger.info("coupon_res："+coupon_res)
            if(i==0):
                asserts.assertIsEmpty(coupon_res, msg="未到账财物1")
            if(i==2):
                asserts.assertIsNotEmpty(coupon_res, msg="到账财物3")
    finally:
        dbA.close()
        dbB.close()

• 领券成功后进行胀大。查询优惠侧账户表，将查询成果作为变量，鄙人一个接口的前置脚本中，进行券到账的校验；

import json
import requests
from util.db_mysql import DBMySQL
from util.db_redis import DbRedis
def call(env_vars, g_vars, l_vars, sys_funcs, asserts, logger, **kwargs):
    call_param = sys_funcs.get_call_param()
    userId = call_param.get('userId')
    activityId = call_param.get('activityId')
    dbB = DBMySQL(env_vars.get("db.B"))
    if not userId:
        user_var = l_vars.get(call_param.get('var_userId'))
        userId = user_var
    if not activityId:
        activityId_var = l_vars.get(call_param.get('var_activityId'))
        activityId = activityId_var
    if not userId and not activityId:
        raise '请传入查询条件'
    try:
        if not activityId:
            sql = "SELECT * FROM a_coupon_%s WHERE user_id = %s;"%(int(userId)%4,userId)
        elif not userId:
            sql = "SELECT * FROM a_coupon_%s WHERE activity_id = %s;"%(n,activityId)
        else:
            sql = "SELECT * FROM a_coupon_%s WHERE user_id = %s and activity_id = %s;"%(int(userId)%4,userId,activityId)
        logger.info(sql)
        res = dbB.select(sql)
        logger.info(res)
        l_vars.set("select_tableB_res",res)
    except Exception as e:
        logger.info(f'查询失利【{str(e)}】')
        raise e
    finally:
        dbB.close()
        return res

import json
import requests
def call(env_vars, g_vars, l_vars, sys_funcs, asserts, logger, **kwargs):
    select_tableB_res = l_vars.get('select_tableB_res')
    asserts.assertIsNotEmpty(select_tableB_res, msg="到账财物1")

• 再次胀大，应胀大失利，校验接口code非200，再次核对券表，校验的确只到账了一张券。

import json
import requests
def call(env_vars, g_vars, l_vars, sys_funcs, asserts, logger, **kwargs):
    select_tableB_res = l_vars.get('select_tableB_res')
    asserts.assertEqual(len(select_tableB_res),1,msg="只到账财物1一张")

• 其他类似的场景，能够经过仿制已有的用例或过程直接运用。

• 2.2公共组件的编写一些需求重复调用的功用，咱们能够写成公共组件，不仅便利自己，也便利他人。在编写组件时，假如有入参，需求考虑参数值有可能是局部变量的场景。以下面的组件为例，完成的功用是经过数据库查询优惠券发放记载表，能够针对用户ID、优惠财物ID进行查询。考虑到这两个参数有可能是局部变量，由于目前公共组件类型的入参不支持${}参数类型，所以换一种办法来完成——设置2个入参，一个为对应的value，一个为局部界说的key。脚本中，假如value未获取到，则去变量空间中获取局部变量。拿到查询成果后也要尽可能的把成果存到变量空间，以供后续过程的运用。

import json
import requests
from util.db_mysql import DBMySQL
from util.db_redis import DbRedis
def call(env_vars, g_vars, l_vars, sys_funcs, asserts, logger, **kwargs):
    call_param = sys_funcs.get_call_param()
    userId = call_param.get('userId')
    activityId = call_param.get('activityId')
    dbA = DBMySQL(env_vars.get("db.A"))
    if not userId:
        user_var = l_vars.get(call_param.get('var_userId'))
        userId = user_var
    if not activityId:
        activityId_var = l_vars.get(call_param.get('var_activityId'))
        activityId = activityId_var
    if not userId and not activityId:
        raise '请传入查询条件'
    try:
        if not activityId:
            sql = "SELECT * FROM a_coupon_%s WHERE user_id = %s;"%(int(userId)%4,userId)
        elif not userId:
            sql = "SELECT * FROM a_coupon_%s WHERE activity_id = %s;"%(n,activityId)
        else:
            sql = "SELECT * FROM a_coupon_%s WHERE user_id = %s and activity_id = %s;"%(int(userId)%4,userId,activityId)
        logger.info(sql)
        res = dbA.select(sql)
        logger.info(res)
        l_vars.set("select_tableA_res",res)
    except Exception as e:
        logger.info(f'查询失利【{str(e)}】')
        raise e
    finally:
        dbA.close()
        return res

2.3测验方案的履行

装备渠道用例方案，挑选依赖运用，按照自己的需求挑选履行频次。然后再修改方案，装备匹配规矩，能够看到相关的主动化用例。

在用例渠道绑定主动化case，在转测单渠道增加主动化方案，已相关的用例在履行完毕后会主动更新履行状况，进步手动履行的功率。

渠道编写case的常用办法

3.1查询DB数据库

• 在环境变量中装备数据库连接信息
• 在脚本中对数据表进行查询

import json
import requests
from util.db_mysql import DBMySQL
from util.db_redis import DbRedis
def call(env_vars, g_vars, l_vars, sys_funcs, asserts, logger, **kwargs):
    call_param = sys_funcs.get_call_param()
    userId = call_param.get('userId')
    activityId = call_param.get('activityId')
    dbA = DBMySQL(env_vars.get("db.A"))
    if not userId:
        user_var = l_vars.get(call_param.get('var_userId'))
        userId = user_var
    if not activityId:
        activityId_var = l_vars.get(call_param.get('var_activityId'))
        activityId = activityId_var
    if not userId and not activityId:
        raise '请传入查询条件'
    try:
        if not activityId:
            sql = "SELECT * FROM a_coupon_%s WHERE user_id = %s;"%(int(userId)%4,userId)
        elif not userId:
            sql = "SELECT * FROM a_coupon_%s WHERE activity_id = %s;"%(n,activityId)
        else:
            sql = "SELECT * FROM a_coupon_%s WHERE user_id = %s and activity_id = %s;"%(int(userId)%4,userId,activityId)
        logger.info(sql)
        res = dbA.select(sql)
        logger.info(res)
        l_vars.set("select_tableA_res",res)
    except Exception as e:
        logger.info(f'查询失利【{str(e)}】')
        raise e
    finally:
        dbA.close()
        return res

3.2获取运用ip地址作为host域名

• 装备host环境变量：[http://$](http://$){sys.container.ip:app_name}:8888，app_name为服务名
• 调用公共组件获取ip，传入服务名，回来ip
• http恳求时，host挑选对应的环境变量即可

3.3一个case下多个随机账号切换恳求

• 随机创立用户后，获取当时登录信息，将恳求头存到本地变量

import json
import requests
def call(env_vars, g_vars, l_vars, sys_funcs, asserts, logger, **kwargs):
    l_vars.set("user1",l_vars.get("sys.public.login.headers"))

• 鄙人一次再次需求运用该账号时，替换恳求头即可

import json
import requests
def call(env_vars, g_vars, l_vars, sys_funcs, asserts, logger, **kwargs):
    l_vars.set("sys.public.login.headers", l_vars.get("user1"))

运用渠道时遇到的一些问题

4.1查询redis，回来的数据带b’

处理办法一：不运用渠道的东西，代码如下：

import redis
redisConn = redis.Redis(host='redis.host', port=666, password='test123',db=1, decode_responses=True)

处理办法二：redis渠道东西回来是数据是 bytes 类型，需求encoding一下

re = DbRedis.ger_redis(link_info)
test = re.get(test_key)
test_str = test.decode(encoding='utf-8')
key = key+test_str
re.set(key,"aaa")

4.2update、insert、delete句子履行成功，数据库却未生效

处理办法：需求db.commit() ，select句子不需求该句子

dbA = DBMySQL(db_A)
sql = "INSERT INTO t(name,age) VALUES (%s, %s);"
try:
    res = db.insert(sql,['lucy', 18])
    db.commit()
finally:
    dbA.close()

备注：delete办法，删除数据量是0.会有error。

4.3http组件json恳求体中有中文，运行报错

处理办法：恳求头装备 application/json;charset=UTF-8

总结

接入主动化渠道后，便利了很多，也还有更多的运用场景待探索和沟通。主动化最主要的目的是提效，时刻节省下来后咱们能够有更多的时刻去考虑反常场景以及杂乱场景，做一些探索测验，削减由于用例规划遗漏而发生的问题。