Backend Development 14 min read

Comprehensive Python API Testing Examples with Requests and Pytest

This article provides a collection of practical Python snippets demonstrating how to send GET and POST requests, validate status codes, response bodies, JSON schemas, handle headers, query parameters, file uploads, redirects, timeouts, sessions, authentication, rate limiting, and parameterized testing using the requests library and pytest framework.

Test Development Learning Exchange
Test Development Learning Exchange
Test Development Learning Exchange
Comprehensive Python API Testing Examples with Requests and Pytest

Send GET request and verify status code

import requests

def test_get_status_code():
    """发送GET请求并验证状态码是否为200"""
    url = "http://api.example.com/resource"
    response = requests.get(url)
    print(f"响应状态码: {response.status_code}")
    assert response.status_code == 200, f"期望的状态码是200,但实际得到的是{response.status_code}"

Send POST request and verify response body

import requests

def test_post_and_check_response_body():
    """发送POST请求,并验证响应体中包含特定值"""
    url = "http://api.example.com/resource"
    payload = {"key": "value"}
    response = requests.post(url, json=payload)
    print(f"响应体: {response.json()}")
    assert "expected_key" in response.json(), "'expected_key' 不在响应体中"

Validate JSON response format

import requests
from jsonschema import validate

schema = {
    "type": "object",
    "properties": {
        "id": {"type": "integer"},
        "name": {"type": "string"}
    },
    "required": ["id", "name"]
}

def test_json_schema_validation():
    """验证API返回的JSON响应是否符合预期的JSON Schema"""
    url = "http://api.example.com/resource"
    response = requests.get(url)
    validate(instance=response.json(), schema=schema)
    print("JSON Schema 验证通过")

Set custom request headers

import requests

def test_with_custom_headers():
    """使用自定义请求头发送请求"""
    url = "http://api.example.com/resource"
    headers = {'Authorization': 'Bearer YOUR_TOKEN'}
    response = requests.get(url, headers=headers)
    print(f"响应状态码: {response.status_code}")
    assert response.status_code == 200, f"期望的状态码是200,但实际得到的是{response.status_code}"

Handle query parameters

import requests

def test_with_query_params():
    """发送带有查询参数的GET请求"""
    url = "http://api.example.com/resource"
    params = {'param1': 'value1', 'param2': 'value2'}
    response = requests.get(url, params=params)
    print(f"完整URL: {response.url}")
    assert response.status_code == 200, f"期望的状态码是200,但实际得到的是{response.status_code}"

File upload

import requests

def test_file_upload():
    """测试文件上传"""
    url = "http://api.example.com/upload"
    files = {'file': open('test.txt', 'rb')}
    response = requests.post(url, files=files)
    print(f"响应体: {response.text}")
    assert response.status_code == 200, f"期望的状态码是200,但实际得到的是{response.status_code}"

Redirect handling

import requests

def test_follow_redirects():
    """测试跟随重定向"""
    url = "http://api.example.com/redirect"
    response = requests.get(url, allow_redirects=True)
    print(f"最终URL: {response.url}")
    assert response.history, "没有发生重定向"

Timeout setting

import requests

def test_request_timeout():
    """设置请求超时时间"""
    url = "http://api.example.com/resource"
    try:
        response = requests.get(url, timeout=0.001)  # 设置一个非常短的超时时间以触发异常
    except requests.exceptions.Timeout:
        print("请求超时,如预期")

Session management

import requests

def test_session_management():
    """使用Session保持会话"""
    session = requests.Session()
    session.headers.update({'Authorization': 'Bearer YOUR_TOKEN'})
    url = "http://api.example.com/resource"
    response = session.get(url)
    print(f"响应状态码: {response.status_code}")
    assert response.status_code == 200, f"期望的状态码是200,但实际得到的是{response.status_code}"

Validate response content type

import requests

def test_content_type():
    """验证响应的内容类型"""
    url = "http://api.example.com/resource"
    response = requests.get(url)
    print(f"Content-Type: {response.headers['Content-Type']}")
    assert 'application/json' in response.headers['Content-Type'], "期望的内容类型是application/json"

Handle redirects and retrieve all URLs

import requests

def test_get_all_redirects():
    """获取所有跳转的URL"""
    url = "http://api.example.com/redirect"
    response = requests.get(url, allow_redirects=True)
    print("所有跳转的URL:")
    for resp in response.history:
        print(resp.url)
    print(f"最终URL: {response.url}")

Parameterized testing with pytest

import pytest
import requests

@pytest.mark.parametrize("test_input, expected", [
    ("http://api.example.com/resource1", 200),
    ("http://api.example.com/resource2", 404)
])
def test_multiple_urls(test_input, expected):
    """使用参数化进行多个URL的测试"""
    response = requests.get(test_input)
    print(f"请求 {test_input} 的响应状态码: {response.status_code}")
    assert response.status_code == expected, f"期望的状态码是{expected},但实际得到的是{response.status_code}"

Basic authentication

import requests
from requests.auth import HTTPBasicAuth

def test_basic_auth():
    """测试Basic认证"""
    url = "http://api.example.com/protected"
    auth = HTTPBasicAuth('username', 'password')
    response = requests.get(url, auth=auth)
    print(f"响应状态码: {response.status_code}")
    assert response.status_code == 200, f"期望的状态码是200,但实际得到的是{response.status_code}"

API rate limiting test

import requests
import time

def test_rate_limiting():
    """测试API的速率限制"""
    url = "http://api.example.com/ratelimited"
    for i in range(10):  # 发送10次请求以触发可能的速率限制
        response = requests.get(url)
        print(f"第{i+1}次请求的响应状态码: {response.status_code}")
        if response.status_code != 200:
            print(f"速率限制被触发,状态码为 {response.status_code}")
            break
        time.sleep(1)  # 等待1秒避免过快触发速率限制

OAuth2 token validation

import requests

def test_oauth2_token():
    """测试OAuth2令牌的有效性"""
    url = "http://api.example.com/resource"
    headers = {'Authorization': 'Bearer YOUR_OAUTH2_TOKEN'}
    response = requests.get(url, headers=headers)
    print(f"响应状态码: {response.status_code}")
    assert response.status_code == 200, f"期望的状态码是200,但实际得到的是{response.status_code}"

Assert specific field value in response

import requests

def test_specific_field_in_response():
    """验证响应体中特定字段的值是否符合预期"""
    url = "http://api.example.com/resource"
    response = requests.get(url)
    data = response.json()
    print(f"响应体: {data}")
    assert data['key'] == 'expected_value', f"期望的值是'expected_value',但实际得到的是{data['key']}"

Upload multiple files

import requests

def test_upload_multiple_files():
    """测试上传多个文件"""
    url = "http://api.example.com/upload"
    files = [('file', open('file1.txt', 'rb')), ('file', open('file2.txt', 'rb'))]
    response = requests.post(url, files=files)
    print(f"响应体: {response.text}")
    assert response.status_code == 200, f"期望的状态码是200,但实际得到的是{response.status_code}"

Custom session object request

import requests

def test_custom_session():
    """使用带有自定义设置的Session对象发送请求"""
    session = requests.Session()
    session.headers.update({'User-Agent': 'CustomUserAgent'})
    url = "http://api.example.com/resource"
    response = session.get(url)
    print(f"响应状态码: {response.status_code}")
    assert response.status_code == 200, f"期望的状态码是200,但实际得到的是{response.status_code}"

Complex request headers

import requests

def test_complex_headers():
    """发送带有复杂请求头的请求"""
    url = "http://api.example.com/resource"
    headers = {
        'Content-Type': 'application/json',
        'Accept-Language': 'zh-CN,zh;q=0.9',
        'Custom-Header': 'customValue'
    }
    response = requests.get(url, headers=headers)
    print(f"响应状态码: {response.status_code}")
    assert response.status_code == 200, f"期望的状态码是200,但实际得到的是{response.status_code}"

API error message handling

import requests

def test_api_error_messages():
    """测试API返回的错误信息"""
    url = "http://api.example.com/error"
    response = requests.get(url)
    error_message = response.json().get('error', '')
    print(f"错误信息: {error_message}")
    assert error_message == 'Expected Error Message', f"期望的错误信息是'Expected Error Message',但实际得到的是'{error_message}'"

Parameterized data‑driven testing with pytest

import pytest
import requests

@pytest.mark.parametrize("url, expected_status", [
    ("http://api.example.com/success", 200),
    ("http://api.example.com/notfound", 404),
    ("http://api.example.com/servererror", 500)
])
def test_api_with_param(url, expected_status):
    """通过参数化的方式对多个API端点进行测试"""
    response = requests.get(url)
    print(f"请求 {url} 的响应状态码: {response.status_code}")
    assert response.status_code == expected_status, f"期望的状态码是{expected_status},但实际得到的是{response.status_code}"

Assert array length in response

import requests

def test_array_length_in_response():
    """验证响应体中某个数组的长度是否符合预期"""
    url = "http://api.example.com/resource"
    response = requests.get(url)
    data = response.json()
    print(f"响应体: {data}")
    assert len(data['items']) == 3, f"期望数组长度为3,但实际长度为{len(data['items'])}"
PythonHTTPAPI testingRequestspytest
Test Development Learning Exchange
Written by

Test Development Learning Exchange

Test Development Learning Exchange

0 followers
Reader feedback

How this landed with the community

login Sign in to like

Rate this article

Was this worth your time?

Sign in to rate
Discussion

0 Comments

Thoughtful readers leave field notes, pushback, and hard-won operational detail here.