一.項目目錄
二.項目結構
三.源碼地址
gitee ：https://gitee.com/your_dad_died/bm_pytest_api
github： https://github.com/Theshy0000/bm_pytest_api
四.conftest文件全局變量設置
1.使用conftest配合@pytest.fixture()使用在用例所在的文件中不需要導入可以直接使用。
conftest.py
import pytest
from data.config_data import *
@pytest.fixture()
def login():
print(u"登錄成功！")
@pytest.fixture()
def global_variable():
'''全局變量配置'''
# 調用config_data文件內配置的信息TEST_IP，DOWNSTREAM_TOKEN，UPSTREAM_TOKEN
variable = {
'test_ip': TEST_IP,
'downstream_token': DOWNSTREAM_TOKEN,
'upstream_token': UPSTREAM_TOKEN
}
return variable
五.參數配置（一）
1.其他文件通過導入進行讀取
config_data.py
# 配置ip，token，headers
#測試環境
TEST_IP = 'https://xxx.cn'
#亞申token
YA_SHENG_TOKEN = 'eyxxcFoxODdrbk1DWG85QTdxY3pwakNoUGVvVEM1Ym1BPSJ9'
#中昌token 下游
DOWNSTREAM_TOKEN='eyxPSJ9'
DOWNSTREAM_HEADERS={
'user-agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/97.0.4692.71 Safari/537.36',
'token':DOWNSTREAM_TOKEN,
'Content-Type':'application/json'
}
#三笑token 上游
UPSTREAM_TOKEN='eyJlbmNyeXB0HFUN3d1Q5QfQ=='
UPSTREAM_HEADERS={
'user-agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/97.0.4692.71 Safari/537.36',
'token':UPSTREAM_TOKEN,
'Content-Type':'application/json'
}
#數據庫配置，使用數據庫操作時使用
MYSQL_CONFIG=("101.xx.xx.xx",3306, 'root', '123456', 'drf')
六.參數配置（二）
1.create_bill_data和delete_bill_data是接口原始參數，存放在data文件下。
2.replace_expression是接口之間依賴表達式替換，直接用變量寫的，也可以寫在配置yaml文件下。
3.rv是提取的參數所儲存的字典。
raw_parameter.py
'''這裡可以分文件管理，全部寫一起不好，我就寫幾個例子就不分了'''
upstream_create_waybill_data={
"waybillNo":"Y22011101026000006",
"orderNo":"",
"projectName":"XM-揚州三笑-揚州康遠",
"projectNo":"320SF000206004",
"customer":"揚州三笑物流有限公司",
"customerNo":"320SF000206",
"customerOrderNo":"客戶訂單號",
"thirdOrderNo":"12",
"shipperName":"喬巴",
"shipperPhone":"18000001111",
"shipperProvince":"海南省",
"shipperProvinceCode":"460000",
"shipperCity":"定安縣",
"shipperCityCode":"469021",
"shipperDistrict":"定安縣",
"shipperDistrictCode":"469021",
"shipperAddress":"青浦區徐泾鎮明珠路1008號",
"receiverName":"路飛",
"receiverPhone":"18000002222",
"receiverProvince":"廣東省",
"receiverProvinceCode":"440000",
"receiverCity":"中山市",
"receiverCityCode":"442000",
"receiverDistrict":"中山市",
"receiverDistrictCode":"442000",
"receiverAddress":"縣政府",
"requiredDeliveryDate":None,
"requiredArrivalDate":None,
"line":"定安縣-中山市",
"hasReceipt":True,
"requirementRemark":"",
"transportMode":"WHOLE_VEHICLE",
"carModel":"GUA_CHE",
"carLength":"5.0",
"freight":555,
"contractFreight":None,
"serviceType":None,
"chargeMode":"CAR_SAY_VALUATION",
"settlement":"D",
"cashPayment":"",
"arrivalPayment":555,
"receiptPayment":"",
"goodsList":[
{
"goodsName":"1",
"goodsWeight":1,
"goodsVolume":1,
"goodsNums":1,
"goodsWorth":1,
"goodsCosts":1,
"goodsUnit":"1"
}
],
"showFreight":None,
"sourceFrom":"PC"
}
create_bill_data={
"contractNo":"HT-91367323584",
"sheetDateEnum":"SEND_DATE",
"year":"2022",
"month":"01",
"projectName":"XM-三笑海運-中昌",
"projectNo":"320SF000206003",
"downCompanyName":"",
"startDate":"2022-01-01 00:00:00",
"endDate":"2022-01-31 23:59:59",
"upCompanyName":"審稿人",
"generateBillOperator":"012001011"
}
delete_bill_data={
'id':1111
}
六.接口參數依賴關聯封裝
1.參數提取：調用parameter.update_rv()方法設置參數提取
2.set()標記的地方先使用了集合去重的特性，最後用列表返回l
parameter_setting.py
# 解決接口之間參數依賴替換,先找出相同的鍵值，然後把接口返回結果的值賦給替換表達式的值，最後把替換表達式的值賦值給請求參數
class ParameterSetting(object):
'''參數相關工具方法集合字典'''
returned_value = {
}
def parameters_depend(self, request_parameter, replace_expression, rv):
''' :param request_parameter: 原請求參數 :param replace_expression: 替換參數表達式 字典 :param rv: 接口返回接口保存字典 :return: '''
key_list = self.__identical_key(request_parameter, replace_expression)
for i in key_list:
replace_expression[i] = rv[i]
request_parameter[i] = replace_expression[i]
return request_parameter
def __identical_key(self, v1, v2):
''' :param v1: 字典 :param v2: 字典 :return: 找出2個字典相同的鍵以列表格式返回鍵值 '''
s = set()
for k in v1.keys():
for k1 in v2.keys():
if k == k1:
s.add(k)
return list(s)
@property
def rv(self):
'''解決接口之間參數依賴,讀取別的接口的返回結果，使用變量名作為字典鍵值讀取'''
return self.returned_value
def update_rv(self, value):
''' 寫入返回結果，使用變量名和值的方式保存 :param value: 字典格式數據 '''
for k, v in value.items():
self.returned_value[k] = v
七.requests二次簡單封裝
1.直接寫get和post的2種方法，提前把ip，headers信前封裝，針對公司業務封裝上下游企業。
requests_.py
# 二次簡單封裝
import json
import requests
from data.config_data import TEST_IP,DOWNSTREAM_HEADERS,UPSTREAM_HEADERS
class Requests:
def __init__(self, up_or_down='up'):
self.ip = TEST_IP # 調用config_data配置文件內的配置信息TEST_IP
self.up_or_down = up_or_down
if self.up_or_down == 'up':
self.headers = UPSTREAM_HEADERS # 調用config_data配置文件內的配置信息UPSTREAM_HEADERS
elif self.up_or_down == 'down':
self.headers = DOWNSTREAM_HEADERS # 調用config_data配置文件內的配置信息DOWNSTREAM_HEADERS
def get(self, url):
# url使用ip+url拼接，並調用headers參數
result = requests.get(url=self.ip+url, headers=self.headers)
return result
def post(self, url, data):
# url使用ip+url拼接，並調用headers參數，對入參數進行轉換json格式
result = requests.post(url=self.ip+url, headers=self.headers, data=json.dumps(data))
return result
八.實例一
bill_test.py
# 賬單相關流程：1.新增賬單,2.刪除賬單
import random
import allure
from data.raw_parameter import *
from tool.requests_ import Requests
from tool.parameter_setting import ParameterSetting
# 調用ParameterSetting類，並進行類的實例化
parameter = ParameterSetting()
# 存儲接口返回參數，接口依賴時讀取（調用實例化parameter類中的rv方法）
rv = parameter.rv
# 調用Requests類，並進行類的實例化（調用Requests類中的各種請求方法）
requests = Requests()
@allure.description(u"賬單操作")
class TestBill:
@allure.description(u"創建賬單")
def test_create_bill(self,global_variable): # 調用全局conftest.py文件中的global_variable方法
print(u'使用下全局變量{global_variable["test_ip"]}')
# 調用requests類中post方法
# 調用raw_parameter類中create_bill_data參數，並請求返回json數據
result = requests.post('/bill/createCommonBill',create_bill_data).json()
print(result)
# 調用ParameterSetting類中update_rv方法，存儲接口返回的值
parameter.update_rv({
'id':result['content']['id']})
@allure.description(u"刪除賬單")
def test_delete_bill(self):
# 使用replace_expression參數化上游接口存儲返回的值，此處replace_expression和調用ParameterSetting類中的表達式一致
# 使用rv表示接口返回結果保存為字典與調用ParameterSetting類中的方法rv一致
replace_expression = {
'id':rv['id']}
# 調用ParameterSetting類中dparameters_depend方法
# 調用raw_parameter類中delete_bill_data參數
parameter.parameters_depend(delete_bill_data,replace_expression,rv)
# 調用requests類中post方法
result = requests.post('/bill/deleteBillCommon',delete_bill_data)
print(result.json()) # 並請求返回json數據
1.原始數據的’id’:1111也被替換。
九.實例二
1.result為接口返回的完整響應主體內容，通過字典鍵值提取值，給與這個值賦值給waybillNo這個變量，後續通過waybillNo可以使用這個提取的值。
2.upstream_create_waybill_data 是接口原始參數，存放在data文件下。
3.replace_expression是依賴表達式，直接用變量寫的，也可以寫在配置文件yaml文件下。
4.rv 之前提取的參數所儲存的字典。
main_process_test.py
''' 主業務流程： 1.獲取運單號單 2.錄單 3.獲取調度單號 4.創建調度單選擇下游承運商（回單付） 5.派單 6.發車 7.到達 9.回單上傳 10.回單確認 11.生成回單付支付訂單 12.獲取支付訂單信息 13.線上支付到卡 14.支付審核信息查詢 15.支付審批通過 16.新增上游賬單 17.確認上游賬單 19.切換班博系統認領流水 20.流水核銷上游賬單 21.生成下游賬單 22.確認下游賬單 23.生成尾款 24.尾款支付 '''
import random
import allure
import pytest
from data.raw_parameter import *
from tool.requests_ import Requests
from tool.parameter_setting import ParameterSetting
# 調用ParameterSetting類，並進行類的實例化
parameter = ParameterSetting()
# 存儲接口返回參數，接口依賴時讀取（調用實例化parameter類中的rv方法）
rv = parameter.rv # 存儲接口返回參數，接口依賴時讀取
# 調用Requests類，並進行類的實例化（調用Requests類中的各種請求方法）
requests = Requests()
@allure.description(u"主流程")
@pytest.mark.waybill
class TestMainProcess:
@allure.description(u"獲取運單號")
def test_get_waybill_no(self):
result = requests.get('/waybill/generateWaybillNo').json()
# 參數提取
# 調用ParameterSetting類中update_rv方法，存儲接口返回的值
# result為接口返回的完整響應主體內容，通過字典鍵值提取值，給與這個值賦值給waybillNo這個變量，
# 後續通過waybillNo可以使用這個提取的值。
parameter.update_rv({
'waybillNo': result['content']})
assert result['code'] == 0
@allure.description(u"創建運單")
def test_create_waybill(self):
# 依賴表達式
# 使用replace_expression參數化上游接口存儲返回的值，此處replace_expression和調用ParameterSetting類中的表達式一致
# 使用rv表示接口返回結果保存為字典與調用ParameterSetting類中的方法rv一致
# upstream_create_waybill_data是接口原始參數，存放在data文件下。
# replace_expression是依賴表達式，直接用變量寫的，也可以寫在配置yaml文件內。
# rv是提取的參數所儲存的字典。
replace_expression = {
'waybillNo': rv['waybillNo']}
# 方法參數替換
# 調用ParameterSetting類中dparameters_depend方法
# 調用raw_parameter類中upstream_create_waybill_data參數
parameter.parameters_depend(upstream_create_waybill_data, replace_expression, rv)
# 字典原始參數替換
upstream_create_waybill_data['customerOrderNo'] = 'ce{random.randint(100, 999)}'
result = requests.post('/waybill/createWaybill', upstream_create_waybill_data).json()
assert result['code'] == 1
5.注意：在截圖中 “waybillNo”:“Y22011101026000006”，經過替換變了。
十.日志封裝
1.顏色文件存儲位置，儲存文件日志級別日志名稱等
2.from tool.log import logger調用日志封裝文件
log.py
import logging
#設置日志顏色的包
import colorlog
import datetime
''' Loggers：記錄器，提供應用程序代碼能直接使用的接口； Handlers：處理器，將記錄器產生的日志發送至目的地； Filters：過濾器，提供更好的粒度控制，決定哪些日志會被輸出； Formatters：格式化器，設置日志內容的組成結構和消息字段。 %(name)s Logger的名字 #也就是其中的.getLogger裡的路徑,或者我們用他的文件名看我們填什麼 %(levelno)s 數字形式的日志級別 #日志裡面的打印的對象的級別 %(levelname)s 文本形式的日志級別 #級別的名稱 %(pathname)s 調用日志輸出函數的模塊的完整路徑名，可能沒有 %(filename)s 調用日志輸出函數的模塊的文件名 %(module)s 調用日志輸出函數的模塊名 %(funcName)s 調用日志輸出函數的函數名 %(lineno)d 調用日志輸出函數的語句所在的代碼行 %(created)f 當前時間，用UNIX標准的表示時間的浮 點數表示 %(relativeCreated)d 輸出日志信息時的，自Logger創建以 來的毫秒數 %(asctime)s 字符串形式的當前時間。默認格式是 “2003-07-08 16:49:45,896”。逗號後面的是毫秒 %(thread)d 線程ID。可能沒有 %(threadName)s 線程名。可能沒有 %(process)d 進程ID。可能沒有 %(message)s用戶輸出的消息 '''
'''日志顏色配置'''
log_colors_config = {
#顏色支持 blue藍，green綠色，red紅色，yellow黃色，cyan青色
'DEBUG': 'cyan',
'INFO': 'green',
'WARNING': 'yellow',
'ERROR': 'red',
'CRITICAL': 'bold_red',
}
'''創建logger記錄器'''
logger = logging.getLogger('test')
# 輸出到控制台
console_handler = logging.StreamHandler()
# 輸出到文件
path = "./log"
'''獲取當前年月日作為日志文件名'''
fileName = str(datetime.datetime.now().year) + '-' + str(datetime.datetime.now().month) + '-' + str(
datetime.datetime.now().day) + '.log'
file_handler = logging.FileHandler(filename=path+'\\'+fileName, mode='a', encoding='utf8')
'''日志級別設置'''
#logger控制最低輸出什麼級別日志(優先級最高)
logger.setLevel(logging.DEBUG)
#console_handler設置控制台最低輸出級別日志
console_handler.setLevel(logging.DEBUG)
#console_handler設置保存到文件最低輸出級別日志
file_handler.setLevel(logging.INFO)
# 日志輸出格式
#保存文件的日志格式
file_formatter = logging.Formatter(
fmt='[%(asctime)s] %(filename)s -> %(funcName)s line:%(lineno)d [%(levelname)s] : %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
#控制台的日志格式
console_formatter = colorlog.ColoredFormatter(
#輸出那些信息，時間，文件名，函數名等等
fmt='%(log_color)s[%(asctime)s] %(filename)s -> %(funcName)s line:%(lineno)d [%(levelname)s] : %(message)s',
#時間格式
datefmt='%Y-%m-%d %H:%M:%S',
log_colors=log_colors_config
)
console_handler.setFormatter(console_formatter)
file_handler.setFormatter(file_formatter)
# 重復日志問題：
# 這裡進行判斷，如果logger.handlers列表為空，則添加，否則，直接去寫日志，解決重復打印的問題
if not logger.handlers:
logger.addHandler(console_handler)
logger.addHandler(file_handler)
console_handler.close()
file_handler.close()
if __name__ == '__main__':
logger.debug(u'顏色')
logger.info(u'綠色測試日志保存路徑{}'.format(path+fileName))
logger.warning(u'顏色')
logger.error('error')
logger.critical('critical')
十一.yaml文件配置
1.實現讀和寫，判斷文件不存在則新建後讀寫
2.這裡使用了’a’ 操作文件，追加
yaml_.py
import os
from ruamel import yaml
class YamlUsage:
def __init__(self,file_path):
self.file_path = file_path
def write_yaml(self,data,mode):
self.__check()
file = open(self.file_path, mode, encoding='utf-8')
yaml.dump(data, file, Dumper=yaml.RoundTripDumper)
file.close()
def read_yaml(self):
self.__check()
file = open(self.file_path, 'r', encoding='utf-8')
with file as doc:
content = yaml.load(doc, Loader=yaml.Loader)
return content
def __check(self):
if os.path.exists(self.file_path):
pass
else:
file = open(self.file_path, 'w')
file.close()
if __name__ == '__main__':
#在這裡寫入了測試bill_search_test.py文件的數據，也可以手動直接再yaml文件裡寫
yml = YamlUsage(r'F:\bm_api\data\bill_search_data.yaml')
data1 = {
"billCommonNo": "",
"status": None,
"makeBillStatus": None,
"projectNos": [],
"commonCompanyName": "",
"year": "2022",
"month": "07",
"billType": "UP",
"size": 20,
"page": 1
}
data2 = {
"billCommonNo": "",
"status": None,
"makeBillStatus": None,
"projectNos": ['320SF000206003'],
"commonCompanyName": "",
"year": "2022",
"month": "07",
"billType": "UP",
"size": 20,
"page": 1
}
data3 = {
"billCommonNo": "",
"status": 'RECONCILIATIONING',
"makeBillStatus": None,
"projectNos": [],
"commonCompanyName": "",
"year": "2022",
"month": "07",
"billType": "UP",
"size": 20,
"page": 1
}
dat = [data1,data2,data3]
yml.write_yaml(data,'a')
print(yml.read_yaml())
bill_search_data.yaml
- billCommonNo: ''
status:
makeBillStatus:
projectNos: []
commonCompanyName: ''
year: '2022'
month: '07'
billType: UP
size: 20
page: 1
- billCommonNo: ''
status:
makeBillStatus:
projectNos:
- 320SF000206003
commonCompanyName: ''
year: '2022'
month: '07'
billType: UP
size: 20
page: 1
- billCommonNo: ''
status: RECONCILIATIONING
makeBillStatus:
projectNos: []
commonCompanyName: ''
year: '2022'
month: '07'
billType: UP
size: 20
page: 1
十二.mysql文件配置
mysql.py
import pymysql
from data.config_data import MYSQL_CONFIG
class MysqlDb():
def __init__(self, host, port, user, password, db_name):
self.db = pymysql.connect(host=host,port=port,user=user,passwd=password,db=db_name)
self.cur = self.db.cursor(cursor=pymysql.cursors.DictCursor)
# 對象資源被釋放時觸發，在對象即將被刪除時的最後操作
def __del__(self):
self.cur.close()
self.db.close()
def select_db(self, sql):
"""查詢"""
self.cur.execute(sql)
data = self.cur.fetchall()
return data
def execute_db(self, sql):
"""更新/插入/刪除"""
try:
self.cur.execute(sql)
self.db.commit()
except Exception as e:
print(u"操作出現錯誤：{}".format(e))
self.db.rollback()
if __name__ == '__main__':
aa = '99' # 參數化sql
mysql_db = MysqlDb(*MYSQL_CONFIG)
print(mysql_db.select_db('SELECT * FROM case_test'))
mysql_db.execute_db("INSERT INTO case_test( aa, `add`) VALUES ('1', '1');")
mysql_db.execute_db("UPDATE case_test SET aa = '2', `add` = {aa} WHERE id = 1;")
mysql_db.execute_db("delete from case_test where aa={aa}")
print(mysql_db.select_db('SELECT * FROM case_test'))
# 多表聯查
print(mysql_db.select_db("SELECT * FROM apitest_environment as e left join auth_user as u on e.user_id=u.id "))
十三.pytest.ini文件配置
pytest.ini
[pytest]
# 更改默認運行pytest命令，設置報告原始數據的目錄
addopts = -vs --alluredir ./report/allure_raw --durations=0 --alluredir ./report/allure_raw
# 設置運行用例目錄
testpaths = case
# 設置標簽，可以根據標簽來運行指定用例
markers =
waybill : run waybill case -m waybill
bill :run bill case -m bill
十四.實例三
bill_search_test.py
'''賬單搜索接口'''
import allure
import pytest
import time
from tool.log import logger
from tool.requests_ import Requests
from tool.yaml_ import YamlUsage
# 調用Requests類，並進行類的實例化（調用Requests類中的各種請求方法）
request = Requests()
# 調用yaml_文件中的配置信息
yml = YamlUsage(r'F:\bm_api\data\bill_search_data.yaml')
# 調用YamlUsage類中的read_yaml方法
data_yml = yml.read_yaml()
@allure.description(u'賬單搜索')
# 給每組參數都給與傳參
@pytest.mark.parametrize('data', data_yml, ids=[u'賬單年月搜索',u'項目搜索',u'狀態搜索'])
def test_bill_search(data):
time.sleep(1) # 搜索接口太頻繁會訪問失敗
result = request.post('/bill/getBillCommonList', data=data).json()
logger.info(u'接口地址:/bill/getBillCommonList ,請求參數:{data},返回結果:{result}')
assert result['code'] == 0
十五.run運行文件
run.py
import pytest
import os
import shutil
if __name__ == '__main__':
try:
# 刪除之前的文件夾
shutil.rmtree("report/allure_raw")
except:
print(u'之前未生成報告原文件')
pytest.main([])
#編譯報告原文件並啟動報告服務
os.system('allure serve report/allure_raw')
1.allure報告
轉載：https://blog.csdn.net/aaaaaaaaanjjj/article/details/122487373
