? ? ? ?經(jīng)過(guò)一周多時(shí)間，基于python+pytest+excel+allure框架的接口自動(dòng)化測(cè)試初版已基本實(shí)現(xiàn)，包括基本配置讀取、用例讀取、用例執(zhí)行、sql讀取執(zhí)行、前置數(shù)據(jù)準(zhǔn)備、后置數(shù)據(jù)清理以及測(cè)試報(bào)告生成等，環(huán)境獨(dú)立運(yùn)行、項(xiàng)目獨(dú)立運(yùn)行、用例獨(dú)立運(yùn)行、jenkins集成、郵件發(fā)送暫未實(shí)現(xiàn)，再后期版本會(huì)再次推出，現(xiàn)在把整個(gè)框架設(shè)計(jì)思路和想法分享給大家來(lái)參考和借鑒。希望大家也能提供更好的思路和方法幫助我進(jìn)行優(yōu)化改進(jìn)。

? ? ? ?實(shí)戰(zhàn)項(xiàng)目是三端交互的項(xiàng)目，所以在設(shè)計(jì)思路上要考慮多項(xiàng)目如何交互，目前只寫了1個(gè)項(xiàng)目的，其它2個(gè)項(xiàng)目都預(yù)留了位置，后期直接添加就可以，思路一樣。

一、整個(gè)代碼目錄及介紹

common

request.py? 封裝post、get請(qǐng)求方法，供所有地方調(diào)用

login.py?封裝各項(xiàng)目、各種方式的登錄方法，供所有地方調(diào)用

readFile.py?封裝讀取yaml中配置文件、excel中測(cè)試用例方法，供所有地方調(diào)用

execSql.py?封裝sql操作增、刪、改、查方法，供所有地方調(diào)用

prefixSqlData.py 封裝前置、后置公共sql，供所有地方調(diào)用

assertion.py?封裝斷言方法，供所有用例中調(diào)用

logs.py?封裝日志方法，供所有地方調(diào)用

config

sql.yaml? 前置數(shù)據(jù)sql語(yǔ)句、后置清理sql語(yǔ)句等

test.yaml? test環(huán)境數(shù)據(jù)庫(kù)配置、接口域名、測(cè)試賬號(hào)、登錄接口數(shù)據(jù)等

uat.yaml? uat環(huán)境數(shù)據(jù)庫(kù)配置、接口域名、測(cè)試賬號(hào)、登錄接口數(shù)據(jù)等

testcase.xls? 測(cè)試用例等

testcase

項(xiàng)目1

? ? 用例1、用例2

項(xiàng)目2

? ? 用例1、用例2

testcase.xls? 接口測(cè)試用例等

conftest.py? 放置了登錄獲取token供所有用例調(diào)用

run_all_case.py 執(zhí)行所有測(cè)試用例并生成測(cè)試報(bào)告

logs

每天的日志數(shù)據(jù)

report

html??測(cè)試報(bào)告index.html

二、yaml文件基本配置

項(xiàng)目一:url: 'https://www.baidu.com/'headers:Content-Type: 'application/json'Authorization: 'token'account:a_account: '17900000000'b_account: '18000000000'c_account: '19900000000'c_account1: '19900000001'c_login:method: posturl: '/user/login'param:type: "7"login_from: 7mobile: "18888888888"code: "123456"c_sms_code:method: posturl: '/1.0/users/login'mysql:host: 127.0.0.1user: testpwd: testtest_db: user項(xiàng)目二:url: 'https://www.baidu.com/'

三、yaml文件sql配置

項(xiàng)目一:查id:- select_list- select id from A.B where create_mobile={}查用戶id:- select_list- select id from A.B where mobile={}查團(tuán)隊(duì)id:- select_list- select id from A.B where company_user_id=(select id from A.B where mobile={})查C端用戶id:- select_list- select user_id from A.B where mobile in({})解除用戶:- update- update A.B set status=2 where mobile={}項(xiàng)目二:查id:- select_list- select id from A.B where create_mobile={}查用戶id:- select_list- select id from A.B where mobile={}查團(tuán)隊(duì)id:- select_list- select id from A.B where company_user_id=(select id from A.B where mobile={})查C端用戶id:- select_list- select user_id from A.B where mobile in({})解除用戶:- update- update A.B set status=2 where mobile={}

四、讀取yaml文件、excel用例文件

#!/usr/bin/env python # _*_coding:utf-8_*_ import yaml,os,sys,xlwt,xlrd from common.logs import Logclass ReadFile(object):log = Log()_instance=Nonedef __new__(cls,*args,**kwargs):if cls._instance is None:cls._instance=super().__new__(cls)return cls._instancedef __init__(self):self.excel_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'config/testcase.xls')self.yaml_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'config/test.yaml')self.sql_yaml_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'config/sql.yaml')def read_yaml(self,path_type):"""讀yaml文件:return:"""try:if path_type=='yaml_path':file_path=self.yaml_pathelif path_type=='sql_yaml_path':file_path=self.sql_yaml_pathwith open(file_path,'r',encoding='utf-8') as f:return yaml.load(f.read())except Exception as e:self.log.error("讀yaml文件報(bào)錯(cuò){}".format(e))def read_excel(self,sheet_name,function,casename=None):"""讀取excel:param sheet_name::param function::return:"""try:book=xlrd.open_workbook(self.excel_path)sheet=book.sheet_by_name(sheet_name)param=[]for i in range(0,sheet.nrows):if casename==None:if sheet.row_values(i)[0]==function and sheet.row_values(i)[3]==1:param.append(sheet.row_values(i))else:if sheet.row_values(i)[0]==function and sheet.row_values(i)[1]==casename and sheet.row_values(i)[3]==1:param.append(sheet.row_values(i))return paramexcept Exception as e:self.log.error("讀取excel報(bào)錯(cuò){}".format(e))if __name__ == '__main__':test=ReadFile()print(test.read_excel('lxk','我的','全部頁(yè)面'))

五、用例模板如下

根據(jù)每個(gè)sheet存放不同項(xiàng)目的測(cè)試用例，然后根據(jù)再根據(jù)sheet去讀取對(duì)應(yīng)項(xiàng)目模塊的測(cè)試用例

Function模塊、CaseName測(cè)試用例名、Type請(qǐng)求類型、Run是否執(zhí)行、URL接口地址、Headers請(qǐng)求頭、Param請(qǐng)求參數(shù)、SQL1、SQL2、SQL3測(cè)試中需用到的前置數(shù)據(jù)或后置數(shù)據(jù)、AssertType斷言類型，因?yàn)榻涌诜祷氐捻憫?yīng)數(shù)據(jù)可能會(huì)多種多樣，所以這里斷言分了幾種情況、Expect1預(yù)期結(jié)果1、Expect2預(yù)期結(jié)果2、Expect3預(yù)期結(jié)果3

六、request方法

#!/usr/bin/env python # _*_coding:utf-8_*_ import requests,urllib3 from urllib3 import encode_multipart_formdata from common.logs import Logclass RunMethod(object):"""request"""log = Log()urllib3.disable_warnings()def post_main(self,url,data,header,file=None):"""post請(qǐng)求:param url::param data::param header::param file::return:"""res=Noneif file!=None:res=requests.post(url=url,json=data,headers=header,verify=False)else:res = requests.post(url=url, json=data,headers=header, files=file, verify=False)return res.json()def get_main(self,url,header,param=None):"""get請(qǐng)求:param url::param header::param param::return:"""res=Noneif param!=None:res=requests.get(url=url,headers=header,verify=False)else:res = requests.get(url=url, headers=header, json=param,verify=False)return res.json()def run_main(self,method,url,header,data=None,file=None):"""被調(diào)用主request:param method::param url::param header::param data::param file::return:"""try:res=Noneif method=='post' or method=='POST' or method=='Post':res=self.post_main(url,data,header,file=None)elif method=='get' or method=='GET' or method=='Get':res=self.get_main(url,header,param=None)else:return "request傳參錯(cuò)誤"return resexcept Exception as e:self.log.error("請(qǐng)求方法報(bào)錯(cuò){}".farmat(e)) if __name__ == '__main__':print(111)

七、登錄方法

#!/usr/bin/env python # _*_coding:utf-8_*_ from common import request from common.readFile import ReadFile from common.logs import Logclass Login(object):"""登錄"""log = Log()request = request.RunMethod()def __init__(self):self.yaml_data = ReadFile().read_yaml('yaml_path')['lxk']self.header = self.yaml_data['headers']self.url = self.yaml_data['url']self.lxk_c_url = self.yaml_data['c_login']['url']self.lxk_c_method = self.yaml_data['c_login']['method']self.lxk_c_param = self.yaml_data['c_login']['param']def lxk_c_login(self,project,mobile):"""藍(lán)薪卡C端登錄:param project::param mobile::return:"""try:if project=='lxk_c':self.lxk_c_param['mobile']=mobileresult=self.request.run_main(self.lxk_c_method, self.url+self.lxk_c_url, self.header, self.lxk_c_param)elif project=='lxk_a':passelif project=='lxk_b':passreturn resultexcept Exception as e:self.log.error('登錄報(bào)錯(cuò){}'.format(e))if __name__ == '__main__':test=Login()print(test.lxk_c_login('lxk_c','18221124104'))

八、操作sql方法

#!/usr/bin/env python # _*_coding:utf-8_*_ from common.readFile import ReadFile import pymysql import sys from common.logs import Logclass ExecSql(object):"""執(zhí)行sql語(yǔ)句類"""log = Log()_instance=Nonedef __new__(cls,*args,**kwargs):if cls._instance is None:cls._instance=super().__new__(cls)return cls._instancedef __init__(self):"""初始化mysql配置:param platform_name:"""#self.sql_conf = self._get_sql_conf(platform_name)self.sql_conf=Nonedef _get_sql_conf(self, project):"""獲取mysql配置:param platform_name::return:"""try:return ReadFile().read_yaml('yaml_path')[project]['mysql']except:self.log.error("找不到對(duì)應(yīng)項(xiàng)目：{0}".format(project))def connect_db(self):"""連接mysql:return:"""host = self.sql_conf['host']user = self.sql_conf['user']pwd = self.sql_conf['pwd']test_db = self.sql_conf['test_db']try:self.conn = pymysql.connect(host=host, user=user, password=pwd, db=test_db, port=3306, charset="utf8")except Exception as e:self.log.error("連接mysql失敗：{0}".format(e))def get_cursor(self):"""獲取游標(biāo):return:"""self.cursor=self.conn.cursor()return self.cursordef exec_sql(self,project,sql_type,sql):"""執(zhí)行sql語(yǔ)句:param sql_type::param sql::return:"""self.sql_conf = self._get_sql_conf(project)try:if sql_type == 'select_one':self.connect_db()cursor = self.get_cursor()cursor.execute(sql)result = cursor.fetchone()elif sql_type == 'select_list':self.connect_db()cursor = self.get_cursor()cursor.execute(sql)result = cursor.fetchall()elif sql_type == 'update' or sql_type == 'del' or sql_type == 'insert':self.connect_db()result = self.get_cursor().execute(sql)self.conn.commit()self.cursor.close()self.conn.close()return resultexcept Exception as e:self.log.error("執(zhí)行sql語(yǔ)句報(bào)錯(cuò)：{0}".format(e))if __name__ == '__main__':test = ExecSql()a=test.exec_sql('lxk',"select_list","sql)print(aaa)

九、日志方法

#!/usr/bin/env python # _*_coding:utf-8 _*_ import os, time, logging log_path=os.path.join(os.path.dirname(os.path.dirname(os.path.realpath(__file__))), 'logs') if not os.path.exists(log_path): os.mkdir(log_path)class Log(object):"""log日志類"""def __init__(self):self.logname = os.path.join(log_path, '%s.log' % time.strftime('%Y_%m_%d'))self.logger = logging.getLogger()self.logger.setLevel(logging.DEBUG)self.formatter = logging.Formatter('[%(asctime)s]-%(filename)s]-%(levelname)s:%(message)s')def __console(self, level, message):fh=logging.FileHandler(self.logname, 'a', 'utf-8')fh.setLevel(logging.DEBUG)fh.setFormatter(self.formatter)self.logger.addHandler(fh)ch = logging.StreamHandler()ch.setLevel(logging.INFO)ch.setFormatter(self.formatter)self.logger.addHandler(ch)if level == 'info':self.logger.info(message)elif level == 'debug':self.logger.debug(message)elif level == 'warning':self.logger.warning(message)elif level == 'error':self.logger.error(message)self.logger.removeHandler(ch)self.logger.removeHandler(fh)fh.close()def debug(self, message):self.__console('debug', message)def info(self, message):self.__console('info', message)def warning(self, message):self.__console('warning', message)def error(self, message):self.__console('error', message)if __name__ == '__main__':log = Log()log.info("---測(cè)試---")

十、斷言方法

#!/usr/bin/env python # _*_coding:utf-8_*_ from common.execSql import ExecSql from common.logs import Logclass Assertion(object):log=Log()sql_values_list = []response_values = []def __init__(self):self.test=ExecSql().exec_sqldef get_sql_data(self,project,sql_type,sql):'''查詢sql數(shù)據(jù)組合list:param project::param sql_type::param sql::return:'''try:sql_values=self.test(project,sql_type,sql)for i in sql_values:for j in i:self.sql_values_list.append(j)except Exception as e:self.log.error("查詢sql數(shù)據(jù)組合list報(bào)錯(cuò){}".format(e))def get_response_data(self,response_data, keys=[]):'''獲取接口響應(yīng)數(shù)據(jù)組合list:param response_data::param keys::return:'''try:if isinstance(response_data, list):for value in response_data:if isinstance(value, list) or isinstance(value, dict):self.get_response_data(value, keys)elif isinstance(response_data, dict):for i, j in sorted(response_data.items()):if i in keys:self.response_values.append(j)else:self.get_response_data(j, keys)else:passexcept Exception as e:self.log.error("獲取接口響應(yīng)數(shù)據(jù)組合list報(bào)錯(cuò){}".format(e))def asser(self,function,casename,expect,response_data,assert_type=None):'''斷言:param assert_type::param expect::param response_data::return:'''try:if assert_type=='type1':assert self.sql_values_list==self.response_valuesself.log.info("查詢sql數(shù)據(jù)組合list為{}".format(self.sql_values_list))self.log.info("接口響應(yīng)數(shù)據(jù)組合list為{}".format(self.response_values))assert eval(expect)['code'] == response_data['code']assert eval(expect)['msg'] == response_data['msg']self.log.info("{}——{}【PASS】".format(function,casename))except Exception as e:self.log.error("{}——{}【PASS】{}".format(function,casename,e))if __name__ == '__main__':# sql="sql"# test=Assertion()# test.get_sql_data(self,project,sql_type,sql)self.log.error("查詢sql數(shù)據(jù)組合list報(bào)錯(cuò){}".format(e))

十一、conftest登錄獲取token

#!/usr/bin/env python # _*_coding:utf-8_*_ import pytest,os,yaml,requests from common.readFile import ReadFile from common.login import Loginyaml_data=ReadFile().read_yaml('yaml_path')@pytest.fixture(scope='session') def get_lxk_c_headers():"""登錄獲取token更新headers:return:"""headers=yaml_data['lxk']['headers']token=Login().lxk_c_login('lxk_c',yaml_data['lxk']['account']['c_account'])['data']['token']headers['Authorization']=tokenreturn headers

十二、測(cè)試用例方法

#!/usr/bin/env python # _*_coding:utf-8_*_import pytest from common.readFile import ReadFile from common.request import RunMethod from common.assertion import Assertion from common.execSql import ExecSql from common.prefixSqlData import MakeSqlData import alluredata = ReadFile().read_excel('lxk', '我的報(bào)名')@pytest.mark.parametrize('function,casename,type,run,url,hearders,param,sql1,sql2,sql3,asserttype,expect1,expect2,expect3',data) class Test(object):'''我的報(bào)名'''request = RunMethod().run_mainassertion = Assertion()exec_sql = ExecSql().exec_sqlyaml_data = ReadFile().read_yaml('yaml_path')['lxk']sql_yaml_data = ReadFile().read_yaml('sql_yaml_path')['lxk']prefix_sql_data = MakeSqlData('lxk').make_sql_datadef setup_class(self):'''數(shù)據(jù)初始化:return:'''prefix_data=self.prefix_sql_data(self.yaml_data['account']['b_account'], self.yaml_data['account']['c_account'])company_id = self.exec_sql('lxk', self.sql_yaml_data['查企業(yè)id'][0], self.sql_yaml_data['查企業(yè)id'][1].format(self.yaml_data['account']['b_account']))[0][0]task_id = self.exec_sql('lxk', self.sql_yaml_data['查任務(wù)id'][0], self.sql_yaml_data['查任務(wù)id'][1].format(self.yaml_data['account']['b_account']))[0][0]self.exec_sql('lxk', self.sql_yaml_data['報(bào)名任務(wù)'][0], self.sql_yaml_data['報(bào)名任務(wù)'][1].format(prefix_data['user_id'], task_id, company_id,self.yaml_data['account']['c_account'],1))self.exec_sql('lxk', self.sql_yaml_data['報(bào)名任務(wù)'][0], self.sql_yaml_data['報(bào)名任務(wù)'][1].format(prefix_data['user_id'], task_id, company_id,self.yaml_data['account']['c_account'],2))self.exec_sql('lxk', self.sql_yaml_data['報(bào)名任務(wù)'][0], self.sql_yaml_data['報(bào)名任務(wù)'][1].format(prefix_data['user_id'], task_id, company_id,self.yaml_data['account']['c_account'],3))self.exec_sql('lxk', self.sql_yaml_data['報(bào)名任務(wù)'][0], self.sql_yaml_data['報(bào)名任務(wù)'][1].format(prefix_data['user_id'], task_id, company_id,self.yaml_data['account']['c_account'],4))self.exec_sql('lxk', self.sql_yaml_data['報(bào)名任務(wù)'][0], self.sql_yaml_data['報(bào)名任務(wù)'][1].format(prefix_data['user_id'], task_id, company_id,self.yaml_data['account']['c_account'],5))self.exec_sql('lxk', self.sql_yaml_data['報(bào)名任務(wù)'][0], self.sql_yaml_data['報(bào)名任務(wù)'][1].format(prefix_data['user_id'], task_id, company_id,self.yaml_data['account']['c_account'],7))self.exec_sql('lxk', self.sql_yaml_data['報(bào)名任務(wù)'][0], self.sql_yaml_data['報(bào)名任務(wù)'][1].format(prefix_data['user_id'], task_id, company_id,self.yaml_data['account']['c_account'], 8))def teardown_class(self):'''數(shù)據(jù)清理:return:'''self.exec_sql('lxk', self.sql_yaml_data['刪除已報(bào)名任務(wù)'][0], self.sql_yaml_data['刪除已報(bào)名任務(wù)'][1].format(self.yaml_data['account']['c_account']))self.exec_sql('lxk', self.sql_yaml_data['解除用戶團(tuán)隊(duì)'][0], self.sql_yaml_data['解除用戶團(tuán)隊(duì)'][1].format(self.yaml_data['account']['c_account']))#@allure.feature('藍(lán)薪卡')@allure.story('lxk_我的報(bào)名')def test_apply_task(self,get_lxk_c_headers,function,casename,type,run,url,hearders,param,sql1,sql2,sql3,asserttype,expect1,expect2,expect3):'''我的報(bào)名:param get_lxk_c_headers::param function::param casename::param type::param run::param url::param hearders::param param::param sql1::param sql2::param sql3::param asserttype::param expect1::param expect2::param expect3::return:'''response_data = self.request(type,self.yaml_data['url']+url,get_lxk_c_headers,eval(param))self.assertion.get_sql_data('lxk',eval(sql1)[0],eval(sql1)[1].format(self.yaml_data['account']['c_account']))self.assertion.get_response_data(response_data,eval(expect2))self.assertion.asser(function,casename,expect1,response_data,asserttype)if __name__ == "__main__":pytest.main(["-s", "test_001_applyTask.py"])

十三、run_all_case主程序執(zhí)行入口

#!/usr/bin/env python # _*_coding:utf-8_*_ import pytest,os,allure if __name__ == "__main__":pytest.main(['-s',''])#生成測(cè)試報(bào)告jsonpytest.main(["-s", "-q", '--alluredir', 'C:/Users/wangli/PycharmProjects/PytestAutomation/report/result'])#將測(cè)試報(bào)告轉(zhuǎn)為html格式split='allure '+'generate '+'C:/Users/wangli/PycharmProjects/PytestAutomation/report/result '+'-o '+'C:/Users/wangli/PycharmProjects/PytestAutomation/report/html '+'--clean'os.system('cd C:/Users/wangli/PycharmProjects/PytestAutomation/report')os.system(split)print(split)

十四、測(cè)試報(bào)告如下

總結(jié)

以上是生活随笔為你收集整理的Pytest+Allure+Jenkins接口自动化项目实战（一）的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。