在開發測試中常常會遇到接口迭代和代碼重構，一個可有可無的改動每每會引發整個項目的運行。現有的接口測試中每每只是針對單一接口的測試，但是業務的連貫性是很是緊密的，好比：用戶從登錄，獲取商品信息，下單，支付回調的處理，以及退款等流程的處理。發現沒有好使的接口測試工具（jmeter太難用了）。因而本身使用python 實現自動化接口測試的一套腳本，該腳本能夠實現單一接口的測試，和流程的測試；支持多個項目之間的測試；主要的功能以下：javascript
- 讀取Excel
- 接口參數（一、地址欄參數，二、body參數）動態化，能夠從請求返回值中提取數據做爲全局參數，供流程下一步驟使用
- 根據接口發送請求
- 結果的診斷，使用jsonpath
- 結果報告郵件
#!/usr/bin/python
# -*- coding: UTF-8 -*-
import xlrd
import requests
import json
import logging
import smtplib
from email.mime.text import MIMEText
from email.utils import formataddr
import jsonpath
import sys
import traceback
#日誌定義
logging.basicConfig(level=logging.DEBUG, # log level
format='[%(asctime)s] %(levelname)s [%(funcName)s: %(filename)s, %(lineno)d] %(message)s', # log格式
datefmt='%Y-%m-%d %H:%M:%S', # 日期格式
filename='log.txt', # 日誌輸出文件
filemode='a') # 追加模式
class Config:
global PHP_API_DOMAIN #php接口域名
global JAVA_API_DOMAIN #java接口域名
global TEST_CASE_PATH #測試用例excel的路徑
global TEST_CASE_LIST #測試用例轉換後的集合<字典>
global PARAMS #存儲結果變量
global CURRENT_REQUEST_HEADER #存儲當前請求的請求頭
global CURRENT_REQUEST_PARAM #存儲當前請求的請求數據
global ALL_REQUEST_RESULT
global CURRENT_REQUEST_STATUS
def __init__(self,php_api_domain,java_api_domain,test_case_path):
self.PHP_API_DOMAIN = php_api_domain
self.TEST_CASE_PATH = test_case_path
self.JAVA_API_DOMAIN = java_api_domain
self.TEST_CASE_LIST = []
self.PARAMS = {}
self.CURRENT_REQUEST_HEADER = {}
self.CURRENT_REQUEST_PARAM = {}
self.ALL_REQUEST_RESULT = []
self.CURRENT_REQUEST_STATUS = True
headers = {'Accept': 'application/json, text/javascript, */*',
'Accept-Encoding':'gzip, deflate, br',
'Accept-Language':'zh-CN,zh;q=0.9',
'Connection':'Keep-Alive',
'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/52.0.2743.116 Safari/537.36 Edge/15.15063'}
class request_handler:
test_case = {}
result_obj = {}
assert_obj = {}
assert_result = True
need_assert = True
def __init__(self, test_case):
self.test_case = test_case
self.main()
self.return_entity()
def main(self):
self.request(self.test_case)
if self.need_assert:
self.setting_param()
def return_entity(self):
return_entity = {}
return_entity['成功狀態'] = self.assert_result
return_entity['請求結果數據'] = self.result_obj
return_entity['斷言結果'] = self.assert_obj
return_entity['用例數據'] = self.test_case
return_entity['請求參數'] = config.CURRENT_REQUEST_PARAM
return_entity['請求頭'] = config.CURRENT_REQUEST_HEADER
logging.info(return_entity)
if not self.assert_result:
config.ALL_REQUEST_RESULT.append(return_entity)
config.CURRENT_REQUEST_STATUS = False
else:
config.CURRENT_REQUEST_STATUS = True
def request(self,test_case):
try:
if test_case['method'] == 'GET':
self.result_obj = self.request_get(test_case)
if test_case['method'] == 'POST':
self.result_obj = self.request_post(test_case)
if test_case['method'] == 'DELETE':
self.result_obj = self.request_delete(test_case)
if test_case['method'] == 'PUT':
self.result_obj = self.request_put(test_case)
self.assert_handler()
except Exception as e:
self.need_assert = False
self.assert_result = False
self.result_obj = {"status_code": "error", "error_message": e}
traceback.print_exc()
info = traceback.format_exc()
print(info)
def request_get(self,object):
url = self.get_url(object['url'], object['type'], object['param_data'])
head = headers
if isinstance(object['header'],list):
head = self.get_config_param(object['header'], head)
request_body = {}
if isinstance(json.loads(object['body_data']),dict):
request_body = self.get_config_param(json.loads(object['body_data']), request_body)
print("GET URL:"+url)
config.CURRENT_REQUEST_HEADER = head
config.CURRENT_REQUEST_PARAM = request_body
response = requests.get(url=url, params=request_body,headers=head)
return {"status_code" : response.status_code, "response_content": response.text}
def request_delete(self,object):
url = self.get_url(object['url'], object['type'], object['param_data'])
head = headers
if isinstance(object['header'],list):
head = self.get_config_param(object['header'], head)
request_body = {}
if isinstance(json.loads(object['body_data']),dict):
request_body = self.get_config_param(json.loads(object['body_data']), request_body)
print("GET URL:"+url)
config.CURRENT_REQUEST_HEADER = head
config.CURRENT_REQUEST_PARAM = request_body
response = requests.get(url=url, params=request_body,headers=head)
return {"status_code" : response.status_code, "response_content": response.text}
def request_put(self,object):
url = self.get_url(object['url'], object['type'], object['param_data'])
head = headers
if isinstance(object['header'],list):
head = self.get_config_param(object['header'], head)
request_body = {}
if isinstance(json.loads(object['body_data']),dict):
request_body = self.get_config_param(json.loads(object['body_data']), request_body)
print("GET URL:"+url + '; 參數：' + request_body)
config.CURRENT_REQUEST_HEADER = head
config.CURRENT_REQUEST_PARAM = request_body
response = requests.get(url=url, params=request_body,headers=head)
return {"status_code" : response.status_code, "response_content": response.text}
def request_post(self,object):
head = headers
#將header中的數據添加到請求中
if isinstance(json.loads(object['header']),dict):
head = self.get_config_param(json.loads(object['header']),head)
url = self.get_url(object['url'], object['type'], object['param_data'])
#將data中的動態參數從共享參數中添加到數據中
request_body = {}
if isinstance(json.loads(object['body_data']),dict):
request_body = self.get_config_param(json.loads(object['body_data']), request_body)
print("POST URL:" + url+";param:"+str(request_body))
config.CURRENT_REQUEST_HEADER = head
config.CURRENT_REQUEST_PARAM = request_body
response = requests.post(url, request_body, headers=head)
return {"status_code": response.status_code, "response_content": response.text}
def get_config_param(self, list, object):
for param_name in list.keys():
param_value = list[param_name]
if not str(param_value).startswith("$P_"):
object[param_name] = param_value
else:
body_value = config.PARAMS[param_value] if param_value in config.PARAMS.keys() else param_value
object[param_name] = body_value
return object
def get_url(self, path, type, param_data):
request_url = (config.PHP_API_DOMAIN + '/' + path) if type == 'php' else (config.JAVA_API_DOMAIN + '/' + path)
param_data = json.loads(param_data)
url_param = self.get_config_param(param_data, {})
request_url = request_url + '?a=a'
for key in url_param.keys():
request_url = request_url + "&" + key + '=' + url_param[key]
return request_url
def assert_handler(self):
test_case = self.test_case
result_obj = self.result_obj
self.assert_obj = {"測試用例名稱:": test_case['case_name']}
# 訪問接口直接報錯直接
if result_obj['status_code'] == 'error':
self.assert_obj = dict(self.assert_obj.items(), {"請求狀態：": "錯誤", "錯誤信息:": result_obj['error_message']}.items())
self.need_assert = False
self.assert_result = False
return
# 狀態大於300的異常直接處理
if result_obj['status_code'] >= 300:
self.assert_obj['請求狀態'] = result_obj['status_code']
self.assert_obj['錯誤信息'] = result_obj['response_content']
self.need_assert = False
self.assert_result = False
return
# 請求狀態爲成功，判斷業務狀態
buiess_content = json.loads(result_obj['response_content'])
expect_res = json.loads(test_case['expect_res'])
# 校驗規則
for ruler in expect_res:
matcher = jsonpath.jsonpath(buiess_content, ruler['rule'])
if isinstance(matcher,bool):
self.assert_obj['請求狀態'] = result_obj['status_code']
self.assert_obj['錯誤信息'] = '規則：' + ruler['rule'] + '值不匹配，指望是：' + str(ruler['value']) + ',返回是：' + str(
matcher)
self.assert_result = False
break
elif len(matcher) == 0:
self.assert_obj['請求狀態'] = result_obj['status_code']
self.assert_obj['錯誤信息'] = '規則：' + ruler['rule'] + '提取出的數組長度爲0'
self.assert_result = False
break
elif ruler['type'] == 'value': # 對值進行比較
if matcher[0] == ruler['value']:
continue
else:
self.assert_obj['請求狀態'] = result_obj['status_code']
self.assert_obj['錯誤信息'] = '規則：' + ruler['rule'] + '值不匹配，指望是：' + str(ruler['value']) + ',返回是：' + str(
matcher[0])
self.assert_result = False
break
elif ruler['type'] == 'length': # 對數組長度進行比較
if len(matcher) == ruler['value']:
continue
else:
self.assert_obj['請求狀態'] = result_obj['status_code']
self.assert_obj['錯誤信息'] = '規則：' + ruler['rule'] + '大小不匹配，指望是：' + str(ruler['value']) + ',返回是：' + str(
len(matcher))
self.assert_result = False
break
else:
self.assert_obj['請求狀態'] = result_obj['status_code']
self.assert_obj['錯誤信息'] = '規則：' + ruler['rule'] + '錯誤'
self.assert_result = False
break
def setting_param(self):
actives = json.loads(self.test_case['active'])
buiess_content = json.loads(self.result_obj['response_content'])
for active in actives:
try:
p_name = active['p_name']
p_value = jsonpath.jsonpath(buiess_content, active['p_value'])
if isinstance(p_value, bool):
p_value = ''
elif len(p_value) == 0:
p_value = ''
else:
p_value = p_value[0]
config.PARAMS[p_name] = p_value
except Exception as e:
traceback.print_exc()
info = traceback.format_exc()
print(info)
def read_excel():
wb = xlrd.open_workbook(config.TEST_CASE_PATH)
sh = wb.sheet_by_name('Sheet1')
#處理 Excel的數據
for row_ndex in range(sh.nrows):
test_case = dict(zip(sh.row_values(0), sh.row_values(row_ndex)))
try:
if row_ndex == 0:
continue
request_handler(test_case)
if test_case['error_continue'] == 0 and not config.CURRENT_REQUEST_STATUS:
break
except Exception as e:
traceback.print_exc()
info = traceback.format_exc()
print(info)
#config.ALL_REQUEST_RESULT.append({"測試用例": test_case, "請求異常": e})
#發送郵件通知成功與否
message = ''
for result_record in config.ALL_REQUEST_RESULT:
message += json.dumps(result_record,ensure_ascii=False) +'\n\n\n'
send_email(message)
def send_email(message):
if len(message) == 0:
return
my_sender = 'xxx@xxx.com' # 發件人郵箱帳號
my_pass = 'xxxxxx' # 發件人郵箱密碼
my_user = 'xxx@xxx.com' # 收件人郵箱帳號，我這邊發送給本身
ret = True
try:
msg = MIMEText(message, 'plain', 'utf-8')
msg['From'] = formataddr(["小課接口測試報告", my_sender]) # 括號裏的對應發件人郵箱暱稱、發件人郵箱帳號
msg['To'] = formataddr(["250", my_user]) # 括號裏的對應收件人郵箱暱稱、收件人郵箱帳號
msg['Subject'] = "小課接口測試報告" # 郵件的主題，也能夠說是標題
server = smtplib.SMTP_SSL("imap.exmail.qq.com", 465) # 發件人郵箱中的SMTP服務器，端口是25
server.login(my_sender, my_pass) # 括號中對應的是發件人郵箱帳號、郵箱密碼
server.sendmail(my_sender, [my_user, ], msg.as_string()) # 括號中對應的是發件人郵箱帳號、收件人郵箱帳號、發送郵件
server.quit() # 關閉鏈接
except Exception as e: # 若是 try 中的語句沒有執行，則會執行下面的 ret=False
traceback.print_exc()
info = traceback.format_exc()
print(info)
ret = False
if ret:
print("發送郵件成功")
else:
print('發送郵件失敗')
#config = Config(r'php域名', r'java域名',r'excel路徑.xlsx')
if __name__ == '__main__':
argv = sys.argv
if len(sys.argv) != 4:
print('參數個數不正確')
php_api = argv[1]
java_api = argv[2]
excel_path = argv[3]
config = Config(php_api, java_api, excel_path)
read_excel()
EXCEL模板案列php
| case_name | header | type | url | method | param_data | body_data | execp_res | memo | active | error_continue |
| 獲取用戶token | {} | java | student-service/get-student-token | GET | {"student_id":25009} | {} | [{"rule":"$.code","type":"value","value":0}] | [{"p_name":"$P_student_id","p_type":"int",java "p_value":"$.data.student_id"},{"p_name":"$P_token","p_type":"string",python "p_value":"$.data.token"}]json |
1 | |
| 獲取商品信息 | {"token":"$P_token"} | php | product/info | POST | {} | {"product_id":3100052} | [{"rule":"$.code","type":"value","value":0}] | [{"p_name":"$P_product_id",api "p_value":"$.data.id"},{"p_name":"$P_course_id",數組 "p_value":"$.data.course_id"},{"p_name":"$P_total_money",服務器 "p_value":"$.data.total_money"},{"p_name":"$P_actual_money",app "p_value":"$.data.actual_money"}]dom |
0 | |
| 下單 | {"token":"$P_token"} | php | order/create | POST | {} | {"product_id":"$P_product_id","student_id":"$P_student_id", "course_id":"$P_course_id", "total_money":"$P_total_money", "actual_money":"$P_actual_money","group_buy_order_id":null} |
[{"rule":"$.code","type":"value","value":0}] | [{"p_name":"$P_order_id", "p_value":"$.data.order_id"}] |
1 |
字段說明：
case_name：測試用例名稱
type：接口提供者（java/php）
url：請求地址path
method：請求接口類型（POST/GET/PUT/DELETE）
param_data：請求參數（須要放在地址欄上的參數），格式json
body_data：請求接口類型不是GET時，請求的body參數
expect_res：對接口返回數據的斷言 ;rule：用jsonpath提取返回字段，type:value/length 標識判斷字段值的長度仍是值匹配，value:若是type == value，則判斷返回字段值和value值是否相同；type==length 判斷返回的數組或字符串長度是否相同
[{"rule":"$.code","type":"value","value":0},{"rule":"$.code","type":"length","value":1}]
active：將返回的數據值保存，做爲下一步驟的請求參數，以下含義案例的含義是：存儲返回值的json中的data下的student_id的值，使用時用 $P_student_id參數使用
[{"p_name":"$P_student_id","p_type":"int","p_value":"$.data.student_id"}]
error_continue：當前步驟診斷錯誤時，是否繼續執行後面步驟
案列說明：
上述excel流程是，獲取用戶token ---》請求商品參數 ----》從商品中獲取參數 ----》建立訂單 流程，固然還有取消，支付，退款就不一一演示了