脚本怎样批量校验接口参数

本文目录导读：

1. Python + requests + JSON Schema
2. Postman + Newman (使用集合批量测试)
3. 使用 YAML 配置文件的通用校验脚本
4. 使用 pytest + pytest-xdist 并行测试
5. 使用 Jenkins Pipeline 自动化
6. 推荐使用方案

Python + requests + JSON Schema

import requests
import json
from jsonschema import validate, ValidationError
import pandas as pd
from typing import Dict, List
class APIParameterValidator:
    def __init__(self, base_url: str):
        self.base_url = base_url
    def generate_test_cases(self, api_spec: Dict) -> List[Dict]:
        """
        根据API规范生成测试用例
        """
        test_cases = []
        # 必填参数测试
        for param, spec in api_spec.get('required_params', {}).items():
            test_cases.append({
                'name': f'缺少必填参数: {param}',
                'params': {k: v for k, v in api_spec.get('default_params', {}).items() 
                          if k != param},
                'expected_status': 400
            })
        # 参数类型测试
        for param, spec in api_spec.get('params', {}).items():
            if spec.get('type') == 'integer':
                test_cases.append({
                    'name': f'参数类型错误: {param}',
                    'params': {**api_spec.get('default_params', {}), 
                              param: 'not_a_number'},
                    'expected_status': 400
                })
        # 边界值测试
        for param, spec in api_spec.get('params', {}).items():
            if 'min' in spec and 'max' in spec:
                test_cases.extend([
                    {
                        'name': f'参数小于最小值: {param}',
                        'params': {**api_spec.get('default_params', {}), 
                                  param: spec['min'] - 1},
                        'expected_status': 400
                    },
                    {
                        'name': f'参数大于最大值: {param}',
                        'params': {**api_spec.get('default_params', {}), 
                                  param: spec['max'] + 1},
                        'expected_status': 400
                    }
                ])
        return test_cases
    def validate_response(self, response: requests.Response, 
                         expected_status: int) -> Dict:
        """验证响应"""
        result = {
            'status': 'pass' if response.status_code == expected_status else 'fail',
            'status_code': response.status_code,
            'expected_status': expected_status,
            'response_time': response.elapsed.total_seconds(),
            'body': response.text[:500] if response.text else ''
        }
        return result
    def run_batch_tests(self, endpoint: str, api_spec: Dict) -> pd.DataFrame:
        """批量运行测试"""
        test_cases = self.generate_test_cases(api_spec)
        results = []
        for test_case in test_cases:
            try:
                response = requests.post(
                    f"{self.base_url}{endpoint}",
                    params=test_case['params'],
                    timeout=10
                )
                result = self.validate_response(response, 
                                               test_case['expected_status'])
                result['test_name'] = test_case['name']
                result['params'] = test_case['params']
                results.append(result)
            except Exception as e:
                results.append({
                    'test_name': test_case['name'],
                    'status': 'error',
                    'error': str(e),
                    'params': test_case['params']
                })
        return pd.DataFrame(results)
# 使用示例
if __name__ == "__main__":
    # API规范定义
    api_spec = {
        'endpoint': '/api/user/create',
        'method': 'POST',
        'required_params': ['username', 'email', 'age'],
        'default_params': {
            'username': 'test_user',
            'email': 'test@example.com',
            'age': 25
        },
        'params': {
            'username': {
                'type': 'string',
                'min_length': 3,
                'max_length': 50
            },
            'email': {
                'type': 'string',
                'pattern': r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
            },
            'age': {
                'type': 'integer',
                'min': 0,
                'max': 150
            }
        }
    }
    validator = APIParameterValidator('https://api.example.com')
    results = validator.run_batch_tests('/api/user/create', api_spec)
    print(results)
    results.to_csv('api_test_results.csv', index=False)

Postman + Newman (使用集合批量测试)

// postman_collection.json 示例
{
    "info": {
        "name": "API参数校验测试",
        "description": "批量验证接口参数"
    },
    "item": [
        {
            "name": "必填参数测试",
            "event": [
                {
                    "listen": "test",
                    "script": {
                        "exec": [
                            "pm.test('状态码检查', function() {",
                            "    pm.response.to.have.status(400);",
                            "});",
                            "",
                            "pm.test('错误信息检查', function() {",
                            "    var jsonData = pm.response.json();",
                            "    pm.expect(jsonData).to.have.property('error');",
                            "});"
                        ]
                    }
                }
            ],
            "request": {
                "method": "POST",
                "header": [
                    {"key": "Content-Type", "value": "application/json"}
                ],
                "body": {
                    "mode": "raw",
                    "raw": "{\"email\": \"test@example.com\"}"
                },
                "url": {
                    "raw": "{{base_url}}/api/user/create",
                    "host": ["{{base_url}}"],
                    "path": ["api", "user", "create"]
                }
            }
        }
    ]
}

使用 YAML 配置文件的通用校验脚本

import yaml
import requests
import json
import re
from typing import Dict, List, Any
from dataclasses import dataclass
from datetime import datetime
@dataclass
class ValidationRule:
    field: str
    required: bool = False
    type: str = 'string'
    min_length: int = None
    max_length: int = None
    pattern: str = None
    enum: List[str] = None
    min_value: float = None
    max_value: float = None
class ParameterValidator:
    def __init__(self, config_path: str):
        with open(config_path, 'r', encoding='utf-8') as f:
            self.config = yaml.safe_load(f)
        self.results = []
    def validate_parameter(self, value: Any, rule: ValidationRule) -> bool:
        """验证单个参数"""
        if rule.required and value is None:
            return False
        if value is not None:
            # 类型检查
            if rule.type == 'string' and not isinstance(value, str):
                return False
            elif rule.type == 'integer' and not isinstance(value, int):
                return False
            elif rule.type == 'float' and not isinstance(value, (int, float)):
                return False
            elif rule.type == 'boolean' and not isinstance(value, bool):
                return False
            # 字符串长度检查
            if rule.type == 'string':
                if rule.min_length and len(value) < rule.min_length:
                    return False
                if rule.max_length and len(value) > rule.max_length:
                    return False
                # 正则表达式检查
                if rule.pattern and not re.match(rule.pattern, value):
                    return False
            # 数值范围检查
            if rule.type in ['integer', 'float']:
                if rule.min_value is not None and value < rule.min_value:
                    return False
                if rule.max_value is not None and value > rule.max_value:
                    return False
            # 枚举值检查
            if rule.enum and value not in rule.enum:
                return False
        return True
    def generate_test_data(self, rules: List[ValidationRule]) -> List[Dict]:
        """生成测试数据"""
        test_cases = []
        # 正常测试数据
        normal_data = {}
        for rule in rules:
            if rule.type == 'string':
                normal_data[rule.field] = 'test_string'
            elif rule.type == 'integer':
                normal_data[rule.field] = 100
            elif rule.type == 'float':
                normal_data[rule.field] = 100.5
            elif rule.type == 'boolean':
                normal_data[rule.field] = True
        test_cases.append({
            'name': '正常参数测试',
            'data': normal_data,
            'expect_pass': True
        })
        # 异常测试数据
        for rule in rules:
            if rule.required:
                # 缺少必填参数
                missing_data = {k: v for k, v in normal_data.items() 
                              if k != rule.field}
                test_cases.append({
                    'name': f'缺少必填参数: {rule.field}',
                    'data': missing_data,
                    'expect_pass': False
                })
            # 类型错误
            if rule.type == 'integer':
                wrong_data = normal_data.copy()
                wrong_data[rule.field] = 'not_a_number'
                test_cases.append({
                    'name': f'参数类型错误: {rule.field}',
                    'data': wrong_data,
                    'expect_pass': False
                })
        return test_cases
    def run_validation(self, config_key: str = None):
        """运行验证"""
        apis = self.config.get('apis', [])
        if config_key:
            apis = [api for api in apis if api['name'] == config_key]
        for api in apis:
            print(f"\n=== 验证API: {api['name']} ===")
            # 解析规则
            rules = []
            for field_config in api.get('parameters', []):
                rule = ValidationRule(
                    field=field_config['name'],
                    required=field_config.get('required', False),
                    type=field_config.get('type', 'string'),
                    min_length=field_config.get('min_length'),
                    max_length=field_config.get('max_length'),
                    pattern=field_config.get('pattern'),
                    enum=field_config.get('enum'),
                    min_value=field_config.get('min_value'),
                    max_value=field_config.get('max_value')
                )
                rules.append(rule)
            # 生成测试数据并验证
            test_cases = self.generate_test_data(rules)
            for test in test_cases:
                is_valid = all(
                    self.validate_parameter(
                        test['data'].get(rule.field), rule
                    )
                    for rule in rules
                )
                status = 'PASS' if is_valid == test['expect_pass'] else 'FAIL'
                print(f"{status}: {test['name']}")
                self.results.append({
                    'api': api['name'],
                    'test': test['name'],
                    'status': status,
                    'timestamp': datetime.now().isoformat()
                })
    def generate_report(self, output_path: str = 'validation_report.json'):
        """生成报告"""
        report = {
            'total_tests': len(self.results),
            'passed_tests': len([r for r in self.results if r['status'] == 'PASS']),
            'failed_tests': len([r for r in self.results if r['status'] == 'FAIL']),
            'results': self.results
        }
        with open(output_path, 'w', encoding='utf-8') as f:
            json.dump(report, f, ensure_ascii=False, indent=2)
        print(f"\n报告已生成: {output_path}")
        print(f"总计: {report['total_tests']} | 通过: {report['passed_tests']} | 失败: {report['failed_tests']}")
# YAML配置文件示例
"""
apis:
  - name: user_create
    endpoint: /api/user/create
    method: POST
    parameters:
      - name: username
        required: true
        type: string
        min_length: 3
        max_length: 50
      - name: email
        required: true
        type: string
        pattern: '^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
      - name: age
        required: true
        type: integer
        min_value: 0
        max_value: 150
      - name: gender
        type: string
        enum: ['male', 'female', 'other']
"""
# 使用示例
if __name__ == "__main__":
    validator = ParameterValidator('api_config.yaml')
    validator.run_validation()
    validator.generate_report()

使用 pytest + pytest-xdist 并行测试

import pytest
import requests
from typing import Dict, List
from dataclasses import dataclass
@dataclass
class TestCase:
    name: str
    params: Dict
    expected_status: int
    expected_response: Dict = None
class APITestGenerator:
    def __init__(self, base_url: str):
        self.base_url = base_url
    @pytest.fixture(autouse=True)
    def setup(self):
        self.session = requests.Session()
        yield
        self.session.close()
    def generate_params_test(self, api_spec: Dict) -> List[TestCase]:
        """生成参数测试用例"""
        test_cases = []
        # 各种参数组合测试
        param_combinations = []
        # 正常参数
        param_combinations.append({
            'name': '正常参数',
            'params': api_spec.get('valid_params', {}),
            'expected_status': 200
        })
        # 空参数
        param_combinations.append({
            'name': '空参数',
            'params': {},
            'expected_status': 400
        })
        # 无效参数
        param_combinations.append({
            'name': '无效参数',
            'params': {'invalid_param': 'value'},
            'expected_status': 400
        })
        for combo in param_combinations:
            test_cases.append(TestCase(**combo))
        return test_cases
# 测试类
class TestAPIEndpoint:
    @pytest.mark.parametrize("test_case", [
        TestCase(name="测试用例1", params={"key": "value"}, 
                expected_status=200),
        TestCase(name="测试用例2", params={}, 
                expected_status=400),
    ])
    def test_api_endpoint(self, test_case):
        response = requests.post(
            "https://api.example.com/endpoint",
            params=test_case.params
        )
        assert response.status_code == test_case.expected_status
# 运行命令
# pytest test_api.py -v -n auto  # 并行运行
# pytest test_api.py --html=report.html  # 生成HTML报告

使用 Jenkins Pipeline 自动化

pipeline {
    agent any
    parameters {
        string(name: 'API_CONFIG', defaultValue: 'api_config.yaml', 
               description: 'API配置文件路径')
        choice(name: 'ENVIRONMENT', choices: ['dev', 'staging', 'prod'], 
               description: '测试环境')
    }
    stages {
        stage('Checkout') {
            steps {
                checkout scm
            }
        }
        stage('Validate Config') {
            steps {
                sh 'python validate_config.py ${API_CONFIG}'
            }
        }
        stage('Run API Tests') {
            parallel {
                stage('Parameter Validation') {
                    steps {
                        sh 'python parameter_validator.py --config ${API_CONFIG}'
                    }
                }
                stage('Response Validation') {
                    steps {
                        sh 'python response_validator.py --config ${API_CONFIG}'
                    }
                }
            }
        }
        stage('Generate Report') {
            steps {
                sh 'python generate_report.py'
                publishHTML(target: [
                    allowMissing: false,
                    alwaysLinkToLastBuild: true,
                    keepAll: true,
                    reportDir: 'reports',
                    reportFiles: 'index.html',
                    reportName: 'API Validation Report'
                ])
            }
        }
    }
    post {
        always {
            junit 'reports/*.xml'
            cleanWs()
        }
    }
}

推荐使用方案

1. 快速原型/小项目：使用 Postman + Newman
2. 复杂项目/CI集成：使用 Python + pytest
3. 企业级应用：使用 Jenkins Pipeline + pytest
4. API文档驱动：使用 OpenAPI/Swagger 自动生成测试

需要我详细解释某个方案的具体实现,或者帮你针对特定场景选择最合适的方案吗？