Commit 75dae9c3 by kangqingyun

接口框架

parents
# -*- coding:utf8 -*- #
# -----------------------------------------------------------------------------------
# ProjectName: LightEditProject
# FileName: db
# Author:
# Datetime: 2023/8/7 18:55
# Description:
# 命名规范:文件名全小写+下划线,类名大驼峰,方法和变量小写+下划线连接,
# 变量用名词,方法用动词
# -----------------------------------------------------------------------------------
import pymysql
from ApiAuotTest.common.read_ini import ReadIni
class DB:
def __init__(self):
read_ini = ReadIni()
self.conn = pymysql.connect(
host=read_ini.get_sql_connect_message("host"),
port=int(read_ini.get_sql_connect_message("port")),
user=read_ini.get_sql_connect_message("user"),
password=read_ini.get_sql_connect_message("pwd"),
database=read_ini.get_sql_connect_message("database"),
charset="utf8")
self.cursor = self.conn.cursor()
def close(self):
self.cursor.close()
self.conn.close()
def delete(self, sql_sentence):
self.cursor.execute(sql_sentence)
self.conn.commit()
def select(self, sql_sentence):
self.cursor.execute(sql_sentence)
result = self.cursor.fetchall()
if result:
return result[0][0]
\ No newline at end of file
# -*- coding:utf8 -*- #
# -----------------------------------------------------------------------------------
# ProjectName: LightEditProject
# FileName: read_excel
# Author:
# Datetime: 2023/8/7 19:19
# Description:
# 命名规范:文件名全小写+下划线,类名大驼峰,方法和变量小写+下划线连接,
# 变量用名词,方法用动词
# -----------------------------------------------------------------------------------
import openpyxl
from ApiAuotTest.common.read_ini import ReadIni
from ApiAuotTest.common.read_json import read_json
from ApiAuotTest.data_config.settings import *
class ReadExcel:
def __init__(self):
self.read_ini = ReadIni()
case_data_path = self.read_ini.get_file_path(CASE)
expect_data_path = self.read_ini.get_file_path(EXPECT)
sql_data_path = self.read_ini.get_file_path(SQL)
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)
excel_path = self.read_ini.get_file_path(EXCEL)
self.wb = openpyxl.load_workbook(excel_path)
self.ws = self.wb[self.read_ini.get_table_name(TABLE)]
def __get_cell_value(self, column: str, row: int) -> str:
value = self.ws[str(column)+str(row)].value
if value is None:
return None
elif value.strip():
return value.strip()
def module_name(self, row):
return self.__get_cell_value(MODULE, row)
def interface_name(self, row):
return self.__get_cell_value(API, row)
def case_title(self, row):
return self.__get_cell_value(TITLE, row)
def case_level(self, row):
return self.__get_cell_value(LEVEL, row)
def case_method(self, row):
return self.__get_cell_value(METHOD, row)
def case_url(self, row):
host = self.read_ini.get_host(HOST)
return host+self.__get_cell_value(PATH, row)
def case_mime(self, row):
value = self.__get_cell_value(MIME, row)
if value:
return value.lower()
def case_data(self, row):
case_data_key = self.__get_cell_value(CASEDATA, row)
if case_data_key:
module_name = self.module_name(row)
interface_name = self.interface_name(row)
return self.case_data_dict[module_name][interface_name][case_data_key]
def expect_data(self, row):
expect_data_key = self.__get_cell_value(EXPECTDATA, row)
if expect_data_key:
module_name = self.module_name(row)
interface_name = self.interface_name(row)
return self.expect_data_dict[module_name][interface_name][expect_data_key]
def sql_type(self, row):
value = self.__get_cell_value(SQLTYPE, row)
if value:
return value.lower()
def sql_sentence(self, row):
sql_data_key = self.__get_cell_value(SQLSENTENCE, row)
if sql_data_key:
module_name = self.module_name(row)
interface_name = self.interface_name(row)
return self.sql_data_dict[module_name][interface_name][sql_data_key]
def update_key(self, row):
return self.__get_cell_value(UPDATEKEY, row)
def get_data(self):
list_data = []
for row in range(2, self.ws.max_row+1):
module = self.module_name(row)
interface = self.interface_name(row)
title = self.case_title(row)
level = self.case_level(row)
method = self.case_method(row)
url = self.case_url(row)
mime = self.case_mime(row)
data = self.case_data(row)
expect = self.expect_data(row)
sql_type = self.sql_type(row)
sql_sentence = self.sql_sentence(row)
update_key = self.update_key(row)
list_data.append([module, interface, title, level, method, url, mime, data, expect, sql_type, sql_sentence, update_key])
else:
return list_data
if __name__ == '__main__':
read_excel = ReadExcel()
print(read_excel.get_data())
\ No newline at end of file
# -*- coding:utf8 -*- #
# -----------------------------------------------------------------------------------
# ProjectName: LightEditProject
# FileName: read_ini
# Author:
# Datetime: 2023/8/7 18:57
# Description:
# 命名规范:文件名全小写+下划线,类名大驼峰,方法和变量小写+下划线连接,
# 变量用名词,方法用动词
# -----------------------------------------------------------------------------------
import configparser
import os
class ReadIni:
def __init__(self):
self.data_config_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), "data_config")
self.ini_path = os.path.join(self.data_config_path, "config.ini")
self.conf = configparser.ConfigParser()
self.conf.read(self.ini_path, encoding="utf-8")
def get_file_path(self, key):
filename = self.conf.get("file", key)
return os.path.join(self.data_config_path, filename)
def get_host(self, key):
return self.conf.get("host", key)
def get_table_name(self, key):
return self.conf.get("table_name", key)
# def get_sql(self, key):
# return self.conf.get("sql", key)
def get_sql_connect_message(self, key):
return self.conf.get("sql", key)
if __name__ == '__main__':
read_ini = ReadIni()
print(read_ini.get_file_path("excel"))
# -*- coding:utf8 -*- #
# -----------------------------------------------------------------------------------
# ProjectName: LightEditProject
# FileName: read_json
# Author:
# Datetime: 2023/8/7 19:37
# Description:
# 命名规范:文件名全小写+下划线,类名大驼峰,方法和变量小写+下划线连接,
# 变量用名词,方法用动词
# -----------------------------------------------------------------------------------
import json
def read_json(filename):
with open(filename, mode="r", encoding="utf-8") as fp:
return json.load(fp)
\ No newline at end of file
{
"认证接口":{
"登录系统":{
"LoginSuccess": {"password": "MTIzNDU2","username": "admin"},
"LoginFailUsernameIsNone":{"password": "MTIzNDU2","username": ""},
"LoginFailUsernameIsLong":{"password": "MTIzNDU2","username": "adminadminadminadminadminadminadminadminadmin"},
"LoginFailUsernameIsShort":{"password": "MTIzNDU2","username": "a"},
"LoginFailUsernameIsSpecialChar":{"password": "MTIzNDU2","username": "▬♦◊◦☼♠♣"},
"LoginFailUsernameIsError":{"password": "MTIzNDU2","username": "tomcat"},
"LoginFailPWDIsNone":{"password": "","username": "admin"},
"LoginFailPWDIsLong":{"password": "MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2MTIzNDU2","username": "admin"},
"LoginFailPWDIsShort":{"password": "1","username": "admin"},
"LoginFailPWDIsSpecial":{"password": "▬♦◊◦☼♠♣","username": "admin"}
}
},
"维度管理": {
"添加维度": {
"AddDemSuccess": {
"code": "超人",
"description": "打怪兽的超人",
"isDefault": 0,
"name": "小心超人"
}
},
"根据维度编码删除维度": {
"DeleteDemSuccess": {"ids": "不知道"}
}
},
"组织管理": {
"添加组织": {
"AddOrgSuccess": {
"code": "超人组织",
"demId": "不知道",
"exceedLimitNum": 0,
"grade": "",
"limitNum": 0,
"name": "打怪兽的超人组织",
"nowNum": 0,
"orderNo": 0,
"parentId": "0"
}
},
"删除组织": {
"DeleteOrgSuccess": "超人组织"
},
"保存组织参数": {
"SaveOrgParamSuccess": {
"query": {"orgCode": "超人组织"},
"body": [
{"alias":"sz","value":0},
{"alias":"zy","value":"math"},
{"alias":"rq","value":"2023-08-04T16:00:00.000Z"},
{"alias":"ah","value":"lq"},
{"alias":"yyyy","value":"越高越好"},
{"alias":"xb","value":"2"}]
}
}
}
}
[file]
excel=APIAutoTest.xlsx
case=case_data.json
expect=expect_data.json
sql=sql.json
[host]
test_bpm=http://120.46.172.186:8080
[table_name]
table=BPM接口
[sql]
host=120.46.172.186
port=3306
user=root
pwd=root@2023
database=eipsaas
\ No newline at end of file
{
"认证接口":{
"登录系统":{
"LoginSuccess": {"username": "超级管理员","account": "admin"},
"LoginFailUsernameIsNone":{"state": false,"message": "账号或密码错误"},
"LoginFailUsernameIsLong":{"state": false,"message": "账号或密码错误"},
"LoginFailUsernameIsShort":{"state": false,"message": "账号或密码错误"},
"LoginFailUsernameIsSpecialChar":{"state": false,"message": "账号或密码错误"},
"LoginFailUsernameIsError":{"state": false,"message": "账号或密码错误"},
"LoginFailPWDIsNone":{"state": false,"message": "账号或密码错误"},
"LoginFailPWDIsLong":{"state": false,"message": "账号或密码错误"},
"LoginFailPWDIsShort":{"state": false,"message": "账号或密码错误"},
"LoginFailPWDIsSpecial":{"state": false,"message": "账号或密码错误"}
},
"刷新token": {
"RefreshSuccess": {"message": "刷新token成功"}
}
},
"维度管理": {
"添加维度": {
"AddDemSuccess": {"state": true, "message": "添加维度成功!"}
},
"根据维度编码删除维度": {
"DeleteDemSuccess": {"state": true, "message": "删除维度成功!"}
}
},
"组织管理": {
"添加组织": {
"AddOrgSuccess": {"state": true, "message": "添加组织成功!"}
},
"删除组织": {
"DeleteOrgSuccess": {"state": true, "message": "删除组织成功!"}
},
"保存组织参数": {
"SaveOrgParamSuccess": {"state": true, "message": "保存组织参数成功!"}
}
}
}
# -*- coding:utf8 -*- #
# -----------------------------------------------------------------------------------
# ProjectName: LightEditProject
# FileName: settings
# Author:
# Datetime: 2023/8/7 20:28
# Description:
# 命名规范:文件名全小写+下划线,类名大驼峰,方法和变量小写+下划线连接,
# 变量用名词,方法用动词
# -----------------------------------------------------------------------------------
# excel列号的配置文件
MODULE = "B"
API = "C"
TITLE = "D"
LEVEL = "E"
PATH = "F"
METHOD = "G"
MIME = "H"
CASEDATA = "I"
EXPECTDATA = "J"
SQLTYPE = "K"
SQLSENTENCE = "L"
UPDATEKEY = "M"
# 配ini文件中file节点下的key
EXCEL = "excel"
CASE = "case"
EXPECT = "expect"
SQL = "sql"
# 配置table_name节点下的key
TABLE = "table"
# 配置host节点下的key
HOST = "test_bpm"
\ No newline at end of file
{
"维度管理": {
"添加维度": {
"AddDemSuccess": "DELETE FROM uc_demension WHERE `CODE_`=\"超人\";"
},
"根据维度编码删除维度": {
"DeleteDemSuccess": "SELECT ID_ FROM uc_demension WHERE `CODE_`=\"超人\";"
}
},
"组织管理": {
"添加组织": {
"AddOrgSuccess": {
"select": "SELECT ID_ FROM uc_demension WHERE `CODE_`=\"超人\";",
"delete": "DELETE FROM uc_org WHERE `CODE_`=\"超人组织\";"
}
}
}
}
\ No newline at end of file
++ "a/ApiAuotTest/report/log/\346\227\245\345\277\227.log"
# -*- coding:utf8 -*- #
# -----------------------------------------------------------------------------------
# ProjectName: LightEditProject
# FileName: requests_method
# Author:
# Datetime: 2023/8/7 22:39
# Description:
# 命名规范:文件名全小写+下划线,类名大驼峰,方法和变量小写+下划线连接,
# 变量用名词,方法用动词
# -----------------------------------------------------------------------------------
import base64
import requests
from ApiAuotTest.common.read_ini import ReadIni
class RequestsMethod:
def __init__(self):
"""关联token"""
url = ReadIni().get_host("test_bpm") + "/auth"
data = {"username": "admin", "password": base64.b64encode("123456".encode("utf-8")).decode("utf-8")}
self.bpm_session = requests.sessions.Session()
self.bpm_session.headers["Authorization"] = "Bearer "+self.bpm_session.post(url=url, json=data).json().get("token")
def request_all(self, req_method, req_url, req_mime, case_data):
if req_mime == "json" or req_mime == "application/json":
return self.bpm_session.request(method=req_method, url=req_url, json=case_data)
elif req_mime == "x-www-form-urlencoded" or req_mime == "application/x-www-form-urlencoded":
return self.bpm_session.request(method=req_method, url=req_url, data=case_data)
elif req_mime == "form-data" or req_mime == "multipart/form-data":
return self.bpm_session.request(method=req_method, url=req_url, files=case_data)
elif req_mime == "query" or req_mime == "params":
return self.bpm_session.request(method=req_method, url=req_url, params=case_data)
elif req_mime == "application/json|query" or req_mime == "json|query" or req_mime == "query|json":
return self.bpm_session.request(method=req_method, url=req_url, params=case_data["query"], json=case_data["body"])
elif req_mime == "text/plain" or req_mime == "text":
return self.bpm_session.request(method=req_method, url=req_url, data=case_data)
elif req_mime is None:
return self.bpm_session.request(method=req_method, url=req_url)
else:
raise NameError("传入的媒体类型的名称错误")
# -*- coding:utf8 -*- #
# -----------------------------------------------------------------------------------
# ProjectName: LightEditProject
# FileName: conftest
# Author:
# Datetime: 2023/8/7 22:34
# Description:
# 命名规范:文件名全小写+下划线,类名大驼峰,方法和变量小写+下划线连接,
# 变量用名词,方法用动词
# -----------------------------------------------------------------------------------
import pytest
from ApiAuotTest.common.db import DB
from ApiAuotTest.requests_method.requests_method import RequestsMethod
@pytest.fixture(scope="session")
def req_fix():
req = RequestsMethod()
yield req
@pytest.fixture(scope="session")
def db_fix():
db = DB()
yield db
db.close()
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
[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/日志.log
;设置日志保存在文件中的级别
log_file_level=error
;设置日志在文件中的信息格式
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:utf8 -*- #
# -----------------------------------------------------------------------------------
# ProjectName: LightEditProject
# FileName: test_bpm
# Author:
# Datetime: 2023/8/7 22:48
# Description:
# 命名规范:文件名全小写+下划线,类名大驼峰,方法和变量小写+下划线连接,
# 变量用名词,方法用动词
# -----------------------------------------------------------------------------------
import logging
import allure
import pytest
from ApiAuotTest.common.read_excel import ReadExcel
class TestBPM:
@allure.epic("BPM系统")
@pytest.mark.parametrize("module, interface, title, level, method, url, mime, data, expect, sql_type, sql_sentence, update_key", ReadExcel().get_data())
def test_bpm(self, req_fix, db_fix, module, interface, title, level, method, url, mime, data, expect, sql_type, sql_sentence, update_key):
allure.dynamic.feature(module)
allure.dynamic.story(interface)
allure.dynamic.title(title)
allure.dynamic.severity(level)
if sql_type == "delete":
db_fix.delete(sql_sentence)
elif sql_type == "select":
select_result = db_fix.select(sql_sentence)
data[update_key] = select_result
elif sql_type == "delete|select" or sql_type == "select|delete":
db_fix.delete(sql_sentence["delete"])
select_result = db_fix.select(sql_sentence["select"])
data[update_key] = select_result
result = req_fix.request_all(req_method=method, req_url=url, req_mime=mime, case_data=data)
try:
for key in expect:
assert expect[key] == result.json().get(key)
except:
print("断言失败")
logging.error("断言失败, 模块名称为:{},接口名称为:{},用例标题:{},用例为:{},期望结果为:{},服务器返回数据为:{}".format(module, interface, title, str(data), str(expect), result.text))
raise AssertionError("断言失败")
else:
print("断言成功")
if __name__ == '__main__':
pytest.main()
\ No newline at end of file
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