本文目录导读：
Postman + Newman（推荐）
准备测试用例
// 创建测试集合，包含多个用例
pm.test("状态码检查", function () {
pm.response.to.have.status(200);
});
pm.test("响应时间检查", function () {
pm.expect(pm.response.responseTime).to.be.below(2000);
});
批量执行命令
# 安装Newman npm install -g newman # 批量执行所有用例 newman run 测试集合.postman_collection.json \ -e 环境变量.json \ -d 测试数据.csv \ --reporters cli,json,junit \ --reporter-json-export report.json \ --reporter-junit-export report.xml
Python + requests + pytest
测试用例文件 (test_api.py)
import pytest
import requests
import json
class TestAPI:
# 测试数据
test_data = [
{"url": "http://api.example.com/users", "method": "GET", "expected": 200},
{"url": "http://api.example.com/users", "method": "POST",
"data": {"name": "test", "email": "test@test.com"}, "expected": 201},
{"url": "http://api.example.com/users/1", "method": "GET", "expected": 200}
]
@pytest.mark.parametrize("case", test_data)
def test_api(self, case):
if case["method"] == "GET":
response = requests.get(case["url"])
elif case["method"] == "POST":
response = requests.post(case["url"], json=case.get("data", {}))
assert response.status_code == case["expected"]
批量执行脚本 (run_tests.sh)
#!/bin/bash # 安装依赖 pip install pytest requests # 批量执行所有测试用例 pytest test_api.py -v --html=report.html # 生成JUnit格式报告 pytest test_api.py --junitxml=report.xml # 并行执行 pytest test_api.py -n 4 --dist loadscope
JMeter 命令行执行
创建测试计划后，使用命令行执行
# 执行单个测试计划
jmeter -n -t 测试计划.jmx -l 结果.jtl -e -o 报告目录/
# 批量执行多个测试计划
for file in /path/to/test/*.jmx; do
jmeter -n -t "$file" -l "${file%.jmx}.jtl" -e -o "${file%.jmx}_report/"
done
Shell脚本批量执行
创建批量执行脚本 (batch_test.sh)
#!/bin/bash
# 测试用例配置
BASE_URL="http://api.example.com"
REPORT_DIR="./reports/$(date +%Y%m%d_%H%M%S)"
# 创建报告目录
mkdir -p $REPORT_DIR
# 读取测试用例文件
while IFS=',' read -r id method endpoint expected_status body_file
do
echo "执行测试用例: $id"
# 构建请求
url="${BASE_URL}${endpoint}"
case $method in
GET)
response=$(curl -s -o /dev/null -w "%{http_code}" $url)
;;
POST)
if [ -f "$body_file" ]; then
response=$(curl -s -o /dev/null -w "%{http_code}" -X POST -d @"$body_file" $url)
fi
;;
esac
# 检查结果
if [ "$response" = "$expected_status" ]; then
echo "PASS: $id"
echo "$id,PASS" >> "$REPORT_DIR/results.csv"
else
echo "FAIL: $id (Expected: $expected_status, Got: $response)"
echo "$id,FAIL" >> "$REPORT_DIR/results.csv"
fi
done < test_cases.csv
使用配置文件管理批量测试
test_config.json
{
"base_url": "http://api.example.com",
"headers": {
"Content-Type": "application/json",
"Authorization": "Bearer token123"
},
"cases": [
{
"id": "TC001",
"name": "获取用户列表",
"method": "GET",
"endpoint": "/users",
"expected_status": 200,
"timeout": 5000
},
{
"id": "TC002",
"name": "创建新用户",
"method": "POST",
"endpoint": "/users",
"body": {"name": "test", "email": "test@test.com"},
"expected_status": 201,
"timeout": 5000
}
]
}
批量执行脚本 (execute_from_config.py)
import json
import requests
import time
class BatchAPITester:
def __init__(self, config_file):
with open(config_file, 'r') as f:
self.config = json.load(f)
self.results = []
def execute_all(self):
for case in self.config['cases']:
start_time = time.time()
try:
url = self.config['base_url'] + case['endpoint']
if case['method'] == 'GET':
response = requests.get(url, headers=self.config['headers'])
elif case['method'] == 'POST':
response = requests.post(
url,
json=case.get('body', {}),
headers=self.config['headers']
)
elapsed_time = (time.time() - start_time) * 1000
# 验证结果
passed = (response.status_code == case['expected_status'] and
elapsed_time < case.get('timeout', 5000))
self.results.append({
'id': case['id'],
'name': case['name'],
'status': 'PASS' if passed else 'FAIL',
'response_time': f"{elapsed_time:.2f}ms",
'status_code': response.status_code
})
except Exception as e:
self.results.append({
'id': case['id'],
'name': case['name'],
'status': 'ERROR',
'error': str(e)
})
def generate_report(self):
# 生成报告
pass_count = sum(1 for r in self.results if r['status'] == 'PASS')
total_count = len(self.results)
print(f"\n=== 测试报告 ===")
print(f"总用例数: {total_count}")
print(f"通过: {pass_count}")
print(f"失败: {total_count - pass_count}")
print(f"通过率: {pass_count/total_count*100:.2f}%")
for result in self.results:
print(f"{result['status']}: {result['id']} - {result.get('name', '')}")
if __name__ == "__main__":
tester = BatchAPITester('test_config.json')
tester.execute_all()
tester.generate_report()
最佳实践建议
项目结构
api_tests/
├── config/
│ ├── dev_config.json
│ └── prod_config.json
├── test_data/
│ └── test_cases.csv
├── tests/
│ ├── test_user_api.py
│ └── test_order_api.py
├── reports/
│ └── (自动生成的报告)
├── utils/
│ └── api_helper.py
├── run_all_tests.sh
└── requirements.txt并行执行方案
# 使用多线程并行执行测试
from concurrent.futures import ThreadPoolExecutor
import threading
def execute_test_in_thread(test_case):
# 执行单个测试用例
pass
def parallel_execute(test_cases, max_workers=4):
with ThreadPoolExecutor(max_workers=max_workers) as executor:
results = list(executor.map(execute_test_in_thread, test_cases))
return results
使用建议
-
根据团队技术栈选择：
- 测试人员 → Postman + Newman
- 开发人员 → Python + pytest
- 性能测试 → JMeter
-
建议集成CI/CD：
# GitHub Actions 示例 name: API Tests on: [push] jobs: test: runs-on: ubuntu-latest steps: - uses: actions/checkout@v2 - name: Run API tests run: | pip install -r requirements.txt pytest tests/ -v --junitxml=report.xml -
错误重试机制：
import time from functools import wraps
def retry(max_attempts=3, delay=1): def decorator(func): @wraps(func) def wrapper(*args, *kwargs): for attempt in range(max_attempts): try: return func(args, **kwargs) except Exception as e: if attempt == max_attempts - 1: raise time.sleep(delay) return None return wrapper return decorator
选择哪种方案取决于你的具体需求和团队经验，建议从简单的Postman+Newman开始，逐步迁移到更自动化的框架。
