api

name	API测试器
description	当测试API、发送HTTP请求、调试响应或验证端点时，在集成前测试HTTP API并验证响应。
license	MIT

API测试器技能

概述

API是系统之间的连接器。损坏的API会破坏应用程序。在集成前彻底测试端点。

核心原则: API契约必须在部署前验证。

何时使用

始终:

• 测试API端点
• 验证响应
• 检查状态码
• 验证头部
• 性能测试
• 安全测试

触发短语:

• "测试这个API"
• "API响应验证"
• "HTTP请求测试"
• "API端点检查"
• "API性能测试"
• "API调试"

API测试功能

请求测试

• HTTP方法支持
• 请求头配置
• 请求体构建
• 认证处理
• 参数传递

响应验证

• 状态码检查
• 响应头验证
• 响应体解析
• JSON Schema验证
• 响应时间测量

性能测试

• 负载测试
• 压力测试
• 并发测试
• 响应时间分析
• 吞吐量测试

常见API问题

状态码错误

问题:
API返回不正确的HTTP状态码

错误示例:
POST /api/users → 200 OK (应该是 201 Created)
GET /api/users/999 → 200 OK (应该是 404 Not Found)
PUT /api/users/999 → 200 OK (应该是 404 Not Found)

解决方案:
- 201: 资源创建成功
- 400: 请求参数错误
- 401: 未认证
- 403: 权限不足
- 404: 资源不存在
- 500: 服务器内部错误

响应格式不一致

问题:
API响应格式不统一

错误示例:
成功响应:
{
  "data": {"id": 1, "name": "John"}
}

错误响应:
{
  "error": "User not found",
  "code": 404
}

解决方案:
统一响应格式:
{
  "success": true,
  "data": {...},
  "message": "操作成功",
  "timestamp": "2023-01-01T10:00:00Z"
}

认证问题

问题:
API认证配置错误

错误示例:
- 缺少认证头
- Token过期处理不当
- 权限检查缺失

解决方案:
1. 实现标准的认证机制
2. 正确处理Token过期
3. 细粒度权限控制
4. 安全的Token生成

代码实现示例

API测试器

import requests
import json
import time
from typing import Dict, List, Any, Optional, Union
from dataclasses import dataclass
from enum import Enum
import concurrent.futures
import statistics
from requests.auth import HTTPBasicAuth, HTTPDigestAuth
import jwt

class HTTPMethod(Enum):
    GET = "GET"
    POST = "POST"
    PUT = "PUT"
    DELETE = "DELETE"
    PATCH = "PATCH"
    HEAD = "HEAD"
    OPTIONS = "OPTIONS"

class AuthType(Enum):
    NONE = "none"
    BASIC = "basic"
    DIGEST = "digest"
    BEARER = "bearer"
    API_KEY = "api_key"

@dataclass
class APIRequest:
    """API请求"""
    method: HTTPMethod
    url: str
    headers: Dict[str, str] = None
    params: Dict[str, Any] = None
    data: Dict[, ] = 
    json_data: [, ] = 
    auth_type: AuthType = AuthType.NONE
    auth_credentials: [, ] = 
    timeout:  = 


 :
    
    status_code: 
    headers: [, ]
    content: 
    json_data: [[, ]] = 
    response_time:  = 
    success:  = 


 :
    
    request: APIRequest
    response: APIResponse
    passed: 
    errors: []
    warnings: []


 :
    
    total_requests: 
    successful_requests: 
    failed_requests: 
    avg_response_time: 
    min_response_time: 
    max_response_time: 
    p95_response_time: 
    requests_per_second: 

 :
    
    
     ():
        .session = requests.Session()
        .default_headers = {
            : ,
            : ,
            : 
        }
    
     () -> APIResponse:
        
        start_time = time.time()
        
        :
            
            headers = {**.default_headers, **(request.headers  {})}
            
            
            auth = ._prepare_auth(request.auth_type, request.auth_credentials)
            
            
            response = .session.request(
                method=request.method.value,
                url=request.url,
                headers=headers,
                params=request.params,
                data=request.data,
                json=request.json_data,
                auth=auth,
                timeout=request.timeout
            )
            
            
            response_time = time.time() - start_time
            
            
            json_data = 
            :
                json_data = response.json()
            :
                
            
            
            success =  <= response.status_code < 
            
             APIResponse(
                status_code=response.status_code,
                headers=(response.headers),
                content=response.text,
                json_data=json_data,
                response_time=response_time,
                success=success
            )
            
         requests.exceptions.Timeout:
             APIResponse(
                status_code=,
                headers={},
                content=,
                response_time=time.time() - start_time,
                success=
            )
         requests.exceptions.ConnectionError:
             APIResponse(
                status_code=,
                headers={},
                content=,
                response_time=time.time() - start_time,
                success=
            )
         Exception  e:
             APIResponse(
                status_code=,
                headers={},
                content=,
                response_time=time.time() - start_time,
                success=
            )
    
     ():
        
         auth_type == AuthType.BASIC:
             HTTPBasicAuth(
                credentials[],
                credentials[]
            )
         auth_type == AuthType.DIGEST:
             HTTPDigestAuth(
                credentials[],
                credentials[]
            )
         auth_type == AuthType.BEARER:
            
             
         auth_type == AuthType.API_KEY:
            
             
        :
             
    
     () -> APITestResult:
        
        
        headers = request.headers  {}
         request.auth_type == AuthType.BEARER:
            headers[] = 
         request.auth_type == AuthType.API_KEY:
            key_name = request.auth_credentials.get(, )
            headers[key_name] = request.auth_credentials[]
        
        request.headers = headers
        
        
        response = .send_request(request)
        
        
        errors = []
        warnings = []
        passed = 
        
        
         response.status_code != expected_status:
            errors.append()
            passed = 
        
        
         response.response_time > :
            warnings.append()
        
        
         expected_schema  response.json_data:
            schema_errors = ._validate_json_schema(response.json_data, expected_schema)
             schema_errors:
                errors.extend(schema_errors)
                passed = 
        
         APITestResult(
            request=request,
            response=response,
            passed=passed,
            errors=errors,
            warnings=warnings
        )
    
     () -> []:
        
        errors = []
        
        
           schema:
             field  schema[]:
                 field   data:
                    errors.append()
        
           schema:
             field, field_schema  schema[].items():
                 field  data:
                    value = data[field]
                    expected_type = field_schema.get()
                    
                     expected_type ==    (value, ):
                        errors.append()
                     expected_type ==    (value, (, )):
                        errors.append()
                     expected_type ==    (value, ):
                        errors.append()
                     expected_type ==    (value, ):
                        errors.append()
                     expected_type ==    (value, ):
                        errors.append()
        
         errors
    
     () -> PerformanceTestResult:
        
        results = []
        
         ():
             .send_request(request)
        
        
         concurrent.futures.ThreadPoolExecutor(max_workers=concurrent_users)  executor:
            
            futures = [executor.submit(send_single_request)  _  (total_requests)]
            
            
             future  concurrent.futures.as_completed(futures):
                :
                    response = future.result()
                    results.append(response)
                 Exception  e:
                    ()
        
        
        successful_requests = [r  r  results  r.success]
        failed_requests = (results) - (successful_requests)
        
        response_times = [r.response_time  r  successful_requests]
        
         response_times:
            avg_response_time = statistics.mean(response_times)
            min_response_time = (response_times)
            max_response_time = (response_times)
            p95_response_time = (response_times)[((response_times) * )]
        :
            avg_response_time = min_response_time = max_response_time = p95_response_time = 
        
        
        total_time = (r.response_time  r  results)  results  
        requests_per_second = (successful_requests) / total_time  total_time >   
        
         PerformanceTestResult(
            total_requests=total_requests,
            successful_requests=(successful_requests),
            failed_requests=failed_requests,
            avg_response_time=avg_response_time,
            min_response_time=min_response_time,
            max_response_time=max_response_time,
            p95_response_time=p95_response_time,
            requests_per_second=requests_per_second
        )
    
     () -> :
        
          test_results:
             
        
        report = []
        
        
        total_tests = (test_results)
        passed_tests = ([r  r  test_results  r.passed])
        failed_tests = total_tests - passed_tests
        
        report.append()
        report.append()
        report.append()
        report.append()
        report.append()
        
        
        report.append()
         i, result  (test_results, ):
            status =   result.passed  
            report.append()
            
             result.response:
                report.append()
                report.append()
            
             result.errors:
                report.append()
                 error  result.errors:
                    report.append()
            
             result.warnings:
                report.append()
                 warning  result.warnings:
                    report.append()
            
            report.append()
        
         .join(report)
    
     () -> :
        
        report = []
        
        report.append()
        report.append()
        report.append()
        
        report.append()
        report.append()
        report.append()
        report.append()
        
        report.append()
        report.append()
        report.append()
        report.append()
        report.append()
        
        report.append()
        report.append()
        
         .join(report)


 ():
    tester = APITester()
    
    
    request = APIRequest(
        method=HTTPMethod.GET,
        url=,
        headers={: }
    )
    
    
    result = tester.test_api_endpoint(request, expected_status=)
    
    
    report = tester.generate_test_report([result])
    (report)
    
    
    perf_result = tester.performance_test(request, concurrent_users=, total_requests=)
    perf_report = tester.generate_performance_report(perf_result)
    ( + perf_report)

 __name__ == :
    main()

API套件测试

import yaml
from typing import List, Dict, Any
import json

class APITestSuite:
    """API测试套件"""
    
    def __init__(self):
        self.tester = APITester()
        self.test_cases = []
    
    def load_test_suite(self, suite_file: str) -> bool:
        """加载测试套件"""
        try:
            with open(suite_file, 'r', encoding='utf-8') as f:
                if suite_file.endswith('.yaml') or suite_file.endswith('.yml'):
                    suite_data = yaml.safe_load(f)
                else:
                    suite_data = json.load(f)
            
            # 解析测试用例
            for test_case in suite_data.get('test_cases', []):
                self.test_cases.append(self._parse_test_case(test_case))
            
            return True
        
        except Exception as e:
            print(f"加载测试套件失败: {str(e)}")
            return 
    
     () -> [, ]:
        
         {
            : test_case_data.get(, ),
            : test_case_data.get(, ),
            : APIRequest(
                method=HTTPMethod(test_case_data[][]),
                url=test_case_data[][],
                headers=test_case_data[].get(),
                params=test_case_data[].get(),
                data=test_case_data[].get(),
                json_data=test_case_data[].get(),
                auth_type=AuthType(test_case_data[].get(, )),
                auth_credentials=test_case_data[].get(),
                timeout=test_case_data[].get(, )
            ),
            : test_case_data.get(, ),
            : test_case_data.get(),
            : test_case_data.get(),
            : test_case_data.get()
        }
    
     () -> [APITestResult]:
        
        results = []
        
         test_case  .test_cases:
            ()
            
            
             test_case.get():
                ._execute_setup(test_case[])
            
            
            result = .tester.test_api_endpoint(
                test_case[],
                test_case[],
                test_case[]
            )
            result.test_name = test_case[]
            results.append(result)
            
            
             test_case.get():
                ._execute_teardown(test_case[])
        
         results
    
     ():
        
        
        
    
     ():
        
        
        
    
     () -> :
        
          results:
             
        
        report = []
        
        
        total_tests = (results)
        passed_tests = ([r  r  results  r.passed])
        failed_tests = total_tests - passed_tests
        
        report.append()
        report.append()
        report.append()
        report.append()
        report.append()
        
        
        report.append()
         result  results:
            test_name = (result, , )
            status =   result.passed  
            
            report.append()
            report.append()
            report.append()
            report.append()
            report.append()
            
             result.errors:
                report.append()
                 error  result.errors:
                    report.append()
            
             result.warnings:
                report.append()
                 warning  result.warnings:
                    report.append()
            
            report.append()
        
         .join(report)


 ():
    suite = APITestSuite()
    
    
     suite.load_test_suite():
        
        results = suite.run_test_suite()
        
        
        report = suite.generate_suite_report(results)
        (report)

 __name__ == :
    main()

API测试最佳实践

测试策略

1. 分层测试: 单元测试、集成测试、端到端测试
2. 覆盖全面: 测试正常流程和异常情况
3. 数据驱动: 使用不同的测试数据
4. 环境隔离: 测试环境与生产环境分离

测试用例设计

1. 正向测试: 验证正常功能
2. 负向测试: 验证错误处理
3. 边界测试: 测试极限情况
4. 安全测试: 验证安全性

性能测试

1. 基准测试: 建立性能基准
2. 负载测试: 测试预期负载
3. 压力测试: 测试极限负载
4. 稳定性测试: 长时间运行测试

相关技能

• api-validator - API验证
• http-debugger - HTTP调试
• performance-testing - 性能测试
• security-testing - 安全测试