西去商城接口自动化项目架构介绍

haoqi416

242人浏览 · 2026-09-16 16:33:43

haoqi416 · 2026-09-16 16:33:43 发布

一.项目介绍

基于Python+pytest+sqlalchemy+requests+allure+jsonpath+yaml+Jenkins+Linux

该项目为在线购物的商城网站，包括用户注册、登录、下单、上架/下架商品、下单支付等相关功能，核心功能如下：

采用 pytest 单元测试框架管理用例，pytest.ini 结合 conftest.py 配置管理用例执行规则
采用 json 提取器以及 yaml 的方式存放接口配置参数，实现 case 与参数的分离，用例解耦
采用 POM 模式分离用例层与驱动层，实现用户管理、商品下单全流程用例，兼容性更佳
内置动态参数引擎 ${函数名(参数)}，支持时间戳、加密、随机值、接口关联参数自动替换
使用 assert 关键字断言接口返回结果、状态码以及数据库断言，保证用例可靠稳定
支持接口关联：extract/extract_list 提取上游返回值（jsonpath + 正则），写入 extract.yaml 供下游接口使用
集成 Allure 可视化报告，支持钉钉机器人、邮件、Jenkins 多渠道通知

二.项目结构说明

pytest-ui/
├── run.py                          # 入口：按 REPORT_TYPE 调 pytest + Allure/tm 报告
├── conftest.py                     # 会话级：清空 extract、清 Allure 临时文件、汇总结果、可选钉钉
├── pytest.ini                      # 用例发现规则（python_files=test_*.py 等）
├── extract.yaml                    # 运行时变量池（token/Cookie/goodsId/orderNumber…）
├── environment.xml                 # 拷到 Allure 的环境信息
├── requirements.txt
├── conf/
│   ├── setting.py                  # 路径、超时、日志级别、报告类型、钉钉开关
│   ├── operationConfig.py          # 读 config.ini
│   └── config.ini                  # host / MySQL / Redis / ClickHouse / MongoDB / EMAIL / SSH
├── base/
│   ├── apiutil.py                  # 单接口引擎 RequestBase
│   ├── apiutil_business.py         # 业务场景引擎（一份 YAML 多接口顺序执行）
│   ├── generateId.py               # Allure 模块/用例编号生成器
│   ├── removefile.py               # 清理 report/temp
│   ├── new_testcase_tools.py       # PyQt5「测试用例生成工具 V2.0」
│   └── new_tools.ui
├── common/                         # 公共能力
│   ├── sendrequest.py              # requests 封装
│   ├── readyaml.py                 # YAML 加载 / extract 读写
│   ├── debugtalk.py                # ${func()} 动态函数（提取、加密、时间戳、CSV）
│   ├── assertions.py               # contains/eq/ne/rv/db 断言
│   ├── connection.py               # MySQL/Redis/ClickHouse/Mongo/SSH（Oracle 为空类）
│   ├── recordlog.py                # 滚动日志
│   ├── dingRobot.py / semail.py / Pjenkins.py
│   ├── handleExcel.py / operationcsv.py / operxml.py / two_dimension_data.py
├── testcase/
│   ├── conftest.py                 # 登录 fixture、日志前后置、预留 DB 清理
│   ├── Single interface/           # 用户增删改查
│   ├── ProductManager/             # 商品列表/详情/下单/支付
│   └── Business interface/         # 列表→详情→下单→支付→校验订单
├── data/
│   ├── loginName.yaml              # session 登录用例
│   ├── login_data.csv / vehicleNo.csv
│   └── sql/*.xml                   # ClickHouse/统计类 SQL 片段
├── logs/  report/  venv/

三.核心代码

1. 程序入口 run.py

import shutil
import pytest
import os
import webbrowser
from conf.setting import REPORT_TYPE
if name == 'main':
if REPORT_TYPE == 'allure':
pytest.main(
['-s', '-v', '--alluredir=./report/temp', './testcase', '--clean-alluredir',
'--junitxml=./report/results.xml'])
shutil.copy('./environment.xml', './report/temp')
os.system(f'allure serve ./report/temp')
elif REPORT_TYPE == 'tm':
pytest.main(['-vs', '--pytest-tmreport-name=testReport.html', '--pytest-tmreport-path=./report/tmreport'])
webbrowser.open_new_tab(os.getcwd() + '/report/tmreport/testReport.html')

代码说明：

主程序入口

if __name__ == '__main__':

Python 标准写法，表示当脚本作为主程序运行时才执行以下代码块
防止模块被其他脚本导入时意外执行测试逻辑

Allure 报告处理逻辑

pytest.main(['-s', '-v', '--alluredir=./report/temp', './testcase', '--clean-alluredir',
             '--junitxml=./report/results.xml'])

参数说明

参数	含义
`-s`	输出所有打印信息（不屏蔽 stdout）
`-v`	显示详细测试结果
`--alluredir=./report/temp`	将 Allure 报告原始数据保存到指定目录
`./testcase`	指定测试用例所在目录
`--clean-alluredir`	每次运行前清空之前的报告数据，避免脏数据
`--junitxml=./report/results.xml`	生成 JUnit XML 格式结果文件，供 Jenkins 等 CI 工具解析

环境信息复制

shutil.copy('./environment.xml', './report/temp')

复制环境信息文件到报告目录，Allure 总览页展示运行环境（浏览器、Python 版本、BaseUrl 等）

启动报告服务

os.system(f'allure serve ./report/temp')

使用 Allure CLI 启动本地服务，并自动在默认浏览器中打开报告页面

2. 全局方法 conftest.py

# -*- coding: utf-8 -*-
import time
import pytest
from common.readyaml import ReadYamlData
from base.removefile import remove_file
from common.dingRobot import send_dd_msg
from conf.setting import dd_msg
import warnings
yfd = ReadYamlData()
定义全局变量，保存测试会话开始时间
session_start_time = 0
@pytest.fixture(scope="session", autouse=True)
def clear_extract():
# 禁用HTTPS告警，ResourceWarning
warnings.simplefilter('ignore', ResourceWarning)
yfd.clear_yaml_data()
remove_file("./report/temp", ['json', 'txt', 'attach', 'properties'])
pytest初始化钩子，测试开始时记录时间
def pytest_configure(config):
global session_start_time
session_start_time = time.time()
def generate_test_summary(terminalreporter):
"""生成测试结果摘要字符串"""
total = terminalreporter._numcollected
passed = len(terminalreporter.stats.get('passed', []))
failed = len(terminalreporter.stats.get('failed', []))
error = len(terminalreporter.stats.get('error', []))
skipped = len(terminalreporter.stats.get('skipped', []))
# 从全局变量拿开始时间，不再读取pytest内部私有属性
duration = time.time() - session_start_time
summary = f"""
自动化测试结果，通知如下，请着重关注测试失败的接口，具体执行结果如下：
测试用例总数：{total}
测试通过数：{passed}
测试失败数：{failed}
错误数量：{error}
跳过执行数量：{skipped}
执行总时长：{duration:.2f}秒
"""
print(summary)
return summary
def pytest_terminal_summary(terminalreporter, exitstatus, config):
"""自动收集pytest框架执行的测试结果并打印摘要信息"""
summary = generate_test_summary(terminalreporter)
if dd_msg:
send_dd_msg(summary)

方法说明

方法	作用
`clear_extract` fixture	session 级自动执行：清空 extract.yaml（防关联数据残留）+ 清理 report/temp 旧数据
`pytest_configure`	pytest 启动瞬间记录时间戳，用于计算执行总时长
`generate_test_summary`	统计通过 / 失败 / 错误 / 跳过数量，生成摘要字符串
`pytest_terminal_summary`	测试结束自动回调，打印摘要并按开关推送钉钉

执行逻辑详解

会话清理：scope="session" 保证整个测试过程只执行一次，autouse=True 自动执行无需调用，先清空 extract.yaml 防止上一次运行的接口关联数据污染本次用例
计时器：pytest_configure 是 pytest 内置初始化钩子，在收集用例前执行，用全局变量记录开始时间（不读取 pytest 内部私有属性如_sessionstarttime，规避版本升级兼容性问题）
结果统计：从terminalreporter.stats中按passed/failed/error/skipped分类统计用例数量
结果推送：pytest_terminal_summary 是测试结束钩子，将摘要打印到控制台，并判断 dd_msg 开关，开启则调用钉钉机器人推送消息

3.测试用例 testcase

单接口模式代码如下

import allure
import pytest
from base.generateId import m_id, c_id
from base.apiutil import RequestBase
from common.readyaml import get_testcase_yaml
@allure.feature(next(m_id) + '商品管理（单接口）')
class TestLogin:
@allure.story(next(c_id) + "获取商品列表")
@pytest.mark.run(order=1)
@pytest.mark.parametrize('base_info,testcase', get_testcase_yaml('./testcase/ProductManager/getProductList.yaml'))
def test_get_product_list(self, base_info, testcase):
    allure.dynamic.title(testcase['case_name'])
    RequestBase().specification_yaml(base_info, testcase)
@allure.story(next(c_id) + "提交订单")
@pytest.mark.run(order=3)
@pytest.mark.parametrize('base_info,testcase', get_testcase_yaml('./testcase/ProductManager/commitOrder.yaml'))
def test_commit_order(self, base_info, testcase):
allure.dynamic.title(testcase['case_name'])
RequestBase().specification_yaml(base_info, testcase)

业务场景模式代码如下

import allure
import pytest
from common.readyaml import get_testcase_yaml
from base.apiutil_business import RequestBase
from base.generateId import m_id, c_id
注意：业务场景的接口测试要调用base目录下的apiutil_business文件
@allure.feature(next(m_id) + '电子商务管理系统（业务场景）')
class TestEBusinessScenario:
@allure.story(next(c_id) + '商品列表到下单支付流程')
@pytest.mark.parametrize('case_info', get_testcase_yaml('./testcase/Business interface/BusinessScenario.yml'))
def test_business_scenario(self, case_info):
    allure.dynamic.title(case_info['baseInfo']['api_name'])
    RequestBase().specification_yaml(case_info)

代码说明

方法	作用
`@allure.feature`	功能模块名（用 generateId 生成的`M01_`编号前缀，保证 Allure 报告按执行顺序展示）
`@allure.story`	子功能场景名
`@pytest.mark.run(order=n)`	控制用例执行顺序，保证依赖接口先执行
`@pytest.mark.parametrize`	参数化：一组 YAML 数据触发一次完整执行
`allure.dynamic.title`	动态设置报告中的用例标题
`get_testcase_yaml`	解析 YAML 文件，返回 `[baseInfo, testCase]` 列表

执行逻辑详解

编号生成：generateId.py 通过生成器 yield 依次产出 M01_、M02_... 模块编号和 C01_、C02_... 用例编号，作为 feature/story 前缀，保证 Allure 报告展示顺序与 pytest 执行顺序一致
参数化：get_testcase_yaml 读取 YAML 文件，每个 testCase 与 baseInfo 组合成一个参数组，一组数据触发一次完整测试执行
动态标题：allure.dynamic.title 把 YAML 中的 case_name / api_name 设置为报告中的用例标题，报告可读性更强
执行引擎：单接口模式调用 apiutil.py 的 RequestBase().specification_yaml(base_info, testcase)；业务场景模式（多个接口串联成一个流程）调用 apiutil_business.py 的同名方法，内部循环处理 testCase 列表

4. YAML 用例模板

- baseInfo:
    api_name: 提交订单
    url: /coupApply/cms/placeAnOrder
    method: post
    header:
      Content-Type: application/json;charset=UTF-8
  testCase:
    - case_name: 详情页面选择规格，提交订单
      json:
        goods_id: ${get_extract_data(goodsIds,1)}
        number: 2
        propertyChildIds: "2:9"
        inviter_id: 127839112
        price: "128"
        consignee_info:
          { "name": "张三","phone": 13800000000,"address": "北京市海淀区西三环北路74号院4栋3单元1008" }
      validation:
        - eq: { 'message': '提交订单成功' }
      extract:
        orderNumber: $.orderNumber
        userId: $.userId
- baseInfo:
    api_name: 订单支付
    url: /coupApply/cms/orderPay
    method: post
    header:
      Content-Type: application/json;charset=UTF-8
  testCase:
    - case_name: 订单支付
      json:
        orderNumber: ${get_extract_data(orderNumber)}
        userId: ${get_extract_data(userId)}
        timeStamp: ${timestamp()}
      validation:
        - eq: { 'message': '订单支付成功' }

关键字段说明

字段	说明
`baseInfo.api_name`	接口名称，会显示在 Allure 报告中
`baseInfo.url`	接口路径（host 在 config.ini 的`[api_envi]`配置，运行时拼接）
`baseInfo.method`	请求方法（get/post）
`baseInfo.header`	请求头，支持动态参数
`testCase.case_name`	测试用例名称
`params / data / json`	请求参数类型三选一：GET→params，表单→data，JSON→json
`validation`	断言规则，支持 contains/eq/ne/rv/db 五种模式
`extract`	提取单个参数（jsonpath：`$.data.token` 或正则：`"token":"(.*?)"`）
`extract_list`	提取多个参数，以列表形式返回（如商品 ID 列表）
`${函数名(参数)}`	动态参数表达式，运行时反射调用 debugtalk.py 中的方法

完整断言示例

validation:
  - contains: {status_code: 200}     # 字符串包含断言模式，有多种断言时，contains必须在前面
  - contains: {'message':'success'}  # 字符串包含断言模式
  - contains: {'message':None}       # 字符串为空断言
  - eq: {'state': '已入网'}          # 相等断言模式，这个要在contains的下面
  - ne: {'state': '已入网'}          # 不相等断言模式
  - rv: {"data":2}                   # 断言接口返回值的任意值模式
  - db: select * from sys_user su where login_name ='test999'  # 数据库断言，直接写SQL

执行逻辑详解

用例与参数分离：脚本只负责执行，所有接口配置、预期结果、关联关系全部写在 YAML 里，新增用例无需改代码
接口关联：第一个用例通过 extract 提取 orderNumber/userId 写入 extract.yaml，第二个用例通过 ${get_extract_data(orderNumber)} 引用，实现接口间数据传递
断言驱动：validation 是断言规则列表，框架遍历执行所有断言，全部通过该用例才判定成功

5. 接口请求处理 specification_yaml () 方法

代码示例

def specification_yaml(self, base_info, test_case):
    """
    接口请求处理基本方法
    :param base_info: yaml文件里面的baseInfo
    :param test_case: yaml文件里面的testCase
    :return:
    """
    try:
        params_type = ['data', 'json', 'params']
        url_host = self.conf.get_section_for_data('api_envi', 'host')
        api_name = base_info['api_name']
        allure.attach(api_name, f'接口名称：{api_name}', allure.attachment_type.TEXT)
        url = url_host + base_info['url']
        allure.attach(api_name, f'接口地址：{url}', allure.attachment_type.TEXT)
        method = base_info['method']
        allure.attach(api_name, f'请求方法：{method}', allure.attachment_type.TEXT)
        header = self.replace_load(base_info['header'])
        allure.attach(api_name, f'请求头：{header}', allure.attachment_type.TEXT)
        # 处理cookie
        cookie = None
        if base_info.get('cookies') is not None:
            cookie = eval(self.replace_load(base_info['cookies']))
        case_name = test_case.pop('case_name')
        allure.attach(api_name, f'测试用例名称：{case_name}', allure.attachment_type.TEXT)
        # 处理断言
        val = self.replace_load(test_case.get('validation'))
        test_case['validation'] = val
        validation = eval(test_case.pop('validation'))
        # 处理参数提取
        extract = test_case.pop('extract', None)
        extract_list = test_case.pop('extract_list', None)
        # 处理接口的请求参数
        for key, value in test_case.items():
            if key in params_type:
                test_case[key] = self.replace_load(value)
    # 处理文件上传接口
    file, files = test_case.pop('files', None), None
    if file is not None:
        for fk, fv in file.items():
            allure.attach(json.dumps(file), '导入文件')
            files = {fk: open(fv, mode='rb')}
res = self.run.run_main(name=api_name, url=url, case_name=case_name, header=header, method=method,
                        file=files, cookies=cookie, **test_case)
status_code = res.status_code
allure.attach(self.allure_attach_response(res.json()), '接口响应信息', allure.attachment_type.TEXT)
try:
res_json = json.loads(res.text)  # 把json格式转换成字典
if extract is not None:
self.extract_data(extract, res.text)
if extract_list is not None:
self.extract_data_list(extract_list, res.text)
# 处理断言
self.asserts.assert_result(validation, res_json, status_code)
except JSONDecodeError as js:
logs.error('系统异常或接口未请求！')
raise js
except Exception as e:
logs.error(e)
raise e
except Exception as e:
raise e

方法解析

该方法用于处理 YAML 文件中定义的接口测试用例，包括请求参数构造、动态变量替换、接口调用、响应断言以及数据提取等核心功能。

定义常量与基础配置

params_type = ['data', 'json', 'params']
url_host = self.conf.get_section_for_data('api_envi', 'host')

params_type：请求参数可能包含的类型字段
url_host：从 config.ini 的[api_envi]段读取环境主机地址

提取接口基本信息并添加 Allure 报告附件

从 base_info 中提取接口名称、URL、请求方法
使用 allure.attach 全部附加到 Allure 报告，测试结束后可逐接口查看请求信息

处理请求头和 Cookie

replace_load 对 header 做动态变量替换（如 ${get_extract_data(token)}）
存在 cookies 时同样替换后 eval 转为字典格式

提取测试用例名称

case_name = test_case.pop('case_name')

用 pop 取出 case_name 并从 test_case 中移除，避免混入请求参数

处理断言逻辑

val = self.replace_load(test_case.get('validation'))
test_case['validation'] = val
validation = eval(test_case.pop('validation'))

先替换断言表达式中的动态变量
用 eval 把 YAML 中的字符串还原为 list/dict 结构

提取数据字段

extract = test_case.pop('extract', None)
extract_list = test_case.pop('extract_list', None)

提前 pop 出提取规则，供请求完成后从响应中提取数据

处理请求参数（data/json/params）

遍历所有用例参数，若为 data、json、params 类型则执行动态变量替换

处理文件上传

files 字段按 rb 模式打开文件流，支持导入文件类接口

发送接口请求

res = self.run.run_main(name=api_name, url=url, case_name=case_name, header=header,
                        method=method, file=files, cookies=cookie, **test_case)

run_main 统一处理 get/post、上传、cookie、超时（60s）、SSL 校验关闭
响应内容 attach 到 Allure 报告

处理响应与断言

res_json = json.loads(res.text)
if extract is not None:
    self.extract_data(extract, res.text)       # 先提取，写入extract.yaml
if extract_list is not None:
    self.extract_data_list(extract_list, res.text)
self.asserts.assert_result(validation, res_json, status_code)   # 后断言

响应转 JSON 字典
先提取后断言：先把接口关联数据存好，再校验本次响应

异常捕获

JSONDecodeError：响应非 JSON，提示 "系统异常或接口未请求"
其余异常记录日志并 raise 抛出，保证 pytest 能正确识别失败用例

6. 动态参数 replace_load ()（接口关联核心）

代码示例

def replace_load(self, data):
    """yaml数据替换解析"""
    str_data = data
    if not isinstance(data, str):
        str_data = json.dumps(data, ensure_ascii=False)
    for i in range(str_data.count('${')):
        if '${' in str_data and '}' in str_data:
            start_index = str_data.index('$')
            end_index = str_data.index('}', start_index)
            ref_all_params = str_data[start_index:end_index + 1]
            # 取出yaml文件的函数名
            func_name = ref_all_params[2:ref_all_params.index("(")]
            # 取出函数里面的参数
            func_params = ref_all_params[ref_all_params.index("(") + 1:ref_all_params.index(")")]
            # 传入替换的参数获取对应的值,类的反射----getattr
            extract_data = getattr(DebugTalk(), func_name)(*func_params.split(',') if func_params else "")
            if extract_data and isinstance(extract_data, list):
                extract_data = ','.join(e for e in extract_data)
            str_data = str_data.replace(ref_all_params, str(extract_data))
    # 还原数据
    if data and isinstance(data, dict):
        data = json.loads(str_data)
    else:
        data = str_data
    return data

执行逻辑详解

字符串化处理：非字符串数据（dict/list）先 json.dumps 转为字符串，统一用字符串处理

定位表达式：通过 index('$') 和 index('}') 截取出完整的 ${函数名(参数)} 表达式

解析函数与参数：截取表达式中间的 函数名 和括号内的 参数（逗号分隔）

反射调用：getattr(DebugTalk(), func_name)(*参数) 动态调用 debugtalk.py 中的任意方法，新增参数函数无需修改框架代码

列表处理：提取结果若为 list，join 成逗号分隔字符串

还原数据：原数据是 dict 则 json.loads 还原为字典，否则保持字符串

常用动态参数示例

表达式	作用
`${get_extract_data(token)}`	读取 extract.yaml 中 token（接口关联）
`${get_extract_data(goodsIds,1)}`	读取列表型参数，取第 1 个值
`${get_extract_data(Cookie,csrfToken)}`	读取二级节点参数
`${timestamp()}` / `${timestamp_thirteen()}`	生成 10 位 / 13 位时间戳
`${start_time()}` / `${end_time()}`	昨天 / 当前标准时间
`${md5_encryption(123456)}`	MD5 加密
`${vehicle_random()}`	从 CSV 随机读取车牌号

7. 接口关联 extract_data ()（正则 + jsonpath 双提取器）

代码示例

def extract_data(self, testcase_extarct, response):
    """
    提取接口的返回值，支持正则表达式和json提取器
    :param testcase_extarct: testcase文件yaml中的extract值
    :param response: 接口的实际返回值
    :return:
    """
    try:
        pattern_lst = ['(.*?)', '(.+?)', r'(\d)', r'(\d*)']
        for key, value in testcase_extarct.items():
            # 处理正则表达式提取
            for pat in pattern_lst:
                if pat in value:
                    ext_lst = re.search(value, response)
                    if pat in [r'(\d+)', r'(\d*)']:
                        extract_data = {key: int(ext_lst.group(1))}
                    else:
                        extract_data = {key: ext_lst.group(1)}
                    self.read.write_yaml_data(extract_data)
            # 处理json提取参数
            if '$' in value:
                ext_json = jsonpath.jsonpath(json.loads(response), value)[0]
                if ext_json:
                    extarct_data = {key: ext_json}
                    logs.info('提取接口的返回值：', extarct_data)
                else:
                    extarct_data = {key: '未提取到数据，请检查接口返回值是否为空！'}
                self.read.write_yaml_data(extarct_data)
    except Exception as e:
        logs.error(e)

执行逻辑详解

正则提取：通过特征匹配 (.*?) (.+?) (\d) (\d*) 自动识别正则表达式，提取数字时自动转为 int 类型

jsonpath 提取：以 $ 开头的表达式走 jsonpath 解析（如 $.data.token、$.goodsList[*].goodsId），支持嵌套结构和数组

写入 extract.yaml：提取结果通过 write_yaml_data() 追加写入 extract.yaml，供下游用例 ${get_extract_data(key)} 引用

容错处理：提取不到数据时写入 "未提取到数据" 提示并记录日志，不中断用例执行

8. 断言模块（五种断言模式）

def assert_result(self, expected, response, status_code):
    """
    断言，通过断言all_flag标记，all_flag==0表示测试通过，否则为失败
    :param expected: 预期结果
    :param response: 实际响应结果
    :param status_code: 响应code码
    :return:
    """
    all_flag = 0
    try:
        for yq in expected:
            for key, value in yq.items():
                if key == "contains":    # 字符串包含断言
                    flag = self.contains_assert(value, response, status_code)
                elif key == "eq":        # 相等断言
                    flag = self.equal_assert(value, response)
                elif key == 'ne':        # 不相等断言
                    flag = self.not_equal_assert(value, response)
                elif key == 'rv':        # 任意值断言
                    flag = self.assert_response_any(actual_results=response, expected_results=value)
                elif key == 'db':        # 数据库断言
                    flag = self.assert_mysql_data(value)
                else:
                    logs.error("不支持此种断言方式")
                all_flag = all_flag + flag
    except Exception as exceptions:
        logs.error('接口断言异常，请检查yaml预期结果值是否正确填写!')
        raise exceptions
if all_flag == 0:
    logs.info("测试成功")
    assert True
else:
    logs.error("测试失败")
    assert False

五种断言模式详解

模式	关键字	实现原理	适用场景
字符串包含断言	`contains`	jsonpath 提取字段 + `in` 包含判断，支持 status_code 和 NONE 空值	校验 msg、状态码
相等断言	`eq`	取两个字典公共 key，`operator.eq()` 深度比较	校验字段值完全一致
不相等断言	`ne`	同上，`operator.ne()` 比较	校验字段值不一致
任意值断言	`rv`	`operator.eq()` 比较指定 key 的 value	校验返回体指定值
数据库断言	`db`	执行 SQL，`query_all()` 查询到数据即通过	校验落库数据

相等断言核心实现

def equal_assert(self, expected_results, actual_results, statuc_code=None):
    flag = 0
    if isinstance(actual_results, dict) and isinstance(expected_results, dict):
        # 找出实际结果与预期结果共同的key
        common_keys = list(expected_results.keys() & actual_results.keys())[0]
        # 根据相同的key去实际结果中获取，并重新生成一个实际结果的字典
        new_actual_results = {common_keys: actual_results[common_keys]}
        eq_assert = operator.eq(new_actual_results, expected_results)
        if eq_assert:
            logs.info(f"相等断言成功：接口实际结果：{new_actual_results}，等于预期结果：" + str(expected_results))
        else:
            flag += 1
            logs.error(f"相等断言失败：接口实际结果{new_actual_results}，不等于预期结果：" + str(expected_results))
    else:
        raise TypeError('相等断言--类型错误，预期结果和接口实际响应结果必须为字典类型！')
    return flag

相等断言执行逻辑

用集合运算符 & 找出预期结果与实际响应的公共 key，取第一个作为比较字段
从实际响应中取出对应字段值，构造新的字典
用 operator.eq() 深度比较（比==更严格可靠，支持嵌套结构）
相等则记录成功日志；不相等则 flag+1 标记失败
返回 flag：0 表示通过，非 0 表示失败

包含断言核心实现

def contains_assert(self, value, response, status_code):
    flag = 0
    for assert_key, assert_value in value.items():
        if assert_key == "status_code":
            if assert_value != status_code:
                flag += 1
                logs.error("contains断言失败：接口返回码【%s】不等于【%s】" % (status_code, assert_value))
        else:
            resp_list = jsonpath.jsonpath(response, "$..%s" % assert_key)
            if isinstance(resp_list[0], str):
                resp_list = ''.join(resp_list)
            if resp_list:
                assert_value = None if assert_value.upper() == 'NONE' else assert_value
                if assert_value in resp_list:
                    logs.info("字符串包含断言成功：预期结果【%s】,实际结果【%s】" % (assert_value, resp_list))
                else:
                    flag = flag + 1
                    logs.error("响应文本断言失败：预期结果为【%s】,实际结果为【%s】" % (assert_value, resp_list))
    return flag

包含断言执行逻辑

遍历断言字典，判断是否为 status_code 状态码断言
状态码断言：预期值与实际状态码不等则失败
字段断言：jsonpath.jsonpath(response, "$..字段名") 递归提取所有同名节点值
提取结果是字符串则 join 合并，支持 'NONE' 字符串自动转 Python 的 None 判断空值
预期值在响应字段中则通过，否则标记失败并记录日志

9. 请求发送封装 sendrequest.py

代码示例

def run_main(self, name, url, case_name, header, method, cookies=None, file=None, **kwargs):
    try:
        # 收集报告日志
        logs.info('接口名称：%s' % name)
        logs.info('请求地址：%s' % url)
        logs.info('请求方式：%s' % method)
        logs.info('测试用例名称：%s' % case_name)
        req_params = json.dumps(kwargs, ensure_ascii=False)
        if "data" in kwargs.keys() or "json" in kwargs.keys() or "params" in kwargs.keys():
            allure.attach(req_params, '请求参数', allure.attachment_type.TEXT)
            logs.info("请求参数：%s" % kwargs)
    except Exception as e:
        logs.error(e)
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
response = self.send_request(method=method,
                             url=url,
                             headers=header,
                             cookies=cookies,
                             files=file,
                             timeout=setting.API_TIMEOUT,
                             verify=False,
                             **kwargs)
return response

封装特点

特点	说明
统一入口	get/post/ 上传 / 带 cookie 全走 `run_main` 一个方法
忽略 SSL	`verify=False` 测试环境 https 不报证书错误
超时控制	`timeout=60` 全局超时，防止用例卡死
日志与报告	请求 / 响应全部记录日志并 attach 到 Allure 报告
Cookie 关联	`send_request` 用 `requests.session()` 保持会话，自动捕获 set-cookie 写入 extract.yaml
异常处理	连接异常自动 `pytest.fail()` 中断，避免用例挂起

10. 通知与 CI 集成

钉钉机器人推送（conftest.py 自动触发）

def send_dd_msg(content_str, at_all=True):
    """
    向钉钉机器人推送结果
    """
    timestamp_and_sign = generate_sign()   # HmacSHA256加签，保证webhook安全
    url = f'https://oapi.dingtalk.com/robot/send?access_token=你的token&timestamp={timestamp_and_sign[0]}&sign={timestamp_and_sign[1]}'
    headers = {'Content-Type': 'application/json;charset=utf-8'}
    data = {
        "msgtype": "text",
        "text": {"content": content_str},
        "at": {"isAtAll": at_all},
    }
    res = requests.post(url, json=data, headers=headers)
    return res.text

采用加签方式（HmacSHA256 加密 timestamp+secret），防止 webhook 地址泄露后被盗刷
conf/setting.py 中 dd_msg 开关控制是否推送
测试结束自动推送：用例总数、通过数、失败数、执行时长

Jenkins 集成（common/Pjenkins.py）

class PJenkins(object):
    def get_build_job_status(self):
        """读取构建完成的状态"""
        build_num = self.get_job_number()
        job_status = self.__server.get_build_info(self.job_name, build_num).get('result')
        return job_status
def report_success_or_fail(self):
    """统计测试报告用例成功数、失败数、跳过数以及成功率、失败率"""
    report_info = self.get_build_report()
    pass_count = report_info.get('passCount')
    fail_count = report_info.get('failCount')
    skip_count = report_info.get('skipCount')
    total_count = int(pass_count) + int(fail_count) + int(skip_count)
    ...
    return report_info