乐趣区

关于项目管理:利用自动化平台可以做的那亿点事-｜得物技术

jiezi

前言

置信大家对接口自动化曾经不生疏了，这是简直咱们每个迭代都会投入的事件，但消耗了这么多精力去编写和保护，理论的收益如何呢？如果收益不好，是不是阐明咱们自动化 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

总结

接入自动化平台后，不便了很多，也还有更多的应用场景待摸索和交换。自动化最次要的目标是提效，工夫节省下来后咱们能够有更多的工夫去思考异样场景以及简单场景，做一些摸索测试，缩小因为用例设计脱漏而产生的问题。

退出移动版