0前言
相信大家對接口自動化已經不陌生了，這是幾乎我們每個迭代都會投入的事情，但耗費了這么多精力去編寫和維護，實際的收益如何呢？如果收益不好，是不是說明我們自動化case的實現方式、使用方式還有改進的地方呢？以下是接入得物接口自動化平臺后的一些實踐和想法，歡迎大家積極交流～
1淺談接口自動化
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在自動化平臺上的實踐
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，在轉測單平臺添加自動化計劃，已關聯的用例在執行結束后會自動更新執行狀態，提高手動執行的效率。
3平臺編寫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://${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使用平臺時遇到的一些問題
4.1 查詢redis，返回的數據帶b'
解決方法一：不使用平臺的工具，代碼如下：
import redis
redisConn = redis.Redis(host='redis.host', port=666, password='test123',db=1, decode_respnotallow=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.2 update、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.3 http組件json請求體中有中文，運行報錯
解決方式：請求頭配置 application/json;charset=UTF-8
5總結
接入自動化平臺后，方便了很多，也還有更多的使用場景待探索和交流。自動化最主要的目的是提效，時間節省下來后我們可以有更多的時間去思考異常場景以及復雜場景，做一些探索測試，減少因為用例設計遺漏而發生的問題。
