floraachy/apiautotest
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1、优化断言 2、优化依赖接口调用

Browse Source
This commit is contained in:
floraachy
2023-12-13 12:14:05 +08:00
parent 4d11d91544
commit 3c713a1ed3
28 changed files with 762 additions and 443 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+87 -1
View File
@@ -149,7 +149,93 @@ case_info: 具体的用例数据，是以列表的形式进行管理
assert_response：响应断言
assert_sql：数据库断言
```
### 6. Excel用例单独说明
### 6. 参数提取说明
目前支持3种方式的参数提取：type_jsonpath, type_re, type_response
#### type_jsonpath
如果采用jsonpath方式从响应数据提取参数，`extract`的key是`type_jsonpath`; `extract[type_jsonpath]`的key是变量名，value是提取表达式；
参考示例：
```
extract:
type_jsonpath:
nickname: $.username
login: $.login
user_id: $.user_id
```
注意：如果提取到的值是长度为1的列表，会自动获取第一个元素，其值类型由list变更为str。例如通过`name: $.data` 提取到的值是：`["flora"]`, 最后name的值会是`"flora"`。
#### type_re
如果采用正则表达式方式从响应数据提取参数，`extract`的key是`type_re`; `extract[type_re]`的key是变量名，value是提取表达式；
参考示例：
```
# 注意：\是用来转译的，保证yaml格式正确.也可以用引号包裹起来
extract:
type_re:
nickname: \"username":"(.*?)"
login: \"login":"(.*?)"
user_id: '"user_id":(.*?),'
```
注意：如果提取到的值是长度为1的列表，会自动获取第一个元素，其值类型由list变更为str。例如通过`name: "username":"(.*?)"` 提取到的值是：`["flora"]`, 最后name的值会是`"flora"`。
#### type_response
如果是直接从响应数据提取参数，`extract`的key是`type_response`; `extract[type_response]`的key是变量名，value是提取表达式；
基本上response所有方法都支持，部分参考：`response.status_code`， `response.cookies`, `response.text`, `response.headers`, `response.is_redirect`
参考示例：
```
extract:
type_response:
cookies: response.cookies
```
### 6. 响应断言说明
#### 断言状态码
如果想要断言接口响应码，直接这样写即可：
参考示例：
```
assert_response:
status_code: 200
```
#### 响应数据断言
响应断言的参数说明：
```
断言标识(自定义，不为空即可，没有实际的意义):
message: 断言信息，非必填，可为空
expect_value: 预期结果
assert_type: 断言类型，支持如下：==, lt, le, gt, ge, not_eq, str_eq, len_eq, len_gt, len_ge, len_lt, len_le, contains, contained_by, startswith, endswith
type_jsonpath: 通过jsonpath表达式通响应数据提取实际结果，与type_re任选其一
type_re: 通过正则表达式通响应数据提取实际结果，与type_jsonpath任选其一
```
注意：在进行断言的时候，左侧是预期结果，右侧是实际结果。比如我断言类型是`lt`， 那么就是预期结果<实际结果
参考示例：
```
assert_response:
user_id:
message: 断言接口返回的user_id
expect_value: ${user_id}
assert_type: ==
type_jsonpath: $.user_id
```
```
assert_response:
user_id:
expect_value: ${user_id}
assert_type: ==
type_jsonpath: $.user_id
```
### 7. 响应断言说明
todo 待补充ing
### 8. Excel用例单独说明
框架支持excel多表单自动生成测试用例，每一个表单作为一个测试用例模块。
例如：
excel表格名称是：test_demo.xlsx
config/path_config.py
+3 -2
View File
@@ -18,7 +18,7 @@ COMMON_DIR = os.path.join(BASE_DIR, "common_utils")
CONF_DIR = os.path.join(BASE_DIR, "config")
# 测试数据模块目录
DATA_DIR = os.path.join(BASE_DIR, "data")
DATA_DIR = os.path.join(BASE_DIR, "interface")
# 测试文件模块目录
FILES_DIR = os.path.join(BASE_DIR, "files")
@@ -47,7 +47,7 @@ if not os.path.exists(MANUAL_CASE_DIR):
os.mkdir(MANUAL_CASE_DIR)
# 测试用例方法模板路径
CASE_TEMPLATE_DIR = os.path.join(CONF_DIR, "../utils/case_data_utils/case_template.txt")
CASE_TEMPLATE_DIR = os.path.join(CONF_DIR, "../utils/case_generate_utils/case_template.txt")
# 自动生成测试用例模块
AUTO_CASE_DIR = os.path.join(CASE_DIR, "test_auto_case")
@@ -57,6 +57,7 @@ LIB_DIR = os.path.join(BASE_DIR, "lib")
# Allure报告，测试结果集目录
ALLURE_RESULTS_DIR = os.path.join(REPORT_DIR, "allure_results")
# Allure报告，HTML测试报告目录
ALLURE_HTML_DIR = os.path.join(REPORT_DIR, "allure_html")
config/settings.py
+2 -2
View File
@@ -33,7 +33,7 @@ ENV_VARS = {
"db_user": "root",
"db_pwd": "**********",
"db_database": "test**********",
"ssh": True,
"ssh": False,
"ssh_host": "xx.xx.xx.xx",
"ssh_port": 3306,
"ssh_user": "root",
@@ -90,7 +90,7 @@ class RunConfig:
CASE_FILE_TYPE = 1
# 0表示默认不发送任何通知， 1 代表钉钉通知，2 代表企业微信通知， 3 代表邮件通知， 4 代表所有途径都发送通知
SEND_RESULT_TYPE = 3
SEND_RESULT_TYPE = 0
# 指定日志收集级别
LOG_LEVEL = "DEBUG" # 可选值：TRACE DEBUG INFO SUCCESS WARNING ERROR CRITICAL
data/gitlink/test_login.yaml
-77
View File
@@ -1,77 +0,0 @@
# 公共参数
case_common:
allure_epic: GitLink接口 # 敏捷里面的概念，定义史诗，相当于module级的标签, 往下是 feature
allure_feature: 登录模块 # 功能点的描述，相当于class级的标签, 理解成模块往下是 story
allure_story: 登录接口 # 故事，可以理解为场景，相当于method级的标签, 往下是 title
case_markers: # pytest框架的标记 pytest.mark.
- gitlink
- login: 登录接口
# 用例数据
case_info:
-
id: case_login_01
title: 用户名密码正确，登录成功(不校验数据库)
run: True
severity: normal
url: /api/accounts/login.json
method: POST
headers: {"Content-Type": "application/json; charset=utf-8;"}
cookies:
request_type: json
payload: { "login": "${login}","password": "${password}","autologin": 1 }
files:
extract:
type_re:
nickname: \"username":"(.*?)"
login: \"login":"(.*?)"
user_id: \"user_id":(.*?),
assert_response:
eq:
http_code: 200
$.user_id: ${user_id}
assert_sql:
-
id: case_login_02
title: 用户名密码正确，登录成功（校验数据库）
run: False
severity: minor
url: /api/accounts/login.json
method: POST
headers: {"Content-Type": "application/json; charset=utf-8;"}
cookies:
request_type: json
payload: { "login": "${login}","password": "${password}","autologin": 1 }
files:
extract:
nickname: $.username
login: $.login
user_id: $.user_id
assert_response:
eq:
http_code: 200
$.user_id: ${user_id}
assert_sql:
eq:
sql: select count(*) from tokens where user_id=${user_id};
len: 1
-
id: case_login_03
title: 用户名正确，密码错误，登录失败
severity: critical
run: False
url: /api/accounts/login.json
method: POST
headers: {"Content-Type": "application/json; charset=utf-8;"}
cookies:
request_type: json
payload: { "login": "chytest10","password": "password111","autologin": 1 }
files:
extract:
assert_response:
eq:
http_code: 200
$.status: -2
assert_sql:
files/demo_test_demo.py
-32
View File
@@ -1,32 +0,0 @@
# -*- coding: utf-8 -*-
# @Version: Python 3.9
# @Time : 2023/1/9 16:41
# @Author : chenyinhua
# @File : demo_test_login.py
# @Software: PyCharm
# @Desc: python脚本编写的测试用例文件
# 第三方库导入
import pytest
from loguru import logger
import allure
# 本地应用/模块导入
from utils.report_utils.allure_handle import allure_title, allure_step
# 读取用例数据
cases = [{"title": "demo用例01", "severity": "blocker1", "user": "flora", "age": 17, "run": True},
{"title": "demo用例02", "severity": "TRIVIAL", "user": "flora", "age": 17, "run": True}]
@allure.story("demo模块(手动用例)")
@pytest.mark.test_demo
@pytest.mark.parametrize("case", cases, ids=["{}".format(case["title"]) for case in cases])
def test_demo(case):
logger.info("\n-----------------------------START-开始执行用例-----------------------------\n")
logger.debug(f"当前执行的用例数据:{case}")
# 添加用例标题作为allure中显示的用例标题
allure_title(case.get("title", ""))
# 在allure报告中显示请求的用例数据
allure_step(step_title="用例数据", content=f"{case}")
assert case["user"] == "flora"
logger.info("\n-----------------------------END-用例执行结束-----------------------------\n")
files/demo_test_login.py
-52
View File
@@ -1,52 +0,0 @@
# -*- coding: utf-8 -*-
# @Version: Python 3.9
# @Time : 2023/1/9 16:41
# @Author : chenyinhua
# @File : demo_test_login.py
# @Software: PyCharm
# @Desc: python脚本编写的测试用例文件
# 标准库导入
import os
# 第三方库导入
import pytest
import allure
from loguru import logger
# 本地应用/模块导入
from utils.files_utils.yaml_handle import YamlHandle
from config.path_config import DATA_DIR
from config.settings import db_info
from config.global_vars import GLOBAL_VARS
from utils.requests_utils.assert_handle import assert_response, assert_sql
from utils.requests_utils.request_control import RequestPreDataHandle, RequestHandle, after_request_extract
from utils.report_utils.allure_handle import allure_title
# 读取用例数据
yaml_data = YamlHandle(filename=os.path.join(DATA_DIR, "gitlink", "test_login.yaml")).read_yaml
case_common = yaml_data["case_common"]
cases = yaml_data["case_info"]
@allure.epic(case_common["allure_epic"])
@allure.feature(case_common["allure_feature"])
class TestLoginDemo:
@allure.story(case_common["allure_story"])
@pytest.mark.test_demo
@pytest.mark.parametrize("case", cases, ids=["{}".format(case["title"]) for case in cases])
def test_login_demo_auto(self, case):
logger.info("\n-----------------------------START-开始执行用例-----------------------------\n")
logger.debug(f"当前执行的用例数据:{case}")
# 添加用例标题作为allure中显示的用例标题
allure_title(case.get("title", ""))
# 处理请求前的用例数据
case_data = RequestPreDataHandle(case).request_data_handle()
# 发送请求
response = RequestHandle(case_data).http_request()
# 进行响应断言
assert_response(response, case_data["assert_response"])
# 进行数据库断言
assert_sql(db_info[GLOBAL_VARS["env_key"]], case_data["assert_sql"])
# 断言成功后进行参数提取
after_request_extract(response, case_data.get("extract", None))
logger.info("\n-----------------------------END-用例执行结束-----------------------------\n")
data/gitlink/glcc/test_get_apply_information.yml → interface/gitlink/glcc/test_get_apply_information.yml
+2 -6
View File
@@ -25,9 +25,7 @@ case_info:
files:
extract:
assert_response:
eq:
http_code: 200
$.message: success
status_code: 200
assert_sql:
-
@@ -47,7 +45,5 @@ case_info:
files:
extract:
assert_response:
eq:
http_code: 200
$.message: success
status_code: 200
assert_sql:
interface/gitlink/projects/test_get_ignores.yaml
+28
View File
@@ -0,0 +1,28 @@
case_common:
allure_epic: GitLink接口
allure_feature: 开源项目模块
allure_story: 项目
case_markers:
- gitlink
- usefixtures: gitlink_login
# 用例数据
case_info:
-
id: gitlink_projects_get_ignores_01
title: 获取gitignore列表数据
severity: normal
run: True
url: /api/ignores.json
method: GET
headers:
Content-Type: application/json; charset=utf-8;
cookies: ${cookies}
cookies:
request_type: params
payload:
files:
extract:
assert_response:
status_code: 200
assert_sql:
data/gitlink/test_new_project_demo.yaml → interface/gitlink/projects/test_new_project_demo.yaml
+53 -24
View File
@@ -2,24 +2,24 @@
case_common:
allure_epic: GitLink接口 # 敏捷里面的概念，定义史诗，相当于module级的标签, 往下是 feature
allure_feature: 开源项目模块 # 功能点的描述，相当于class级的标签, 理解成模块往下是 story
allure_story: 新建项目接口 # 故事，可以理解为场景，相当于method级的标签, 往下是 title
allure_story: 项目 # 故事，可以理解为场景，相当于method级的标签, 往下是 title
case_markers:
- gitlink
- new_project
- usefixtures: login_init
- usefixtures: gitlink_login
# 用例数据
case_info:
-
id: case_new_project_01
title: 正确输入各项必填参数，新建项目成功（不校验数据库, header里面传cookies）
id: gitlink_projects_new_project_01
title: 正确输入各项必填参数，新建公开项目成功
severity: critical
run: True
url: /api/projects.json
method: POST
headers:
Content-Type: application/json; charset=utf-8;
cookies: ${login_cookie}
cookies: ${cookies}
cookies:
request_type: json
payload:
@@ -28,51 +28,76 @@ case_info:
repository_name: ${generate_identifier()}
files:
extract:
type_json:
type_jsonpath:
project_id: $.id
project_name: $.name
project_identifier: $.identifier
assert_response:
eq:
http_code: 200
$.login: ${login}
status_code: 200
login:
message: 断言接口返回的login
expect_value: ${login}
assert_type: ==
type_jsonpath: $.login
assert_sql:
-
id: case_new_project_02
title: 正确输入各项必填参数，新建项目成功（不校验数据库，单独传cookies）
run: True
id: gitlink_projects_new_project_02
title: 正确输入所有参数，新建私有项目成功
run: false
url: /api/projects.json
method: POST
headers:
Content-Type: application/json; charset=utf-8;
cookies: ${login_cookie}
cookies: ${cookies}
request_type: json
payload:
user_id: ${user_id}
name: ${generate_name(lan='zh')}_${generate_identifier()}
repository_name: ${generate_identifier()}
description:
private: true
ignoreFlag: true
ignore_id:
ignore:
licenseFlag: true
license_id:
license:
categoreFlag: true
project_category_id:
project_category:
languageFlag: true
project_language_id:
project_language:
blockchain: false
blockchain_token_all: 10000
blockchain_init_token:
auth_password:
files:
extract:
type_json:
type_jsonpath:
project_id: $.id
project_name: $.name
project_identifier: $.identifier
assert_response:
eq:
http_code: 200
$.login: ${login}
status_code: 200
login:
message: 断言接口返回的login
expect_value: ${login}
assert_type: ==
type_jsonpath: $.login
assert_sql:
-
id: case_new_project_03
id: gitlink_projects_new_project_03
title: 正确输入各项必填参数，新建项目成功（校验数据库）
severity: normal
run: False
run: false
url: /api/projects.json
method: POST
headers: {"Content-Type": "application/json; charset=utf-8;"}
cookies:
headers:
Content-Type: application/json; charset=utf-8;
cookies: ${cookies}
request_type: json
payload:
user_id: ${user_id}
@@ -80,13 +105,17 @@ case_info:
repository_name: ${generate_identifier()}
files:
extract:
type_re:
project_id: $.id
project_name: $.name
project_identifier: $.identifier
assert_response:
eq:
http_code: 200
$.login: ${login}
status_code: 200
login:
message: 断言接口返回的login
expect_value: ${login}
assert_type: ==
type_jsonpath: $.login
assert_sql:
eq:
sql: select id,`name`, identifier from projects where user_id=${user_id} ORDER BY created_on DESC;
data/gitlink/test_upload_files.yaml → interface/gitlink/projects/test_upload_files.yaml
+3 -5
View File
@@ -6,7 +6,7 @@ case_common:
case_markers:
- gitlink
- upload_file
- usefixtures: login_init
- usefixtures: gitlink_login
# 用例数据
case_info:
@@ -24,10 +24,8 @@ case_info:
payload:
files: TOC出库订单导入模板(2).xlsx
extract:
type_json:
type_jsonpath:
file_id: $.id
assert_response:
eq:
http_code: 200
status_code: 200
assert_sql:
interface/gitlink/test_login.yaml
+124
View File
@@ -0,0 +1,124 @@
# 公共参数
case_common:
allure_epic: GitLink接口 # 敏捷里面的概念，定义史诗，相当于module级的标签, 往下是 feature
allure_feature: 登录模块 # 功能点的描述，相当于class级的标签, 理解成模块往下是 story
allure_story: 登录接口 # 故事，可以理解为场景，相当于method级的标签, 往下是 title
case_markers: # pytest框架的标记 pytest.mark.
- gitlink
- login: 登录接口
# 用例数据
case_info:
-
id: gitlink_login_01
title: 用户名密码正确，登录成功(不校验数据库)
run: True
severity: normal
url: /api/accounts/login.json
method: POST
headers:
Content-Type: application/json; charset=utf-8;
cookies:
request_type: json
payload:
login: ${login}
password: ${password}
autologin: 1
files:
extract:
type_re:
nickname: \"username":"(.*?)"
login: \"login":"(.*?)"
user_id: \"user_id":(.*?),
type_response:
cookies: response.cookies
assert_response:
status_code: 200
user_id:
message: 断言接口返回的user_id
expect_value: ${user_id}
assert_type: ==
type_jsonpath: $.user_id
login:
message: 断言接口返回的login
expect_value: ${login}
assert_type: ==
type_jsonpath: $.login
assert_sql:
-
id: case_login_02
title: 用户名密码正确，登录成功（校验数据库）
run: false
severity: minor
url: /api/accounts/login.json
method: POST
headers:
Content-Type: application/json; charset=utf-8;
cookies:
request_type: json
payload:
login: ${login}
password: ${password}
autologin: 1
files:
extract:
type_jsonpath:
nickname: $.username
login: $.login
user_id: $.user_id
assert_response:
status_code: 200
user_id:
expect_value: ${user_id}
assert_type: ==
type_jsonpath: $.user_id
login:
message: 断言接口返回的login
expect_value: ${login}
assert_type: ==
type_jsonpath: $.login
assert_sql:
contains_user:
message: 断言数据库：查询tokens表，表中存在该登录用户记录；预期查询结果的长度为1
sql: select * from tokens where user_id=${user_id};
expect_value: 1
assert_type: len_eq
contains_user2:
message: 断言数据库：查询tokens表，表中存在该登录用户记录;预期 ${user_id} in 查询结果（jsonpath式匹配后的结果）
sql: select * from tokens where user_id=${user_id};
type_jsonpath: $..user_id
expect_value: ${user_id}
assert_type: ==
contains_user3:
message: 断言数据库：查询tokens表，表中存在该登录用户记录；预期 ${user_id} in 查询结果（正则表达式匹配后的结果）
sql: select * from tokens;
type_re: "'user_id': (.*?),"
expect_value: ${user_id}
assert_type: contains
-
id: case_login_03
title: 用户名正确，密码错误，登录失败
severity: critical
run: true
url: /api/accounts/login.json
method: POST
headers:
Content-Type: application/json; charset=utf-8;
cookies:
request_type: json
payload:
login: ${login}
password: 12345678900
autologin: 1
files:
extract:
assert_response:
status_code: 200
user_id:
message: 断言接口返回的status
expect_value: -2
assert_type: ==
type_jsonpath: $.status
assert_sql:
run.py
+1 -1
View File
@@ -40,7 +40,7 @@ import pytest
from loguru import logger
import click
# 本地应用/模块导入
from utils.case_data_utils.case_fun_generate import generate_cases
from utils.case_generate_utils.case_fun_generate import generate_cases
from utils.report_utils.send_result_handle import send_result
from config.path_config import REPORT_DIR, LOG_DIR, CONF_DIR, ALLURE_RESULTS_DIR, ALLURE_HTML_DIR, AUTO_CASE_DIR
from config.settings import LOG_LEVEL, RunConfig, ENV_VARS
test_case/conftest.py
+7 -26
View File
@@ -4,6 +4,7 @@
# @File : conftest.py
# @Software: PyCharm
# @Desc:
import os.path
# 标准库导入
# 第三方库导入
@@ -12,8 +13,9 @@ import allure
from loguru import logger
# 本地应用/模块导入
from config.global_vars import GLOBAL_VARS
from utils.requests_utils.base_request import BaseRequest
from config.path_config import DATA_DIR
from utils.report_utils.allure_handle import allure_title
from utils.requests_utils.api_workflow import get_api_data, api_work_flow
@pytest.fixture(scope="function", autouse=True)
@@ -70,33 +72,12 @@ def pytest_collection_modifyitems(config, items):
@pytest.fixture(scope="session")
def login_init():
def gitlink_login():
"""
获取登录的cookie
:return:
"""
host = GLOBAL_VARS.get("host")
login = GLOBAL_VARS.get('login')
password = GLOBAL_VARS.get('password')
# 兼容一下host后面多一个斜线的情况
if host[-1] == "/":
host = host[:len(host) - 1]
req_data = {
"url": host + "/api/accounts/login.json",
"method": "POST",
"request_type": "json",
"headers": {"Content-Type": "application/json; charset=utf-8;"},
"payload": {"login": login, "password": password, "autologin": 1}
}
# 请求登录接口
try:
res = BaseRequest.send_request(req_data=req_data)
res.raise_for_status()
logger.debug(f"获取用户：{login}登录的cookies成功：{type(res.cookies)} || {res.cookies}")
GLOBAL_VARS["login_cookie"] = res.cookies
GLOBAL_VARS["login"] = res.json()["login"]
GLOBAL_VARS["nickname"] = res.json()["username"]
GLOBAL_VARS["user_id"] = res.json()["user_id"]
except Exception as e:
GLOBAL_VARS["login_cookie"] = None
logger.error(f"获取用户：{login}登录的cookies失败：{e}")
login_api = get_api_data(os.path.join(DATA_DIR, "gitlink", "test_login.yaml"), "gitlink_login_01")
api_work_flow(login_api, GLOBAL_VARS)
utils/assertion_utils/__init__.py
+6
View File
@@ -0,0 +1,6 @@
# -*- coding: utf-8 -*-
# @Time : 2023/12/12 10:59
# @Author : floraachy
# @File : __init__.py
# @Software: PyCharm
# @Desc:
utils/assertion_utils/assert_control.py
+165
View File
@@ -0,0 +1,165 @@
# -*- coding: utf-8 -*-
# @Version: Python 3.9
# @Time : 2023/12/12 14:31
# @Author : floraachy
# @File : assert_control.py
# @Software: PyCharm
# @Desc: 断言类型封装，支持json响应断言、正则表达式响应断言、数据库断言
# 标准库导入
import types
# 第三方库导入
import allure
from requests import Response
from loguru import logger
# 本地应用/模块导入
from utils.models import AssertMethod
from utils.assertion_utils import assert_function
from utils.data_utils.extract_data_handle import json_extractor, re_extract
from utils.database_utils.mysql_handle import MysqlServer
class AssertUtils:
def __init__(self, assert_data, response: Response = None, db_info: dict = None):
"""
断言处理
:param assert_data: 断言数据
:param response: 接口响应数据
:param db_info: 数据库信息
"""
self.assert_data = assert_data
self.response = response
if db_info:
self.db_connect = MysqlServer(**db_info)
@property
def get_message(self):
"""
获取断言描述，如果未填写，则返回 `None`
:return:
"""
return self.assert_data.get("message", "")
@property
def get_assert_type(self):
"""
检查assert_type是否是模型类AssertMethod中指定的值
"""
assert 'assert_type' in self.assert_data.keys(), (
" 断言数据: '%s' 中缺少 `assert_type` 属性 " % self.assert_data
)
# 获取断言类型对应的枚举值
name = AssertMethod(self.assert_data.get("assert_type")).name
return name
@property
def get_sql_result(self):
"""
通过数据库查询获取查询结果
"""
if "sql" not in self.assert_data.keys() or self.assert_data["sql"] is None:
logger.error(f"断言数据: {self.assert_data} 缺少 'sql' 属性或 'sql' 为空")
raise ValueError("断言数据: {self.assert_data} 缺少 'sql' 属性或 'sql' 为空")
return self.db_connect.query_all(sql=self.assert_data["sql"])
def get_actual_value_by_response(self):
"""
通过jsonpath表达式从响应数据中获取实际结果
通过jsonpath表达式从响应数据中获取实际结果
"""
if "type_jsonpath" in self.assert_data and self.assert_data["type_jsonpath"]:
return json_extractor(obj=self.response.json(), expr=self.assert_data["type_jsonpath"])
if "type_re" in self.assert_data and self.assert_data["type_re"]:
return re_extract(obj=self.response.text, expr=self.assert_data["type_re"])
else:
assert 'type_re' or 'type_jsonpath' in self.assert_data.keys(), (
" 断言数据: '%s' 中缺少 `type_re` 或 `type_jsonpath` 属性 " % self.assert_data
)
def get_actual_value_by_sql(self):
"""
通过jsonpath表达式从数据库查询结果中获取实际结果
通过正则表达式从数据库查询结果中获取实际结果
"""
if "type_jsonpath" in self.assert_data and self.assert_data["type_jsonpath"]:
return json_extractor(obj=self.get_sql_result, expr=self.assert_data["type_jsonpath"])
elif "type_re" in self.assert_data and self.assert_data["type_re"]:
return re_extract(obj=str(self.get_sql_result), expr=self.assert_data["type_re"])
else:
return self.get_sql_result
@property
def get_expect_value(self):
"""
获取预期结果， 断言数据中应该存在key=expect_value
"""
assert 'expect_value' in self.assert_data.keys(), (
f"断言数据: {self.assert_data} 中缺少 `value` 属性 "
)
return self.assert_data.get("expect_value")
@property
def assert_function_mapping(self):
"""
断言方法匹配, 获取utils\assertion_utils\assert_function.py中的方法并返回
"""
# 从 module中获取方法的名称和所在的内存地址 """
module_functions = {}
for name, item in vars(assert_function).items():
if isinstance(item, types.FunctionType):
module_functions[name] = item
return module_functions
def assert_handle(self):
"""
断言处理
"""
if "sql" in self.assert_data.keys():
actual_value = self.get_actual_value_by_sql()
else:
actual_value = self.get_actual_value_by_response()
expect_value = self.get_expect_value
message = str(self.get_message)
assert_type = self.get_assert_type
logger.debug(f"\nmessage: {message}\n"
f"assert_type: {assert_type}\n"
f"expect_value: {expect_value}\n"
f"actual_value: {actual_value}\n")
message = message or f"断言 --> 预期结果：{type(expect_value)} || {expect_value} 实际结果：{type(actual_value)} || {actual_value}"
with allure.step(message):
# 调用utils.assertion_utils.assert_type里面的方法
self.assert_function_mapping[assert_type](expect_value=expect_value, actual_value=actual_value,
message=message)
class AssertHandle(AssertUtils):
def get_assert_data_list(self):
"""
获取所有的断言数据，并以列表的形式返回
"""
assert_list = []
if self.assert_data and isinstance(self.assert_data, dict):
for k, v in self.assert_data.items():
if k.lower() == "status_code":
with allure.step("断言 --> 响应状态码"):
assert_function.equals(expect_value=v, actual_value=self.response.status_code)
else:
assert_list.append(v)
else:
logger.debug(f"断言数据为空或者不是字典格式，跳过断言！\n"
f"断言数据：{self.assert_data}")
return assert_list
def assert_handle(self):
"""
将收集到的断言数据逐一进行断言
"""
for value in self.get_assert_data_list():
self.assert_data = value
super().assert_handle()
utils/assertion_utils/assert_function.py
+161
View File
@@ -0,0 +1,161 @@
# -*- coding: utf-8 -*-
# @Version: Python 3.9
# @Time : 2023/12/12 14:31
# @Author : floraachy
# @File : assert_function.py
# @Software: PyCharm
# @Desc: 断言方法
# 标准库导入
from typing import Any, Union, Text
# 第三方库导入
import allure
@allure.step("预期结果：{expect_value} == 实际结果：{actual_value}")
def equals(expect_value: Any, actual_value: Any, message: Text = ""):
"""
判断是否相等
"""
assert expect_value == actual_value, message
@allure.step("预期结果：{expect_value} < 实际结果：{actual_value}")
def less_than(expect_value: Union[int, float], actual_value: Union[int, float], message: Text = ""):
"""
判断预期结果小于实际结果
"""
assert expect_value < actual_value, message
@allure.step("预期结果：{expect_value} <= 实际结果：{actual_value}")
def less_than_or_equals(expect_value: Union[int, float], actual_value: Union[int, float], message: Text = ""):
"""
判断预期结果小于等于实际结果
"""
assert expect_value <= actual_value, message
@allure.step("预期结果：{expect_value} > 实际结果：{actual_value}")
def greater_than(expect_value: Union[int, float], actual_value: Union[int, float], message: Text = ""):
"""
判断预期结果大于实际结果
"""
assert expect_value > actual_value, message
@allure.step("预期结果：{expect_value} >= 实际结果：{actual_value}")
def greater_than_or_equals(expect_value: Union[int, float], actual_value: Union[int, float], message: Text = ""):
"""
判断预期结果大于等于实际结果
"""
assert expect_value >= actual_value, message
@allure.step("预期结果：{expect_value} != 实际结果：{actual_value}")
def not_equals(expect_value: Any, actual_value: Any, message: Text = ""):
"""
判断预期结果不等于实际结果
"""
assert expect_value != actual_value, message
@allure.step("预期结果：{expect_value} == 实际结果：{actual_value}")
def string_equals(expect_value: Any, actual_value: Text, message: Text = ""):
"""
判断字符串是否相等
"""
assert expect_value == actual_value, message
@allure.step("长度相等 --> 预期结果：{expect_value} == 实际结果：{actual_value}")
def length_equals(expect_value: int, actual_value: Text, message: Text = ""):
"""
判断长度是否相等
"""
assert isinstance(
expect_value, int
), "expect_value 需要为 int 类型"
assert expect_value == len(actual_value), message
@allure.step("长度大于 --> 预期结果：{expect_value} > 实际结果：{actual_value}")
def length_greater_than(expect_value: Union[int, float], actual_value: Text, message: Text = ""):
"""
判断长度大于
"""
assert isinstance(
expect_value, (float, int)
), "expect_value 需要为 float/int 类型"
assert len(str(actual_value)) > expect_value, message
@allure.step("长度大于等于 --> 预期结果：{expect_value} >= 实际结果：{actual_value}")
def length_greater_than_or_equals(expect_value: Union[int, float], actual_value: Text, message: Text = ""):
"""
判断长度大于等于
"""
assert isinstance(
expect_value, (int, float)
), "expect_value 需要为 float/int 类型"
assert expect_value >= len(actual_value), message
@allure.step("长度小于 --> 预期结果：{expect_value} < 实际结果：{actual_value}")
def length_less_than(expect_value: Union[int, float], actual_value: Text, message: Text = ""):
"""
判断长度小于
"""
assert isinstance(
expect_value, (int, float)
), "expect_value 需要为 float/int 类型"
assert expect_value < len(actual_value), message
@allure.step("长度小于等于 --> 预期结果：{expect_value} <= 实际结果：{actual_value}")
def length_less_than_or_equals(expect_value: Union[int, float], actual_value: Text, message: Text = ""):
"""判断长度小于等于"""
assert isinstance(
expect_value, (int, float)
), "expect_value 需要为 float/int 类型"
assert expect_value <= len(actual_value), message
@allure.step("预期结果：{expect_value} in 实际结果：{actual_value}")
def contains(expect_value: Any, actual_value: Any, message: Text = ""):
"""
判断预期结果内容包含在实际结果中
"""
assert isinstance(
actual_value, (list, tuple, dict, str, bytes)
), "actual_value 需要为 list/tuple/dict/str/bytes 类型"
assert expect_value in actual_value, message
@allure.step("实际结果：{expect_value} in 预期结果：{actual_value}")
def contained_by(expect_value: Any, actual_value: Any, message: Text = ""):
"""
判断实际结果包含在预期结果中
"""
assert isinstance(
actual_value, (list, tuple, dict, str, bytes)
), "actual_value 需要为 list/tuple/dict/str/bytes 类型"
assert actual_value in expect_value, message
@allure.step("实际结果：{actual_value} 是以 预期结果： {expect_value} 开头的")
def startswith(expect_value: Any, actual_value: Any, message: Text = ""):
"""
检查实际结果的开头是否和预期结果内容的开头相等
"""
assert str(actual_value).startswith(str(expect_value)), message
@allure.step("实际结果：{actual_value} 是以 预期结果： {expect_value} 结尾的")
def endswith(expect_value: Any, actual_value: Any, message: Text = ""):
"""
检查实际结果的结尾是否和预期结果内容相等
"""
assert str(actual_value).endswith(str(expect_value)), message
utils/case_data_utils/__init__.py → utils/case_generate_utils/__init__.py
View File
utils/case_data_utils/case_data_analysis.py → utils/case_generate_utils/case_data_analysis.py
+1 -1
View File
@@ -8,7 +8,7 @@
# 标准库导入
from typing import Text
# 第三方库导入
from utils.case_data_utils.models import TestCaseEnum, Method, RequestType, Severity
from utils.models import TestCaseEnum, Method, RequestType, Severity
class CaseDataCheck:
utils/case_data_utils/case_fun_generate.py → utils/case_generate_utils/case_fun_generate.py
+2 -2
View File
@@ -16,11 +16,11 @@ from loguru import logger
from utils.files_utils.excel_handle import ExcelHandle
from utils.files_utils.yaml_handle import YamlHandle
from utils.files_utils.files_handle import get_files, get_relative_path
from utils.case_data_utils.models import CaseFileType
from utils.models import CaseFileType
from config.settings import CASE_FILE_TYPE
from config.path_config import DATA_DIR, CASE_TEMPLATE_DIR, AUTO_CASE_DIR
from config.global_vars import CUSTOM_MARKERS
from utils.case_data_utils.case_data_analysis import CaseDataCheck
from utils.case_generate_utils.case_data_analysis import CaseDataCheck
"""
主要步骤：
utils/case_data_utils/case_template.txt → utils/case_generate_utils/case_template.txt
+3 -5
View File
@@ -5,12 +5,10 @@
# 第三方库导入
import pytest
import allure
from loguru import logger
# 本地应用/模块导入
from config.global_vars import GLOBAL_VARS
from utils.requests_utils.assert_handle import assert_response, assert_sql
from utils.requests_utils.request_control import RequestPreDataHandle, RequestHandle, after_request_extract
from utils.report_utils.allure_handle import allure_title
from utils.assertion_utils.assert_control import AssertHandle
# 用例数据
@@ -30,9 +28,9 @@ class ${class_title}Auto:
# 发送请求
response = RequestHandle(case_data=case_data, global_var=GLOBAL_VARS).http_request()
# 进行响应断言
assert_response(response, case_data["assert_response"])
AssertHandle(assert_data=case_data["assert_response"], response=response).assert_handle()
# 进行数据库断言
assert_sql(db_info=GLOBAL_VARS["db_info"], expected=case_data["assert_sql"])
AssertHandle(assert_data=case_data["assert_sql"], db_info=GLOBAL_VARS["db_info"]).assert_handle()
# 断言成功后进行参数提取
res = after_request_extract(response, case_data.get("extract", None))
for k, v in res.items():
utils/data_utils/extract_data_handle.py
+21 -5
View File
@@ -11,9 +11,11 @@ import re
from jsonpath import jsonpath
from loguru import logger
from requests import Response
# 本地应用/模块导入
from utils.data_utils.data_handle import data_handle
def json_extractor(obj: dict, expr: str = '.'):
def json_extractor(obj, expr: str = '.'):
"""
从目标对象obj, 根据表达式expr提取指定的值
:param obj :json/dict类型数据
@@ -22,12 +24,14 @@ def json_extractor(obj: dict, expr: str = '.'):
"""
try:
result = jsonpath(obj, expr)[0] if len(jsonpath(obj, expr)) == 1 else jsonpath(obj, expr)
logger.trace(f"\n提取表达式： {expr} \n"
logger.debug(f"\n提取对象：{obj}\n"
f"提取表达式： {expr} \n"
f"提取值类型： {type(result)}\n"
f"提取值：{result}\n")
return result
except Exception as e:
logger.trace(f"\n提取表达式： {expr}\n"
logger.debug(f"\n提取对象：{obj}\n"
f"提取表达式： {expr}\n"
f"提取对象： {obj}\n"
f"错误信息：{e}\n")
@@ -42,12 +46,16 @@ def re_extract(obj: str, expr: str = '.'):
try:
# 如果提取后的数据长度为1，则取第一个元素（返回str），否则返回列表
result = re.findall(expr, obj)[0] if len(re.findall(expr, obj)) == 1 else re.findall(expr, obj)
logger.debug(f"\n提取表达式： {expr}\n"
# 由于提取出来的数据都是str格式，将eval一样，还原数据格式
result = data_handle(obj=result)
logger.debug(f"\n提取对象：{obj}\n"
f"提取表达式： {expr}\n"
f"提取值类型： {type(result)}\n"
f"提取值：{result}\n")
return result
except Exception as e:
logger.debug(f"\n提取表达式： {expr}\n"
logger.debug(f"\n提取对象：{obj}\n"
f"提取表达式： {expr}\n"
f"提取对象： {obj}\n"
f"错误信息：{e}\n")
@@ -69,3 +77,11 @@ def response_extract(response: Response, expr: str = '.'):
logger.debug(f"\n提取表达式： {expr}\n"
f"提取对象： {response}\n"
f"错误信息：{e}\n")
if __name__ == '__main__':
obj = [{'id': 1, 'user_id': 102, 'action': 'autologin', 'value': '3734462a398eedd9ab7448c4e2880ddd3f9bb2cb'}]
expre = "'user_id': (.*?),"
res = re_extract(obj=str(obj), expr=expre)
print(res)
utils/database_utils/mysql_handle.py
+2 -2
View File
@@ -92,7 +92,7 @@ class MysqlServer:
f"SQL: {sql}\n" \
f"result: {data}\n" \
"=====================================================")
return self.verify(data)
return data
except Exception as e:
logger.error(f"{sql} --> 报错: {e}")
raise e
@@ -112,7 +112,7 @@ class MysqlServer:
f"SQL: {sql}\n" \
f"result: {data}\n" \
"=====================================================")
return self.verify(data)
return data
except Exception as e:
logger.error(f"{sql} --> 报错: {e}")
raise e
utils/case_data_utils/models.py → utils/models.py
+24
View File
@@ -135,3 +135,27 @@ class RequestType(Enum):
FILE = 'FILE'
EXPORT = "EXPORT"
NONE = "NONE"
@unique
class AssertMethod(Enum):
"""
断言类型
注意：这里的类型与assert_type.py中的方法名相同，不要随意改动
"""
equals = "=="
less_than = "lt"
less_than_or_equals = "le"
greater_than = "gt"
greater_than_or_equals = "ge"
not_equals = "not_eq"
string_equals = "str_eq"
length_equals = "len_eq"
length_greater_than = "len_gt"
length_greater_than_or_equals = 'len_ge'
length_less_than = "len_lt"
length_less_than_or_equals = 'len_le'
contains = "contains"
contained_by = 'contained_by'
startswith = 'startswith'
endswith = 'endswith'
utils/report_utils/allure_handle.py
+1 -1
View File
@@ -11,7 +11,7 @@ import json
# 第三方库导入
import allure
# 本地应用/模块导入
from utils.case_data_utils.models import AllureAttachmentType
from utils.models import AllureAttachmentType
from utils.report_utils.platform_handle import PlatformHandle
from utils.files_utils.files_handle import zip_file, copy_file
utils/report_utils/send_result_handle.py
+1 -1
View File
@@ -8,7 +8,7 @@
# 第三方模块
from loguru import logger
# 本地应用/模块导入
from utils.case_data_utils.models import NotificationType
from utils.models import NotificationType
from config.settings import SEND_RESULT_TYPE, email, ding_talk, wechat, email_subject, email_content, ding_talk_title, \
ding_talk_content, wechat_content
from utils.data_utils.data_handle import data_handle
utils/requests_utils/api_workflow.py
+57 -25
View File
@@ -7,43 +7,75 @@
# 标准库导入
import os
# 第三方库导入
from loguru import logger
# 本地应用/模块导入
from utils.files_utils.yaml_handle import YamlHandle
from utils.requests_utils.request_control import RequestPreDataHandle, RequestHandle, after_request_extract
from utils.requests_utils.assert_handle import assert_response
from utils.files_utils.files_handle import get_files
from utils.data_utils.eval_data_handle import eval_data
from utils.assertion_utils.assert_control import AssertHandle
from config.global_vars import GLOBAL_VARS
class ApiWorkFlow:
def __init__(self, api_file_path: str):
"""
:param:api_file_path 接口yaml文件路径
:param:key 对应接口的id
:param:global_var 全局变量，保存接口相关变量的实际值， 例如接口中的${login}，会从GLOBAL_VAR中找到key=login的元素进行替换
"""
self.api_file_path = api_file_path
def get_api_mode(self, key):
def get_api_data(api_file_path: str, key: str = None):
"""
根据指定的yaml文件路径，以及key值，获取对应的接口
:param:api_file_path 接口yaml文件路径
:param:key 对应接口的id
"""
api_modes = YamlHandle(filename=self.api_file_path).read_yaml
matching_api = next((item for item in api_modes["case_info"] if item["id"] == key), None)
if matching_api:
return matching_api
else:
logger.error(f"根据{key}, 未能找到对应的接口")
raise KeyError(f"根据{key}, 未能找到对应的接口")
api_data = []
if os.path.isdir(api_file_path):
logger.debug(f"目标路径是一个目录：{api_file_path}")
api_files = get_files(target=api_file_path, end=".yaml")
for api_file in api_files:
api_data.append(YamlHandle(filename=api_file).read_yaml)
def api_work_flow(self, key: str, global_var: dict = {}) -> dict:
api_mode = self.get_api_mode(key)
api_info = RequestPreDataHandle(request_data=api_mode, global_var=global_var).request_data_handle()
elif os.path.isfile(api_file_path):
logger.debug(f"目标路径是一个文件：{api_file_path}")
api_data.append(YamlHandle(filename=api_file_path).read_yaml)
else:
logger.error(f"目标路径错误，请检查！api_file_path={api_file_path}")
return None
for api in api_data:
matching_api = next((item for item in api["case_info"] if item["id"] == key), None)
if matching_api:
logger.info("\n----------匹配到的api----------\n"
f"类型：{type(matching_api)}"
f"值：{matching_api}\n")
return matching_api
# 在找不到匹配的情况下，返回一个默认值且记录一条错误日志
logger.warning(f"未找到id为{key}的接口， 返回值是None")
return None # 或者根据需求返回其他默认值
def api_work_flow(req_data: dict, source: dict) -> dict:
"""
请求过程：请求前用例数据处理，发送请求，断言，参数提取
:param:req_data 接口请求数据
:param:source 全局变量，保存接口相关变量的实际值， 例如接口中的${login}，会从source中找到key=login的元素进行替换
"""
logger.debug(f"\n----------------api_work_flow-----------------\n"
f"接口请求数据：{req_data}\n"
f"全局变量：{source}\n")
if req_data:
extract_result = {}
api_data = RequestPreDataHandle(request_data=req_data, global_var=source).request_data_handle()
# 发送请求
response = RequestHandle(api_info, global_var).http_request()
response = RequestHandle(case_data=api_data, global_var=source).http_request()
# 进行响应断言
assert_response(response, api_info["assert_response"])
AssertHandle(assert_data=api_data["assert_response"], response=response).assert_handle()
# 断言成功后进行参数提取
res = after_request_extract(response, api_info["extract"])
return res
res = after_request_extract(response, api_data["extract"])
for k, v in res.items():
extract_result[k] = eval_data(v)
GLOBAL_VARS.update(extract_result)
return extract_result
else:
logger.error(f"接口请求数据不能为空！\n"
f"req_data = {req_data}")
raise f"接口请求数据不能为空！\nreq_data = {req_data}"
utils/requests_utils/assert_handle.py
-165
View File
@@ -1,165 +0,0 @@
# -*- coding: utf-8 -*-
# @Version: Python 3.9
# @Time : 2023/2/1 14:31
# @Author : chenyinhua
# @File : assert_handle.py
# @Software: PyCharm
# @Desc: 断言
# 第三方库导入
from loguru import logger
from requests import Response
import allure
# 本地应用/模块导入
from utils.data_utils.extract_data_handle import json_extractor, re_extract
from utils.report_utils.allure_handle import allure_step
from utils.database_utils.mysql_handle import MysqlServer
@allure.step("响应断言 --> 响应数据：{response} - 预期结果：{expected}")
def assert_response(response: Response, expected: dict) -> None:
""" 断言方法
:param response: 实际响应对象
:param expected: 预期响应内容，从excel中或者yaml读取、或者手动传入,格式如下：
{
'eq':
{'http_code': 200, '$.user_id': '85392'},
"in":
{'message': "success"}
}
return None
"""
logger.info("\n======================================================\n" \
f"-------------Start：响应断言--------------------\n"
f"响应断言预期结果：{expected}, {type(expected)}")
if expected is None:
return
index = 0
for k, v in expected.items():
# 获取需要断言的实际结果部分
for _k, _v in v.items():
if _k == "http_code":
actual = response.status_code
else:
if response_type(response) == "json":
# 如果响应数据是json格式
actual = json_extractor(response.json(), _k)
else:
# 响应数据不是json格式
actual = re_extract(response.text, _k)
index += 1
logger.info(
f'第{index}个响应断言 -|- 预期结果: {_k}: {_v}, {type(_v)} {k} 实际结果: {actual}, {type(actual)}')
allure_step(
step_title=f'第{index}个响应断言数据---->预期结果: {_k}: {_v}, {type(_v)} {k} 实际结果: {actual}, {type(actual)}')
try:
if k == "eq": # 预期结果 = 实际结果
assert _v == actual
logger.success(f"预期结果: {_k}: {_v} == 实际结果: {actual}, 断言成功！")
elif k == "in": # 实际结果 包含 预期结果
assert _v in actual
logger.success(f"预期结果: {_k}: {_v} in 实际结果: {actual}, 断言成功！")
elif k == "gt": # 预期结果 > 实际结果 (值应该为数值型)
assert _v > actual
logger.success(f"预期结果: {_k}: {_v} > 实际结果: {actual}, 断言成功！")
elif k == "lt": # 预期结果 < 实际结果 (值应该为数值型)
assert _v < actual
logger.success(f"预期结果: {_k}: {_v} < 实际结果: {actual}, 断言成功！")
elif k == "not": # 预期结果 != 实际结果
assert _v != actual
logger.success(f"预期结果: {_k}: {_v} != 实际结果: {actual}, 断言成功！")
else:
logger.error(f"判断关键字: {k} 错误！, 目前仅支持如下关键字：eq, in, gt, lt, not")
allure_step(step_title=f'判断关键字: {k} 错误！',
content='目前仅支持如下关键字：eq, in, gt, lt, not')
except AssertionError:
logger.error(
f"第{index}个响应断言失败 -|- 预期结果: {_k}: {_v}, {type(_v)} {k} 实际结果: {actual}, {type(actual)}")
allure_step(
step_title=f'第{index}个响应断言失败---->预期结果: {_k}: {_v}, {type(_v)} {k} 实际结果: {actual}, {type(actual)}')
logger.info('\n-------------End：响应断言--------------------\n' \
"=====================================================")
raise AssertionError(
f"第{index}个响应断言失败 -|- 预期结果: {_k}: {_v}, {type(_v)} {k} 实际结果: {actual}, {type(actual)}")
logger.info('\n-------------End：响应断言--------------------\n' \
"=====================================================")
@allure.step("数据库断言")
def assert_sql(db_info, expected: dict):
"""
数据库断言
:param db_info: 数据库配置信息
:param expected: 预期结果，从excel中或者yaml读取、或者手动传入,格式如下：
{
'eq':
{'sql': 'select count(*) from users where user_id=1;', 'len': '1'},
'eq':
{'sql': 'select * from users where user_id=1;', '$.username': '${username}'},
}
"""
logger.info("\n======================================================\n" \
f"-------------Start：数据库断言--------------------")
if expected is None:
logger.info("判断是否存在数据库断言---->当前用例无数据库断言")
allure_step(step_title='判断是否存在数据库断言---->当前用例无数据库断言')
return
if not db_info:
logger.error("判断是否存在数据库配置---->当前环境无数据库配置，跳过数据库断言！")
allure_step(step_title='判断是否存在数据库配置---->当前环境无数据库配置，跳过数据库断言！')
logger.info('\n-------------End：数据库断言--------------------\n'
"=====================================================")
return
for k, v in expected.items():
sql_result = None
for _k, _v in v.items():
if _k == "sql":
try:
# 查询数据库，获取查询结果
sql_result = MysqlServer(**db_info).query_one(_v)
logger.info(f'数据库响应断言 -|- SQL：{_v} || 查询结果：{sql_result}')
allure_step(step_title=f'数据库断言---->SQL：{_v} ',
content=f'查询结果：{sql_result}')
except Exception as e:
logger.error(f'数据库服务报错：{e}')
allure_step(step_title=f'数据库服务报错',
content=f'{e} ')
logger.info('\n-------------End：数据库断言--------------------\n'
"=====================================================")
raise AssertionError(f"数据库服务报错：{e}")
try:
if k == "eq": # 预期结果 = 实际结果
if _k == "len":
assert _v == len(sql_result)
logger.success(f"预期结果: {_v} == 实际结果: {len(sql_result)}, 断言成功！")
allure_step(
step_title=f'数据库断言结果---->预期结果: {_v} == 实际结果: {len(sql_result)}, 断言成功！')
# 如果时$.开头，则从数据库查询结果中提取相应的值作为实际结果
elif _k.startswith("$."):
actual = json_extractor(sql_result, _k)
assert _v == actual
logger.success(f"预期结果: {_v} == 实际结果: {actual}, 断言成功！")
allure_step(step_title=f'数据库断言结果---->预期结果: {_v} == 实际结果: {actual}, 断言成功！')
except AssertionError:
logger.error(f"数据库断言失败 -|- 预期结果:{_v} {k} 实际结果: {sql_result})")
allure_step(step_title=f'数据库断言失败---->预期结果:{_v} {k} 实际结果: {sql_result}')
logger.info('\n-------------End：数据库断言--------------------\n' \
"=====================================================")
raise AssertionError(f"数据库断言失败 -|-预期结果: {_v} {k} 实际结果: {sql_result}")
logger.info('\n-------------End：数据库断言--------------------\n' \
"=====================================================")
def response_type(response: Response) -> str:
"""
:param response: requests 返回
:return: 返回响应数据类型 json或者str
"""
try:
response.json()
return "json"
except:
return "str"
utils/requests_utils/request_control.py
+3 -3
View File
@@ -279,10 +279,10 @@ def after_request_extract(response: Response, extract):
re_result = {}
response_result = {}
if extract:
if extract.get("type_json"):
if extract.get("type_jsonpath"):
# 如果响应数据是json格式，则将按照json方式对后置提取参数进行处理
res = response.json()
for k, v in extract["type_json"].items():
for k, v in extract["type_jsonpath"].items():
json_result[k] = json_extractor(res, v)
logger.debug("\n-------------从response.json()中通过jsonpath方式提取到的结果--------------------\n"
f"后置提取参数（新）: {json_result}\n")
@@ -290,7 +290,7 @@ def after_request_extract(response: Response, extract):
# 如果响应数据是str格式，则将按照str方式对后置提取参数进行处理
res = response.text
for k, v in extract["type_re"].items():
re_result[k] = re_extract(res, v)
re_result[k] = data_handle(obj=re_extract(res, v))
logger.debug("\n-------------从response.text中通过正则表达式提取到的结果--------------------\n"
f"后置提取参数（新）: {re_result}\n")