结合单元测试框架pytest+数据驱动模型+allure
目录
api： 存储测试接口conftest.py :设置前置操作目前前置操作：1、获取token并传入headers，2、获取命令行参数给到环境变量,指定运行环境commmon：存储封装的公共方法connect_mysql.py：连接数据库http_requests.py: 封装自己的请求方法logger.py: 封装输出日志文件read_yaml.py：读取yaml文件测试用例数据read_save_data.py：读取保存的数据文件case: 存放所有的测试用例data：存放测试需要的数据save_data: 存放接口返回数据、接口下载文件test_data: 存放测试用例依赖数据upload_data: 存放上传接口文件logs: 存放输出的日志文件report: 存放测试输出报告getpathinfo.py :封装项目测试路径pytest.int :配置文件requirement.txt: 本地python包（pip install -r requirements.txt 安装项目中所有python包）run_main.py: 项目运行文件
结构设计
1.每一个接口用例组合在一个测试类里面生成一个py文件2.将每个用例调用的接口封装在一个测试类里面生成一个py文件3.将测试数据存放在yml文件中通过parametrize进行参数化，实现数据驱动4.通过allure生成测试报告
代码展示
api/api_service.py #需要测试的一类接口
'''
Code description:服务相关接口
Create time: 2020/12/3
Developer: 叶修
'''
import os
from common.http_requests import HttpRequestsclass Api_Auth_Service(object):def __init__(self):self.headers = HttpRequests().headersdef api_home_service_list(self):# 首页服务列表# url = "http://192.168.2.199:9092/v1/auth/auth_service/findAuthService"url = os.environ["host"] + "/v1/auth/auth_service/findAuthService" # 读取conftest.py文件地址进行拼接response = HttpRequests().get(url, headers=self.headers, verify=False)# print(response.json())return responsedef get_service_id(self):#获取银行卡三要素认证服务idurl = "http://192.168.2.199:9092/v1/auth/auth_service/findAuthService"#url = os.environ["host"] + "/v1/auth/auth_service/findAuthService" # 读取conftest.py文件地址进行拼接response = HttpRequests().get(url,headers=self.headers)#print(response.json()['data'][0]['service_list'][0]['id'])service_id = response.json()['data'][0]['service_list'][1]['id']return service_iddef api_service_info(self,serviceId='0b6cf45bec757afa7ee7209d30012ce1',developerId=''):#服务详情body = {"serviceId" :serviceId,"developerId":developerId}url = "http://192.168.2.199:9092/v1/auth/auth_service/findServiceDetail"#url = os.environ["host"] + "/v1/auth/auth_service/findServiceDetail"#读取conftest.py文件地址进行拼接response = HttpRequests().get(url,headers=self.headers,params = body,verify=False)#print(response.json())return responsedef api_add_or_update_service(self,api_param_req,api_param_res,description,error_code,icon,id,interface_remarks,name,product_info,request_method,sample_code,sort,type,url):#服务添加或者更新body={"api_param_req": api_param_req,"api_param_res": api_param_res,"description": description,"error_code": error_code,"icon": icon,"id": id,"interface_remarks": interface_remarks,"name": name,"product_info": product_info,"request_method": request_method,"sample_code": sample_code,"sort": sort,"type": type,"url": url,}#url = "http://192.168.2.199:9092/v1/auth/auth_service/insertOrUpdateService"url = os.environ["host"] + "/v1/auth/auth_service/insertOrUpdateService" # 读取conftest.py文件地址进行拼接response = HttpRequests().post(url,json=body,headers=self.headers,verify=False)return responsedef api_add_service_price(self,id,max_number,money,service_id,small_number):#服务价格添加body = {"id": id,"max_number": max_number,"money": money,"service_id": service_id,"small_number": small_number}# url = "http://192.168.2.199:9092/v1/auth/auth_service/insertServicePrice"url = os.environ["host"] + "/v1/auth/auth_service/insertServicePrice" # 读取conftest.py文件地址进行拼接response = HttpRequests().post(url, json=body, headers=self.headers, verify=False)return responsedef api_apply_service(self,developer_id,service_id):#申请服务body ={"developer_id": developer_id,"service_id": service_id}# url = "http://192.168.2.199:9092/v1/auth/auth_service/applyService"url = os.environ["host"] + "/v1/auth/auth_service/applyService" # 读取conftest.py文件地址进行拼接response = HttpRequests().post(url, json=body, headers=self.headers, verify=False)return responseif __name__ == '__main__':#Auth_Service().api_home_service_list()Api_Auth_Service().get_service_id()#Auth_Service().api_service_info()api/get_token.py#获取登录token
'''
Code description：获取token
Create time：2020-12-03
Developer：叶修
'''
import osimport urllib3
from common.http_requests import HttpRequestsclass Get_Token(object):def get_token(self,account='****',password='****'):#url = "http://192.168.2.199:9092/v1/auth/developer/accountLogin"url = os.environ["host"]+"/v1/auth/developer/accountLogin"body = {"account": account,"password": password,}urllib3.disable_warnings()r = HttpRequests().post(url, json=body,verify=False)#print(r.json())token = r.json()['data']['token']params = {"access_token": token}HttpRequests().params.update(params)#更新token到sessionreturn tokenif __name__ == '__main__':print(Get_Token().get_token())case/test_service_info.py #上面接口某一测试用例
'''
Code description: 服务详情
Create time: 2020/12/3
Developer: 叶修
'''
import sys
import allure
import pytest
from common.logger import Log
from common.read_yaml import ReadYaml
from api.api_auth_service.api_auth_service import Api_Auth_Servicetestdata = ReadYaml("auth_service.yml").get_yaml_data() # 读取数据@allure.feature('服务详情')
class Test_Service_Info(object):log = Log()@pytest.mark.process@pytest.mark.parametrize('serviceId,developerId,expect', testdata['service_info'],ids=['服务详情'])def test_service_info(self,serviceId,developerId,expect):self.log.info('%s{%s}' % ((sys._getframe().f_code.co_name,'------服务详情接口-----')))with allure.step('获取服务id'):serviceId = Api_Auth_Service().get_service_id()with allure.step('服务详情'):msg = Api_Auth_Service().api_service_info(serviceId,developerId)self.log.info('%s:%s' % ((sys._getframe().f_code.co_name, '获取请求结果：%s' % msg.json())))# 断言assert msg.json()["result_message"] == expect['result_message']assert msg.json()['result_code'] == expect['result_code']assert 'url' in msg.json()['data']conftest.py
'''
Code description:配置信息
Create time: 2020/12/3
Developer: 叶修
'''
import os
import pytest
from api.get_token import Get_Token
from common.http_requests import HttpRequests@pytest.fixture(scope="session")
def get_token():'''前置操作获取token并传入headers'''Get_Token().get_token()if not HttpRequests().params.get("access_token", ""):#没有get到token，跳出用例pytest.skip("未获取token跳过用例")yield HttpRequests().reqHttpRequests().req.close()def pytest_addoption(parser):parser.addoption("--cmdhost", action="store", default="http://192.168.1.54:32099",help="my option: type1 or type2")
@pytest.fixture(scope="session",autouse=True)
def host(request):'''获取命令行参数'''#获取命令行参数给到环境变量#pytest --cmdhost 运行指定环境os.environ["host"] = request.config.getoption("--cmdhost")print("当前用例运行测试环境:%s" % os.environ["host"])common/connect_mysql.py
'''
Code description: 配置连接数据库
Create time: 2020/12/3
Developer: 叶修
'''
import pymysqldbinfo = {"host":"******","user":"root","password":"******","port":31855
}
class DbConnect():def __init__(self,db_conf,database=""):self.db_conf = db_conf#打开数据库self.db = pymysql.connect(database = database,cursorclass = pymysql.cursors.DictCursor,**db_conf)#使用cursor()方式获取操作游标self.cursor = self.db.cursor()def select(self,sql):#sql查询self.cursor.execute(sql)#执行sqlresults = self.cursor.fetchall()return resultsdef execute(self,sql):#sql 删除 提示 修改try:self.cursor.execute(sql)#执行sqlself.db.commit()#提交修改except:#发生错误时回滚self.db.rollback()def close(self):self.db.close()#关闭连接def select_sql(select_sql):'''查询数据库'''db = DbConnect(dbinfo,database='auth_platform')result = db.select(select_sql)db.close()return resultdef execute_sql(sql):'''执行SQL'''db = DbConnect(dbinfo,database='auth_platform')db.execute(sql)db.close()if __name__ == '__main__':sql = "SELECT * FROM auth_platform.auth_service where name='四要素认证'"sel = select_sql(sql)[0]['name']print(sel)common/http_requests.py
'''
Code description: 封装自己的请求类型
Create time: 2020/12/3
Developer: 叶修
'''import requests# 定义一个HttpRequests的类
class HttpRequests(object):req = requests.session()#定义session会话# 定义公共请求头headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/84.0.4147.125 Safari/537.36','cookie':''}params = {'access_token':''}# 封装自己的get请求,获取资源def get(self, url='', params='', data='', headers=None, cookies=None,stream=None,verify=None):response = self.req.get(url,params=params,data=data,headers=headers,cookies=cookies,stream=stream,verify=verify)return response# 封装自己的post方法，创建资源def post(self, url='', params='',data='', json='', headers=None, cookies=None,stream=None,verify=None):response = self.req.post(url,params=params,data=data,json=json,headers=headers,cookies=cookies,stream=stream,verify=verify)return response# 封装自己的put方法，更新资源def put(self, url='', params='', data='', headers=None, cookies=None,verify=None):response = self.req.put(url, params=params, data=data, headers=headers, cookies=cookies,verify=verify)return response# 封装自己的delete方法，删除资源def delete(self, url='', params='', data='', headers=None, cookies=None,verify=None):response = self.req.delete(url, params=params, data=data, headers=headers, cookies=cookies,verify=verify)return responsecommon/logger.py
'''
Code description: 封装输出日志文件
Create time: 2020/12/3
Developer: 叶修
'''
import logging,time
import os
import getpathinfopath = getpathinfo.get_path()#获取本地路径
log_path = os.path.join(path,'logs')# log_path是存放日志的路径
# 如果不存在这个logs文件夹，就自动创建一个
if not os.path.exists(log_path):os.mkdir(log_path)class 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 - %(name)s - %(levelname)s - %(message)s')def __console(self,level,message):#创建一个fileHander，用于写入本地fh = logging.FileHandler(self.logname,'a',encoding='utf-8')fh.setLevel(logging.DEBUG)fh.setFormatter(self.formatter)self.logger.addHandler(fh)#创建一个StreamHandler,用于输入到控制台ch = logging.StreamHandler()ch.setLevel(logging.DEBUG)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(fh)self.logger.removeHandler(ch)#关闭打开文件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('测试')log.debug('测试')log.warning('测试')log.error('测试')common/read_save_data.py
'''
Code description: 读取保存数据
Create time: 2020/12/8
Developer: 叶修
'''
import os
import yaml
import getpathinfo
class Read_Save_Date():def __init__(self):path = getpathinfo.get_path()#获取本地路径self.head_img_path = os.path.join(path,'data','save_data')+"/"+'head_img_path.txt'# head_img_path文件地址self.order_id_path = os.path.join(path, 'data', 'save_data') + "/" + 'order_id.txt' # order_id.txt文件地址def get_head_img_path(self):# 获取head_img_pathwith open(self.head_img_path, "r", encoding="utf-8")as f:return f.read()def get_order_id(self):# 获取order_idwith open(self.order_id_path, "r", encoding="utf-8")as f:return f.read()if __name__ == '__main__':print(Read_Save_Date().get_head_img_path())print(Read_Save_Date().get_order_id())common/read_yaml.py
'''
Code description: 读取yml文件测试数据
Create time: 2020/12/3
Developer: 叶修
'''
import os
import yaml
import getpathinfo
class ReadYaml():def __init__(self,filename):path = getpathinfo.get_path()#获取本地路径self.filepath = os.path.join(path,'data','test_data')+"/"+filename#拼接定位到data文件夹def get_yaml_data(self):with open(self.filepath, "r", encoding="utf-8")as f:# 调用load方法加载文件流return yaml.load(f,Loader=yaml.FullLoader)if __name__ == '__main__':data = ReadYaml("auth_service.yml").get_yaml_data()['add_or_update_service']print(data)data/
data/test_data/auth_service.yml
home_service_list:- [{'result_code': '0', 'result_message': '处理成功'}]
service_info:- ['','',{'result_code': '0', 'result_message': '处理成功'}]
add_or_update_service:- [['1','1','1','1','1','123456','1','测试','1','1','1','1','1','1'],{'result_code': '0', 'result_message': '处理成功'}]
add_service_price:- ['123456789','10','0','','0',{'result_code': '0', 'result_message': '处理成功'}]
apply_service:- ['','',{'result_code': '0', 'result_message': '处理成功'}]logs/
report/
getpathinfo.py
'''
Code description:配置文件路径
Create time: 2020/12/3
Developer: 叶修
'''
import osdef get_path():# 获取当前路径curpath = os.path.dirname(os.path.realpath(__file__))return curpathif __name__ == '__main__':# 执行该文件，测试下是否OKprint('测试路径是否OK,路径为：', get_path())pytest.ini
#pytest.ini
[pytest]
markers = process
addopts = -p no:warnings
#addopts = -v --reruns 1 --html=./report/report.html --self-contained-html
#addopts = -v --reruns 1 --alluredir ./report/allure_raw
#addopts = -v -s -p no:warnings --reruns 1 --pytest_report ./report/Pytest_Report.htmlrequirements.txt
allure-pytest==2.8.18
allure-python-commons==2.8.18
BeautifulReport==0.1.3
beautifulsoup4==4.9.3
ddt==1.4.1
Faker==4.18.0
Flask==1.1.1
httpie==1.0.3
httplib2==0.9.2
HttpRunner==1.5.8
py==1.9.0
PyMySQL==0.10.1
pytest==6.1.1
pytest-base-url==1.4.2
pytest-cov==2.10.1
pytest-forked==1.3.0
pytest-html==2.1.1
pytest-instafail==0.4.2
pytest-metadata==1.10.0
pytest-mock==3.3.1
pywin32==228
PyYAML==5.3.1
requests==2.22.0
requests-oauthlib==1.3.0
requests-toolbelt==0.9.1run_main.py
'''
Code description: 运行主流程测试用例
Create time: 2020/11/5
Developer: 叶修
'''
import os
import pytest
if __name__ == '__main__':pytest.main(['-m','process', '-s','--alluredir', 'report/tmp'])#-m运行mark标记文件os.system('allure generate report/tmp -o report/html --clean') # /report/tmp 为存放报告的源文件目录
