Python+Pytest+Allure 零基础接口自动化框架完整流程（可直接落地方案）

API 接口自动化框架：结构、功能与运行时序

说明：本说明与代码均基于仓库当前的 Python/pytest/YAML 驱动实现。框架目标是：用户只需要编写 data/***_test.yaml 用例，框架自动扫描、变量替换、发送请求、参数提取关联与断言，并生成 Allure/简易 HTML 报告；CI（Jenkins）中由 scripts/ci_run.sh 调用 run.py 完成全流程。

0. 先对齐这 4 条约定

• config/config.yaml 当前环境 key 以 admin_url、sale_admin_url 为主；YAML 中 base_url_key 必须与之一致。
• 参数提取路径使用“点路径”写法（如 data.auth、data.orderNo），后续引用用 ${变量名}。
• Jenkins 使用 Pipeline script from SCM，脚本路径固定为项目根目录 Jenkinsfile。
• 登录类用例建议文件名前缀 01_，确保在排序后优先执行，避免后置接口拿不到 auth_token。

1. 代码层级结构

核心执行链路（按文件归属）：

• run.py
  • 入口脚本：初始化/清理目录、执行 pytest、生成/归档 Allure 报告或降级 HTML 报告、打开浏览器。
• conftest.py
  • pytest 会话初始化：每次执行前清空 config/extract.yaml，避免 token/关联变量脏数据。
• test_api.py
  • pytest 用例加载与执行器：
    • 扫描 data/** 下所有 *_test.yaml
    • 按文件名排序（确保登录类用例优先）
    • 逐条用例执行：构建 URL、变量替换、发送请求、extract 关联、执行断言
• common/*（框架通用能力）
  • common/config_loader.py：读取 config/config.yaml 作为 base_url 配置源
  • common/utils.py：上下文变量（${xxx}）替换、响应 extract 解析与持久化到 config/extract.yaml
  • common/request_handler.py：requests 请求封装（need_auth 注入、json/data/form-data 兼容）
  • common/assertion.py：断言规则执行（eq/ne/contains/not_contains/gt/ge/lt/le），支持 json.* 路径取值
  • common/logger.py：日志（loguru）写入 logs/run_YYYY-MM-DD.log + 控制台
• scripts/send_report_mail.py
  • Jenkins 结束后发送邮件：拼接报告路径/最新日志/构建链接（通过环境变量注入）
• scripts/ci_run.sh
  • CI wrapper：直接调用 python3 run.py

1.1 ApiFramework/ 建议目录结构

ApiFramework/
├── common/ # 核心封装层
│ ├── __init__.py
│ ├── logger.py # 日志封装
│ ├── request_util.py # 请求封装（含鉴权、关联）
│ └── yaml_util.py # YAML读写工具
├── data/ # 测试用例目录 (用户只需操作这里)
│ ├── test_login.yaml # 登录模块用例
│ └── test_order.yaml # 订单模块用例
├── logs/ # 运行日志存放目录 (自动创建)
├── report/ # 测试报告临时文件夹 (自动创建)
├── config.yaml # 全局配置 (环境地址、账号密码等)
├── conftest.py # Pytest 钩子 (处理 Allure 动态标题)
├── pytest.ini # Pytest 配置文件
├── requirements.txt # 依赖库
└── run.py # 执行入口文件

说明：仓库实际文件命名与路径略有差异，但能力模块一致；例如：

• common/request_util.py 对应仓库的 common/request_handler.py
• common/yaml_util.py 对应仓库的 common/utils.py（含 ${…} 替换与 extract.yaml 读写）
• config.yaml 对应仓库的 config/config.yaml

2. YAML 用例驱动方式（数据结构概览）

框架会递归扫描 data/** 下所有以 *_test.yaml 结尾的文件，并把文件内容当作“用例列表”加载：

单个用例的大致字段（来自 test_api.py 取值）：

• project / module / api_name / title：用于 Allure 的分组展示
• base_url_key：从 config/config.yaml 取 base_url
• request
  • method：HTTP 方法
  • url：相对路径（会与 base_url 拼接）
  • need_auth：是否需要鉴权（会从上下文注入 auth/Authorization）
  • headers
  • params / json / data：请求参数
  • form_data_json_field：当 data 是 dict 且网关要求 json 字符串时进行包装
• extract（可选）：从响应 JSON 中提取字段，写入 config/extract.yaml
• validate（可选）：断言规则列表

2.1 常见请求方式写法

• GET：通常放 params
• POST(JSON)：放 json
• POST(Form)：放 data
• 文件上传：放 files（当前框架透传给 requests，不同网关格式可能不同）

提示：若接口要求真实文件流，需确认后端的 multipart 格式要求；必要时在框架层补“路径转文件对象”封装后统一使用。

断言规则结构（来自 common/assertion.py）示例：

• `validate:
  • eq: [“status_code”, 200]
  • eq: [“json.code”, 0]
  • contains: [“json.msg”, “success”]`

2.2 鉴权与参数关联最小模板

前置接口提取：

extract:
auth_token: "data.auth"

后置接口复用（任选其一）：

request:
need_auth: true

request:
headers:
Authorization: "Bearer ${auth_token}"

3. 框架运行时序图（Sequence）

send_report_mail.pyAllure CLI (allure generate)requests.SessionAssertionHandlerRequestHandlerContextConfigLoadertest_api.pypytestconftest.py (pytest_sessionstart)run.py用户/CIsend_report_mail.pyAllure CLI (allure generate)requests.SessionAssertionHandlerRequestHandlerContextConfigLoadertest_api.pypytestconftest.py (pytest_sessionstart)run.py用户/CI#mermaid-svg-sLdXopMQdrsAJhyF{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-sLdXopMQdrsAJhyF .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-sLdXopMQdrsAJhyF .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-sLdXopMQdrsAJhyF .error-icon{fill:#552222;}#mermaid-svg-sLdXopMQdrsAJhyF .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-sLdXopMQdrsAJhyF .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-sLdXopMQdrsAJhyF .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-sLdXopMQdrsAJhyF .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-sLdXopMQdrsAJhyF .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-sLdXopMQdrsAJhyF .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-sLdXopMQdrsAJhyF .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-sLdXopMQdrsAJhyF .marker{fill:#333333;stroke:#333333;}#mermaid-svg-sLdXopMQdrsAJhyF .marker.cross{stroke:#333333;}#mermaid-svg-sLdXopMQdrsAJhyF svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-sLdXopMQdrsAJhyF p{margin:0;}#mermaid-svg-sLdXopMQdrsAJhyF .actor{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-sLdXopMQdrsAJhyF text.actor>tspan{fill:black;stroke:none;}#mermaid-svg-sLdXopMQdrsAJhyF .actor-line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);}#mermaid-svg-sLdXopMQdrsAJhyF .innerArc{stroke-width:1.5;stroke-dasharray:none;}#mermaid-svg-sLdXopMQdrsAJhyF .messageLine0{stroke-width:1.5;stroke-dasharray:none;stroke:#333;}#mermaid-svg-sLdXopMQdrsAJhyF .messageLine1{stroke-width:1.5;stroke-dasharray:2,2;stroke:#333;}#mermaid-svg-sLdXopMQdrsAJhyF #arrowhead path{fill:#333;stroke:#333;}#mermaid-svg-sLdXopMQdrsAJhyF .sequenceNumber{fill:white;}#mermaid-svg-sLdXopMQdrsAJhyF #sequencenumber{fill:#333;}#mermaid-svg-sLdXopMQdrsAJhyF #crosshead path{fill:#333;stroke:#333;}#mermaid-svg-sLdXopMQdrsAJhyF .messageText{fill:#333;stroke:none;}#mermaid-svg-sLdXopMQdrsAJhyF .labelBox{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-sLdXopMQdrsAJhyF .labelText,#mermaid-svg-sLdXopMQdrsAJhyF .labelText>tspan{fill:black;stroke:none;}#mermaid-svg-sLdXopMQdrsAJhyF .loopText,#mermaid-svg-sLdXopMQdrsAJhyF .loopText>tspan{fill:black;stroke:none;}#mermaid-svg-sLdXopMQdrsAJhyF .loopLine{stroke-width:2px;stroke-dasharray:2,2;stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);}#mermaid-svg-sLdXopMQdrsAJhyF .note{stroke:#aaaa33;fill:#fff5ad;}#mermaid-svg-sLdXopMQdrsAJhyF .noteText,#mermaid-svg-sLdXopMQdrsAJhyF .noteText>tspan{fill:black;stroke:none;}#mermaid-svg-sLdXopMQdrsAJhyF .activation0{fill:#f4f4f4;stroke:#666;}#mermaid-svg-sLdXopMQdrsAJhyF .activation1{fill:#f4f4f4;stroke:#666;}#mermaid-svg-sLdXopMQdrsAJhyF .activation2{fill:#f4f4f4;stroke:#666;}#mermaid-svg-sLdXopMQdrsAJhyF .actorPopupMenu{position:absolute;}#mermaid-svg-sLdXopMQdrsAJhyF .actorPopupMenuPanel{position:absolute;fill:#ECECFF;box-shadow:0px 8px 16px 0px rgba(0,0,0,0.2);filter:drop-shadow(3px 5px 2px rgb(0 0 0 / 0.4));}#mermaid-svg-sLdXopMQdrsAJhyF .actor-man line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-sLdXopMQdrsAJhyF .actor-man circle,#mermaid-svg-sLdXopMQdrsAJhyF line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;stroke-width:2px;}#mermaid-svg-sLdXopMQdrsAJhyF :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}alt[需要变量替换]alt[need_auth=true]alt[存在 extract]loop[每条用例 case_data]Jenkins 后置阶段通常会发送邮件python3 run.py1init_dirs() / clean_dirs()2os.system(pytest -s -v –alluredir=./report …)3pytest_sessionstart()4Context.clear_extract_file()5pytest_generate_tests(metafunc)6扫描 data/**/*_test.yaml 并排序7parametrize(case_data)8test_api_runner(case_data)9get_base_url(base_url_key)10replace_vars(request fields)11send(method, url, **kwargs)12Context.get("auth_token")13注入 headers: auth / Authorization Bearer14session.request(method, url, timeout=10, **kwargs)15response16resp17extract_and_set(resp.text, extract_rules)18set(var_name, val) 并落盘到 config/extract.yaml19AssertionHandler.run(resp, validate)20pytest 结束（allure runtime results 已写入 ./report）21allure generate ./report -o ./html_report …22inject_allure_logo() / archive_html_report()23打开 html_report/index.html (若可用)24python scripts/send_report_mail.py25

4. 每个模块的完整代码

下面按“框架模块”逐个贴出源码（保持与仓库一致）。你可以直接对照文件路径阅读。

4.1 run.py

import os
import subprocess
import shutil
import webbrowser
from pathlib import Path
from datetime import datetime

PROJECT_ROOT = Path(__file__).resolve().parent
LOG_DIR = PROJECT_ROOT / "logs"
REPORT_DIR = PROJECT_ROOT / "report"
HTML_REPORT_DIR = PROJECT_ROOT / "html_report"
ARCHIVE_REPORT_ROOT = PROJECT_ROOT / "report_history"
CUSTOM_LOGO_PATH = PROJECT_ROOT / "assets" / "allure_logo.png"
LOCAL_JAVA_HOME = PROJECT_ROOT / "tools" / "jdk-17.0.18+8-jre" / "Contents" / "Home"
LOCAL_ALLURE = PROJECT_ROOT / "tools" / "allure-2.29.0" / "bin" / "allure"

def init_dirs():
"""初始化必要的目录：logs, report, html_report"""
dirs = [LOG_DIR, REPORT_DIR, HTML_REPORT_DIR, ARCHIVE_REPORT_ROOT]
for d in dirs:
if not d.exists():
d.mkdir(parents=True, exist_ok=True)
print(f"📂 创建目录: {d}")
else:
print(f"📂 目录已存在: {d}")

def clean_dirs():
"""清理旧的测试报告数据，防止干扰"""
sub_dirs = [REPORT_DIR, HTML_REPORT_DIR]
for d in sub_dirs:
if d.exists():
shutil.rmtree(d)
d.mkdir(parents=True, exist_ok=True)
print(f"🧹 已清理旧报告: {d}")

def inject_allure_logo():
"""
可选: 自定义 Allure 顶部 logo。
将 assets/allure_logo.png 复制到报告目录并注入样式。
"""
if not CUSTOM_LOGO_PATH.exists():
print("ℹ️ 未检测到自定义 logo，使用 Allure 默认样式")
return

app_js = HTML_REPORT_DIR / "app.js"
logo_target = HTML_REPORT_DIR / "allure_logo.png"
if not app_js.exists():
print("⚠️ 未找到 app.js，跳过 logo 注入")
return

shutil.copyfile(CUSTOM_LOGO_PATH, logo_target)
js_text = app_js.read_text(encoding="utf-8")
marker = "/* custom_logo_injected */"
if marker in js_text:
return

inject_code = (
"/* custom_logo_injected */\\n"
"(function(){\\n"
" var style = document.createElement('style');\\n"
" style.innerHTML = '.side-nav__brand{background-image:url(allure_logo.png)!important;"
"background-repeat:no-repeat;background-size:140px auto;background-position:center;}'\\n"
".side-nav__brand svg{opacity:0;}';\\n"
" document.head.appendChild(style);\\n"
"})();\\n"
)
app_js.write_text(js_text + "\\n" + inject_code, encoding="utf-8")
print("✅ 已注入自定义 Allure logo")

def archive_html_report():
"""归档本次 html_report，避免历史报告被覆盖。"""
if not HTML_REPORT_DIR.exists():
return
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
archive_dir = ARCHIVE_REPORT_ROOT / f"html_report_{timestamp}"
shutil.copytree(HTML_REPORT_DIR, archive_dir)
print(f"🗂️ 已归档报告: {archive_dir}")

def get_allure_runtime():
"""优先使用系统 allure，不存在则使用项目内置 portable allure。"""
env = os.environ.copy()
try:
subprocess.check_output("allure –version", shell=True, stderr=subprocess.STDOUT)
return "allure", env
except Exception:
if LOCAL_ALLURE.exists() and LOCAL_JAVA_HOME.exists():
env["JAVA_HOME"] = str(LOCAL_JAVA_HOME)
env["PATH"] = f"{LOCAL_JAVA_HOME / 'bin'}:{env.get('PATH', '')}"
return str(LOCAL_ALLURE), env
return None, env

def run():
print("="*40)
print("🚀 自动化测试框架启动")
print("="*40)

# 1. 初始化与清理
init_dirs()
clean_dirs()

# 2. 执行 Pytest
print("\\n🧪 正在执行测试用例…")
# 同时输出 allure 临时数据 + pytest-html 报告，避免二次重复执行用例
pytest_cmd = (
"python3 -m pytest -s -v –alluredir=./report "
"–html=./html_report/report.html –self-contained-html"
)
cmd_status = os.system(pytest_cmd)
exit_code = os.WEXITSTATUS(cmd_status) if os.name != "nt" else cmd_status

print("\\n" + "="*40)
print("📊 正在生成测试报告…")

# 3. 生成报告 (智能降级策略)
allure_cmd, allure_env = get_allure_runtime()
try:
if not allure_cmd:
raise RuntimeError("allure 不可用")
subprocess.run(
f"\\"{allure_cmd}\\" generate ./report -o ./html_report –clean –single-file",
shell=True,
check=True,
env=allure_env,
)
inject_allure_logo()
archive_html_report()
print("✅ Allure 报告生成成功！正在打开…")

# 自动打开浏览器
index_path = os.path.join(os.getcwd(), "html_report", "index.html")
webbrowser.open("file:///" + index_path)

except Exception:
print("⚠️ 未检测到可用的 Allure 命令行工具，无法生成 Allure 报告。")
print("💡 提示：你可以安装 Allure 以获得更好的体验，或者查看 logs 文件夹中的日志。")
archive_html_report()
print("✅ 已生成简易 HTML 报告: ./html_report/report.html")
return exit_code

if __name__ == '__main__':
raise SystemExit(run())

4.2 test_api.py

"""
API 测试用例文件
"""
import pytest
import yaml
import os
import allure
from common.logger import logger
from common.utils import Context
from common.request_handler import RequestHandler
from common.assertion import AssertionHandler
from common.config_loader import ConfigLoader

def load_yaml_cases(file_path):
with open(file_path, "r", encoding="utf-8") as f:
return yaml.safe_load(f)

def pytest_generate_tests(metafunc):
"""
自动扫描 data 目录，并优先执行登录类用例
"""
if "case_data" in metafunc.fixturenames:
base_dir = os.path.dirname(__file__)
data_dir = os.path.join(base_dir, "data")

all_cases = []

if os.path.exists(data_dir):
# 递归获取所有 yaml 文件，支持 data 下按接口分文件夹管理
yaml_files = []
for root, _, files in os.walk(data_dir):
for file_name in files:
if file_name.endswith("_test.yaml"):
full_path = os.path.join(root, file_name)
rel_path = os.path.relpath(full_path, data_dir)
yaml_files.append(rel_path)

# ⭐ 关键：排序，确保 01_login… 排在前面
yaml_files.sort()

for yaml_file in yaml_files:
file_path = os.path.join(data_dir, yaml_file)
cases = load_yaml_cases(file_path)
if cases:
for case in cases:
case['_source_file'] = yaml_file
all_cases.append(case)

metafunc.parametrize(
"case_data",
all_cases,
ids=[f"{c.get('title')}_{c.get('_source_file')}" for c in all_cases],
)

def test_api_runner(case_data):
# 在报告中按 project/module/api_name/title 四级展示
project = case_data.get("project", "未分组项目")
module = case_data.get("module", "未分组模块")
api_name = case_data.get("api_name", "未分组接口")
title = case_data.get("title", "未命名用例")

# Behaviors 视图分组：project -> module -> api_name
allure.dynamic.epic(project)
allure.dynamic.feature(module)
allure.dynamic.story(api_name)
# Suites 视图分组：project -> module -> api_name
allure.dynamic.parent_suite(project)
allure.dynamic.suite(module)
allure.dynamic.sub_suite(api_name)
# 用例标题
allure.dynamic.title(title)

# 1. 获取 URL
base_url_key = case_data.get("base_url_key", "base_url")
base_url = ConfigLoader.get_base_url(base_url_key)

# 2. 变量替换
if 'json' in case_data['request']:
case_data['request']['json'] = Context.replace_vars(case_data['request']['json'])
if 'params' in case_data['request']:
case_data['request']['params'] = Context.replace_vars(case_data['request']['params'])
if 'data' in case_data['request']:
case_data['request']['data'] = Context.replace_vars(case_data['request']['data'])
if 'headers' in case_data['request']:
case_data['request']['headers'] = Context.replace_vars(case_data['request']['headers'])

# 3. 拼接与发送
url = base_url + case_data["request"]["url"]
method = case_data["request"]["method"]

kwargs = case_data['request'].copy()
kwargs.pop('url')
kwargs.pop('method')

resp = RequestHandler.send(method, url, **kwargs)
assert resp is not None, f"接口请求失败: {method} {url}"

# 4. 关联
if 'extract' in case_data:
Context.extract_and_set(resp.text, case_data['extract'])

# 5. 断言
AssertionHandler.run(resp, case_data.get('validate'))

4.3 conftest.py

"""
pytest 全局初始化:
1. 清理接口关联数据文件，避免旧 token 干扰
2. 让每次执行都从干净上下文开始
"""
from common.utils import Context

def pytest_sessionstart(session):
Context.clear_extract_file()

4.4 common/config_loader.py

import os
import yaml

class ConfigLoader:
_config = None

@classmethod
def get_config(cls):
if cls._config is None:
base_dir = os.path.dirname(os.path.dirname(__file__))
config_path = os.path.join(base_dir, "config", "config.yaml")
with open(config_path, "r", encoding="utf-8") as f:
cls._config = yaml.safe_load(f)
return cls._config

@classmethod
def get_base_url(cls, env="base_url"):
return cls.get_config().get(env)

4.5 common/request_handler.py

import requests
import json
from common.logger import logger
from common.utils import Context

class RequestHandler:
session = requests.Session()

@staticmethod
def send(method, url, **kwargs):
method = str(method).upper()

# 1. 自动鉴权处理
if kwargs.pop("need_auth", False):
# 获取登录接口提取的 auth_token
token = Context.get("auth_token")
if token:
headers = kwargs.get("headers", {})
# 默认注入 auth 头，也兼容 Authorization。
headers.setdefault("auth", str(token))
headers.setdefault("Authorization", f"Bearer {token}")
kwargs["headers"] = headers
logger.info(f"🔍 自动注入 Token: {token}")
else:
logger.warning("⚠️ 需要鉴权但未找到 Token，请检查登录接口是否执行成功")

# 2. 统一处理 JSON/Form 请求体
if "json" in kwargs and "data" in kwargs:
logger.warning("⚠️ 同时存在 json 和 data，将优先使用 json，忽略 data")
kwargs.pop("data", None)

# 3. 某些网关要求 form-data 的 data 字段为 JSON 字符串:
# data={"k":"v"} -> data={"data":"{\\"k\\":\\"v\\"}"}
if kwargs.pop("form_data_json_field", False) and isinstance(kwargs.get("data"), dict):
kwargs["data"] = {
"data": json.dumps(kwargs["data"], ensure_ascii=False, separators=(",", ":"))
}

# 4. 发送请求
logger.info(f"👉 请求: [{method}] {url}")
# 仅记录非空参数，避免日志过长
params_log = (
kwargs.get("json")
or kwargs.get("data")
or kwargs.get("params")
or kwargs.get("files")
or "无参数"
)
logger.info(f"👉 参数: {params_log}")

try:
resp = RequestHandler.session.request(method, url, timeout=10, **kwargs)
logger.info(f"👈 响应: [{resp.status_code}] {resp.text[:150]}…")
return resp
except Exception as e:
logger.error(f"❌ 请求异常: {e}")
return None

4.6 common/assertion.py

import json
from common.logger import logger

class AssertionHandler:
@staticmethod
def _get_value(data, key_path):
if isinstance(data, str):
try: data = json.loads(data)
except: return None
keys = key_path.split('.')
current = data
for k in keys:
try: current = current[k]
except: return None
return current

@classmethod
def run(cls, resp, validate_rules):
if not validate_rules: return

try: resp_json = resp.json()
except: resp_json = {}

for rule in validate_rules:
for op, val_list in rule.items():
check_item, expected_val = val_list[0], val_list[1]

actual_val = None
if check_item == "status_code": actual_val = resp.status_code
elif check_item.startswith("json."): actual_val = cls._get_value(resp_json, check_item.replace("json.", ""))

logger.info(f"🔍 断言: {check_item} {op} {expected_val} (实际: {actual_val})")

if op == "eq": assert actual_val == expected_val, f"期望 {expected_val}, 实际 {actual_val}"
elif op == "ne": assert actual_val != expected_val
elif op == "contains": assert expected_val in str(actual_val)
elif op == "not_contains": assert expected_val not in str(actual_val)
elif op == "gt": assert actual_val > expected_val
elif op == "ge": assert actual_val >= expected_val
elif op == "lt": assert actual_val < expected_val
elif op == "le": assert actual_val <= expected_val
else: raise AssertionError(f"不支持的断言操作符: {op}")

4.7 common/utils.py

import json
import re
import os
import yaml
from common.logger import logger

class Context:
_variables = {}
_base_dir = os.path.dirname(os.path.dirname(__file__))
_extract_file = os.path.join(_base_dir, "config", "extract.yaml")

@classmethod
def _ensure_extract_file(cls):
if not os.path.exists(cls._extract_file):
with open(cls._extract_file, "w", encoding="utf-8") as f:
yaml.safe_dump({}, f, allow_unicode=True, sort_keys=False)

@classmethod
def clear_extract_file(cls):
cls._variables = {}
cls._ensure_extract_file()
with open(cls._extract_file, "w", encoding="utf-8") as f:
yaml.safe_dump({}, f, allow_unicode=True, sort_keys=False)
logger.info("🧹 已清空接口关联数据文件 config/extract.yaml")

@classmethod
def _load_extract_file(cls):
cls._ensure_extract_file()
with open(cls._extract_file, "r", encoding="utf-8") as f:
data = yaml.safe_load(f) or {}
if isinstance(data, dict):
cls._variables.update(data)

@classmethod
def _save_extract_file(cls):
cls._ensure_extract_file()
with open(cls._extract_file, "w", encoding="utf-8") as f:
yaml.safe_dump(cls._variables, f, allow_unicode=True, sort_keys=False)

@classmethod
def set(cls, key, value):
if not cls._variables:
cls._load_extract_file()
cls._variables[key] = value
cls._save_extract_file()
logger.info(f"📝 提取变量: {key} = {value}")

@classmethod
def get(cls, key):
if not cls._variables:
cls._load_extract_file()
return cls._variables.get(key)

@classmethod
def replace_vars(cls, data):
if isinstance(data, dict):
return {k: cls.replace_vars(v) for k, v in data.items()}
elif isinstance(data, list):
return [cls.replace_vars(item) for item in data]
elif isinstance(data, str):
pattern = r'\\$\\{(.+?)\\}'
matches = re.findall(pattern, data)
for match in matches:
val = cls.get(match)
if val is not None:
data = data.replace(f"${{{match}}}", str(val))
return data
else:
return data

@classmethod
def extract_and_set(cls, response_text, extract_rules):
if not extract_rules: return
try:
json_data = json.loads(response_text)
except:
logger.error("❌ 响应不是 JSON，无法提取变量")
return

# 检查是否登录成功（根据 code 字段判断）
code = json_data.get('code', –1)
if code != 0:
logger.warning("⚠️ 登录失败，跳过变量提取")
return

for var_name, json_path in extract_rules.items():
keys = json_path.split('.')
val = json_data
try:
for k in keys:
val = val[k]
cls.set(var_name, val)
except Exception as e:
logger.error(f"❌ 提取变量 {var_name} 失败: {e}")

4.8 common/logger.py

import os
from loguru import logger

log_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), "logs")
if not os.path.exists(log_path):
os.makedirs(log_path)

log_file = os.path.join(log_path, "run_{time:YYYY-MM-DD}.log")

logger.remove()
logger.add(log_file, format="{time:YYYY-MM-DD HH:mm:ss} | {level} | {message}", level="INFO", rotation="1 day", encoding="utf-8")
logger.add(lambda msg: print(msg, end=""), format="{time:HH:mm:ss} | {level: <8} | {message}", level="INFO")

4.9 scripts/send_report_mail.py

import os
import smtplib
from email.mime.text import MIMEText
from email.header import Header
from email.utils import formataddr
from pathlib import Path
from datetime import datetime

def _env(name: str, default: str = "") –> str:
value = os.getenv(name, default).strip()
return value

def build_mail_content() –> str:
root = Path(__file__).resolve().parent.parent
workspace = _env("WORKSPACE")
# Prefer Jenkins workspace so mail displays server-side absolute path.
server_root = Path(workspace) if workspace else root

report_path = server_root / "html_report" / "index.html"
if not report_path.exists():
report_path = server_root / "html_report" / "report.html"

latest_log = sorted((server_root / "logs").glob("*.log"))[–1] if list((server_root / "logs").glob("*.log")) else None
build_url = _env("BUILD_URL")
html_report_url = f"{build_url}HTML_20Report/" if build_url else ""

lines = [
"接口自动化测试已执行完成。",
f"执行时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
"",
f"报告路径: {report_path}",
]
if latest_log:
lines.append(f"日志路径: {latest_log}")
if build_url:
lines.append(f"Jenkins构建链接: {build_url}")
if html_report_url:
lines.append(f"Jenkins报告入口: {html_report_url}")

lines.extend(
[
"",
"说明:",
"1) 若 Jenkins 配置了 HTML Publisher，可直接在构建页面查看报告。",
"2) 若 Jenkins 报告入口无法打开，请检查 HTML Publisher 的 Report Name 是否为 HTML Report。",
"3) 若报告为空，请检查服务器工作空间中的 html_report 目录。",
]
)
return "\\n".join(lines)

def send_mail() –> None:
host = _env("SMTP_HOST")
port = int(_env("SMTP_PORT", "465"))
user = _env("SMTP_USER")
password = _env("SMTP_PASS")
sender = _env("MAIL_SENDER", user)
receivers = _env("MAIL_RECEIVERS")

if not host or not user or not password or not receivers:
print("[MAIL] Skip: SMTP env is not fully configured.")
return

receiver_list = [r.strip() for r in receivers.split(",") if r.strip()]
if not receiver_list:
print("[MAIL] Skip: MAIL_RECEIVERS is empty.")
return
if "@" not in sender:
print(f"[MAIL] Skip: MAIL_SENDER is invalid: {sender!r}")
return

subject = _env("MAIL_SUBJECT", "接口自动化测试报告")
content = build_mail_content()

message = MIMEText(content, "plain", "utf-8")
# RFC-compliant headers: keep a valid mailbox in angle brackets
message["From"] = formataddr((str(Header("API自动化", "utf-8")), sender))
message["To"] = ", ".join(receiver_list)
message["Subject"] = Header(subject, "utf-8")

with smtplib.SMTP_SSL(host, port, timeout=30) as server:
server.login(user, password)
server.sendmail(sender, receiver_list, message.as_string())
print("[MAIL] Sent successfully.")

if __name__ == "__main__":
send_mail()

4.10 scripts/ci_run.sh

#!/usr/bin/env bash
set -euo pipefail

ROOT_DIR="$(cd "$(dirname "$0")/.." && pwd)"
cd "$ROOT_DIR"

echo "[CI] Start API test run at $(date '+%F %T')"
python3 run.py
echo "[CI] Finished API test run at $(date '+%F %T')"

4.11 config/config.yaml

# 后台管理系统域名
admin_url: "https://admin.example.com"
# 销售系统后台域名
sale_admin_url: "https://sale-admin.example.com"

4.12 config/extract.yaml

auth_token: "EXAMPLE_AUTH_TOKEN"

注：conftest.py 在每次会话开始时会清空 config/extract.yaml，然后由登录接口用 extract 规则把实际 token 写回。

4.13 pytest.ini

[pytest]
addopts = -s -v –alluredir=./report
testpaths = ./
python_files = test_*.py *_test.py conftest.py
python_classes = Test*
python_functions = test_*

4.14 requirements.txt

pytest
requests
PyYAML
allure-pytest
loguru
pytest-html

4.15 ci/env.example

## Jenkins 环境变量示例（在 Jenkins Job 或凭据里配置）

# Git
GIT_REPO_URL=https://github.com/your-org/Api_Framework.git
GIT_BRANCH=main

# 邮件配置（示例）
SMTP_HOST=smtp.example.com
SMTP_PORT=465
SMTP_USER=your_mail@example.com
SMTP_PASS=your_smtp_authorization_code
MAIL_SENDER=your_mail@example.com
MAIL_RECEIVERS=qa1@example.com,qa2@example.com
MAIL_SUBJECT=销售平台接口自动化测试报告

5. 运行入口与 CI 对接要点（简要）

• 本地/CI 都以 python3 run.py 为统一入口。
• 每次执行都会清空 config/extract.yaml，保证登录提取的关联变量只来自本次 run 的执行结果。
• test_api.py 通过 *_test.yaml 的文件名排序保证登录用例在前（例如 01_*），从而让后续 need_auth: true 用例能从 Context 中拿到 auth_token。
• Jenkins 运行时调用 scripts/ci_run.sh，后置阶段再调用 scripts/send_report_mail.py 根据环境变量发送报告邮件。

6. 从 0开始 到 Jenkins 持续集成

7. 常见卡点快速对照

• Jenkinsfile not found：仓库根目录缺文件，或 Job 的 Script Path 不是 Jenkinsfile
• ${auth_token} 未替换：登录用例未先执行 / extract 路径错误 / 变量名不一致
• base_url_key 取不到地址：YAML key 与 config/config.yaml 不一致
• 本地能过 Jenkins 失败：优先对照 Jenkins Console、归档日志、工作空间 config/extract.yaml

具体执行测试报告结果