python pytest 进阶二
·
一、数据驱动测试(DDT)深度优化
基础参数化满足简单场景,复杂场景需要更灵活的数据源管理:
1. 从外部文件读取测试数据(JSON/Excel/YAML)
(1)从 JSON 文件读取(最常用)
# 步骤1:创建测试数据文件 test_data.json
{
"login_success": [
{"username": "admin", "pwd": "123456", "expected": "登录成功"},
{"username": "test", "pwd": "654321", "expected": "登录成功"}
],
"login_fail": [
{"username": "", "pwd": "123456", "expected": "用户名不能为空"},
{"username": "admin", "pwd": "", "expected": "密码不能为空"}
]
}
# 步骤2:编写测试用例
import pytest
import json
# 读取JSON数据的工具函数
def load_json_data(file_path):
with open(file_path, encoding="utf-8") as f:
return json.load(f)
# 从JSON读取数据并参数化
@pytest.mark.parametrize("case", load_json_data("test_data.json")["login_success"])
def test_login_success(case):
print(f"测试用例:{case}")
# 模拟登录逻辑
assert case["expected"] == "登录成功"
@pytest.mark.parametrize("case", load_json_data("test_data.json")["login_fail"])
def test_login_fail(case):
print(f"测试用例:{case}")
assert case["expected"] in ["用户名不能为空", "密码不能为空"]
(2)从 Excel 读取(适合测试人员维护)
先安装依赖:pip install openpyxl
import pytest
import openpyxl
# 读取Excel数据的工具函数
def load_excel_data(file_path, sheet_name):
wb = openpyxl.load_workbook(file_path)
ws = wb[sheet_name]
data = []
# 跳过表头,读取所有行数据
for row in ws.iter_rows(min_row=2, values_only=True):
data.append({
"username": row[0],
"pwd": row[1],
"expected": row[2]
})
return data
# 从Excel读取数据(假设Excel有3列:用户名、密码、预期结果)
@pytest.mark.parametrize("case", load_excel_data("test_data.xlsx", "login"))
def test_login_from_excel(case):
assert case["expected"] is not None
2. 动态生成测试用例(适配批量场景)
import pytest
# 动态生成测试数据(比如从数据库查询)
def generate_test_cases():
# 模拟从数据库查询100个用户数据
cases = []
for i in range(1, 101):
cases.append((f"user{i}", f"pwd{i}", f"user{i}登录成功"))
return cases
# 动态参数化
@pytest.mark.parametrize("username, pwd, expected", generate_test_cases())
def test_dynamic_cases(username, pwd, expected):
assert expected == f"{username}登录成功"
二、接口自动化测试(pytest+requests)
这是 pytest 最核心的实战场景,结合 requests 实现接口自动化:
1. 封装通用请求夹具(全局复用)
# conftest.py
import pytest
import requests
@pytest.fixture(scope="session")
def api_client():
"""封装requests会话,全局复用(自动保持cookie)"""
session = requests.Session()
# 设置全局请求头
session.headers.update({
"Content-Type": "application/json",
"User-Agent": "pytest-test/1.0"
})
# 设置超时时间
session.timeout = 10
yield session
# 后置操作:关闭会话
session.close()
@pytest.fixture(scope="session")
def base_url():
"""全局基础URL(支持多环境切换)"""
# 可通过命令行参数/环境变量控制环境
return "https://api.test.com/v1"
2. 完整接口测试案例(含 token、断言、异常处理)
import pytest
import json
# 登录接口(获取token)
@pytest.fixture(scope="module")
def login_token(api_client, base_url):
"""获取登录token,供其他接口使用"""
login_data = {
"username": "admin",
"password": "123456"
}
resp = api_client.post(
url=f"{base_url}/login",
json=login_data
)
# 基础断言
assert resp.status_code == 200
assert resp.json()["code"] == 0
# 返回token
return resp.json()["data"]["token"]
# 测试用户列表接口(依赖token)
def test_user_list(api_client, base_url, login_token):
# 添加token到请求头
api_client.headers.update({"Authorization": f"Bearer {login_token}"})
# 发送GET请求(带查询参数)
resp = api_client.get(
url=f"{base_url}/users",
params={"page": 1, "size": 10}
)
# 多维度断言(接口自动化核心)
# 1. 状态码断言
assert resp.status_code == 200
# 2. 响应体字段断言
resp_json = resp.json()
assert resp_json["code"] == 0
assert isinstance(resp_json["data"]["list"], list)
assert resp_json["data"]["total"] >= 0
# 3. 响应时间断言(性能基线)
assert resp.elapsed.total_seconds() < 1
# 测试异常场景(参数错误)
def test_user_list_error(api_client, base_url, login_token):
api_client.headers.update({"Authorization": f"Bearer {login_token}"})
# 传入非法page参数
resp = api_client.get(
url=f"{base_url}/users",
params={"page": "abc", "size": 10}
)
# 断言异常响应
assert resp.status_code == 400
assert resp.json()["msg"] == "参数错误"
3. 接口测试进阶:请求 / 响应钩子(统一处理)
# conftest.py
import pytest
import logging
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@pytest.fixture(scope="session")
def api_client_with_hooks():
session = requests.Session()
session.headers.update({"Content-Type": "application/json"})
# 请求钩子:记录所有发送的请求
def request_hook(request):
logger.info(f"请求URL:{request.url}")
logger.info(f"请求头:{json.dumps(dict(request.headers), indent=2)}")
if request.body:
logger.info(f"请求体:{request.body.decode('utf-8')}")
# 响应钩子:记录所有响应
def response_hook(response, *args, **kwargs):
logger.info(f"响应状态码:{response.status_code}")
logger.info(f"响应体:{json.dumps(response.json(), indent=2, ensure_ascii=False)}")
return response
# 注册钩子
session.hooks["request"].append(request_hook)
session.hooks["response"].append(response_hook)
yield session
session.close()
三、测试配置与环境管理(企业级规范)
1. 命令行参数扩展(自定义运行参数)
# conftest.py
def pytest_addoption(parser):
"""添加自定义命令行参数"""
parser.addoption(
"--env", # 参数名
action="store",
default="test", # 默认值
choices=["dev", "test", "prod"], # 可选值
help="指定测试环境:dev/test/prod"
)
parser.addoption(
"--rerun",
action="store",
default=0,
type=int,
help="失败重试次数"
)
# 将命令行参数转为fixture,供用例使用
@pytest.fixture(scope="session")
def env(request):
return request.config.getoption("--env")
# 用例中使用
def test_env_config(env):
print(f"当前运行环境:{env}")
# 根据环境加载不同配置
config = {
"dev": "https://api.dev.com",
"test": "https://api.test.com",
"prod": "https://api.prod.com"
}
assert config[env] is not None
运行命令:pytest -v --env prod --rerun 2 test_demo.py
2. 配置文件分层(不同环境隔离)
# 项目结构
pytest_project/
├── config/ # 配置目录
│ ├── dev_config.yaml # 开发环境配置
│ ├── test_config.yaml # 测试环境配置
│ └── prod_config.yaml # 生产环境配置
├── conftest.py
└── test_demo.py
# 读取配置的fixture(conftest.py)
import pytest
import yaml
import os
@pytest.fixture(scope="session")
def config(env):
"""根据环境加载对应配置"""
config_path = os.path.join(os.path.dirname(__file__), f"config/{env}_config.yaml")
with open(config_path, encoding="utf-8") as f:
return yaml.safe_load(f)
# 用例中使用
def test_use_config(config):
assert config["database"]["host"] is not None
assert config["api"]["timeout"] == 10
四、性能测试与并发执行
1. 用例并发执行(提升测试效率)
安装依赖:pip install pytest-xdist
# 按CPU核心数并发执行(推荐)
pytest -v -n auto test_demo.py
# 指定并发数(比如4个进程)
pytest -v -n 4 test_demo.py
2. 接口性能基准测试(简单压测)
import pytest
import time
import threading
# 简单并发请求测试(模拟10个用户同时请求)
def test_api_performance(api_client, base_url):
request_count = 10 # 并发数
results = [] # 存储每个请求的响应时间
def send_request():
start_time = time.time()
resp = api_client.get(f"{base_url}/users")
end_time = time.time()
results.append(end_time - start_time)
assert resp.status_code == 200
# 创建并启动线程
threads = []
for _ in range(request_count):
t = threading.Thread(target=send_request)
threads.append(t)
t.start()
# 等待所有线程完成
for t in threads:
t.join()
# 性能断言
avg_time = sum(results) / len(results)
max_time = max(results)
print(f"平均响应时间:{avg_time:.2f}s,最大响应时间:{max_time:.2f}s")
assert avg_time < 0.5 # 平均响应时间<500ms
assert max_time < 1 # 最大响应时间<1s
五、CI/CD 集成(自动化运行)
1. 编写 pytest 运行脚本(run_tests.py)
import pytest
import sys
if __name__ == "__main__":
# 构建运行参数
args = [
"test_case/", # 测试用例目录
"-v", # 详细输出
"-s", # 显示打印内容
"--html=report.html", # 生成HTML报告
"--self-contained-html",
"--alluredir=allure-results", # 生成Allure数据
"-n auto", # 并发执行
"--env=test" # 指定测试环境
]
# 执行pytest
exit_code = pytest.main(args)
# 退出码:0=全部通过,1=部分失败,2=用户中断,3=内部错误
sys.exit(exit_code)
2. GitLab CI/CD 配置示例(.gitlab-ci.yml)
yaml
stages:
- test
pytest_test:
stage: test
image: python:3.9
before_script:
- pip install -r requirements.txt # 安装依赖
- pip install pytest pytest-html allure-pytest pytest-xdist requests
script:
- python run_tests.py # 运行测试脚本
artifacts:
paths:
- report.html # 保存HTML报告
- allure-results/ # 保存Allure数据
expire_in: 7 days # 报告有效期7天
only:
- main # 仅main分支触发
总结
- 数据驱动:优先用 JSON/YAML/Excel 管理测试数据,实现用例与数据分离,降低维护成本;
- 接口自动化:封装
requests.Session作为夹具,结合多维度断言(状态码、响应体、响应时间),是企业级接口测试的核心; - 环境管理:通过命令行参数 + 分层配置文件,实现多环境一键切换,符合企业级规范;
- 效率优化:用
pytest-xdist并发执行用例,结合 CI/CD 实现自动化运行,提升测试效率。
更多推荐


所有评论(0)