Commit f2679039 by chenyao

第一次提交

parents
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: Python1
# FileName: __init__.py
# Author: chenyao
# Datetime: 2025/3/12 18:20
# Description:
#
# ---------------------------------------------------------------------------
import functools
import logging
import os
import time
def log_func(file_path):
"""获取写日志的logger"""
logger = logging.getLogger()
logger.setLevel(logging.INFO)
handler = logging.FileHandler(file_path, mode="a", encoding="utf-8")
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)
return logger
print(__file__)
# 动态获取日志文件存放目录
log_dir = os.path.join(os.path.join(os.path.dirname(__file__), "report"), "log")
# 获取日志文件名称
log_name = "apiAutoTest_"+time.strftime("%Y_%m_%d_%H_%M_%S")+".log"
# 日志文件的路径
log_path = os.path.join(log_dir, log_name)
# 获取写日志的logger
log = log_func(log_path)
# 创建写日志的装饰器
def log_decorator(func_name):
@functools.wraps(func_name)
def inner(*args, **kwargs):
log.info(f"执行的功能为:{func_name.__name__}, 功能描述为:{func_name.__doc__}, 所在的文件:{func_name.__code__.co_filename}, 所在行为:{func_name.__code__.co_firstlineno}")
try:
result = func_name(*args, **kwargs)
except Exception as e:
log.error(f"产生错误为:{type(e)}, 错误的描述为:{e},执行的功能为:{func_name.__name__}, 功能描述为:{func_name.__doc__}, 所在的文件:{func_name.__code__.co_filename}, 所在行为:{func_name.__code__.co_firstlineno}")
raise e
else:
return result
return inner
\ No newline at end of file
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: Python1
# FileName: __init__.py
# Author: chenyao
# Datetime: 2025/3/12 18:21
# Description:
#
# ---------------------------------------------------------------------------
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: Python1
# FileName: db.py
# Author: chenyao
# Datetime: 2025/3/12 20:22
# Description:
#
# ---------------------------------------------------------------------------
import pymysql
from requests import delete
from Auto_TestV4 import log_decorator
from Auto_TestV4.commom.read_ini import ReadIni
from Review.pymysql_faker.sql_practice import select_result
class DB:
@log_decorator
def __init__(self):
ini=ReadIni()
self.conn=pymysql.connect(
host=ini.get_sql_conn("host"),
port=int(ini.get_sql_conn("port")),
user=ini.get_sql_conn("user"),
password=ini.get_sql_conn("password"),
database=ini.get_sql_conn("database"),
charset="utf8"
)
self.curosr=self.conn.cursor()
@log_decorator
def delete_method(self,delete_sql):
if isinstance(delete_sql,str) and delete_sql.lower().startswith("delete"):
self.curosr.execute(delete_sql)
self.conn.commit()
else:
raise ValueError("删除的sql语句出错")
@log_decorator
def select_method(self,select_sql):
if isinstance(select_sql,str) and select_sql.lower().startswith("select"):
self.curosr.execute(select_sql)
select_result=self.curosr.fetchall()
if select_result:
return select_result[0][0]
else:
raise ValueError("执行的sql语句错误")
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: Python1
# FileName: read_excel.py
# Author: chenyao
# Datetime: 2025/3/12 19:02
# Description:
#
# ---------------------------------------------------------------------------
import openpyxl
from Auto_TestV4 import log_decorator
from Auto_TestV4.commom.read_ini import ReadIni
from Auto_TestV4.commom.read_json import ReadJson
class ReadExcel:
@log_decorator
def __init__(self):
"""获取基本数据,存放二维列表中"""
ini=ReadIni()
excel_path=ini.get_file_path("excel")
case_data_path=ini.get_file_path("case")
expect_data_path=ini.get_file_path("expect")
sql_data_path=ini.get_file_path("sql")
self.server_path=ini.get_server_value("host")
table_name=ini.get_table_name("tablename")
self.case_data_dict=ReadJson(case_data_path)
self.expect_data_dict=ReadJson(expect_data_path)
self.sql_data_dict=ReadJson(sql_data_path)
wb=openpyxl.load_workbook(excel_path)
self.ws=wb[table_name]
@log_decorator
def __get_cell_value(self,colomn:str,row:int) :
value=self.ws[colomn+str(row)].value
if isinstance(value,str) and value.strip():
return value.strip()
@log_decorator
def get_module_name(self,row):
return self.__get_cell_value("b",row)
@log_decorator
def get_api_name(self,row):
return self.__get_cell_value("c",row)
@log_decorator
def get_title(self,row):
return self.__get_cell_value("d",row)
@log_decorator
def get_level(self,row):
return self.__get_cell_value("e",row)
@log_decorator
def get_method(self,row):
return self.__get_cell_value("f",row)
@log_decorator
def get_case_url(self,row):
value=self.__get_cell_value("g",row)
if value:
return self.server_path+value
@log_decorator
def get_type(self,row):
value=self.__get_cell_value("h",row)
if value:
return value.lower()
@log_decorator
def get_case_data(self,row):
value=self.__get_cell_value("i",row)
if value:
module_name=self.get_module_name(row)
api_name=self.get_api_name(row)
case_data=self.case_data_dict[module_name][api_name][value]
return case_data
@log_decorator
def get_expect_data(self, row):
value = self.__get_cell_value("j", row)
if value:
module_name = self.get_module_name(row)
api_name = self.get_api_name(row)
expect_data = self.expect_data_dict[module_name][api_name][value]
return expect_data
@log_decorator
def get_sql_data(self, row):
value = self.__get_cell_value("k", row)
if value:
module_name = self.get_module_name(row)
api_name = self.get_api_name(row)
sql_data= self.sql_data_dict[module_name][api_name][value]
return sql_data
@log_decorator
def get_sql_type(self,row):
value=self.__get_cell_value("l",row)
if value:
return value.lower()
@log_decorator
def get_updatekey(self,row):
value=self.__get_cell_value("m",row)
return value
@log_decorator
def get_all_data(self):
data_list=[]
for row in range(2,self.ws.max_row+1):
req_method=self.get_method(row)
req_url=self.get_case_url(row)
req_type=self.get_type(row)
req_case_data=self.get_case_data(row)
req_expect_data=self.get_expect_data(row)
req_sql_data=self.get_sql_data(row)
req_sql_type=self.get_sql_type(row)
req_updatekey=self.get_updatekey(row)
if req_method and req_url and req_type:
data_list.append([req_method,req_url,req_type,req_case_data,req_expect_data,req_sql_data,req_sql_type,req_updatekey])
else:
return data_list
if __name__ == '__main__':
ec=ReadExcel()
print(ec.get_all_data())
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: Python1
# FileName: read_ini.py
# Author: chenyao
# Datetime: 2025/3/12 18:23
# Description:
#
# ---------------------------------------------------------------------------
import configparser
import os
from Auto_TestV4 import log_decorator
class ReadIni:
def __init__(self):
self.dir_path=os.path.join(os.path.dirname(os.path.dirname(__file__)),"data_config")
self.ini_path=os.path.join(self.dir_path,"config.ini")
self.conf=configparser.ConfigParser()
self.conf.read(self.ini_path,encoding="utf-8")
print(self.ini_path)
@log_decorator
def get_file_path(self,key):
"""获取文件路径"""
file_path=self.conf.get("file",key)
return os.path.join(self.dir_path,file_path)
@log_decorator
def get_table_name(self,key):
"""获取工作表名字"""
return self.conf.get("table",key)
@log_decorator
def get_server_value(self,key):
"""获取服务器信息"""
return self.conf.get("server",key)
@log_decorator
def get_sql_conn(self,key):
"""获取数据库连接信息"""
return self.conf.get("MySql",key)
if __name__ == '__main__':
ini=ReadIni()
print(ini.get_file_path("excel"))
print(ini.get_table_name("tablename"))
print(ini.get_server_value("host"))
print(ini.get_sql_conn("host"))
\ No newline at end of file
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: Python1
# FileName: read_json.py
# Author: chenyao
# Datetime: 2025/3/12 18:51
# Description:
#
# ---------------------------------------------------------------------------
import json
import os.path
from Auto_TestV4 import log_decorator
@log_decorator
def ReadJson(file_path):
"""从json文件中获取数据"""
if isinstance(file_path,str) and os.path.isfile(file_path) and file_path.endswith(".json"):
with open(file_path,mode="r",encoding="utf-8") as f:
return json.load(f)
else:
raise FileNotFoundError("未找到json文件或者传入路径错误")
if __name__ == '__main__':
print(ReadJson(r"D:\PythonDoc\Python1\Auto_TestV4\data_config\case_data.json"))
\ No newline at end of file
{
"认证接口":{
"登录系统":{
"LoginSuccess":"{\"username\":\"admin\",\"password\":\"K9YOaFEcqlvmh75272mg4z7u7MT/UHPJkaLClDRptg3MPquHhHDRikFyjGtDgf9KcuGAvKQrCbC2HAcsl9oAUdQPQwxgNfBd0CI8YirWpXxwouXu+Q9i1Xnyal+FwYZY6vszsgXbtrHdCknjYL5hzxkLpzKpL/2ahZY45p9R61I=\"}",
"LoginErrorPasswordParamIsNone":"{\"username\":\"admin\",\"\":\"K9YOaFEcqlvmh75272mg4z7u7MT/UHPJkaLClDRptg3MPquHhHDRikFyjGtDgf9KcuGAvKQrCbC2HAcsl9oAUdQPQwxgNfBd0CI8YirWpXxwouXu+Q9i1Xnyal+FwYZY6vszsgXbtrHdCknjYL5hzxkLpzKpL/2ahZY45p9R61I=\"}",
"LoginErrorPasswordIsNone":"{\"username\":\"admin\",\"password\":\"\"}",
"LoginErrorPasswordIsSpecial":"{\"username\":\"admin\",\"password\":\"☻☼♠♥♡♣♤♥♦♧♨\"}",
"LoginErrorPasswordIsLong":"{\"username\":\"admin\",\"password\":\"K9YOaFEcqlvmh75272mg4z7u7MT/UHPJkaLClDRptg3MPquHhHDRikFyjGtDgf9KcuGAvKQrCbC2HAcsl9oAUdQPQwxgNfBd0CI8YirWpXxwouXu+Q9i1Xnyal+FwYZY6vszsgXbtrHdCknjYL5hzxkLpzKpL/2ahZY45p9R61I=K9YOaFEcqlvmh75272mg4z7u7MT/UHPJkaLClDRptg3MPquHhHDRikFyjGtDgf9KcuGAvKQrCbC2HAcsl9oAUdQPQwxgNfBd0CI8YirWpXxwouXu+Q9i1Xnyal+FwYZY6vszsgXbtrHdCknjYL5hzxkLpzKpL/2ahZY45p9R61I=\"}",
"LoginErrorPasswordIsShort":"{\"username\":\"admin\",\"password\":\"1\"}",
"LoginErrorUsernameIsNone":"{\"username\":\"\",\"password\":\"K9YOaFEcqlvmh75272mg4z7u7MT/UHPJkaLClDRptg3MPquHhHDRikFyjGtDgf9KcuGAvKQrCbC2HAcsl9oAUdQPQwxgNfBd0CI8YirWpXxwouXu+Q9i1Xnyal+FwYZY6vszsgXbtrHdCknjYL5hzxkLpzKpL/2ahZY45p9R61I=\"}",
"LoginErrorUsernameIsErr":"{\"username\":\"adminABC123\",\"password\":\"K9YOaFEcqlvmh75272mg4z7u7MT/UHPJkaLClDRptg3MPquHhHDRikFyjGtDgf9KcuGAvKQrCbC2HAcsl9oAUdQPQwxgNfBd0CI8YirWpXxwouXu+Q9i1Xnyal+FwYZY6vszsgXbtrHdCknjYL5hzxkLpzKpL/2ahZY45p9R61I=\"}",
"LoginErrorUsernameIsSpecial":"{\"username\":\"☻☼♠♥♡♣♤♥♦♧♨\",\"password\":\"K9YOaFEcqlvmh75272mg4z7u7MT/UHPJkaLClDRptg3MPquHhHDRikFyjGtDgf9KcuGAvKQrCbC2HAcsl9oAUdQPQwxgNfBd0CI8YirWpXxwouXu+Q9i1Xnyal+FwYZY6vszsgXbtrHdCknjYL5hzxkLpzKpL/2ahZY45p9R61I=\"}",
"LoginErrorUsernameIsShort":"{\"username\":\"a\",\"password\":\"K9YOaFEcqlvmh75272mg4z7u7MT/UHPJkaLClDRptg3MPquHhHDRikFyjGtDgf9KcuGAvKQrCbC2HAcsl9oAUdQPQwxgNfBd0CI8YirWpXxwouXu+Q9i1Xnyal+FwYZY6vszsgXbtrHdCknjYL5hzxkLpzKpL/2ahZY45p9R61I=\"}"
}
},
"维度管理": {
"添加维度": {
"AddDemSuccess": {
"code": "test_dem",
"description": "测试添加的维度",
"isDefault": 0,
"name": "测试添加的维度"
},
"AddDemErrorDemCodeParamIsNone": "{\"\": \"test_demA\",\"description\": \"测试添加的维度A\",\n \"isDefault\": 0,\n \"name\": \"测试添加的维度A\"\n }"
},
"设置默认维度": {
"SetDefaultDemSuccess": {"code": "test_dem"},
"SetDefaultDemFailCodeIsClass": "class=test_dem"
},
"根据维度编码删除维度": {
"DeleteDemSuccess": {"ids": "需要更新"},
"DeleteDemFailIsIsId": "id=需要更新"
}
},
"组织管理": {
"添加组织": {
"AddOrgSuccess": {
"code": "test_org",
"demId": "需要更新",
"exceedLimitNum": 0,
"grade": "高级",
"limitNum": 0,
"name": "测试组织",
"nowNum": 0,
"orderNo": 0,
"parentId": "0"
}
},
"添加组织参数": {
"AddOrgParamSuccess": {
"query": "orgCode=test_org",
"body": [{"alias":"pz","value":9999}]
},
"AddOrgParamSuccessQueryDataTypeDict": {
"query": {"orgCode": "test_org"},
"body": "[{\"alias\":\"pz\",\"value\":8848}]"
}
},
"删除组织": {
"DeleteOrgSuccess": "test_org"
}
}
}
# file节点主要配置数据配置层中存放测试数据的文件
[file]
# excel配置管理用例数据的excel文件
excel=AutoTestAPI.xlsx
# case配置存放用例数据的json文件
case=case_data.json
# expect配置存放期望数据的json文件
expect=expect_data.json
# sql语句的存放文件
sql=sql_data.json
# 配置工作表的名称
[table]
# 配置工作表的名称
tableName=BPM
# 配置被测系统的服务相关的选项
[server]
# host配置被测系统的域名
host=http://36.139.193.99:8088
# 配置mysql数据的信息
[MySql]
host=36.139.193.99
port=3306
user=root
password=Rhrc@2024
database=eip8
\ No newline at end of file
{
"认证接口":{
"登录系统":{
"LoginSuccess":{"loginStatus": true},
"LoginErrorPasswordParamIsNone": {"message":"账号或密码错误"},
"LoginErrorPasswordIsNone":{"message":"账号或密码错误"},
"LoginErrorPasswordIsSpecial":{"message":"账号或密码错误"},
"LoginErrorPasswordIsLong":{"message":"账号或密码错误"},
"LoginErrorPasswordIsShort":{"message":"账号或密码错误"},
"LoginErrorUsernameIsNone":{"message":"账号或密码错误"},
"LoginErrorUsernameIsErr":{"message":"账号或密码错误"},
"LoginErrorUsernameIsSpecial":{"message":"账号或密码错误"},
"LoginErrorUsernameIsShort":{"message":"账号或密码错误"}
},
"刷新token": {
"RefreshSuccess": {"message": "刷新token成功"}
}
},
"维度管理": {
"添加维度": {
"AddDemSuccess": {"message": "添加维度成功!"},
"AddDemErrorDemCodeParamIsNone": {"message": "添加维度失败!"}
},
"设置默认维度": {
"SetDefaultDemSuccess": {"message": "设置默认维度成功!"},
"SetDefaultDemFailCodeIsClass": {"message": "设置默认维度失败!"}
},
"根据维度编码删除维度": {
"DeleteDemSuccess": {"message": "删除维度成功!"},
"DeleteDemFailIsIsId": {"message": "删除维度失败,维度ids必填"}
}
},
"组织管理": {
"添加组织": {
"AddOrgSuccess": {"message": "添加组织成功"}
},
"添加组织参数": {
"AddOrgParamSuccess": {"state":true,"message":"保存组织参数成功!","value":"","code":200},
"AddOrgParamSuccessQueryDataTypeDict": {"state":true,"message":"保存组织参数成功!","value":"","code":200}
},
"删除组织": {
"DeleteOrgSuccess": {"state":true,"message":"删除组织成功!","value":"","code":200}
}
}
}
{
"维度管理":{
"添加维度":{
"AddDemSuccess":"delete from uc_demension where CODE_=\"test_dem\";"
},
"根据维度编码删除维度": {
"DeleteDemSuccess": "select ID_ from uc_demension where CODE_=\"test_dem\";",
"DeleteDemFailIsIsId": "select ID_ from uc_demension where CODE_=\"test_dem\";"
}
},
"组织管理": {
"添加组织": {
"AddOrgSuccess": {
"select": "select ID_ from uc_demension where CODE_=\"test_dem\";",
"delete": "delete from uc_org where CODE_=\"test_org\";"
}
}
}
}
This source diff could not be displayed because it is too large. You can view the blob instead.
This source diff could not be displayed because it is too large. You can view the blob instead.
This source diff could not be displayed because it is too large. You can view the blob instead.
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: Python1
# FileName: __init__.py
# Author: chenyao
# Datetime: 2025/3/12 18:21
# Description:
#
# ---------------------------------------------------------------------------
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: Python1
# FileName: request_method.py
# Author: chenyao
# Datetime: 2025/3/12 21:48
# Description:
#
# ---------------------------------------------------------------------------
from time import sleep
import requests
from Auto_TestV4 import log_decorator
from Auto_TestV4.commom.db import DB
from Auto_TestV4.commom.read_excel import ReadExcel
from Review.pymysql_faker.sql_practice import select_result
class RequestMethod:
@log_decorator
def __init__(self):
self.db=DB()
self.bpm_session=requests.sessions.Session()
login_url="http://36.139.193.99:8088/auth"
login_data={"username":"admin","password":"f3lQcxTepVVR+2ykUc1s3nXGrn7fZWhPyfUW6PxhPr1piAFXSukfUbwHEmeJsPrJ2sJKw0hHqh8PsEqFOb+K35FnU/kohDJ//fu0gCzGg/GGAzXN3Lwshby3eTHPLnUijhDJD2f6x/T5Cd2k/rKkqLDgrJ9SIPMxJBRbt7snjg4="}
self.headers={"Authorization":"Bearer "+self.bpm_session.post(url=login_url,json=login_data).json().get("token")}
self.bpm_session.headers.update(self.headers)
@log_decorator
def request_all_data(self,r_method,r_url,r_type,r_case,r_expect,r_sql,r_sql_type,r_upkey):
if r_sql_type=="delete":
self.db.delete_method(r_sql)
elif r_sql_type=="select":
value=self.db.select_method(r_sql)
if r_upkey=="需要更新":
r_case.replace("需要更新",value)
elif isinstance(r_case,dict):
r_case[r_upkey]=value
elif r_sql_type=="select|delete" or r_sql_type=="delete|select":
self.db.delete_method(r_sql["delete"])
value=self.db.select_method(r_sql["select"])
if r_upkey=="需要更新":
r_case.replace("需要更新",value)
elif isinstance(r_case,dict):
r_case[r_upkey]=value
if r_type=="application/json" or r_type=="json":
self.bpm_session.headers["content-type"]="application/json"
if isinstance(r_case,str):
return self.bpm_session.request(method=r_method,url=r_url,data=r_case)
elif isinstance(r_case,(dict,list)):
return self.bpm_session.request(method=r_method,url=r_url,json=r_case)
elif r_type=="application/x-www-form-urlencoded" or r_type=="x-www-form-urlencoded":
if isinstance(r_case,(str,dict,list)):
return self.bpm_session.request(method=r_method,url=r_url,data=r_case)
elif r_type=="multipart/form-data" or r_type=="form-data":
self.bpm_session.headers["content-type"]="multipart/form-data"
return requests.request(method=r_method,url=r_url,files=r_case,headers=self.headers)
elif r_type=="query":
if isinstance(r_case,str):
r_url=r_url+"?"+r_case
return self.bpm_session.request(method=r_method,url=r_url)
elif isinstance(r_case,dict):
return self.bpm_session.request(method=r_method,url=r_url,params=r_case)
elif r_type is None:
return self.bpm_session.request(method=r_method,url=r_url)
elif r_type=="text/plain":
self.bpm_session.headers["content-type"]="text/plain"
return self.bpm_session.request(method=r_method,url=r_url,data=r_case)
elif r_type=="query|json" or r_type=="json|query":
query_data=r_case["query"]
json_data=r_case["json"]
if isinstance(query_data,str):
r_url=r_url+"?"+r_case
if isinstance(json_data,str):
return self.bpm_session.request(method=r_method,url=r_url,data=json_data)
elif isinstance(json_data,(dict,list)):
return self.bpm_session.request(method=r_method, url=r_url,json=json_data)
elif isinstance(query_data,(dict,list)):
if isinstance(json_data,str):
return self.bpm_session.request(method=r_method,url=r_url,data=json_data,params=query_data)
elif isinstance(json_data,(dict,list)):
return self.bpm_session.request(method=r_method, url=r_url,json=json_data,params=query_data)
else:
raise ValueError("传入媒体类型错误,请更改")
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: Python1
# FileName: __init__.py
# Author: chenyao
# Datetime: 2025/3/12 18:21
# Description:
#
# ---------------------------------------------------------------------------
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: Python1
# FileName: conftest.py
# Author: chenyao
# Datetime: 2025/3/15 15:32
# Description:
#
# ---------------------------------------------------------------------------
import pytest
from Auto_TestV4.request_method.request_method import RequestMethod
@pytest.fixture(scope="session")
def req_fix():
req = RequestMethod()
yield req
req.db.close()
\ No newline at end of file
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName: Python1
# FileName: test_bpm.py
# Author: chenyao
# Datetime: 2025/3/15 14:01
# Description:
#
# ---------------------------------------------------------------------------
import pytest
from Auto_TestV4 import log_decorator
from Auto_TestV4.commom.read_excel import ReadExcel
from Auto_TestV4.request_method.request_method import RequestMethod
data=ReadExcel().get_all_data()
class Testbpm:
@log_decorator
@pytest.mark.parametrize("r_method,r_url,r_type,r_case,r_expect,r_s,r_st,r_uk",data)
def test_bpm(self,req_fix,r_method,r_url,r_type,r_case,r_expect,r_s,r_st,r_uk):
res= RequestMethod().request_all_data(r_method=r_method,r_url=r_url,r_type=r_type,r_case=r_case,r_expect=r_expect,r_sql=r_s,r_sql_type=r_st,r_upkey=r_uk)
print(res)
try:
for key in r_expect:
assert r_expect[key]==res.json().get(key)
except Exception as e:
raise AssertionError(f"断言失败,请求方法为:{r_method}, url:{r_url}, 媒体类型:{r_type}, 用例数据:{r_case}, 期望数据:{r_expect}, sql语句为:{r_s}, sql语句类型:{r_st}, 更新为key:{r_uk}")
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