Commit 9b5e26af by shenjiamin

api.1

parent ff2f409f
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: ${PROJECT_NAME}
# FileName: ${FILE_NAME}
# Author: shen
# Datetime: ${DATE} ${TIME}
# Description:
#
# ---------------------------------------------------------------------------
import functools
import logging
import os.path
import time
log_path = os.path.join(os.path.join(os.path.join(os.path.dirname(__file__),"report"),"log"),time.strftime("%Y_%m_%d_%H_%M_%S",time.localtime())+".log")
def project_log(log_file):
#创建logger对象
logger = logging.getLogger()
#设置日志的级别
logger.setLevel("INFO")
#创建日志文件的handler
handler = logging.FileHandler(log_file,mode="a",encoding="utf-8")
#设置日志显示的格式
formatter = logging.Formatter("%(asctime)s - %(levelname)s: %(message)s")
#给日志文件添加日志的格式
handler.setFormatter(formatter)
#给logger对象添加handler
logger.addHandler(handler)
return logger
project_log = project_log(log_path)
def log_decorator(func_name):
@functools.wraps(func_name)
def inner(*args, **kwargs):
try:
project_log.info(f"执行的功能为:{func_name.__name__}, 功能的描述为:{func_name.__doc__}, 所在的文件路径为:{func_name.__code__.co_filename}, 所做的行为:{func_name.__code__.co_firstlineno}")
res = func_name(*args, **kwargs)
except Exception as e:
project_log.error(f"执行的功能为:{func_name.__name__}, 功能的描述为:{func_name.__doc__}, 所在的文件路径为:{func_name.__code__.co_filename}, 所做的行为:{func_name.__code__.co_firstlineno}, 产生错误,错误的类型为:{type(e)}, 错误的描述为:{e}")
raise e
else:
return res
return inner
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: ${PROJECT_NAME}
# FileName: ${FILE_NAME}
# Author: shen
# Datetime: ${DATE} ${TIME}
# Description:
#
# ---------------------------------------------------------------------------
import pymysql
from APIAutoTest_shen_2 import log_decorator
from APIAutoTest_shen_2.common.read_basic_ini import ReadBasicIni
class DB:
@log_decorator
def __init__(self):
"""连接数据库,获取连接对象和游标对象"""
# 创建ReadBasicIni对象
ini = ReadBasicIni()
#创建连接对象
self.conn = pymysql.connect(
host=ini.get_database_conn_msg("host"),
port=int(ini.get_database_conn_msg("port")),
user=ini.get_database_conn_msg("user"),
password=ini.get_database_conn_msg("pwd"),
database=ini.get_database_conn_msg("database"),
charset="utf8"
)
#创建游标对象
self.cursor = self.conn.cursor()
@log_decorator
def close(self):
"""关闭连接对象和游标对象"""
#先关闭游标对象再关闭连接对象
self.cursor.close()
self.conn.close()
@log_decorator
def delete(self,sql_data):
"""执行sql的删除语句"""
#执行删除语句
self.cursor.execute(sql_data)
#连接对象提交
self.conn.commit()
@log_decorator
def select(self,sql_data):
"""执行sql的查询语句"""
#执行查询语句
self.cursor.execute(sql_data)
#获取查询结果并返回
select_res = self.cursor.fetchall()
#判断是否有查询结果,有就返回第一个结果,没有就返回none
if select_res:
return select_res[0][0]
if __name__ == '__main__':
db = DB()
sql = """DELETE FROM uc_demension where CODE_ = "test_dem";"""
sql1 = """select ID_ from uc_demension where CODE_="test_dem1";"""
print(db.select(sql1))
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: ${PROJECT_NAME}
# FileName: ${FILE_NAME}
# Author: shen
# Datetime: ${DATE} ${TIME}
# Description:
#
# ---------------------------------------------------------------------------
import configparser
import os.path
from APIAutoTest_shen_2 import log_decorator
class ReadBasicIni:
@log_decorator
def __init__(self):
"""获取项目下基础配置ini配置文件的路径,创建Configparser对象,读取ini文件"""
self.data_config_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), "data_config")
ini_path = os.path.join(self.data_config_path, "config.ini")
self.conf = configparser.ConfigParser()
self.conf.read(ini_path, encoding="utf-8")
@log_decorator
def get_host(self, key):
"""根据key,获取host节点下系统的域名"""
return self.conf.get("host", key)
@log_decorator
def get_database_conn_msg(self, key):
"""根据key,获取sql节点下数据库的连接信息"""
return self.conf.get("sql", key)
@log_decorator
def get_user_data_dir_name(self, key):
"""根据key,获取用户存放数据的目录路径"""
# 获取用户存放数据的目录名称
user_data_dir_name = self.conf.get("user", key)
# 用户存放数据的目录名称和data_config的目录路径进行路径拼接,再返回
return os.path.join(self.data_config_path, user_data_dir_name)
if __name__ == '__main__':
ini = ReadBasicIni()
#print(ini.get_file_path("excel"))
print(ini.get_host("bpm"))
#print(ini.get_table_name("name"))
# print(ini.get_database_conn_msg("host"))
# print(ini.get_user_data_dir_name("zs"))
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: ${PROJECT_NAME}
# FileName: ${FILE_NAME}
# Author: shen
# Datetime: ${DATE} ${TIME}
# Description:
#
# ---------------------------------------------------------------------------
import openpyxl
from APIAutoTest_shen_2 import log_decorator
from APIAutoTest_shen_2.common.read_basic_ini import ReadBasicIni
from APIAutoTest_shen_2.common.read_json import read_json
from APIAutoTest_shen_2.common.read_user_ini import ReadUserIni
class ReadExcel:
@log_decorator
def __init__(self,username):
"""获取excel和所有json文件的路径以及excel工作表的名称,获取工作薄和工作表,获取所有json文件数据对应的python对象"""
#创建ReadBasicIni的对象,主要是用来获取被测系统的域名
self.basic_ini = ReadBasicIni()
#创建ReadUserIni的对象,来获取用户的数据
self.ini = ReadUserIni(username)
# 获取excel文件的路径和所有json文件的路径,获取工作表的名称
case_data_path = self.ini.get_file_path("case")
expect_data_path = self.ini.get_file_path("expect")
excel_path = self.ini.get_file_path("excel")
sql_data_path = self.ini.get_file_path("sql")
table_name = self.ini.get_table_name("name")
#获取工作薄和工作表
#获取excel工作薄的路径
wb = openpyxl.load_workbook(excel_path)
#获取工作表
self.ws = wb[table_name]
#调用read_json函数,获取json文件的数据
#获取用例文件的数据
self.case_data_dict = read_json(case_data_path)
#获取期望用例的数据
self.expect_data_dict = read_json(expect_data_path)
#获取sql语句的数据
self.sql_data_dict = read_json(sql_data_path)
@log_decorator
#使用私有方法根据行和列获取单元格的数据
def __get_cell_value(self,column,row):
"""根据行和列获取指定单元格的数据"""
value = self.ws[column+str(row)].value
#判断单元格数据是否不为none,如果不为none,将单元格数据去掉前后空格之后再返回,如果为none,将去掉空格为空的字符串返回none
if isinstance(value,str) and value.strip():
return value.strip()
@log_decorator
def module_name(self,row):
"""根据行,获取模块名称"""
#调用私有方法__get_cell_value,根据行来获取模块名称
return self.__get_cell_value("b",row)
@log_decorator
def api_name(self,row):
"""根据行,获取接口名称"""
#调用私有方法__get_cell_value,根据行来获取接口名称
return self.__get_cell_value("c",row)
@log_decorator
def case_url(self,row):
"""根据行,获取url"""
# 调用私有方法__get_cell_value,根据行来获取url
path = self.__get_cell_value("f",row)
#判断路径是否不为none,如果不为none,和域名进行拼接
if path:
return self.basic_ini.get_host("bpm") + path
@log_decorator
def case_mime(self,row):
"""根据行,获取用例的媒体类型"""
# 调用私有方法__get_cell_value,根据行来获取媒体类型
mime = self.__get_cell_value("g",row)
#判断媒体类型的值,如果不为none,全部转小写再返回
if mime:
return mime.lower()
@log_decorator
def case_req(self,row):
"""根据行获取用例的请求方法"""
# 调用私有方法__get_cell_value,根据行来获取请求方法
return self.__get_cell_value("h",row)
@log_decorator
def case_data(self,row):
"""根据行,获取用例的数据"""
# 调用私有方法__get_cell_value,根据行来获取用例数据
case_data_key = self.__get_cell_value("i",row)
# 判断用例数据的key是否不为None,如果不为None,就提取用例数据,如果为None,就返回None
if case_data_key:
#获取模块名称
module_name = self.module_name(row)
#获取接口名称
api_name = self.api_name(row)
#提取用例数据并返回
return self.case_data_dict[module_name][api_name][case_data_key]
@log_decorator
def expect_data(self,row):
"""根据行获取期望数据"""
# 调用私有方法__get_cell_value,根据行来获取期望数据
expect_data_key = self.__get_cell_value("j",row)
#判断期望数据的key是否不为none,如果不为none,提取期望数据,如果为none,就返回none
if expect_data_key:
#获取模块名称
module_name = self.module_name(row)
api_name = self.api_name(row)
#提取期望数据并返回
return self.expect_data_dict[module_name][api_name][expect_data_key]
@log_decorator
def sql_data(self,row):
"""根据行获取sql数据"""
# 调用私有方法__get_cell_value,根据行来获取sql数据
sql_data_key = self.__get_cell_value("l",row)
#判断sql数据是否不为none,如果不为none,就提取数据,如果为none,就返回none
if sql_data_key:
#获取模块名称
module_name = self.module_name(row)
#获取接口名称
api_name = self.api_name(row)
#提取sql数据并返回
return self.sql_data_dict[module_name][api_name][sql_data_key]
@log_decorator
def sql_type(self,row):
"""根据行获取sql类型"""
# 调用私有方法__get_cell_value,根据行来获取sql类型
sql_statement =self.__get_cell_value("k",row)
#判断sql语句类型的值是否不为none,如果不为none,就将值转小写再返回
if sql_statement:
return sql_statement.lower()
@log_decorator
def update_key(self,row):
"""根据行获取更新的值"""
# 调用私有方法__get_cell_value,根据行来获取更新的值
return self.__get_cell_value("m",row)
@log_decorator
def get_data(self):
"""获取所有的测试数据,放在二维列表中"""
list_data = []
for row in range(2,self.ws.max_row+1):
url = self.case_url(row)
mime = self.case_mime(row)
req = self.case_req(row)
case = self.case_data(row)
expect = self.expect_data(row)
sql = self.sql_data(row)
sql_type1 = self.sql_type(row)
update = self.update_key(row)
#过滤空行
if url is not None and req is not None and expect is not None:
#将每行数据放在列表中,再将列表追加到之前创建的空列表中
list_data.append([url,mime,req,case,expect,sql,sql_type1,update])
else:
return list_data
if __name__ == '__main__':
excel = ReadExcel("zs")
print(excel.get_data())
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: ${PROJECT_NAME}
# FileName: ${FILE_NAME}
# Author: shen
# Datetime: ${DATE} ${TIME}
# Description:
#
# ---------------------------------------------------------------------------
import json
from APIAutoTest_shen_2 import log_decorator
@log_decorator
def read_json(filepath):
"""读取json文件,将json文件的内容序列化为python对象再返回"""
with open(filepath,mode="r",encoding="utf-8") as f:
#序列化为python对象再返回
return json.load(f)
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: ${PROJECT_NAME}
# FileName: ${FILE_NAME}
# Author: shen
# Datetime: ${DATE} ${TIME}
# Description:
#
# ---------------------------------------------------------------------------
import configparser
import os.path
from APIAutoTest_shen_2 import log_decorator
from APIAutoTest_shen_2.common.read_basic_ini import ReadBasicIni
class ReadUserIni:
@log_decorator
def __init__(self,username):
"""获取项目下用户ini配置文件的路径,创建Configparser对象,读取ini文件"""
#创建ReadBasicIni的对象,调用get_user_data_dir_name方法,获取用户存放数据的目录路径
self.user_data_dir_path = ReadBasicIni().get_user_data_dir_name(username)
# 获取ini文件的路径
ini_path = os.path.join(self.user_data_dir_path,"config.ini")
# 创建Configparser对象
self.conf = configparser.ConfigParser()
# Configparser对象对象调用read方法读取ini文件
self.conf.read(ini_path,encoding="utf-8")
@log_decorator
def get_file_path(self,key):
"""根据key,获取file节点下文件的路径"""
return os.path.join(self.user_data_dir_path,self.conf.get("file",key))
@log_decorator
def get_table_name(self,key):
"""根据key,获取table节点下工作表的名称"""
return self.conf.get("table",key)
if __name__ == '__main__':
res = ReadUserIni("zs")
print(res.get_table_name("name"))
[host]
# 配置项目的域名
bpm= http://36.139.193.99:8088
[sql]
# 数据库的连接信息
host=36.139.193.99
port=3306
user=root
pwd=Rhrc@2024
database=eip8
[user]
# 随着用户的增加而增加选项,配置的是数据的目录名称
zs=zhangsan_data
{
"认证接口": {
"登录系统": {
"LoginSuccess": {
"username": "admin",
"password": "WuHRoZXGrg1LHOix8EVOhqGcu3CEEd0hOWNPRENTScZMb2ekrEtBGGBJweUxGKi36k346+OrS5mD3nnjyrocZOfzo4UqoshNoqJgx09u81LV2vc53Sy83DsBWKGoZZZRUhsDrpRj8feUlakJMCpgqswG0y9jm95Lk9auWpI146I="
},
"LoginErrorUsernameIsNone": {
"username": "",
"password": "WuHRoZXGrg1LHOix8EVOhqGcu3CEEd0hOWNPRENTScZMb2ekrEtBGGBJweUxGKi36k346+OrS5mD3nnjyrocZOfzo4UqoshNoqJgx09u81LV2vc53Sy83DsBWKGoZZZRUhsDrpRj8feUlakJMCpgqswG0y9jm95Lk9auWpI146I="
},
"LoginErrorUsernameIsShort": {
"username": "a",
"password": "WuHRoZXGrg1LHOix8EVOhqGcu3CEEd0hOWNPRENTScZMb2ekrEtBGGBJweUxGKi36k346+OrS5mD3nnjyrocZOfzo4UqoshNoqJgx09u81LV2vc53Sy83DsBWKGoZZZRUhsDrpRj8feUlakJMCpgqswG0y9jm95Lk9auWpI146I="
},
"LoginErrorUsernameIsLong": {
"username": "adminadminadminadminadminadminadmin",
"password": "WuHRoZXGrg1LHOix8EVOhqGcu3CEEd0hOWNPRENTScZMb2ekrEtBGGBJweUxGKi36k346+OrS5mD3nnjyrocZOfzo4UqoshNoqJgx09u81LV2vc53Sy83DsBWKGoZZZRUhsDrpRj8feUlakJMCpgqswG0y9jm95Lk9auWpI146I="
},
"LoginErrorUsernameIsSpecial": {
"username": "▣▤▥▦▩◘◙",
"password": "WuHRoZXGrg1LHOix8EVOhqGcu3CEEd0hOWNPRENTScZMb2ekrEtBGGBJweUxGKi36k346+OrS5mD3nnjyrocZOfzo4UqoshNoqJgx09u81LV2vc53Sy83DsBWKGoZZZRUhsDrpRj8feUlakJMCpgqswG0y9jm95Lk9auWpI146I="
},
"LoginErrorUsernameIsError": {
"username": "Admin123",
"password": "WuHRoZXGrg1LHOix8EVOhqGcu3CEEd0hOWNPRENTScZMb2ekrEtBGGBJweUxGKi36k346+OrS5mD3nnjyrocZOfzo4UqoshNoqJgx09u81LV2vc53Sy83DsBWKGoZZZRUhsDrpRj8feUlakJMCpgqswG0y9jm95Lk9auWpI146I="
},
"LoginErrorPwdIsNone": {
"username": "admin",
"password": ""
},
"LoginErrorPwdIsShort": {
"username": "admin",
"password": "1"
},
"LoginErrorPwdIsSpecial": {
"username": "admin",
"password": "▣▤▥▦▩◘◙"
},
"LoginErrorPwdIsLong": {
"username": "admin",
"password": "WuHRoZXGrg1LHOix8EVOhqGcu3CEEd0hOWNPRENTScZMb2ekrEtBGGBJweUxGKi36k346+OrS5mD3nnjyrocZOfzo4UqoshNoqJgx09u81LV2vc53Sy83DsBWKGoZZZRUhsDrpRj8feUlakJMCpgqswG0y9jm95Lk9auWpI146I=WuHRoZXGrg1LHOix8EVOhqGcu3CEEd0hOWNPRENTScZMb2ekrEtBGGBJweUxGKi36k346+OrS5mD3nnjyrocZOfzo4UqoshNoqJgx09u81LV2vc53Sy83DsBWKGoZZZRUhsDrpRj8feUlakJMCpgqswG0y9jm95Lk9auWpI146I="
}
}
},
"维度管理": {
"添加维度": {
"AddDemSuccess": {
"code": "test_dem1",
"description": "测试添加的维度",
"isDefault": 1,
"name": "测试维度"
}
},
"根据维度编码删除维度": {
"DeleteDemSuccess": {
"ids": "需要更新"
}
},
"更新维度": {
"PutDemSuccess": {
"code": "test_dem1",
"description": "测试添加的维度",
"isDefault": 1,
"name": "测试维度-更新"
}
}
},
"组织管理": {
"添加组织": {
"AddOrgSuccess": {
"code": "test_org_code",
"demId": "需要更新",
"exceedLimitNum": 0,
"grade": "",
"limitNum": 0,
"name": "测试组织",
"nowNum": 0,
"orderNo": 0,
"parentId": "0"
}
},
"组织加入用户": {
"UserJoinOrgSuccess": {"orgCode": "test_org_code", "accounts": "admin"}
},
"保存组织参数": {
"SaveOrgParamSuccess": {
"query": {"orgCode": "test_org_code"},
"json": [{"alias":"sz","value":100000}]
}
},
"删除组织": {
"DeleteOrgSuccess": "test_org_code"
}
}
}
[file]
#配置数据层中工作表的名称
excel=apiAutoTest.xlsx
case=case_data.json
expect=expect_data.json
sql=sql_data.json
[table]
#配置数据层中工作表的名称
name=BPM
{
"认证接口": {
"登录系统": {
"LoginSuccess": {
"username": "超级管理员",
"account": "admin",
"userId": "1",
"expiration": 86400
},
"LoginErrorUsernameIsNone": {
"message": "账号或密码错误"
},
"LoginErrorUsernameIsShort": {
"message": "账号或密码错误"
},
"LoginErrorUsernameIsLong": {
"message": "账号或密码错误"
},
"LoginErrorUsernameIsSpecial": {
"message": "账号或密码错误"
},
"LoginErrorUsernameIsError": {
"message": "账号或密码错误"
},
"LoginErrorPwdIsNone": {
"message": "账号或密码错误"
},
"LoginErrorPwdIsShort": {
"message": "账号或密码错误"
},
"LoginErrorPwdIsSpecial": {
"message": "账号或密码错误"
},
"LoginErrorPwdIsLong": {
"message": "账号或密码错误"
}
},
"刷新token": {
"RefreshSuccess": {
"message": "刷新成功"
}
}
},
"维度管理": {
"添加维度": {
"AddDemSuccess": {
"message": "添加维度成功!"
}
},
"根据维度编码删除维度": {
"DeleteDemSuccess": {
"message": "删除维度成功!"
}
},
"更新维度": {
"PutDemSuccess": {
"message": "更新维度成功!"
}
}
},
"组织管理": {
"添加组织": {
"AddOrgSuccess": {
"message": "添加组织成功"
}
},
"组织加入用户": {
"UserJoinOrgSuccess": {"message": "加入成功"}
},
"保存组织参数": {
"SaveOrgParamSuccess": {"message":"保存组织参数成功!"}
},
"删除组织": {
"DeleteOrgSuccess": {"message":"删除组织成功!"}
}
}
}
{
"维度管理": {
"添加维度": {
"AddDemSuccess": "delete from uc_demension where CODE_=\"test_dem1\";"
},
"根据维度编码删除维度": {
"DeleteDemSuccess": "select ID_ from uc_demension where CODE_=\"test_dem1\";"
}
},
"组织管理": {
"添加组织": {
"AddOrgSuccess": {
"select": "select ID_ from uc_demension where CODE_=\"test_org_code\"; ",
"delete": "delete from uc_org where CODE_=\"test_org_code\";"
}
}
}
}
body {
font-family: Helvetica, Arial, sans-serif;
font-size: 12px;
/* do not increase min-width as some may use split screens */
min-width: 800px;
color: #999;
}
h1 {
font-size: 24px;
color: black;
}
h2 {
font-size: 16px;
color: black;
}
p {
color: black;
}
a {
color: #999;
}
table {
border-collapse: collapse;
}
/******************************
* SUMMARY INFORMATION
******************************/
#environment td {
padding: 5px;
border: 1px solid #e6e6e6;
vertical-align: top;
}
#environment tr:nth-child(odd) {
background-color: #f6f6f6;
}
#environment ul {
margin: 0;
padding: 0 20px;
}
/******************************
* TEST RESULT COLORS
******************************/
span.passed,
.passed .col-result {
color: green;
}
span.skipped,
span.xfailed,
span.rerun,
.skipped .col-result,
.xfailed .col-result,
.rerun .col-result {
color: orange;
}
span.error,
span.failed,
span.xpassed,
.error .col-result,
.failed .col-result,
.xpassed .col-result {
color: red;
}
.col-links__extra {
margin-right: 3px;
}
/******************************
* RESULTS TABLE
*
* 1. Table Layout
* 2. Extra
* 3. Sorting items
*
******************************/
/*------------------
* 1. Table Layout
*------------------*/
#results-table {
border: 1px solid #e6e6e6;
color: #999;
font-size: 12px;
width: 100%;
}
#results-table th,
#results-table td {
padding: 5px;
border: 1px solid #e6e6e6;
text-align: left;
}
#results-table th {
font-weight: bold;
}
/*------------------
* 2. Extra
*------------------*/
.logwrapper {
max-height: 230px;
overflow-y: scroll;
background-color: #e6e6e6;
}
.logwrapper.expanded {
max-height: none;
}
.logwrapper.expanded .logexpander:after {
content: "collapse [-]";
}
.logwrapper .logexpander {
z-index: 1;
position: sticky;
top: 10px;
width: max-content;
border: 1px solid;
border-radius: 3px;
padding: 5px 7px;
margin: 10px 0 10px calc(100% - 80px);
cursor: pointer;
background-color: #e6e6e6;
}
.logwrapper .logexpander:after {
content: "expand [+]";
}
.logwrapper .logexpander:hover {
color: #000;
border-color: #000;
}
.logwrapper .log {
min-height: 40px;
position: relative;
top: -50px;
height: calc(100% + 50px);
border: 1px solid #e6e6e6;
color: black;
display: block;
font-family: "Courier New", Courier, monospace;
padding: 5px;
padding-right: 80px;
white-space: pre-wrap;
}
div.media {
border: 1px solid #e6e6e6;
float: right;
height: 240px;
margin: 0 5px;
overflow: hidden;
width: 320px;
}
.media-container {
display: grid;
grid-template-columns: 25px auto 25px;
align-items: center;
flex: 1 1;
overflow: hidden;
height: 200px;
}
.media-container--fullscreen {
grid-template-columns: 0px auto 0px;
}
.media-container__nav--right,
.media-container__nav--left {
text-align: center;
cursor: pointer;
}
.media-container__viewport {
cursor: pointer;
text-align: center;
height: inherit;
}
.media-container__viewport img,
.media-container__viewport video {
object-fit: cover;
width: 100%;
max-height: 100%;
}
.media__name,
.media__counter {
display: flex;
flex-direction: row;
justify-content: space-around;
flex: 0 0 25px;
align-items: center;
}
.collapsible td:not(.col-links) {
cursor: pointer;
}
.collapsible td:not(.col-links):hover::after {
color: #bbb;
font-style: italic;
cursor: pointer;
}
.col-result {
width: 130px;
}
.col-result:hover::after {
content: " (hide details)";
}
.col-result.collapsed:hover::after {
content: " (show details)";
}
#environment-header h2:hover::after {
content: " (hide details)";
color: #bbb;
font-style: italic;
cursor: pointer;
font-size: 12px;
}
#environment-header.collapsed h2:hover::after {
content: " (show details)";
color: #bbb;
font-style: italic;
cursor: pointer;
font-size: 12px;
}
/*------------------
* 3. Sorting items
*------------------*/
.sortable {
cursor: pointer;
}
.sortable.desc:after {
content: " ";
position: relative;
left: 5px;
bottom: -12.5px;
border: 10px solid #4caf50;
border-bottom: 0;
border-left-color: transparent;
border-right-color: transparent;
}
.sortable.asc:after {
content: " ";
position: relative;
left: 5px;
bottom: 12.5px;
border: 10px solid #4caf50;
border-top: 0;
border-left-color: transparent;
border-right-color: transparent;
}
.hidden, .summary__reload__button.hidden {
display: none;
}
.summary__data {
flex: 0 0 550px;
}
.summary__reload {
flex: 1 1;
display: flex;
justify-content: center;
}
.summary__reload__button {
flex: 0 0 300px;
display: flex;
color: white;
font-weight: bold;
background-color: #4caf50;
text-align: center;
justify-content: center;
align-items: center;
border-radius: 3px;
cursor: pointer;
}
.summary__reload__button:hover {
background-color: #46a049;
}
.summary__spacer {
flex: 0 0 550px;
}
.controls {
display: flex;
justify-content: space-between;
}
.filters,
.collapse {
display: flex;
align-items: center;
}
.filters button,
.collapse button {
color: #999;
border: none;
background: none;
cursor: pointer;
text-decoration: underline;
}
.filters button:hover,
.collapse button:hover {
color: #ccc;
}
.filter__label {
margin-right: 10px;
}
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: ${PROJECT_NAME}
# FileName: ${FILE_NAME}
# Author: shen
# Datetime: ${DATE} ${TIME}
# Description:
#
# ---------------------------------------------------------------------------
import requests.sessions
from APIAutoTest_shen_2 import log_decorator
from APIAutoTest_shen_2.common.db import DB
from APIAutoTest_shen_2.common.read_basic_ini import ReadBasicIni
from APIAutoTest_shen_2.common.read_excel import ReadExcel
class RequestsMethod:
@log_decorator
def __init__(self):
"""管理被测系统的登录状态"""
#创建db类对象
self.db = DB()
#创建session对象
self.bpm_session =requests.sessions.Session()
#配置登录数据
login_url = ReadBasicIni().get_host("bpm") + "/auth"
login_data = {"username":"admin","password":"WuHRoZXGrg1LHOix8EVOhqGcu3CEEd0hOWNPRENTScZMb2ekrEtBGGBJweUxGKi36k346+OrS5mD3nnjyrocZOfzo4UqoshNoqJgx09u81LV2vc53Sy83DsBWKGoZZZRUhsDrpRj8feUlakJMCpgqswG0y9jm95Lk9auWpI146I="}
#获取token
token = self.bpm_session.request(method="post",url=login_url,json=login_data).json().get("token")
#将获取到的token更新到session对象的headers中
self.bpm_session.headers["Authorization"] = "Bearer "+token
@log_decorator
def close(self):
"""关闭数据的连接"""
self.db.close()
#
@log_decorator
def request_all(self,req_url,req_mime,req_method,case_data,sql_data,update_key,sql_type):
#封装一个公共的请求方法,先执行sql语句再发送请求
#执行删除sql语句
if sql_type =="delete":
self.db.delete(sql_data)
#再执行查询的sql语句
elif sql_type == "select":
select_result = self.db.select(sql_data)
#将查询结果更新到用例数据中
case_data[update_key] = select_result
#判断sql语句是否为select|delete,或者是否为delete|select
elif sql_type == "select|delete" or sql_type == "delete|select":
#执行删除语句
self.db.delete(sql_data["delete"])
#执行查询语句
select_result = self.db.select(sql_data["select"])
#将查询到的结果更新到用例中
case_data[update_key] = select_result
#发送请求
# requests传参关心的是传参的方式,不关心请求的方法
#判断媒体类型是否为query|json,json|query表示请求体使用json传参,同时也在地址栏传参
if req_mime == "query|json" or req_mime == "json|query":
#requests支持地址栏和请求体同时传参,请求体中每次只能使用一个关键字传参
return self.bpm_session.request(method=req_method,url=req_url,json=case_data["json"],params=case_data["query"])
#判断媒体类型是否为application/json或json,使用关键字json传参
elif req_mime == "application/json" or req_mime == "json":
return self.bpm_session.request(method=req_method,url=req_url,json=case_data)
#判断媒体类型是否为application/x-www-form-urlencoded或form,使用关键字data传参
elif req_mime == "application/x-www-form-urlencoded" or req_mime == "form":
return self.bpm_session.request(method=req_method,url=req_url,data=case_data)
#判断媒体类型是否为multipart/form-data或form-data,使用关键字files传参
elif req_mime == "multipart/form-data" or req_mime == "form-data":
return self.bpm_session.request(method=req_method,url=req_url,files=case_data)
elif req_mime == "query":
return self.bpm_session.request(method=req_method, url=req_url,params=case_data)
elif req_mime is None:
return self.bpm_session.request(method=req_method, url=req_url)
else:
return ValueError("媒体类型的值错误")
if __name__ == '__main__':
req = RequestsMethod()
print(req.bpm_session.headers)
# data =['http://36.139.193.99:8088/api/demension/v1/dem/addDem', 'json', 'post', {'code': 'test_dem', 'description': '测试添加的维度', 'isDefault': 1, 'name': '测试维度'}, {'message': '添加维度成功!'}, 'delete from uc_demension where CODE_="test_dem";', 'delete', None]
# res = req.request_all(req_url=data[0],req_method=data[2],req_mime=data[1],case_data=data[3],sql_type=data[-2],sql_data=data[-3],update_key=data[-1])
# print(res.text)
# res = ReadExcel("zs").get_data()
# for data in res:
# res1 = req.request_all(req_url=data[0],req_method=data[2],req_mime=data[1],case_data=data[3],sql_type=data[-2],sql_data=data[-3],update_key=data[-1])
# print(res1.json())
import pytest
from APIAutoTest_shen_2 import log_decorator
from APIAutoTest_shen_2.requests_method.requests_method import RequestsMethod
@log_decorator
#用例的自定义固件,一般设置为最高级别
@pytest.fixture(scope="session")
def req_fix1():
#创建RequestsMethod对象
req = RequestsMethod()
yield req
req.close()
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: ${PROJECT_NAME}
# FileName: ${FILE_NAME}
# Author: shen
# Datetime: ${DATE} ${TIME}
# Description:
#
# ---------------------------------------------------------------------------
import pytest
from APIAutoTest_shen_2 import log_decorator
from APIAutoTest_shen_2.common.read_excel import ReadExcel
#获取张三的数据
case_datas = ReadExcel("zs").get_data()
class TestBPM:
#调用装饰器,来获取日志的输出到report目录文件
@log_decorator
@pytest.mark.parametrize("url,mime,req,case,expect,sql,sql_type1,update",case_datas)
#用例函数
def test_bpm(self,req_fix1,url,mime,req,case,expect,sql,sql_type1,update):
#调用封装的公共请求方法发送请求
res = req_fix1.request_all(req_url=url,req_method=req,req_mime=mime,case_data=case,sql_type=sql_type1,sql_data=sql,update_key=update)
print([url,mime,req,case,sql,sql_type1,update,expect])
try:
#将key循环取到期望数据的值
for key in expect:
#断言,判断获取到的res中期望数据的key是否等于期望数据中的key
assert res.json().get(key) == expect[key]
except AssertionError:
#如果不等于,就抛出断言类型的错误
raise AssertionError(f"断言失败,用例数据为:{case},期望数据为:{expect}, 服务器返回的数据为:{res.text}")
except Exception as e:
raise e
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment