接口自动化测试框架整合简洁版后续可根据需求进一步扩充框架目录如下projectAPITest_framework/ ├── config/ │ └── settings.py # 全局配置 ├── core/ │ ├── api_client.py # HTTP请求封装 │ ├── response_extractor.py # 响应数据提取 │ └── assert_util.py # 断言工具 ├── data/ │ └── test_data.yaml # 测试数据 ├── utils/ │ └── yaml_loader.py # YAML文件加载 ├── tests/ │ ├── conftest.py # pytest fixtures Allure配置 │ └── test_machine.py # 测试用例 ├── reports/ # 测试报告 │ ├── allure-results/ # Allure原始数据 │ └── allure-report/ # 生成的HTML报告 ├── requirements.txt └── run.py # 执行入口config/settings 配置相关主要是一个公共常量配置方便后续去调用如请求根地址、 发送邮件相关地址、项目路径相关若不存在则进行递归创建import os from pathlib import Path class Config: 全局配置 BASE_URL http://IP:port TIMEOUT 10 # 邮件配置 EMAIL_HOST smtp.qq.com EMAIL_PORT 465 # SSL端口 EMAIL_SENDER os.getenv(EMAIL_SENDER, xxxxxqq.com) # 发件人邮箱 EMAIL_AUTH_CODE os.getenv(EMAIL_PASSWORD, xxxx) # 授权码环境变量 EMAIL_RECEIVER [xxxxqq.com, xxxxx163.com] # 收件人 EMAIL_SUBJECT 接口自动化测试报告 EMAIL_BODY_TEMPLATE 测试执行完成详情如下 【测试结果摘要】 ───────────────── 总用例数 : {total} 通过 : {passed} ✅ 失败 : {failed} ❌ 跳过 : {skipped} ⏭️ 通过率 : {pass_rate:.2f}% 详细报告请查看附件Allure报告压缩包。 有疑问请联系xxx Telphone:xxxxxxxxxxx 生成时间: {timestamp} # 项目根目录 ROOT_DIR Path(__file__).parent.parent DATA_DIR ROOT_DIR / data REPORT_DIR ROOT_DIR / reports ALLURE_RESULTS_DIR REPORT_DIR / allure-results ALLURE_REPORT_DIR REPORT_DIR / allure-report # 确保目录存在 classmethod def ensure_dirs(cls): cls.ALLURE_RESULTS_DIR.mkdir(parentsTrue, exist_okTrue) cls.ALLURE_REPORT_DIR.mkdir(parentsTrue, exist_okTrue)api/api_client 主要是请求方式的处理方便后续调用还包括Session更新共享即后续每 次调用都实现了Session共用以减少连接次数来节约资源在conftest类里会有一个相关的资 源关闭和释放另外还包括值的set和get的调用方便后续值的注入和获取用于接口串联的 依赖调用import requests import json from config.settings import Config class ApiClient: API客户端封装requests并支持变量存储用于接口关联 def __init__(self): self.session requests.Session() self.base_url Config.BASE_URL self.timeout Config.TIMEOUT self._vars {} # 存储动态变量如token def request(self, method, url, **kwargs): 统一请求方法 full_url f{self.base_url}{url} if not url.startswith(http) else url kwargs.setdefault(timeout, self.timeout) # 如果传递了headers合并到session if headers in kwargs: self.session.headers.update(kwargs.pop(headers)) response self.session.request(method, full_url, **kwargs) # 打印日志便于调试 print(f\n【请求】{method} {full_url}) print(f【请求头】{dict(self.session.headers)}) if json in kwargs: print(f【请求体】{json.dumps(kwargs[json], ensure_asciiFalse)}) elif data in kwargs: print(f【请求体】{kwargs[data]}) print(f【响应状态】{response.status_code}) print(f【响应内容】{response.text[:300]}...) return response def get(self, url, **kwargs): return self.request(GET, url, **kwargs) def post(self, url, **kwargs): return self.request(POST, url, **kwargs) def put(self, url, **kwargs): return self.request(PUT, url, **kwargs) def delete(self, url, **kwargs): return self.request(DELETE, url, **kwargs) def set_header(self, key, value): 设置全局请求头所有请求自动携带 self.session.headers[key] value def set_var(self, key, value): 保存变量用于接口关联 self._vars[key] value def get_var(self, key): 获取变量 return self._vars.get(key) def clear_vars(self): 清空变量 self._vars.clear()api/assert_util 值的提取和断言与response_extractor.py强关联from core.response_extractor import ResponseExtractor class AssertUtil: 断言工具 staticmethod def assert_status_code(response, expected_code): 断言HTTP状态码 assert response.status_code expected_code, \ f状态码错误: 期望{expected_code}, 实际{response.status_code} print(f✓ 状态码验证通过: {expected_code}) staticmethod def assert_json_value(response, path, expected, methodpath): 断言JSON中指定字段的值 data response.json() if method path: actual ResponseExtractor.extract_by_path(data, path) elif method jsonpath: actual ResponseExtractor.extract_by_jsonpath(data, path) else: raise ValueError(method必须为path或jsonpath) assert actual expected, \ f字段{path}值错误: 期望{expected}, 实际{actual} print(f✓ 字段验证通过: {path} {expected}) staticmethod def assert_json_contains(response, key): 断言响应JSON中包含某个字段仅顶级 data response.json() assert key in data, f响应中不包含字段: {key} print(f✓ 字段存在: {key}) staticmethod def assert_not_none(value, msg值不应为None): 断言值不为None assert value is not None, msg print(f✓ 值不为None: {value})api/response_extractor 根据请求路径来进行提取的目前主要支持常见的两种from jsonpath import jsonpath class ResponseExtractor: 响应提取工具支持点号路径和JSONPath staticmethod def extract_by_path(data, path, defaultNone): 通过点号路径提取如 data.user.name 支持列表索引: orders[0].id 或 orders.0.id keys path.split(.) value data for key in keys: if isinstance(value, dict): value value.get(key) elif isinstance(value, list): # 如果key是数字则作为索引 if key.isdigit(): try: value value[int(key)] except IndexError: return default else: return default else: return default if value is None: return default return value staticmethod def extract_by_jsonpath(data, expr, defaultNone): 通过JSONPath提取 result jsonpath(data, expr) if result is False: return default return result[0] if len(result) 1 else result classmethod def extract(cls, response, rule): 统一提取入口 rule格式: {method: path|jsonpath, path: ..., default: None} data response.json() method rule.get(method, path) path rule.get(path) default rule.get(default) if method path: return cls.extract_by_path(data, path, default) elif method jsonpath: return cls.extract_by_jsonpath(data, path, default) return Nonedata/test_data 这里存放的是请求入参的yaml文件当然还可考虑excel和数据库存放# 登录数据 login_data: - case_id: TC001 name: xxx password: xxxxxx expected_code: 200 expected_token_prefix: Bearer # 机器查询数据 machine_query: - case_id: TC002 projectid: 1 expected_code: 200 expected_data_type: intreports/allure-results/environment.properties 报告的环境相关的信息的设置BASE_URLhttp://IP:port PythonVersion3.10 TestEnvtest Platformnttest/conftest 存放各种测试前置资源如测试结果、Session、数据库连接、随机数生成之类的import pytest import allure from core.api_client import ApiClient from utils.yaml_loader import load_yaml import os import json from config.settings import Config # 用于存储测试结果的全局变量 test_results { total: 0, passed: 0, failed: 0, skipped: 0 } # 确保Allure结果目录存在 Config.ensure_dirs() # Fixtures pytest.fixture(scopesession) def api_client(): 全局API客户端每个会话只创建一次 client ApiClient() yield client client.session.close() pytest.fixture(scopesession) def test_data(): 加载测试数据YAML yaml_path Config.DATA_DIR / test_data.yaml return load_yaml(yaml_path) pytest.fixture(scopefunction) def logged_client(api_client, test_data): 已登录的客户端自动执行登录并提取token设置到请求头中 此fixture供需要token的测试用例使用 login_info test_data[login_data][0] # 取第一组登录数据 with allure.step(登录获取token): response api_client.post( /account/token, json{name: login_info[name], password: login_info[password]} ) # 断言登录成功 assert response.status_code 200, 登录失败 # 提取token token response.json().get(data) assert token, 未返回token # 设置全局请求头 Authorization api_client.set_header(Authorization, token) api_client.set_var(token, token) # 将token附加到Allure报告 allure.attach(token, nameToken, attachment_typeallure.attachment_type.TEXT) yield api_client # 返回已登录的客户端 # 清理可选 # api_client.session.headers.pop(Authorization, None) # Allure环境信息 pytest.fixture(scopesession, autouseTrue) def allure_environment(): 将测试环境信息写入Allure报告 env_file Config.ALLURE_RESULTS_DIR / environment.properties with open(env_file, w) as f: f.write(fBASE_URL{Config.BASE_URL}\n) f.write(fPythonVersion3.10\n) f.write(fTestEnvtest\n) f.write(fPlatform{os.name}\n) yield # 失败时附加信息 pytest.hookimpl(tryfirstTrue, hookwrapperTrue) def pytest_runtest_makereport(item, call): 收集每个测试用例的执行结果 global test_results outcome yield report outcome.get_result() if report.when call: test_results[total] 1 if report.failed: test_results[failed] 1 elif report.skipped: test_results[skipped] 1 else: test_results[passed] 1 def pytest_sessionfinish(session, exitstatus): 测试会话结束时将结果写入文件 global test_results # 计算通过率 total test_results[total] if total 0: test_results[pass_rate] test_results[passed] / total * 100 else: test_results[pass_rate] 0.0 # 写入结果文件 result_file Config.REPORT_DIR / test_summary.json result_file.parent.mkdir(parentsTrue, exist_okTrue) with open(result_file, w, encodingutf-8) as f: json.dump(test_results, f, ensure_asciiFalse, indent2) print(f测试结果已保存: {result_file}) print(f总用例: {test_results[total]}, 通过: {test_results[passed]}, f失败: {test_results[failed]}, 通过率: {test_results[pass_rate]:.2f}%)test/test_machine 发送接口请求响应断言测试日志输出等import pytest import allure from core.assert_util import AssertUtil from core.response_extractor import ResponseExtractor from utils.yaml_loader import load_yaml from config.settings import Config # 加载测试数据 test_data load_yaml(Config.DATA_DIR / test_data.yaml) login_data test_data[login_data] machine_data test_data[machine_query] allure.epic(机器管理) allure.feature(认证与查询) class TestMachine: 机器相关接口测试 allure.story(登录获取Token) allure.title(测试登录接口并提取token) pytest.mark.parametrize(case, login_data) def test_login_and_get_token(self, api_client, case): 测试登录接口提取token并保存 with allure.step(f发送登录请求: {case[name]}): response api_client.post( /account/token, json{name: case[name], password: case[password]} ) # 断言 AssertUtil.assert_status_code(response, case[expected_code]) if case[expected_code] 200: with allure.step(提取并验证token): token ResponseExtractor.extract_by_path(response.json(), data) AssertUtil.assert_not_none(token, 未获取到token) assert token.startswith(case[expected_token_prefix]), \ ftoken应以{case[expected_token_prefix]}开头实际: {token[:20]}... # 保存到客户端变量供后续使用 api_client.set_var(token, token) api_client.set_header(Authorization, token) allure.attach(token, nameToken, attachment_typeallure.attachment_type.TEXT) allure.story(查询机器数量) allure.title(使用登录后的token查询机器数量) def test_get_machine_num(self, logged_client): 使用登录后的客户端获取机器数量接口关联 with allure.step(发送GET请求 /machine/getmachinenum): response logged_client.get( /machine/getmachinenum, params{projectid: 1} ) # 断言 AssertUtil.assert_status_code(response, 200) with allure.step(验证返回的data为整数): data ResponseExtractor.extract_by_path(response.json(), data) assert isinstance(data, int), fdata应为整数实际为{type(data)} allure.attach(str(data), name机器数量, attachment_typeallure.attachment_type.TEXT) print(f✓ 机器数量: {data}) allure.story(查询机器数量参数化) allure.title(使用不同projectid查询机器数量) pytest.mark.parametrize(projectid, [1, 2, 3]) def test_machine_projectid_param(self, logged_client, projectid): 参数化测试不同projectid with allure.step(f查询projectid{projectid}): response logged_client.get( /machine/getmachinenum, params{projectid: projectid} ) AssertUtil.assert_status_code(response, 200) data ResponseExtractor.extract_by_path(response.json(), data) assert data is not None allure.attach(str(data), namef项目{projectid}数量, attachment_typeallure.attachment_type.TEXT) print(f项目{projectid}数量: {data})utils/email_sender 封装测试报告发送邮件的工具类import smtplib import os from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from email.mime.application import MIMEApplication from pathlib import Path from datetime import datetime from typing import Union, List, Optional from config.settings import Config def send_email_with_attachment( subject: str, body: str, receiver: Union[str, List[str]], attachments: Optional[List[Union[str, Path]]] None ) - bool: 发送带附件的邮件支持多个收件人 :param subject: 邮件主题 :param body: 邮件正文 :param receiver: 收件人可以是单个邮箱字符串或邮箱列表 :param attachments: 附件路径列表可选 :return: 发送成功返回True否则False sender Config.EMAIL_SENDER auth_code Config.EMAIL_AUTH_CODE if not sender or not auth_code: print(邮件发送者或授权码未配置跳过邮件发送) return False # ---------- 统一收件人为列表 ---------- if isinstance(receiver, str): receiver_list [receiver] else: receiver_list list(receiver) # 确保是列表 if not receiver_list: print(收件人为空跳过发送) return False # ---------- 构建邮件 ---------- msg MIMEMultipart() msg[From] sender msg[To] , .join(receiver_list) # 邮件头显示多个收件人 msg[Subject] subject # 添加正文 msg.attach(MIMEText(body, plain, utf-8)) # 添加附件 if attachments: for file_path in attachments: file_path Path(file_path) if not file_path.exists(): print(f附件不存在跳过: {file_path}) continue try: with open(file_path, rb) as f: part MIMEApplication(f.read(), Namefile_path.name) part[Content-Disposition] fattachment; filename{file_path.name} msg.attach(part) print(f已添加附件: {file_path.name}) except Exception as e: print(f添加附件失败 {file_path}: {e}) # ---------- 发送邮件 ---------- try: # 使用 SSL 连接QQ 邮箱使用 465 端口 with smtplib.SMTP_SSL(Config.EMAIL_HOST, Config.EMAIL_PORT) as server: server.login(sender, auth_code) server.sendmail(sender, receiver_list, msg.as_string()) print(f邮件发送成功 - {, .join(receiver_list)}) return True except Exception as e: print(f邮件发送失败: {e}) return False def send_test_report( report_summary: dict, allure_report_dir: Optional[Union[str, Path]] None, zip_path: Optional[Union[str, Path]] None ) - bool: 发送测试报告邮件封装函数 :param report_summary: 测试结果摘要字典包含 total, passed, failed, skipped, pass_rate :param allure_report_dir: Allure 报告目录未使用仅保留兼容 :param zip_path: 已打包的 zip 文件路径 :return: 发送成功返回 True # 格式化正文 body Config.EMAIL_BODY_TEMPLATE.format( totalreport_summary.get(total, 0), passedreport_summary.get(passed, 0), failedreport_summary.get(failed, 0), skippedreport_summary.get(skipped, 0), pass_ratereport_summary.get(pass_rate, 0.0), timestampdatetime.now().strftime(%Y-%m-%d %H:%M:%S) ) # 构建附件列表 attachments [] if zip_path and Path(zip_path).exists(): attachments.append(zip_path) # 发送邮件收件人从 Config 读取支持列表 return send_email_with_attachment( subjectConfig.EMAIL_SUBJECT, bodybody, receiverConfig.EMAIL_RECEIVER, # 现在可以是列表或字符串 attachmentsattachments )utils/yaml_loader 封装读取YAML的工具类import yaml from pathlib import Path def load_yaml(file_path): 加载YAML文件并返回数据 with open(file_path, r, encodingutf-8) as f: return yaml.safe_load(f)run.py 测试执行的启动类执行测试并输出测试结果报告import pytest import sys import subprocess import zipfile from pathlib import Path from config.settings import Config from utils.email_sender import send_test_report import json from datetime import datetime def generate_allure_report(): 生成Allure报告并打包为zip # 1. 生成HTML报告 cmd [ allure, generate, str(Config.ALLURE_RESULTS_DIR), -o, str(Config.ALLURE_REPORT_DIR), --clean ] try: subprocess.run(cmd, checkTrue, capture_outputTrue, textTrue) print(fAllure报告生成成功: {Config.ALLURE_REPORT_DIR}) except subprocess.CalledProcessError as e: print(fAllure报告生成失败: {e.stderr}) return None except FileNotFoundError: print(未找到allure命令请确保已安装Allure并配置环境变量) return None # 2. 打包为zip zip_path Config.REPORT_DIR / fallure_report_{datetime.now().strftime(%Y%m%d_%H%M%S)}.zip with zipfile.ZipFile(zip_path, w, zipfile.ZIP_DEFLATED) as zipf: for file in Config.ALLURE_REPORT_DIR.rglob(*): if file.is_file(): arcname file.relative_to(Config.ALLURE_REPORT_DIR) zipf.write(file, arcname) print(fAllure报告已打包: {zip_path}) return zip_path def get_test_summary(): 从结果文件中读取测试摘要 summary_file Config.REPORT_DIR / test_summary.json if not summary_file.exists(): print(测试摘要文件不存在) return {total: 0, passed: 0, failed: 0, skipped: 0, pass_rate: 0} with open(summary_file, r, encodingutf-8) as f: return json.load(f) def run_tests(): 执行测试并发送报告邮件 Config.ensure_dirs() print( 开始执行测试 ) # 1. 执行pytest args [ tests/, -v, --alluredir, str(Config.ALLURE_RESULTS_DIR), --maxfail1, --tbshort, --strict-markers ] exit_code pytest.main(args) print(f测试执行完毕退出码: {exit_code}) # 2. 生成Allure报告并打包 zip_path generate_allure_report() # 3. 读取测试结果摘要 summary get_test_summary() # 4. 发送邮件 if zip_path: send_test_report(summary, Config.ALLURE_REPORT_DIR, zip_path) else: # 如果没有zip报告只发摘要 send_test_report(summary, None, None) # 5. 输出提示 print(\n * 60) if exit_code 0: print(✅ 所有测试通过) else: print(❌ 有测试失败请查看报告) print(f Allure结果: {Config.ALLURE_RESULTS_DIR}) print(f 邮件已发送至: {Config.EMAIL_RECEIVER}) print( * 60) return exit_code if __name__ __main__: sys.exit(run_tests())运行run.py测试启动类即可启动执行测试用例并发送测试报告到相关邮箱具体详情解析可查看这篇文章Python版--测试报告邮件发送配置-CSDN博客测试报告生成方法 # 方式1直接serve自动打开浏览器 allure serve reports/allure-results # 方式2先生成再打开(推荐因为可能第一种方法会将前面的测试结果的合集加起来可能存在测试用例执行条数变多) allure generate reports/allure-results -o reports/allure-report --clean allure open reports/allure-report Python接口自动化测试框架
接口自动化测试框架整合简洁版后续可根据需求进一步扩充框架目录如下projectAPITest_framework/ ├── config/ │ └── settings.py # 全局配置 ├── core/ │ ├── api_client.py # HTTP请求封装 │ ├── response_extractor.py # 响应数据提取 │ └── assert_util.py # 断言工具 ├── data/ │ └── test_data.yaml # 测试数据 ├── utils/ │ └── yaml_loader.py # YAML文件加载 ├── tests/ │ ├── conftest.py # pytest fixtures Allure配置 │ └── test_machine.py # 测试用例 ├── reports/ # 测试报告 │ ├── allure-results/ # Allure原始数据 │ └── allure-report/ # 生成的HTML报告 ├── requirements.txt └── run.py # 执行入口config/settings 配置相关主要是一个公共常量配置方便后续去调用如请求根地址、 发送邮件相关地址、项目路径相关若不存在则进行递归创建import os from pathlib import Path class Config: 全局配置 BASE_URL http://IP:port TIMEOUT 10 # 邮件配置 EMAIL_HOST smtp.qq.com EMAIL_PORT 465 # SSL端口 EMAIL_SENDER os.getenv(EMAIL_SENDER, xxxxxqq.com) # 发件人邮箱 EMAIL_AUTH_CODE os.getenv(EMAIL_PASSWORD, xxxx) # 授权码环境变量 EMAIL_RECEIVER [xxxxqq.com, xxxxx163.com] # 收件人 EMAIL_SUBJECT 接口自动化测试报告 EMAIL_BODY_TEMPLATE 测试执行完成详情如下 【测试结果摘要】 ───────────────── 总用例数 : {total} 通过 : {passed} ✅ 失败 : {failed} ❌ 跳过 : {skipped} ⏭️ 通过率 : {pass_rate:.2f}% 详细报告请查看附件Allure报告压缩包。 有疑问请联系xxx Telphone:xxxxxxxxxxx 生成时间: {timestamp} # 项目根目录 ROOT_DIR Path(__file__).parent.parent DATA_DIR ROOT_DIR / data REPORT_DIR ROOT_DIR / reports ALLURE_RESULTS_DIR REPORT_DIR / allure-results ALLURE_REPORT_DIR REPORT_DIR / allure-report # 确保目录存在 classmethod def ensure_dirs(cls): cls.ALLURE_RESULTS_DIR.mkdir(parentsTrue, exist_okTrue) cls.ALLURE_REPORT_DIR.mkdir(parentsTrue, exist_okTrue)api/api_client 主要是请求方式的处理方便后续调用还包括Session更新共享即后续每 次调用都实现了Session共用以减少连接次数来节约资源在conftest类里会有一个相关的资 源关闭和释放另外还包括值的set和get的调用方便后续值的注入和获取用于接口串联的 依赖调用import requests import json from config.settings import Config class ApiClient: API客户端封装requests并支持变量存储用于接口关联 def __init__(self): self.session requests.Session() self.base_url Config.BASE_URL self.timeout Config.TIMEOUT self._vars {} # 存储动态变量如token def request(self, method, url, **kwargs): 统一请求方法 full_url f{self.base_url}{url} if not url.startswith(http) else url kwargs.setdefault(timeout, self.timeout) # 如果传递了headers合并到session if headers in kwargs: self.session.headers.update(kwargs.pop(headers)) response self.session.request(method, full_url, **kwargs) # 打印日志便于调试 print(f\n【请求】{method} {full_url}) print(f【请求头】{dict(self.session.headers)}) if json in kwargs: print(f【请求体】{json.dumps(kwargs[json], ensure_asciiFalse)}) elif data in kwargs: print(f【请求体】{kwargs[data]}) print(f【响应状态】{response.status_code}) print(f【响应内容】{response.text[:300]}...) return response def get(self, url, **kwargs): return self.request(GET, url, **kwargs) def post(self, url, **kwargs): return self.request(POST, url, **kwargs) def put(self, url, **kwargs): return self.request(PUT, url, **kwargs) def delete(self, url, **kwargs): return self.request(DELETE, url, **kwargs) def set_header(self, key, value): 设置全局请求头所有请求自动携带 self.session.headers[key] value def set_var(self, key, value): 保存变量用于接口关联 self._vars[key] value def get_var(self, key): 获取变量 return self._vars.get(key) def clear_vars(self): 清空变量 self._vars.clear()api/assert_util 值的提取和断言与response_extractor.py强关联from core.response_extractor import ResponseExtractor class AssertUtil: 断言工具 staticmethod def assert_status_code(response, expected_code): 断言HTTP状态码 assert response.status_code expected_code, \ f状态码错误: 期望{expected_code}, 实际{response.status_code} print(f✓ 状态码验证通过: {expected_code}) staticmethod def assert_json_value(response, path, expected, methodpath): 断言JSON中指定字段的值 data response.json() if method path: actual ResponseExtractor.extract_by_path(data, path) elif method jsonpath: actual ResponseExtractor.extract_by_jsonpath(data, path) else: raise ValueError(method必须为path或jsonpath) assert actual expected, \ f字段{path}值错误: 期望{expected}, 实际{actual} print(f✓ 字段验证通过: {path} {expected}) staticmethod def assert_json_contains(response, key): 断言响应JSON中包含某个字段仅顶级 data response.json() assert key in data, f响应中不包含字段: {key} print(f✓ 字段存在: {key}) staticmethod def assert_not_none(value, msg值不应为None): 断言值不为None assert value is not None, msg print(f✓ 值不为None: {value})api/response_extractor 根据请求路径来进行提取的目前主要支持常见的两种from jsonpath import jsonpath class ResponseExtractor: 响应提取工具支持点号路径和JSONPath staticmethod def extract_by_path(data, path, defaultNone): 通过点号路径提取如 data.user.name 支持列表索引: orders[0].id 或 orders.0.id keys path.split(.) value data for key in keys: if isinstance(value, dict): value value.get(key) elif isinstance(value, list): # 如果key是数字则作为索引 if key.isdigit(): try: value value[int(key)] except IndexError: return default else: return default else: return default if value is None: return default return value staticmethod def extract_by_jsonpath(data, expr, defaultNone): 通过JSONPath提取 result jsonpath(data, expr) if result is False: return default return result[0] if len(result) 1 else result classmethod def extract(cls, response, rule): 统一提取入口 rule格式: {method: path|jsonpath, path: ..., default: None} data response.json() method rule.get(method, path) path rule.get(path) default rule.get(default) if method path: return cls.extract_by_path(data, path, default) elif method jsonpath: return cls.extract_by_jsonpath(data, path, default) return Nonedata/test_data 这里存放的是请求入参的yaml文件当然还可考虑excel和数据库存放# 登录数据 login_data: - case_id: TC001 name: xxx password: xxxxxx expected_code: 200 expected_token_prefix: Bearer # 机器查询数据 machine_query: - case_id: TC002 projectid: 1 expected_code: 200 expected_data_type: intreports/allure-results/environment.properties 报告的环境相关的信息的设置BASE_URLhttp://IP:port PythonVersion3.10 TestEnvtest Platformnttest/conftest 存放各种测试前置资源如测试结果、Session、数据库连接、随机数生成之类的import pytest import allure from core.api_client import ApiClient from utils.yaml_loader import load_yaml import os import json from config.settings import Config # 用于存储测试结果的全局变量 test_results { total: 0, passed: 0, failed: 0, skipped: 0 } # 确保Allure结果目录存在 Config.ensure_dirs() # Fixtures pytest.fixture(scopesession) def api_client(): 全局API客户端每个会话只创建一次 client ApiClient() yield client client.session.close() pytest.fixture(scopesession) def test_data(): 加载测试数据YAML yaml_path Config.DATA_DIR / test_data.yaml return load_yaml(yaml_path) pytest.fixture(scopefunction) def logged_client(api_client, test_data): 已登录的客户端自动执行登录并提取token设置到请求头中 此fixture供需要token的测试用例使用 login_info test_data[login_data][0] # 取第一组登录数据 with allure.step(登录获取token): response api_client.post( /account/token, json{name: login_info[name], password: login_info[password]} ) # 断言登录成功 assert response.status_code 200, 登录失败 # 提取token token response.json().get(data) assert token, 未返回token # 设置全局请求头 Authorization api_client.set_header(Authorization, token) api_client.set_var(token, token) # 将token附加到Allure报告 allure.attach(token, nameToken, attachment_typeallure.attachment_type.TEXT) yield api_client # 返回已登录的客户端 # 清理可选 # api_client.session.headers.pop(Authorization, None) # Allure环境信息 pytest.fixture(scopesession, autouseTrue) def allure_environment(): 将测试环境信息写入Allure报告 env_file Config.ALLURE_RESULTS_DIR / environment.properties with open(env_file, w) as f: f.write(fBASE_URL{Config.BASE_URL}\n) f.write(fPythonVersion3.10\n) f.write(fTestEnvtest\n) f.write(fPlatform{os.name}\n) yield # 失败时附加信息 pytest.hookimpl(tryfirstTrue, hookwrapperTrue) def pytest_runtest_makereport(item, call): 收集每个测试用例的执行结果 global test_results outcome yield report outcome.get_result() if report.when call: test_results[total] 1 if report.failed: test_results[failed] 1 elif report.skipped: test_results[skipped] 1 else: test_results[passed] 1 def pytest_sessionfinish(session, exitstatus): 测试会话结束时将结果写入文件 global test_results # 计算通过率 total test_results[total] if total 0: test_results[pass_rate] test_results[passed] / total * 100 else: test_results[pass_rate] 0.0 # 写入结果文件 result_file Config.REPORT_DIR / test_summary.json result_file.parent.mkdir(parentsTrue, exist_okTrue) with open(result_file, w, encodingutf-8) as f: json.dump(test_results, f, ensure_asciiFalse, indent2) print(f测试结果已保存: {result_file}) print(f总用例: {test_results[total]}, 通过: {test_results[passed]}, f失败: {test_results[failed]}, 通过率: {test_results[pass_rate]:.2f}%)test/test_machine 发送接口请求响应断言测试日志输出等import pytest import allure from core.assert_util import AssertUtil from core.response_extractor import ResponseExtractor from utils.yaml_loader import load_yaml from config.settings import Config # 加载测试数据 test_data load_yaml(Config.DATA_DIR / test_data.yaml) login_data test_data[login_data] machine_data test_data[machine_query] allure.epic(机器管理) allure.feature(认证与查询) class TestMachine: 机器相关接口测试 allure.story(登录获取Token) allure.title(测试登录接口并提取token) pytest.mark.parametrize(case, login_data) def test_login_and_get_token(self, api_client, case): 测试登录接口提取token并保存 with allure.step(f发送登录请求: {case[name]}): response api_client.post( /account/token, json{name: case[name], password: case[password]} ) # 断言 AssertUtil.assert_status_code(response, case[expected_code]) if case[expected_code] 200: with allure.step(提取并验证token): token ResponseExtractor.extract_by_path(response.json(), data) AssertUtil.assert_not_none(token, 未获取到token) assert token.startswith(case[expected_token_prefix]), \ ftoken应以{case[expected_token_prefix]}开头实际: {token[:20]}... # 保存到客户端变量供后续使用 api_client.set_var(token, token) api_client.set_header(Authorization, token) allure.attach(token, nameToken, attachment_typeallure.attachment_type.TEXT) allure.story(查询机器数量) allure.title(使用登录后的token查询机器数量) def test_get_machine_num(self, logged_client): 使用登录后的客户端获取机器数量接口关联 with allure.step(发送GET请求 /machine/getmachinenum): response logged_client.get( /machine/getmachinenum, params{projectid: 1} ) # 断言 AssertUtil.assert_status_code(response, 200) with allure.step(验证返回的data为整数): data ResponseExtractor.extract_by_path(response.json(), data) assert isinstance(data, int), fdata应为整数实际为{type(data)} allure.attach(str(data), name机器数量, attachment_typeallure.attachment_type.TEXT) print(f✓ 机器数量: {data}) allure.story(查询机器数量参数化) allure.title(使用不同projectid查询机器数量) pytest.mark.parametrize(projectid, [1, 2, 3]) def test_machine_projectid_param(self, logged_client, projectid): 参数化测试不同projectid with allure.step(f查询projectid{projectid}): response logged_client.get( /machine/getmachinenum, params{projectid: projectid} ) AssertUtil.assert_status_code(response, 200) data ResponseExtractor.extract_by_path(response.json(), data) assert data is not None allure.attach(str(data), namef项目{projectid}数量, attachment_typeallure.attachment_type.TEXT) print(f项目{projectid}数量: {data})utils/email_sender 封装测试报告发送邮件的工具类import smtplib import os from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from email.mime.application import MIMEApplication from pathlib import Path from datetime import datetime from typing import Union, List, Optional from config.settings import Config def send_email_with_attachment( subject: str, body: str, receiver: Union[str, List[str]], attachments: Optional[List[Union[str, Path]]] None ) - bool: 发送带附件的邮件支持多个收件人 :param subject: 邮件主题 :param body: 邮件正文 :param receiver: 收件人可以是单个邮箱字符串或邮箱列表 :param attachments: 附件路径列表可选 :return: 发送成功返回True否则False sender Config.EMAIL_SENDER auth_code Config.EMAIL_AUTH_CODE if not sender or not auth_code: print(邮件发送者或授权码未配置跳过邮件发送) return False # ---------- 统一收件人为列表 ---------- if isinstance(receiver, str): receiver_list [receiver] else: receiver_list list(receiver) # 确保是列表 if not receiver_list: print(收件人为空跳过发送) return False # ---------- 构建邮件 ---------- msg MIMEMultipart() msg[From] sender msg[To] , .join(receiver_list) # 邮件头显示多个收件人 msg[Subject] subject # 添加正文 msg.attach(MIMEText(body, plain, utf-8)) # 添加附件 if attachments: for file_path in attachments: file_path Path(file_path) if not file_path.exists(): print(f附件不存在跳过: {file_path}) continue try: with open(file_path, rb) as f: part MIMEApplication(f.read(), Namefile_path.name) part[Content-Disposition] fattachment; filename{file_path.name} msg.attach(part) print(f已添加附件: {file_path.name}) except Exception as e: print(f添加附件失败 {file_path}: {e}) # ---------- 发送邮件 ---------- try: # 使用 SSL 连接QQ 邮箱使用 465 端口 with smtplib.SMTP_SSL(Config.EMAIL_HOST, Config.EMAIL_PORT) as server: server.login(sender, auth_code) server.sendmail(sender, receiver_list, msg.as_string()) print(f邮件发送成功 - {, .join(receiver_list)}) return True except Exception as e: print(f邮件发送失败: {e}) return False def send_test_report( report_summary: dict, allure_report_dir: Optional[Union[str, Path]] None, zip_path: Optional[Union[str, Path]] None ) - bool: 发送测试报告邮件封装函数 :param report_summary: 测试结果摘要字典包含 total, passed, failed, skipped, pass_rate :param allure_report_dir: Allure 报告目录未使用仅保留兼容 :param zip_path: 已打包的 zip 文件路径 :return: 发送成功返回 True # 格式化正文 body Config.EMAIL_BODY_TEMPLATE.format( totalreport_summary.get(total, 0), passedreport_summary.get(passed, 0), failedreport_summary.get(failed, 0), skippedreport_summary.get(skipped, 0), pass_ratereport_summary.get(pass_rate, 0.0), timestampdatetime.now().strftime(%Y-%m-%d %H:%M:%S) ) # 构建附件列表 attachments [] if zip_path and Path(zip_path).exists(): attachments.append(zip_path) # 发送邮件收件人从 Config 读取支持列表 return send_email_with_attachment( subjectConfig.EMAIL_SUBJECT, bodybody, receiverConfig.EMAIL_RECEIVER, # 现在可以是列表或字符串 attachmentsattachments )utils/yaml_loader 封装读取YAML的工具类import yaml from pathlib import Path def load_yaml(file_path): 加载YAML文件并返回数据 with open(file_path, r, encodingutf-8) as f: return yaml.safe_load(f)run.py 测试执行的启动类执行测试并输出测试结果报告import pytest import sys import subprocess import zipfile from pathlib import Path from config.settings import Config from utils.email_sender import send_test_report import json from datetime import datetime def generate_allure_report(): 生成Allure报告并打包为zip # 1. 生成HTML报告 cmd [ allure, generate, str(Config.ALLURE_RESULTS_DIR), -o, str(Config.ALLURE_REPORT_DIR), --clean ] try: subprocess.run(cmd, checkTrue, capture_outputTrue, textTrue) print(fAllure报告生成成功: {Config.ALLURE_REPORT_DIR}) except subprocess.CalledProcessError as e: print(fAllure报告生成失败: {e.stderr}) return None except FileNotFoundError: print(未找到allure命令请确保已安装Allure并配置环境变量) return None # 2. 打包为zip zip_path Config.REPORT_DIR / fallure_report_{datetime.now().strftime(%Y%m%d_%H%M%S)}.zip with zipfile.ZipFile(zip_path, w, zipfile.ZIP_DEFLATED) as zipf: for file in Config.ALLURE_REPORT_DIR.rglob(*): if file.is_file(): arcname file.relative_to(Config.ALLURE_REPORT_DIR) zipf.write(file, arcname) print(fAllure报告已打包: {zip_path}) return zip_path def get_test_summary(): 从结果文件中读取测试摘要 summary_file Config.REPORT_DIR / test_summary.json if not summary_file.exists(): print(测试摘要文件不存在) return {total: 0, passed: 0, failed: 0, skipped: 0, pass_rate: 0} with open(summary_file, r, encodingutf-8) as f: return json.load(f) def run_tests(): 执行测试并发送报告邮件 Config.ensure_dirs() print( 开始执行测试 ) # 1. 执行pytest args [ tests/, -v, --alluredir, str(Config.ALLURE_RESULTS_DIR), --maxfail1, --tbshort, --strict-markers ] exit_code pytest.main(args) print(f测试执行完毕退出码: {exit_code}) # 2. 生成Allure报告并打包 zip_path generate_allure_report() # 3. 读取测试结果摘要 summary get_test_summary() # 4. 发送邮件 if zip_path: send_test_report(summary, Config.ALLURE_REPORT_DIR, zip_path) else: # 如果没有zip报告只发摘要 send_test_report(summary, None, None) # 5. 输出提示 print(\n * 60) if exit_code 0: print(✅ 所有测试通过) else: print(❌ 有测试失败请查看报告) print(f Allure结果: {Config.ALLURE_RESULTS_DIR}) print(f 邮件已发送至: {Config.EMAIL_RECEIVER}) print( * 60) return exit_code if __name__ __main__: sys.exit(run_tests())运行run.py测试启动类即可启动执行测试用例并发送测试报告到相关邮箱具体详情解析可查看这篇文章Python版--测试报告邮件发送配置-CSDN博客测试报告生成方法 # 方式1直接serve自动打开浏览器 allure serve reports/allure-results # 方式2先生成再打开(推荐因为可能第一种方法会将前面的测试结果的合集加起来可能存在测试用例执行条数变多) allure generate reports/allure-results -o reports/allure-report --clean allure open reports/allure-report