Python+Pytest 接口自动化测试框架
一.项目介绍
技术栈：Python+pytest+sqlalchemy+requests+allure+jsonpath+yaml+Jenkins+Linux
针对当当购书平台核心业务链路，搭建接口自动化框架，封装请求逻辑，实现接口关联、响应与数据库双维度校验。
二、项目结构
框架按「解耦分层」设计，每个目录只负责一件事，改代码、改数据不用到处找，核心结构及作用如下：
1 pythonproject/ # 项目根目录 2 ├── base/ # 基础工具层（封装接口、用例生成等核心工具） 3 │ ├── __init__.py # 包标识文件 4 │ ├── apiutil.py # 基础接口请求工具 5 │ ├── apiutil_business.py # 业务接口请求封装 6 │ ├── generateId.py # 测试数据ID生成工具 7 │ ├── new_testcase_tools.py # 自动化用例生成辅助工具 8 │ ├── new_tools.ui # UI类辅助工具 9 │ └── removefile.py # 测试文件清理工具 10 ├── common/ # 公共组件层（封装断言、数据库、通知等公共能力） 11 │ ├── __init__.py # 包标识文件 12 │ ├── assertions.py # 断言工具（接口+数据库校验） 13 │ ├── connection.py # 数据库连接工具（基于SQLAlchemy） 14 │ ├── debugtalk.py # 调试辅助工具 15 │ ├── dingRobot.py # 钉钉机器人通知工具 16 │ ├── handleExcel.py # Excel测试数据处理工具 17 │ ├── operationscsv.py # CSV数据处理工具 18 │ ├── operxml.py # XML文件解析工具 19 │ ├── Pjenkins.py # Jenkins持续集成对接工具 20 │ ├── readyaml.py # YAML配置读取工具 21 │ ├── recordlog.py # 日志记录工具 22 │ ├── semail.py # 邮件通知工具 23 │ ├── sendrequest.py # 请求发送二次封装 24 │ └── two_dimension_data.py # 二维数据处理工具 25 ├── conf/ # 配置层（管理项目环境、系统参数） 26 │ ├── __init__.py # 包标识文件 27 │ ├── config.ini # 项目全局配置文件 28 │ ├── operationConfig.py # 配置文件操作工具 29 │ └── setting.py # 配置项常量定义 30 ├── data/ # 数据层（存储测试数据、SQL脚本） 31 │ ├── sql/ # SQL脚本目录 32 │ │ └── __init__.py # 包标识文件 33 │ ├── login_data.csv # 登录场景测试数据 34 │ ├── loginName.yaml # 登录账号配置 35 │ ├── vehicleNo.csv # 物流模块车辆信息数据 36 │ └── 测试数据.xls # 通用测试数据表格 37 ├── logs/ # 日志层（存储测试执行日志） 38 │ ├── test.20260105.log # 2026-01-05测试日志 39 │ └── test.20260106.log # 2026-01-06测试日志 40 ├── report/ # 报告层（存储测试报告及临时文件） 41 │ ├── allureReport/ # Allure可视化报告生成目录 42 │ ├── temp/ # 报告临时文件目录 43 │ ├── tmreport/ # 自定义格式报告目录 44 │ └── results.xml # Pytest测试结果XML（用于生成Allure报告） 45 ├── test_env/ # 环境层（测试环境依赖配置） 46 │ ├── Include/ # 环境依赖包Include目录 47 │ ├── Lib/ # 环境依赖库目录 48 │ ├── Scripts/ # 环境脚本目录 49 │ └── pyvenv.cfg # 虚拟环境配置文件 50 ├── testcase/ # 用例层（POM模式，按业务拆分） 51 │ ├── Business interface/ # 业务全链路用例目录 52 │ ├── ProductManager/ # 产品模块用例目录 53 │ ├── Single interface/ # 单接口用例目录 54 │ ├── __init__.py # 包标识文件 55 │ └── conftest.py # 用例层局部夹具配置 56 ├── venv/ # 虚拟环境目录（隔离项目依赖） 57 ├── __init__.py # 项目根包标识文件 58 ├── conftest.py # Pytest全局夹具配置 59 ├── environment.xml # Allure报告环境信息配置 60 ├── extract.yaml # 接口关联数据存储文件 61 ├── pytest.ini # Pytest执行规则配置 62 ├── requirements.txt # 项目依赖包清单 63 ├── run.py # 框架执行入口脚本 64 └── 使用前请阅读此文件.md # 项目使用说明文档附：项目依赖清单（requirements.txt）
pytest==7.4.2
requests==2.31.0
PyYAML==6.0.1
jsonpath-ng==1.6.0
sqlalchemy==2.0.21
pymysql==1.1.0
allure-pytest==2.13.2
三.核心代码模块
1.Pytest 统一配置管理（用例执行规则）
核心文件：
- 根目录
pytest.ini：定义 Pytest 执行规则； - 根目录
conftest.py：全局夹具（如接口请求会话、数据库连接初始化）； - testcase 目录
conftest.py：用例层局部夹具（如链路用例前置登录）。
核心实现（pytest.ini）：
[pytest] ;addopts = -vs --alluredir ./report/temp -p no:warnings --clean-alluredir ; ;testpaths = ./testcase/ filterwarnings = error ignore::UserWarning python_files = test_*.py python_classes = Test* python_functions = test 核心实现（根目录 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() @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']) 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', [])) duration = time.time() - terminalreporter._sessionstarttime summary = f""" 自动化测试结果，通知如下，请着重关注测试失败的接口，具体执行结果如下： 测试用例总数：{total} 测试通过数：{passed} 测试失败数：{failed} 错误数量：{error} 跳过执行数量：{skipped} 执行总时长：{duration} """ print(summary) return summary def pytest_terminal_summary(terminalreporter, exitstatus, config): """自动收集pytest框架执行的测试结果并打印摘要信息""" summary = generate_test_summary(terminalreporter) if dd_msg: send_dd_msg(summary) 2.YAML+JsonPath 实现用例与参数解耦
核心文件：
- common 目录
readyaml.py：YAML 文件读取工具； - 根目录
extract.yaml：接口关联数据存储； - data 目录
*.yaml：接口请求参数（如下单接口参数文件order_params.yaml）。
核心实现（common/readyaml.py）：
import yaml import traceback import os from common.recordlog import logs from conf.operationConfig import OperationConfig from conf.setting import FILE_PATH from yaml.scanner import ScannerError def get_testcase_yaml(file): testcase_list = [] try: with open(file, 'r', encoding='utf-8') as f: data = yaml.safe_load(f) if len(data) <= 1: yam_data = data[0] base_info = yam_data.get('baseInfo') for ts in yam_data.get('testCase'): param = [base_info, ts] testcase_list.append(param) return testcase_list else: return data except UnicodeDecodeError: logs.error(f"[{file}]文件编码格式错误，--尝试使用utf-8编码解码YAML文件时发生了错误，请确保你的yaml文件是UTF-8格式！") except FileNotFoundError: logs.error(f'[{file}]文件未找到，请检查路径是否正确') except Exception as e: logs.error(f'获取【{file}】文件数据时出现未知错误: {str(e)}') class ReadYamlData: """读写接口的YAML格式测试数据""" def __init__(self, yaml_file=None): if yaml_file is not None: self.yaml_file = yaml_file else: pass self.conf = OperationConfig() self.yaml_data = None @property def get_yaml_data(self): """ 获取测试用例yaml数据 :param file: YAML文件 :return: 返回list """ # Loader=yaml.FullLoader表示加载完整的YAML语言，避免任意代码执行，无此参数控制台报Warning try: with open(self.yaml_file, 'r', encoding='utf-8') as f: self.yaml_data = yaml.safe_load(f) return self.yaml_data except Exception: logs.error(str(traceback.format_exc())) #部分代码示例核心实现（common/assertions.py）：
import traceback import allure import jsonpath import operator from common.recordlog import logs from common.connection import ConnectMysql class Assertions: """" 接口断言模式，支持 1）响应文本字符串包含模式断言 2）响应结果相等断言 3）响应结果不相等断言 4）响应结果任意值断言 5）数据库断言 """ def contains_assert(self, value, response, status_code): """ 字符串包含断言模式，断言预期结果的字符串是否包含在接口的响应信息中 :param value: 预期结果，yaml文件的预期结果值 :param response: 接口实际响应结果 :param status_code: 响应状态码 :return: 返回结果的状态标识 """ # 断言状态标识，0成功，其他失败 flag = 0 for assert_key, assert_value in value.items(): if assert_key == "status_code": if assert_value != status_code: flag += 1 allure.attach(f"预期结果：{assert_value}\n实际结果：{status_code}", '响应代码断言结果:失败', attachment_type=allure.attachment_type.TEXT) logs.error("contains断言失败：接口返回码【%s】不等于【%s】" % (status_code, assert_value)) else: resp_list = jsonpath.jsonpath(response, "$..%s" % assert_key) if isinstance(resp_list[0], str):.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 allure.attach(f"预期结果：{assert_value}\n实际结果：{resp_list}", '响应文本断言结果：失败', attachment_type=allure.attachment_type.TEXT) logs.error("响应文本断言失败：预期结果为【%s】,实际结果为【%s】" % (assert_value, resp_list)) return flag #部分代码示例4.POM 模式分离用例层与驱动层
核心文件：
- testcase 目录
Business interface/test_order_link.py：业务链路用例（仅关注业务逻辑）； - base 目录
apiutil_business.py：业务驱动层（封装接口调用逻辑）。
驱动层实现（base/apiutil_business.py）：
# -*- coding: utf-8 -*- # sys.path.insert(0, "..") from common.sendrequest import SendRequest from common.readyaml import ReadYamlData from common.recordlog import logs from conf.operationConfig import OperationConfig from common.assertions import Assertions from common.debugtalk import DebugTalk import allure import json import jsonpath import re import traceback from json.decoder import JSONDecodeError assert_res = Assertions() class RequestBase(object): def __init__(self): self.run = SendRequest() self.read = ReadYamlData() self.conf = OperationConfig() def handler_yaml_list(self, data_dict): """处理yaml文件测试用例请求参数为list情况，以数组形式""" try: for key, value in data_dict.items(): if isinstance(value, list): value_lst = ','.join(value).split(',') data_dict[key] = value_lst return data_dict except Exception: logs.error(str(traceback.format_exc())) 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: # index检测字符串是否子字符串，并找到字符串的索引位置 start_index = str_data.index('$') end_index = str_data.index('}', start_index) # yaml文件的参数，如：${get_yaml_data(loginname)} ref_all_params = str_data[start_index:end_index + 1] # 函数名，获取Debugtalk的方法 func_name = ref_all_params[2:ref_all_params.index("(")] # 函数里的参数 func_params = ref_all_params[ref_all_params.index("(") + 1:ref_all_params.index(")")] # 传入替换的参数获取对应的值,*func_params按,分割重新得到一个字符串 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) self.handler_yaml_list(data) else: data = str_data return data #部分代码示例用例层实现（testcase/Business interface/test_business_scenario.py）：
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) 5.parametrize+YAML 适配多场景测试
核心文件：
- data 目录
order_link_data.yaml：多场景测试数据； - testcase 目录
Business interface/test_order_link.py：parametrize 读取数据。
多场景数据（data/loginName.yaml）：
- baseInfo: api_name: 用户登录 url: /dar/user/login method: post header: Content-Type: application/x-www-form-urlencoded;charset=UTF-8 testCase: - case_name: 用户名和密码正确登录验证 data: user_name: test01 passwd: admin123 validation: - contains: { 'error_code': none } - eq: { 'msg': '登录成功' } extract: token: $.token6.框架执行入口（run.py）
核心文件：根目录 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') 四.项目结语
电商接口自动化的核心是 “高可用、易维护、全覆盖”，本框架通过 “解耦式设计 + 双维度校验 + POM 模式”，既适配了当当购书的核心业务链路，也可快速迁移至其他电商平台。希望本文的实战思路能为测试同学提供参考，让接口自动化真正落地并产生价值。
