脚本怎样批量执行接口测试用例

本文目录导读：

1. Postman + Newman（推荐）
2. Python + requests + pytest
3. JMeter 命令行执行
4. Shell脚本批量执行
5. 使用配置文件管理批量测试
6. 最佳实践建议
7. 使用建议

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

使用建议

1. 根据团队技术栈选择：

   • 测试人员 → Postman + Newman
   • 开发人员 → Python + pytest
   • 性能测试 → JMeter

2. 建议集成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

3. 错误重试机制：

   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开始，逐步迁移到更自动化的框架。