Commit 78932117 by xiaozhou

bpm接口自动化_v1.0

parent 6311f39a
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: test59
# FileName: __init__.py
# Author: laozhao
# Datetime: 2024/3/18 16:11
# Description:
#
# ---------------------------------------------------------------------------
import functools
import logging
import os
import time
def my_log(filename):
# 创建logger对象
logger = logging.getLogger()
# 设置日志的显示级别
logger.level = logging.NOTSET
# 设置日志写入的文件
handler = logging.FileHandler(filename, mode="a", encoding="utf-8")
# 设置日志文件中日志的级别
handler.setLevel(logging.NOTSET)
# 设置日志文件中日志的格式
formatter = logging.Formatter('%(asctime)s - %(levelname)s: %(message)s')
# 将格式设置到日志文件中
handler.setFormatter(formatter)
# 将日志文件加载到logger中
logger.addHandler(handler)
return logger
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")+".log")
# log_path = os.path.join(os.path.join(os.path.join(os.path.dirname(__file__), "report"), "log"), "BPM.log")
log = my_log(log_path)
def log_decorator(func_name):
@functools.wraps(func_name)
def inner(*args, **kwargs):
try:
log.info("执行的函数或方法为:"+str(func_name.__name__)+",其功能为:"+str(func_name.__doc__))
return func_name(*args, **kwargs)
except Exception as e:
log.error(e)
raise e
return inner
\ No newline at end of file
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: test59
# FileName: db.py
# Author: laozhao
# Datetime: 2024/3/18 15:27
# Description:
#
# ---------------------------------------------------------------------------
import pymysql
from InterfaceAutoTest import log_decorator
from InterfaceAutoTest.common.read_ini import ReadIni
class DB:
@log_decorator
def __init__(self):
"""链接是数据库,获取链接对象和游标对象"""
read_ini = ReadIni()
try:
self.conn = pymysql.Connect(
host=read_ini.get_sql_connect_msg("host"),
port=int(read_ini.get_sql_connect_msg("port")),
user=read_ini.get_sql_connect_msg("user"),
password=read_ini.get_sql_connect_msg("password"),
database=read_ini.get_sql_connect_msg("database"),
charset="utf8"
)
self.cursor = self.conn.cursor()
except:
raise ConnectionError("链接数据库失败")
@log_decorator
def close(self):
self.cursor.close()
self.conn.close()
@log_decorator
def delete(self, sql_sentence):
"""使用游标对象执行删除的sql语句"""
try:
self.cursor.execute(sql_sentence)
except:
raise ValueError("sql语句错误")
else:
self.conn.commit()
@log_decorator
def select(self, sql_sentence):
"""使用游标对象执行查询的sql语句,并返回查询的结果"""
try:
self.cursor.execute(sql_sentence)
except:
raise ValueError("sql语句错误")
else:
select_result = self.cursor.fetchall()
if select_result:
return select_result[0][0]
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: test59
# FileName: read_excel.py
# Author: laozhao
# Datetime: 2024/3/18 15:32
# Description:
#
# ---------------------------------------------------------------------------
import openpyxl
from InterfaceAutoTest import log_decorator, log
from InterfaceAutoTest.common.read_ini import ReadIni
from InterfaceAutoTest.common.read_json import read_json
from InterfaceAutoTest.data_config.settings import *
class ReadExcel:
@log_decorator
def __init__(self,username, table_name, excel_path=None, case_data_path=None, expect_data_path=None, sql_data_path=None):
"""获取所有的json文件的路径,在使用read_json读取所有的json文件,再获取excel文件路径,加载工作簿,获取工作表"""
self.read_ini = ReadIni()
# 判断table_name的值是否为None如果为None使用ini配置文件中的值,如果不为None使用传入的值
if table_name is None:
table_name = self.read_ini.get_table_name("name")
if excel_path is None and case_data_path is None and expect_data_path is None or sql_data_path is None:
case_data_path = self.read_ini.get_file_path("case", username)
expect_data_path = self.read_ini.get_file_path("expect", username)
sql_data_path = self.read_ini.get_file_path("sql", username)
excel_path = self.read_ini.get_file_path("excel", username)
self.case_data_dict = read_json(case_data_path)
self.expect_data_dict = read_json(expect_data_path)
self.sql_data_dict = read_json(sql_data_path)
try:
wb = openpyxl.load_workbook(excel_path)
self.ws = wb[table_name]
except:
raise ValueError("配置文件中的excel名称或工作表名称错误")
def __get_cell_value(self, column: str, row: int) -> str:
"""获取指定单元格数据"""
log.info("功能为:获取指定单元格数据"+"方法名称为:"+"__get_cell_value")
try:
value = self.ws[column+str(row)].value
except:
log.error("指定的行和列错误")
raise KeyError("指定的行和列错误")
else:
if value is None:
return None
elif value.strip():
return value.strip()
@log_decorator
def module_name(self, row):
"""根据行号获取模块名称"""
return self.__get_cell_value(MODULE, row)
@log_decorator
def api_name(self, row):
"""根据行号获取接口名称"""
return self.__get_cell_value(API, row)
@log_decorator
def case_title(self, row):
"""根据行号获取用例标题"""
return self.__get_cell_value(TITLE, row)
@log_decorator
def case_level(self, row):
"""根据行号获取用例等级"""
return self.__get_cell_value(LEVEL, row)
@log_decorator
def case_url(self, row):
"""根据行号,获取用例的url"""
host = self.read_ini.get_host("bpm_host")
path = self.__get_cell_value(URL, row)
if path:
return host+path
@log_decorator
def case_req_method(self, row):
"""根据行号,获取用例的请求方法"""
return self.__get_cell_value(METHOD, row)
@log_decorator
def case_mime(self, row):
"""根据行号,获取用例请求的媒体类型"""
value = self.__get_cell_value(MIME, row)
if value:
return value.lower()
@log_decorator
def case_data(self, row):
"""根据行号,获取用例数据"""
case_data_key = self.__get_cell_value(CASE, row)
if case_data_key:
module_name = self.module_name(row)
api_name = self.api_name(row)
try:
return self.case_data_dict[module_name][api_name][case_data_key]
except:
raise KeyError("根据用例的key获取用例数据失败,请察看用例json文件中,是否有配置")
@log_decorator
def expect_data(self, row):
"""根据行号,获取期望数据"""
expect_data_key = self.__get_cell_value(EXPECT, row)
if expect_data_key:
module_name = self.module_name(row)
api_name = self.api_name(row)
try:
return self.expect_data_dict[module_name][api_name][expect_data_key]
except:
raise KeyError("根据期望的key获取期望数据失败,请察看期望json文件中,是否有配置")
@log_decorator
def sql_data(self, row):
"""根据行号,获取sql语句"""
sql_data_key = self.__get_cell_value(SENTENCE, row)
if sql_data_key:
module_name = self.module_name(row)
api_name = self.api_name(row)
try:
return self.sql_data_dict[module_name][api_name][sql_data_key]
except:
raise KeyError("根据sql语句的key获取sql语句失败,请察看sql语句json文件中,是否有配置")
@log_decorator
def sql_type(self, row):
"""根据行号,获取sql语句类型"""
value = self.__get_cell_value(TYPE, row)
if value:
return value.lower()
@log_decorator
def update_key(self, row):
"""根据行号,获取更新的key"""
return self.__get_cell_value(UPDATE_KEY, row)
def get_data(self):
"""获取所有的测试数据,存放在一个二维列表中"""
log.info("功能:获取所有的测试数据,存放在一个二维列表中"+"方法名称为:"+"get_data")
list_data = []
for row in range(2, self.ws.max_row+1):
# url
url = self.case_url(row)
# method
method = self.case_req_method(row)
if url and method:
# mime
mime = self.case_mime(row)
# case_data
case_data = self.case_data(row)
# expect_data
expect_data = self.expect_data(row)
# sql_type
sql_type = self.sql_type(row)
# sql_data
sql_data = self.sql_data(row)
# update_key
update_key = self.update_key(row)
# 模块名称
module_name = self.module_name(row)
# 接口名称
api_name = self.api_name(row)
# 用例标题
case_title = self.case_title(row)
# 用例等级
case_level = self.case_level(row)
list_data.append([module_name, api_name, case_title, case_level, url, method, mime, case_data, expect_data, sql_type, sql_data, update_key])
else:
return list_data
if __name__ == '__main__':
excel = ReadExcel(username="basic", table_name="登录-维度管理")
print(excel.get_data())
\ No newline at end of file
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: test59
# FileName: read_ini.py
# Author: laozhao
# Datetime: 2024/3/18 15:20
# Description:
#
# ---------------------------------------------------------------------------
import configparser
import os
from InterfaceAutoTest import log_decorator
class ReadIni:
@log_decorator
def __init__(self):
"""获取ini文件的路径,使用Configparser对象的read方法读取ini文件"""
self.data_config = os.path.join(os.path.dirname(os.path.dirname(__file__)), "data_config")
ini_path = os.path.join(self.data_config, "config.ini")
self.conf = configparser.ConfigParser()
self.conf.read(ini_path, encoding="utf-8")
@log_decorator
def get_file_path(self, key, username):
"""根据key获取file节点下文件的路径"""
try:
value = self.conf.get("file", key)
except:
raise KeyError("输入的key错误")
else:
return os.path.join(os.path.join(self.data_config, username), value)
@log_decorator
def get_table_name(self, key):
"""根据key获取table节点下key对应的工作表名称"""
try:
value = self.conf.get("table", key)
except:
raise KeyError("输入的key错误")
else:
return value
@log_decorator
def get_host(self, key):
"""根据key获取host节点下key对应的域名"""
try:
value = self.conf.get("host", key)
except:
raise KeyError("输入的key错误")
else:
return value
@log_decorator
def get_sql_connect_msg(self, key):
"""根据key获取sql节点下key对应的数据库链接信息"""
try:
value = self.conf.get("sql", key)
except:
raise KeyError("输入的key错误")
else:
return value
if __name__ == '__main__':
ini = ReadIni()
print(ini.get_file_path("excel", "张三"))
\ No newline at end of file
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: test59
# FileName: read_json.py
# Author: laozhao
# Datetime: 2024/3/18 15:26
# Description:
#
# ---------------------------------------------------------------------------
import json
import os
from InterfaceAutoTest import log_decorator
@log_decorator
def read_json(filename: str):
if os.path.isfile(filename) and filename.endswith(".json"):
try:
with open(filename, mode="r", encoding="utf-8") as fp:
return json.loads(fp.read())
except:
raise IOError("文件内容有错误")
else:
raise FileNotFoundError("传入的文件路径错误")
{
"认证接口":{
"登录系统":{
"LoginSuccess": {"username":"admin", "password":"MTIzNDU2"},
"LoginErrorPasswordIsNone":{"username":"admin", "password":""},
"LoginErrorPasswordIsShort":{"username":"admin", "password":"s"},
"LoginErrorPasswordIsLong":{"username":"admin", "password":"MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2"},
"LoginErrorPasswordIsSpecialChar":{"username":"admin", "password":"▣▤▥▦▩◘◈"},
"LoginErrorPasswordIsError":{"username":"admin", "password":"MTIzNDU232"},
"LoginErrorUsernameIsError":{"username":"admin123", "password":"MTIzNDU2"},
"LoginErrorUsernameIsSpecialChar":{"username":"▣▤▥▦▩◘◈", "password":"MTIzNDU2"},
"LoginErrorUsernameParamIsUser":{"user":"admin", "password":"MTIzNDU2"},
"LonginErrorUsernameIsLong":{"username":"adminadminadminadminadminadminadminadminadminadminadminadminadminadminadminadminadminadminadmin", "password":"MTIzNDU2"}
}
},
"维度管理": {
"添加维度": {
"AddDemSuccess": {"code": "abc_123_xyz","description": "abc_123_xyz","isDefault": 0,"name": "接口自动化框架"}
},
"根据维度编码获取维度信息": {
"GetDemMessageSuccess": {"code": "abc_123_xyz"}
},
"根据维度编码删除维度": {
"DeleteDemSuccess": {"ids": "不知道"}
}
},
"组织管理": {
"添加组织": {
"AddOrgSuccess": {
"code": "add_org_test",
"demId": "不知道",
"exceedLimitNum": 0,
"grade": "",
"limitNum": 0,
"name": "接口自动化框架添加的组织",
"nowNum": 0,
"orderNo": 0,
"parentId": "0"
}
},
"组织加入用户": {
"OrgAddUserSuccess": {"orgCode":"add_org_test", "accounts": "admin"}
}
}
}
{
"认证接口":{
"登录系统":{
"LoginSuccess": {"username":"超级管理员"},
"LoginErrorPasswordIsNone":{"message":"账号或密码错误"},
"LoginErrorPasswordIsShort":{"message":"账号或密码错误"},
"LoginErrorPasswordIsLong":{"message":"账号或密码错误"},
"LoginErrorPasswordIsSpecialChar":{"message":"账号或密码错误"},
"LoginErrorPasswordIsError":{"message":"账号或密码错误"},
"LoginErrorUsernameIsError":{"message":"账号或密码错误"},
"LoginErrorUsernameIsSpecialChar":{"message":"账号或密码错误"},
"LoginErrorUsernameParamIsUser":{"message":"账号或密码错误"},
"LonginErrorUsernameIsLong":{"message":"账号或密码错误"}
},
"刷新token": {
"RefreshTokenSuccess": {"message": "刷新token成功"}
}
},
"维度管理": {
"添加维度": {
"AddDemSuccess": {"message": "添加维度成功"}
},
"根据维度编码获取维度信息": {
"GetDemMessageSuccess": {"isDelete": "0"}
},
"根据维度编码删除维度": {
"DeleteDemSuccess": {"message": "删除维度成功"}
}
},
"组织管理": {
"添加组织": {
"AddOrgSuccess": {"message": "添加组织成功"}
},
"组织加入用户": {
"OrgAddUserSuccess": {"state":true,"message":"加入成功","value":""}
}
}
}
{
"维度管理": {
"添加维度": {
"AddDemSuccess": "DELETE FROM uc_demension WHERE `CODE_`=\"abc_123_xyz\";"
},
"根据维度编码删除维度": {
"DeleteDemSuccess": "select ID_ FROM uc_demension WHERE `CODE_`=\"abc_123_xyz\";"
}
},
"组织管理": {
"添加组织": {
"AddOrgSuccess": {
"delete": "DELETE FROM uc_org WHERE `CODE_`=\"add_org_test\";",
"select": "select ID_ FROM uc_demension WHERE `CODE_`=\"abc_123_xyz\";"
}
}
}
}
[节点名称]
# 注释
; 注释
键=值
[file]
# 编辑data_config目录下的文件
excel=APIAutoTest.xlsx
case=case_data.json
expect=expect_data.json
sql=sql_data.json
[table]
# excel中工作表名称
name=BPM
[host]
# 被测系统的域名
bpm_host=http://120.46.172.186:8080
[sql]
# 数据库的链接信息
host=120.46.172.186
port=3306
user=root
password=root@2023
database=eipsaas
\ No newline at end of file
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: test59
# FileName: settings.py
# Author: laozhao
# Datetime: 2024/3/19 15:40
# Description:
#
# ---------------------------------------------------------------------------
# ==========excel文件列号的配置=============================
MODULE = "B"
API = "C"
TITLE = "D"
LEVEL = "E"
URL = "F"
METHOD = "G"
MIME = "H"
CASE = "I"
EXPECT = "J"
TYPE = "K"
SENTENCE = "L"
UPDATE_KEY = "M"
\ No newline at end of file
每个用户的数据文件名称必须一直
每个用户的数据文件名称必须一直
在用例层中,在获取数据时,需要指定哪个用户的哪个表
@pytest.mark.parametrize("忽略的字段名...", ReadExcel(username="指定用户名", table_name="指定工作表名称").get_data())
def test_bpm(忽略的形参名名...):
\ No newline at end of file
{
{
"认证接口":{
"登录系统":{
"LoginSuccess": {"username":"admin", "password":"MTIzNDU2"},
"LoginErrorPasswordIsNone":{"username":"admin", "password":""},
"LoginErrorPasswordIsShort":{"username":"admin", "password":"s"},
"LoginErrorPasswordIsLong":{"username":"admin", "password":"MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2"},
"LoginErrorPasswordIsSpecialChar":{"username":"admin", "password":"▣▤▥▦▩◘◈"},
"LoginErrorPasswordIsError":{"username":"admin", "password":"MTIzNDU232"},
"LoginErrorUsernameIsError":{"username":"admin123", "password":"MTIzNDU2"},
"LoginErrorUsernameIsSpecialChar":{"username":"▣▤▥▦▩◘◈", "password":"MTIzNDU2"},
"LoginErrorUsernameParamIsUser":{"user":"admin", "password":"MTIzNDU2"},
"LonginErrorUsernameIsLong":{"username":"adminadminadminadminadminadminadminadminadminadminadminadminadminadminadminadminadminadminadmin", "password":"MTIzNDU2"}
}
},
"维度管理": {
"添加维度": {
"AddDemSuccess": {"code": "abc_123_xyz","description": "abc_123_xyz","isDefault": 0,"name": "接口自动化框架"}
},
"根据维度编码获取维度信息": {
"GetDemMessageSuccess": {"code": "abc_123_xyz"}
},
"根据维度编码删除维度": {
"DeleteDemSuccess": {"ids": "不知道"}
}
},
"组织管理": {
"添加组织": {
"AddOrgSuccess": {
"code": "add_org_test",
"demId": "不知道",
"exceedLimitNum": 0,
"grade": "",
"limitNum": 0,
"name": "接口自动化框架添加的组织",
"nowNum": 0,
"orderNo": 0,
"parentId": "0"
}
},
"组织加入用户": {
"OrgAddUserSuccess": {"orgCode":"add_org_test", "accounts": "admin"}
},
"删除组织": {
"DeleteOrgSuccess": "add_org_test"
}
}
}
{
{
"认证接口":{
"登录系统":{
"LoginSuccess": {"username":"超级管理员"},
"LoginErrorPasswordIsNone":{"message":"账号或密码错误"},
"LoginErrorPasswordIsShort":{"message":"账号或密码错误"},
"LoginErrorPasswordIsLong":{"message":"账号或密码错误"},
"LoginErrorPasswordIsSpecialChar":{"message":"账号或密码错误"},
"LoginErrorPasswordIsError":{"message":"账号或密码错误"},
"LoginErrorUsernameIsError":{"message":"账号或密码错误"},
"LoginErrorUsernameIsSpecialChar":{"message":"账号或密码错误"},
"LoginErrorUsernameParamIsUser":{"message":"账号或密码错误"},
"LonginErrorUsernameIsLong":{"message":"账号或密码错误"}
},
"刷新token": {
"RefreshTokenSuccess": {"message": "刷新token成功"}
}
},
"维度管理": {
"添加维度": {
"AddDemSuccess": {"message": "添加维度成功"}
},
"根据维度编码获取维度信息": {
"GetDemMessageSuccess": {"isDelete": "0"}
},
"根据维度编码删除维度": {
"DeleteDemSuccess": {"message": "删除维度成功"}
}
},
"组织管理": {
"添加组织": {
"AddOrgSuccess": {"message": "添加组织成功"}
},
"组织加入用户": {
"OrgAddUserSuccess": {"state":true,"message":"加入成功","value":""}
},
"删除组织": {
"DeleteOrgSuccess": {"state":true,"message":"删除组织成功!","value":""}
}
}
}
{
{
"维度管理": {
"添加维度": {
"AddDemSuccess": "DELETE FROM uc_demension WHERE `CODE_`=\"abc_123_xyz\";"
},
"根据维度编码删除维度": {
"DeleteDemSuccess": "select ID_ FROM uc_demension WHERE `CODE_`=\"abc_123_xyz\";"
}
},
"组织管理": {
"添加组织": {
"AddOrgSuccess": {
"delete": "DELETE FROM uc_org WHERE `CODE_`=\"add_org_test\";",
"select": "select ID_ FROM uc_demension WHERE `CODE_`=\"abc_123_xyz\";"
}
}
}
}
{
{
"认证接口":{
"登录系统":{
"LoginSuccess": {"username":"admin", "password":"MTIzNDU2"},
"LoginErrorPasswordIsNone":{"username":"admin", "password":""},
"LoginErrorPasswordIsShort":{"username":"admin", "password":"s"},
"LoginErrorPasswordIsLong":{"username":"admin", "password":"MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2"},
"LoginErrorPasswordIsSpecialChar":{"username":"admin", "password":"▣▤▥▦▩◘◈"},
"LoginErrorPasswordIsError":{"username":"admin", "password":"MTIzNDU232"},
"LoginErrorUsernameIsError":{"username":"admin123", "password":"MTIzNDU2"},
"LoginErrorUsernameIsSpecialChar":{"username":"▣▤▥▦▩◘◈", "password":"MTIzNDU2"},
"LoginErrorUsernameParamIsUser":{"user":"admin", "password":"MTIzNDU2"},
"LonginErrorUsernameIsLong":{"username":"adminadminadminadminadminadminadminadminadminadminadminadminadminadminadminadminadminadminadmin", "password":"MTIzNDU2"}
}
},
"维度管理": {
"添加维度": {
"AddDemSuccess": {"code": "abc_123_xyz","description": "abc_123_xyz","isDefault": 0,"name": "接口自动化框架"}
},
"根据维度编码获取维度信息": {
"GetDemMessageSuccess": {"code": "abc_123_xyz"}
},
"根据维度编码删除维度": {
"DeleteDemSuccess": {"ids": "不知道"}
}
},
"组织管理": {
"添加组织": {
"AddOrgSuccess": {
"code": "add_org_test",
"demId": "不知道",
"exceedLimitNum": 0,
"grade": "",
"limitNum": 0,
"name": "接口自动化框架添加的组织",
"nowNum": 0,
"orderNo": 0,
"parentId": "0"
}
},
"组织加入用户": {
"OrgAddUserSuccess": {"orgCode":"add_org_test", "accounts": "admin"}
},
"删除组织": {
"DeleteOrgSuccess": "add_org_test"
},
"保存组织参数": {
"SaveOrgParamSuccess": {
"query": {"orgCode": "add_org_test"},
"body": [{"alias":"ah","value":"1,2"},{"alias":"sz","value":"成都"}]
}
}
}
}
{
{
"认证接口":{
"登录系统":{
"LoginSuccess": {"username":"超级管理员"},
"LoginErrorPasswordIsNone":{"message":"账号或密码错误"},
"LoginErrorPasswordIsShort":{"message":"账号或密码错误"},
"LoginErrorPasswordIsLong":{"message":"账号或密码错误"},
"LoginErrorPasswordIsSpecialChar":{"message":"账号或密码错误"},
"LoginErrorPasswordIsError":{"message":"账号或密码错误"},
"LoginErrorUsernameIsError":{"message":"账号或密码错误"},
"LoginErrorUsernameIsSpecialChar":{"message":"账号或密码错误"},
"LoginErrorUsernameParamIsUser":{"message":"账号或密码错误"},
"LonginErrorUsernameIsLong":{"message":"账号或密码错误"}
},
"刷新token": {
"RefreshTokenSuccess": {"message": "刷新token成功"}
}
},
"维度管理": {
"添加维度": {
"AddDemSuccess": {"message": "添加维度成功"}
},
"根据维度编码获取维度信息": {
"GetDemMessageSuccess": {"isDelete": "0"}
},
"根据维度编码删除维度": {
"DeleteDemSuccess": {"message": "删除维度成功"}
}
},
"组织管理": {
"添加组织": {
"AddOrgSuccess": {"message": "添加组织成功"}
},
"组织加入用户": {
"OrgAddUserSuccess": {"state":true,"message":"加入成功","value":""}
},
"删除组织": {
"DeleteOrgSuccess": {"state":true,"message":"删除组织成功!","value":""}
},
"保存组织参数": {
"SaveOrgParamSuccess": {"state":true,"message":"保存组织参数成功!","value":""}
}
}
}
{
{
"维度管理": {
"添加维度": {
"AddDemSuccess": "DELETE FROM uc_demension WHERE `CODE_`=\"abc_123_xyz\";"
},
"根据维度编码删除维度": {
"DeleteDemSuccess": "select ID_ FROM uc_demension WHERE `CODE_`=\"abc_123_xyz\";"
}
},
"组织管理": {
"添加组织": {
"AddOrgSuccess": {
"delete": "DELETE FROM uc_org WHERE `CODE_`=\"add_org_test\";",
"select": "select ID_ FROM uc_demension WHERE `CODE_`=\"abc_123_xyz\";"
}
}
}
}
[pytest]
;开启日志
log_cli=true
;设置日志的级别,如果不设置级别的话,可以设置为NOTSET,如果要设置级别,级别可以有debug,info,warning,error,致命
log_level=NOTSET
;设置日志显示的信息格式
log_format=%(levelname)s--%(asctime)s--%(message)s
;设置日志中时间显示的格式
log_date_format=%Y-%m-%d %H:%M:%S
;每个py文件运行的时候追加的命令
;addopts=-vs
;设置日志保存的文件
;log_file=../report/log/bpm_test.log
;设置日志保存在文件中的级别
log_file_level=NOTSET
;设置日志在文件中的信息格式
log_file_format=%(levelname)s--%(asctime)s--%(message)s
;设置文件日志中时间显示的格式
log_file_date_format=%Y-%m-%d %H:%M:%S
\ No newline at end of file
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: test59
# FileName: requests_method.py
# Author: laozhao
# Datetime: 2024/3/15 16:12
# Description:
#
# ---------------------------------------------------------------------------
import base64
import requests
from InterfaceAutoTest.common.read_ini import ReadIni
class RequestsMethod:
def __init__(self):
"""关联登录成功的状态--关联token"""
# 获取登录的url
login_url = ReadIni().get_host("bpm_host")+"/auth"
# 配置登录数据
login_data = {"username": "admin", "password": base64.b64encode("123456".encode()).decode()}
# 创建Session对象
self.bpm_session = requests.sessions.Session()
# 使用Session对象发送登录请求
res = self.bpm_session.request(method="post", url=login_url, json=login_data)
# 提取token,将token更新到Session对象的headers中
self.bpm_session.headers["Authorization"] = "Bearer "+res.json().get("token")
def request_all(self, req_url, req_method, req_mime, case_data):
"""
公共的请求方法
:param req_url: 请求的url,字符串格式
:param req_method: 请求的方法,字符串格式
:param req_mime: 请求的媒体类型,字符串格式
:param case_data: 用例数据
:return: Response type
"""
# 判断请求的媒体类型的值是否为application/x-www-form-urlencoded或x-www-form-urlencoded,使用data关键字传参
# 除了json和上传文件以外,在请求体传参的使用data
if req_mime == 'application/x-www-form-urlencoded' or req_mime == 'x-www-form-urlencoded':
return self.bpm_session.request(method=req_method, url=req_url, data=case_data)
# 判断请求的媒体类型的值是否为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)
# 判断请求的媒体类型的值是否为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)
# 判断请求的媒体类型的值是否为query或params,使用params关键字传参
elif req_mime == 'query' or req_mime == 'params':
return self.bpm_session.request(method=req_method, url=req_url, params=case_data)
# 判断请求的媒体类型的值为None,表示没有传参
elif req_mime is None:
return self.bpm_session.request(method=req_method, url=req_url)
# 判断媒体类型是否为query|json 或者为json|query, 表示请求体和地址栏同时传参
elif req_mime == "query|json" or req_mime == "json|query":
return self.bpm_session.request(method=req_method, url=req_url, params=case_data["query"], json=case_data["body"])
# 其他类型--返回一个错误的提示信息
else:
raise ValueError("传入的媒体类型,没封装,请自行封装")
if __name__ == '__main__':
req = RequestsMethod()
\ No newline at end of file
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: test59
# FileName: __init__.py
# Author: laozhao
# Datetime: 2024/3/18 17:15
# Description:
#
# ---------------------------------------------------------------------------
"""
pytest输出报告,使用插件:pytest-html. pip install pytest-html
pytest 用例模块/用例包 -vv --html=报告保存的路径 --self-contained-html
"""
"""
失败重跑插件:pip install pytest-rerunfailures
pytest 用例模块/用例包 -vv --html=报告保存的路径 --self-contained-html --reruns 失败重跑的次数 --reruns-delay 失败重跑的间隔时间
"""
# allure
"""
1: 配置allure的环境。验证环境是否正确:allure --version
2: 需要按照allure插件:pip install allure-pytest
生成报告:
1:生成测试结果的json文件
pytest 用例模块/用例包 --alluredir=json文件存放的目录 --clean-alluredir
2:根据第一步生成的json文件生成报告:
方式1:生成本地服务
allure serve 第一步存放json文件的目录
方式2:生成静态的报告
allure generate 第一步存放json文件的目录 -o 报告存放的目录
本地打开allure生成的报告:allure open 报告存放的目录
"""
\ No newline at end of file
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: test59
# FileName: __init__.py
# Author: laozhao
# Datetime: 2024/3/18 17:15
# Description:
#
# ---------------------------------------------------------------------------
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: test59
# FileName: conftest.py
# Author: laozhao
# Datetime: 2024/3/18 16:52
# Description:
#
# ---------------------------------------------------------------------------
import pytest
from InterfaceAutoTest.common.db import DB
from InterfaceAutoTest.requests_method.requests_method import RequestsMethod
@pytest.fixture(scope="session")
def db_fix():
db = DB()
yield db
db.close()
@pytest.fixture(scope="session")
def req_fix():
req = RequestsMethod()
yield req
# def pytest_collection_modifyitems(items):
# # item表示每个测试用例,解决用例名称中文显示问题
# for item in items:
# item.name = item.name.encode("utf-8").decode("unicode-escape")
# item._nodeid = item._nodeid.encode("utf-8").decode("unicode-escape")
\ No newline at end of file
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: test59
# FileName: test_bpm.py
# Author: laozhao
# Datetime: 2024/3/18 16:49
# Description:
#
# ---------------------------------------------------------------------------
import pytest
from InterfaceAutoTest.common.read_excel import ReadExcel
from InterfaceAutoTest import log
import allure
class TestBPM:
@allure.epic("BPM项目-basic") # 项目名称
# @allure.feature("模块名称") # 模块名称
# @allure.story("接口名称") # 接口名称
# @allure.title("接口用例标题") # 接口用例标题
# @allure.severity("接口用例等级") # 接口用例等级
@pytest.mark.parametrize("module_name, api_name, case_title, case_level, url, method, mime, case_data, expect_data, sql_type, sql_data, update_key", ReadExcel(username="basic", table_name="登录-维度管理").get_data())
def test_bpm(self, db_fix, req_fix, module_name, api_name, case_title, case_level, url, method, mime, case_data, expect_data, sql_type, sql_data, update_key):
# allure报告中展示的结果
allure.dynamic.feature(module_name)
allure.dynamic.story(api_name)
allure.dynamic.title(case_title)
allure.dynamic.severity(case_level)
# 1.1 判断sql语句类型是否为delete
if sql_type == "delete":
# 1.2 使用DB类对象使用delete方法执行删除的sql语句。DB类对象就是自定义固件db_fix
db_fix.delete(sql_data)
# 1.3 使用RequestsMethod类对象request_all方法发生请求
# 2.1 判断sql语句类型是否为select
elif sql_type == "select":
# 2.2 使用DB类对象使用select方法执行查询的sql语句,并接收查询的结果。DB类对象就是自定义固件db_fix
select_result = db_fix.select(sql_data)
# 2.3 将查询结果更新到用例数据中
case_data[update_key] = select_result
# 2.3 使用RequestsMethod类对象request_all方法发生请求
# 3.1 判断sql语句的类型是否为select|delete或者为delete|select
elif sql_type == "select|delete" or sql_type == "delete|select":
# 3.2 使用DB类对象使用delete方法执行删除的sql语句。DB类对象就是自定义固件db_fix
db_fix.delete(sql_data["delete"])
# 3.3 使用DB类对象使用select方法执行查询的sql语句,并接收查询的结果。DB类对象就是自定义固件db_fix
select_result = db_fix.select(sql_data["select"])
# 3.4 将查询结果更新到用例数据中
case_data[update_key] = select_result
# 3.5 使用RequestsMethod类对象request_all方法发生请求
# 使用RequestsMethod类对象request_all方法发生请求
res = req_fix.request_all(req_url=url, req_method=method, req_mime=mime, case_data=case_data)
# 断言
try:
for key in expect_data.keys():
assert expect_data[key] == res.json().get(key)
except AssertionError:
log.error(f"断言失败,请求的url为:{url},请求的方法为:{method},请求的媒体类型为:{mime}, 请求的用例数据:{case_data}, 期望数据为:{expect_data},服务器返回的数据为:{res.text}")
raise AssertionError("断言失败")
else:
log.error(f"断言成功,请求的url为:{url},请求的方法为:{method},请求的媒体类型为:{mime}, 请求的用例数据:{case_data}, 期望数据为:{expect_data},服务器返回的数据为:{res.text}")
# -*-coding:utf-8 -*- #
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: test59
# FileName: __init__.py
# Author: laozhao
# Datetime: 2024/3/18 17:17
# Description:
#
# ---------------------------------------------------------------------------
# -*-coding:utf-8 -*- #
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: test59
# FileName: conftest.py
# Author: laozhao
# Datetime: 2024/3/18 16:52
# Description:
#
# ---------------------------------------------------------------------------
import pytest
from InterfaceAutoTest.common.db import DB
from InterfaceAutoTest.requests_method.requests_method import RequestsMethod
@pytest.fixture(scope="session")
def db_fix():
db = DB()
yield db
db.close()
@pytest.fixture(scope="session")
def req_fix():
req = RequestsMethod()
yield req
#
# def pytest_collection_modifyitems(items):
# # item表示每个测试用例,解决用例名称中文显示问题
# for item in items:
# item.name = item.name.encode("utf-8").decode("unicode-escape")
# item._nodeid = item._nodeid.encode("utf-8").decode("unicode-escape")
# -*-coding:utf-8 -*- #
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: test59
# FileName: test_login_addDem_getDemMsg.py
# Author: laozhao
# Datetime: 2024/3/18 17:17
# Description:
#
# ---------------------------------------------------------------------------
import base64
import allure
import pytest
import requests
from faker import Faker
depends_data = {}
# 登录
@allure.epic("BPM-场景")
@allure.feature("认证接口")
@allure.story("登录系统")
@allure.title("场景用例的正向用例")
@allure.severity("高")
@pytest.mark.dependency()
def test_login():
url = "http://120.46.172.186:8080/auth"
data = {"username": "admin", "password": base64.b64encode("123456".encode()).decode()}
res = requests.post(url=url, json=data)
token = res.json().get("token")
depends_data["Authorization"] = "Bearer "+token
assert "超级" in res.text
@allure.epic("BPM-场景")
@allure.feature("维度管理")
@allure.story("添加维度")
@allure.title("添加维度场景用例的正向用例")
@allure.severity("高")
@pytest.mark.dependency(depends=["test_login"])
def test_add_dem():
url = "http://120.46.172.186:8080/api/demension/v1/dem/addDem"
data = Faker(locale="zh_cn")
code = data.phone_number()
data = {
"code": code,
"description": "test_add_dem_1234",
"isDefault": 0,
"name": "test_add_dem_1234"
}
res = requests.post(url=url, json=data, headers={"Authorization": depends_data["Authorization"]})
depends_data["dem_code"] = code
assert "成功" in res.text
@allure.epic("BPM-场景")
@allure.feature("维度管理")
@allure.story("根据维度编码获取维度信息")
@allure.title("根据维度编码获取维度信息的正向用例")
@allure.severity("高")
@pytest.mark.dependency(depends=["test_add_dem"])
def test_get_dem_msg():
url = "http://120.46.172.186:8080/api/demension/v1/dem/getDem"
data = {"code": depends_data["dem_code"]}
res = requests.get(url=url, params=data, headers={"Authorization": depends_data["Authorization"]})
assert "isDelete" in res.text
# -*-coding:utf-8 -*- #
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: test59
# FileName: __init__.py
# Author: laozhao
# Datetime: 2024/3/18 17:15
# Description:
#
# ---------------------------------------------------------------------------
# -*-coding:utf-8 -*- #
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: test59
# FileName: conftest.py
# Author: laozhao
# Datetime: 2024/3/18 16:52
# Description:
#
# ---------------------------------------------------------------------------
import pytest
from InterfaceAutoTest.common.db import DB
from InterfaceAutoTest.requests_method.requests_method import RequestsMethod
@pytest.fixture(scope="session")
def db_fix():
db = DB()
yield db
db.close()
@pytest.fixture(scope="session")
def req_fix():
req = RequestsMethod()
yield req
# def pytest_collection_modifyitems(items):
# # item表示每个测试用例,解决用例名称中文显示问题
# for item in items:
# item.name = item.name.encode("utf-8").decode("unicode-escape")
# item._nodeid = item._nodeid.encode("utf-8").decode("unicode-escape")
\ No newline at end of file
# -*-coding:utf-8 -*- #
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: test59
# FileName: test_bpm.py
# Author: laozhao
# Datetime: 2024/3/18 16:49
# Description:
#
# ---------------------------------------------------------------------------
import allure
import pytest
from InterfaceAutoTest.common.read_excel import ReadExcel
from InterfaceAutoTest import log
class TestBPM:
@allure.epic("BPM项目-张三") # 项目名称
# @allure.feature("模块名称") # 模块名称
# @allure.story("接口名称") # 接口名称
# @allure.title("接口用例标题") # 接口用例标题
# @allure.severity("接口用例等级") # 接口用例等级
@pytest.mark.parametrize("module_name, api_name, case_title, case_level, url, method, mime, case_data, expect_data, sql_type, sql_data, update_key", ReadExcel(username="张三", table_name="组织管理").get_data())
def test_bpm(self, db_fix, req_fix, module_name, api_name, case_title, case_level, url, method, mime, case_data, expect_data, sql_type, sql_data, update_key):
# allure报告中展示的结果
allure.dynamic.feature(module_name)
allure.dynamic.story(api_name)
allure.dynamic.title(case_title)
allure.dynamic.severity(case_level)
# 1.1 判断sql语句类型是否为delete
if sql_type == "delete":
# 1.2 使用DB类对象使用delete方法执行删除的sql语句。DB类对象就是自定义固件db_fix
db_fix.delete(sql_data)
# 1.3 使用RequestsMethod类对象request_all方法发生请求
# 2.1 判断sql语句类型是否为select
elif sql_type == "select":
# 2.2 使用DB类对象使用select方法执行查询的sql语句,并接收查询的结果。DB类对象就是自定义固件db_fix
select_result = db_fix.select(sql_data)
# 2.3 将查询结果更新到用例数据中
case_data[update_key] = select_result
# 2.3 使用RequestsMethod类对象request_all方法发生请求
# 3.1 判断sql语句的类型是否为select|delete或者为delete|select
elif sql_type == "select|delete" or sql_type == "delete|select":
# 3.2 使用DB类对象使用delete方法执行删除的sql语句。DB类对象就是自定义固件db_fix
db_fix.delete(sql_data["delete"])
# 3.3 使用DB类对象使用select方法执行查询的sql语句,并接收查询的结果。DB类对象就是自定义固件db_fix
select_result = db_fix.select(sql_data["select"])
# 3.4 将查询结果更新到用例数据中
case_data[update_key] = select_result
# 3.5 使用RequestsMethod类对象request_all方法发生请求
# 使用RequestsMethod类对象request_all方法发生请求
res = req_fix.request_all(req_url=url, req_method=method, req_mime=mime, case_data=case_data)
# 断言
try:
for key in expect_data.keys():
assert expect_data[key] == res.json().get(key)
except AssertionError:
log.error(f"断言失败,请求的url为:{url},请求的方法为:{method},请求的媒体类型为:{mime}, 请求的用例数据:{case_data}, 期望数据为:{expect_data},服务器返回的数据为:{res.text}")
raise AssertionError("断言失败")
else:
log.error(f"断言成功,请求的url为:{url},请求的方法为:{method},请求的媒体类型为:{mime}, 请求的用例数据:{case_data}, 期望数据为:{expect_data},服务器返回的数据为:{res.text}")
# -*-coding:utf-8 -*- #
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: test59
# FileName: __init__.py
# Author: laozhao
# Datetime: 2024/3/18 17:15
# Description:
#
# ---------------------------------------------------------------------------
# -*-coding:utf-8 -*- #
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: test59
# FileName: conftest.py
# Author: laozhao
# Datetime: 2024/3/18 16:52
# Description:
#
# ---------------------------------------------------------------------------
import pytest
from InterfaceAutoTest.common.db import DB
from InterfaceAutoTest.requests_method.requests_method import RequestsMethod
@pytest.fixture(scope="session")
def db_fix():
db = DB()
yield db
db.close()
@pytest.fixture(scope="session")
def req_fix():
req = RequestsMethod()
yield req
# def pytest_collection_modifyitems(items):
# # item表示每个测试用例,解决用例名称中文显示问题
# for item in items:
# item.name = item.name.encode("utf-8").decode("unicode-escape")
# item._nodeid = item._nodeid.encode("utf-8").decode("unicode-escape")
\ No newline at end of file
# -*-coding:utf-8 -*- #
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: test59
# FileName: test_bpm.py
# Author: laozhao
# Datetime: 2024/3/18 16:49
# Description:
#
# ---------------------------------------------------------------------------
import allure
import pytest
from InterfaceAutoTest.common.read_excel import ReadExcel
from InterfaceAutoTest import log
class TestBPM:
@allure.epic("BPM项目-李四") # 项目名称
# @allure.feature("模块名称") # 模块名称
# @allure.story("接口名称") # 接口名称
# @allure.title("接口用例标题") # 接口用例标题
# @allure.severity("接口用例等级") # 接口用例等级
@pytest.mark.parametrize("module_name, api_name, case_title, case_level, url, method, mime, case_data, expect_data, sql_type, sql_data, update_key", ReadExcel(username="李四", table_name="组织管理").get_data())
def test_bpm(self, db_fix, req_fix, module_name, api_name, case_title, case_level, url, method, mime, case_data, expect_data, sql_type, sql_data, update_key):
# allure报告中展示的结果
allure.dynamic.feature(module_name)
allure.dynamic.story(api_name)
allure.dynamic.title(case_title)
allure.dynamic.severity(case_level)
# 1.1 判断sql语句类型是否为delete
if sql_type == "delete":
# 1.2 使用DB类对象使用delete方法执行删除的sql语句。DB类对象就是自定义固件db_fix
db_fix.delete(sql_data)
# 1.3 使用RequestsMethod类对象request_all方法发生请求
# 2.1 判断sql语句类型是否为select
elif sql_type == "select":
# 2.2 使用DB类对象使用select方法执行查询的sql语句,并接收查询的结果。DB类对象就是自定义固件db_fix
select_result = db_fix.select(sql_data)
# 2.3 将查询结果更新到用例数据中
case_data[update_key] = select_result
# 2.3 使用RequestsMethod类对象request_all方法发生请求
# 3.1 判断sql语句的类型是否为select|delete或者为delete|select
elif sql_type == "select|delete" or sql_type == "delete|select":
# 3.2 使用DB类对象使用delete方法执行删除的sql语句。DB类对象就是自定义固件db_fix
db_fix.delete(sql_data["delete"])
# 3.3 使用DB类对象使用select方法执行查询的sql语句,并接收查询的结果。DB类对象就是自定义固件db_fix
select_result = db_fix.select(sql_data["select"])
# 3.4 将查询结果更新到用例数据中
case_data[update_key] = select_result
# 3.5 使用RequestsMethod类对象request_all方法发生请求
# 使用RequestsMethod类对象request_all方法发生请求
res = req_fix.request_all(req_url=url, req_method=method, req_mime=mime, case_data=case_data)
# 断言
try:
for key in expect_data.keys():
assert expect_data[key] == res.json().get(key)
except AssertionError:
log.error(f"断言失败,请求的url为:{url},请求的方法为:{method},请求的媒体类型为:{mime}, 请求的用例数据:{case_data}, 期望数据为:{expect_data},服务器返回的数据为:{res.text}")
raise AssertionError("断言失败")
else:
log.error(f"断言成功,请求的url为:{url},请求的方法为:{method},请求的媒体类型为:{mime}, 请求的用例数据:{case_data}, 期望数据为:{expect_data},服务器返回的数据为:{res.text}")
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