# 通过YAML文件实现Pytest接口自动化测试用例的配置与管理
## 1. 整体架构设计
基于Pytest和YAML的接口自动化测试框架采用分层架构设计,将测试用例、配置信息与执行逻辑分离,实现高度的可维护性和扩展性。核心组件包括:
| 组件模块 | 功能职责 | 技术实现 |
|---------|---------|---------|
| YAML用例管理 | 存储测试用例、配置、变量 | PyYAML库解析 |
| Pytest执行引擎 | 用例收集、执行、报告 | pytest框架+自定义插件 |
| HTTP请求封装 | 发送接口请求、处理响应 | requests库二次封装 |
| 数据驱动机制 | 参数化测试、变量替换 | pytest parametrize+Template |
| 断言验证系统 | 结果校验、多断言处理 | pytest-assume插件 |
| 报告生成模块 | 测试报告展示 | allure-pytest |
## 2. YAML文件结构规范
### 2.1 完整的YAML测试用例文件示例
```yaml
config: # 全局配置部分
baseurl: "https://www.zhixue.com"
timeout: 30.0
headers:
Accept: "application/json, text/javascript, */*; q=0.01"
Accept-Encoding: "gzip, deflate, br"
Accept-Language: "zh-CN,zh;q=0.9"
Connection: "keep-alive"
User-Agent: "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
X-Requested-With: "XMLHttpRequest"
variable: # 全局变量定义
none: "none"
tests: # 测试用例集合
test_login:
description: "用户登录接口"
method: "post"
route: "/weakPwdLogin/?from=web_login"
RequestData:
data:
loginName: "18291900215"
password: "dd636482aca022"
code: ""
description: "encrypt"
Validate:
expectcode: 200
resultcheck: '"result":"success"'
regularcheck: '[\\d]{16}'
Extract:
- "data"
test_login_verify:
description: "登录验证接口"
method: "post"
route: "/loginSuccess/"
RequestData:
data:
userId: "${data}" # 使用提取的变量
Validate:
expectcode: 200
resultcheck: '"result":"success"'
```
### 2.2 YAML结构详解
**config部分**:定义全局请求配置,包括基础URL、超时时间、请求头等,这些配置会被所有测试用例继承[ref_1]。
**variable部分**:预定义全局变量,可在多个测试用例中复用,支持动态替换[ref_1]。
**tests部分**:具体的测试用例定义,每个用例包含:
- `description`:用例描述信息
- `method`:HTTP请求方法
- `route`:接口路径
- `RequestData`:请求参数
- `Validate`:断言验证规则
- `Extract`:响应数据提取规则
## 3. Pytest集成实现
### 3.1 自定义文件收集器
通过重写`pytest_collect_file`方法,让Pytest能够识别YAML文件作为测试用例:
```python
# conftest.py
import yaml
import pytest
from common.cache import cache
from common.exceptions import YamlException
def pytest_collect_file(parent, path):
"""识别YAML格式的测试文件"""
if path.ext in (".yaml", ".yml") and path.basename.startswith("test"):
return YamlFile.from_parent(parent, fspath=path)
class YamlFile(pytest.File):
def collect(self):
"""解析YAML文件并生成测试项"""
raw = yaml.safe_load(self.fspath.open(encoding='utf-8'))
# 检查是否存在测试用例
if not any(k.startswith('test') for k in raw.keys()):
raise YamlException("{}yaml non test found".format(self.fspath))
# 加载全局变量到缓存
if variable := raw.get('variable'):
for k, v in variable.items():
cache.set(k, v)
# 加载配置信息到缓存
if config := raw.get('config'):
for k, v in config.items():
cache.set(k, v)
# 生成测试用例
if tests := raw.get('tests'):
for name, spec in tests.items():
yield YamlTest.from_parent(self,
name=spec.get('description') or name,
spec=spec)
```
### 3.2 测试项执行类
```python
class YamlTest(pytest.Item):
def __init__(self, name, parent, spec):
super(YamlTest, self).__init__(name, parent)
self.spec = spec
self.request = HttpRequest(exception=(RequestException, Exception))
def runtest(self):
"""执行测试用例"""
r = self.request.send_request(**self.spec)
self.response_handle(r,
self.spec.get('Validate'),
self.spec.get('Extract'))
def response_handle(self, r, validate, extract):
"""处理响应结果"""
if validate:
check_results(r, validate) # 结果验证
if extract:
get_result(r, extract) # 数据提取
def repr_failure(self, excinfo):
"""失败异常处理"""
logger.critical(excinfo.value)
logger.critical(excinfo.traceback[-6:-1])
def reportinfo(self):
return self.fspath, 0, f"usecase: {self.name}"
```
## 4. 核心组件实现
### 4.1 HTTP请求封装
```python
# common/request.py
import urllib3
from requests import Session
from common.cache import cache
from utils.logger import logger
from common.regular import findalls, sub_var
urllib3.disable_warnings()
class HttpRequest(Session):
"""requests库二次封装"""
def __init__(self, *args, **kwargs):
super(HttpRequest, self).__init__()
self.exception = kwargs.get("exception", Exception)
def send_request(self, **kwargs):
"""发送HTTP请求"""
try:
# 变量替换处理
kwargs = self.replace_variables(kwargs)
method = kwargs.get('method', 'GET').upper()
url = cache.get('baseurl') + kwargs.get('route')
logger.info(f"Request Url: {url}")
logger.info(f"Request Method: {method}")
# 合并请求数据
request_data = HttpRequest.mergedict(
kwargs.get('RequestData', {}),
headers=cache.get('headers'),
timeout=cache.get('timeout')
)
# 分发请求
response = self.dispatch(method, url, **request_data)
logger.info(f"Request Result: {response}{response.text}")
return response
except self.exception as e:
logger.exception(format(e))
raise e
def replace_variables(self, kwargs):
"""替换请求参数中的变量"""
import json
kwargs_str = json.dumps(kwargs)
variables = findalls(kwargs_str)
if variables:
replaced_str = sub_var(variables, kwargs_str)
kwargs = json.loads(replaced_str)
return kwargs
def dispatch(self, method, *args, **kwargs):
"""请求方法分发"""
handler = getattr(self, method.lower())
return handler(*args, **kwargs)
@staticmethod
def mergedict(args, **kwargs):
"""字典合并"""
for k, v in args.items():
if k in kwargs:
kwargs[k] = {**args[k], **kwargs.pop(k)}
args.update(kwargs)
return args
```
### 4.2 响应结果处理
```python
# common/result.py
import re
import pytest
from common.cache import cache
from common.regular import get_var
from utils.logger import logger
def check_results(r, validate):
"""验证响应结果"""
expectcode = validate.get('expectcode')
resultcheck = validate.get('resultcheck')
regularcheck = validate.get('regularcheck')
# 状态码断言
if expectcode:
pytest.assume(expectcode == r.status_code)
# 响应内容断言
if resultcheck:
pytest.assume(resultcheck in r.text)
# 正则表达式断言
if regularcheck:
pytest.assume(re.findall(regularcheck, r.text))
def get_result(r, extract):
"""从响应中提取数据"""
for key in extract:
value = get_var(key, r.text)
logger.debug(f"正则提取结果值:{key}={value}")
cache.set(key, value) # 存入缓存供后续用例使用
pytest.assume(key in cache) # 验证提取成功
```
### 4.3 变量处理工具
```python
# common/regular.py
import re
from string import Template
from common.cache import cache
from common.json import is_json_str
from utils.logger import logger
def get_var(key, raw_str):
"""从字符串中提取变量值"""
if is_json_str(raw_str):
return re.compile(r'\"%s\":\"(.*?)\"' % key).findall(raw_str)[0]
return re.compile(r'%s' % key).findall(raw_str)[0]
def findalls(string):
"""查找所有需要替换的变量"""
key = re.compile(r"\${(.*?)\}").findall(string)
res = {k: cache.get(k) for k in key}
logger.debug(f"需要替换的变量:{res}")
return res
def sub_var(keys, string):
"""替换变量"""
s = Template(string)
res = s.safe_substitute(keys)
logger.debug(f"替换结果:{res}")
return res
```
## 5. 数据驱动与参数化
通过YAML文件实现数据驱动测试,支持多种测试场景:
### 5.1 多环境配置支持
```yaml
# env.yaml
dev:
baseurl: "https://dev-api.example.com"
timeout: 10
test:
baseurl: "https://test-api.example.com"
timeout: 20
prod:
baseurl: "https://api.example.com"
timeout: 30
```
### 5.2 参数化测试用例
```yaml
tests:
test_user_login:
description: "用户登录参数化测试"
method: "post"
route: "/login"
parameters:
- {username: "user1", password: "pass1", expected: "success"}
- {username: "user2", password: "pass2", expected: "success"}
- {username: "invalid", password: "wrong", expected: "fail"}
RequestData:
data:
username: "${username}"
password: "${password}"
Validate:
resultcheck: '"status":"${expected}"'
```
## 6. 最佳实践建议
### 6.1 目录结构规范
```
project/
├── common/ # 公共组件
│ ├── cache.py # 缓存管理
│ ├── request.py # 请求封装
│ ├── result.py # 结果处理
│ └── regular.py # 正则工具
├── tests/ # 测试用例
│ ├── test_api.yaml
│ └── test_user.yaml
├── conftest.py # Pytest配置
├── pytest.ini # Pytest配置
└── requirements.txt # 依赖包
```
### 6.2 依赖配置
```txt
# requirements.txt
pytest==6.2.5
PyYAML==5.4.1
requests==2.26.0
allure-pytest==2.9.43
pytest-assume==2.4.3
pytest-html==3.1.1
```
### 6.3 执行命令
```bash
# 运行所有测试
pytest -v
# 生成Allure报告
pytest --alluredir=./allure-results
# 指定YAML文件运行
pytest tests/test_api.yaml -v
```
通过以上完整的实现方案,可以构建一个基于YAML配置的、高度可维护的Pytest接口自动化测试框架。这种架构不仅提高了测试用例的可读性,还实现了测试数据与测试逻辑的彻底分离,大大提升了自动化测试的效率和维护性[ref_1][ref_2]。