15 Ready‑to‑Use API Testing Templates with Full Pytest Code Samples
This article provides a comprehensive collection of fifteen reusable API testing templates covering CRUD operations, authentication, idempotency, file upload security, pagination, WebSocket communication, rate limiting, GraphQL, gRPC, OpenAPI contracts, i18n, caching, circuit breaking, data masking, and version compatibility, each accompanied by ready‑to‑run pytest code examples.
Template 1: Basic CRUD Full Coverage
Design pytest test cases for a user management RESTful API with POST, GET, PUT, DELETE methods, covering normal, error, and boundary scenarios, including database state verification using SQLAlchemy.
import pytest
from sqlalchemy import create_engine, text
engine = create_engine("sqlite:///test.db")
@pytest.mark.parametrize(
"name,email,expected_status",
[
("张三", "[email protected]", 201), # positive
("a", "invalid-email", 400), # boundary + format error
("李四" * 15, "[email protected]", 400) # overflow
]
)
def test_create_user(name, email, expected_status):
resp = requests.post("/api/v1/users", json={"name": name, "email": email})
assert resp.status_code == expected_status
if expected_status == 201:
with engine.connect() as conn:
result = conn.execute(text("SELECT COUNT(*) FROM users WHERE email = :email"), {"email": email})
assert result.scalar() == 1Template 2: Authentication & Authorization Testing
Generate security test cases for a JWT‑protected order endpoint, checking token acquisition, missing/invalid tokens, and insufficient permissions.
def get_token(username, password):
resp = requests.post("/auth/login", json={"username": username, "password": password})
return resp.json()["access_token"]
@pytest.mark.parametrize(
"user_role,expected_status",
[
("admin", 200),
("guest", 403) # no order:read permission
]
)
def test_order_access_control(user_role, expected_status):
token = get_token(f"{user_role}_user", "password123")
headers = {"Authorization": f"Bearer {token}"}
resp = requests.get("/api/v1/orders", headers=headers)
assert resp.status_code == expected_statusTemplate 3: Idempotency Verification
Validate that repeated payment requests with the same X-Idempotency-Key return identical results.
import uuid
@pytest.fixture
def idempotency_key():
return str(uuid.uuid4())
def test_payment_idempotency(idempotency_key):
headers = {"X-Idempotency-Key": idempotency_key}
payload = {"amount": 100, "currency": "CNY"}
# first request
resp1 = requests.post("/api/v1/payments", json=payload, headers=headers)
assert resp1.status_code == 201
payment_id = resp1.json()["id"]
# duplicate request
resp2 = requests.post("/api/v1/payments", json=payload, headers=headers)
assert resp2.status_code == 200
assert resp2.json()["id"] == payment_idTemplate 4: File Upload Security Testing
Create high‑risk test cases for a file‑upload endpoint that only accepts .jpg/.png files up to 10 MB, including double‑extension and magic‑number attacks.
from requests_toolbelt.multipart.encoder import MultipartEncoder
@pytest.mark.parametrize(
"filename,content_type,expected_status",
[
("test.jpg", "image/jpeg", 200),
("malicious.pdf.exe", "application/octet-stream", 400),
("large_file.zip", "application/zip", 413) # oversized
]
)
def test_file_upload_security(filename, content_type, expected_status):
with open(f"test_files/{filename}", "rb") as f:
encoder = MultipartEncoder(fields={"file": (filename, f, content_type)})
resp = requests.post(
"/api/v1/upload",
data=encoder,
headers={"Content-Type": encoder.content_type}
)
assert resp.status_code == expected_statusTemplate 5: Pagination & Sorting Validation
Check that a paginated user‑list endpoint returns correct total counts, respects sorting order, and handles edge cases such as page=0 or size=1000.
from faker import Faker
fake = Faker()
# seed 100 users
for _ in range(100):
requests.post("/api/v1/users", json={"name": fake.name(), "email": fake.email()})
def test_pagination_and_sorting():
resp = requests.get("/api/v1/users?page=1&size=10&sort=name,desc")
data = resp.json()
assert data["total"] == 100
assert len(data["items"]) == 10
names = [item["name"] for item in data["items"]]
assert names == sorted(names, reverse=True)Template 6: WebSocket Real‑Time Communication Testing
Design asynchronous tests for a chat WebSocket endpoint, verifying message send/receive and reconnection after a simulated disconnect.
import asyncio
import websockets
async def test_websocket_chat():
token = get_valid_token()
uri = f"wss://api.example.com/chat?token={token}"
async with websockets.connect(uri) as websocket:
await websocket.send('{"type": "text", "content": "Hello"}')
response = await websocket.recv()
assert "Hello" in response
await websocket.close()
# reconnection logic can be added hereTemplate 7: Third‑Party Callback Verification
Test a payment‑gateway notification endpoint with HMAC‑SHA256 signature validation and idempotent handling of duplicate callbacks.
import hmac, hashlib
def generate_signature(payload, secret):
return hmac.new(secret.encode(), payload.encode(), hashlib.sha256).hexdigest()
def test_payment_callback():
payload = '{"order_id": "123", "status": "paid"}'
signature = generate_signature(payload, "SECRET_KEY")
resp = requests.post(
"/payment/notify",
data=payload,
headers={"X-Signature": signature}
)
assert resp.status_code == 200
order = db.get_order("123")
assert order.status == "paid"Template 8: Rate‑Limiting Test
Generate a Locust load‑testing script to verify that the API enforces a limit of 100 requests per minute per IP, returning HTTP 429 when exceeded.
from locust import HttpUser, task, between
class RateLimitUser(HttpUser):
wait_time = between(0.1, 0.5)
@task
def hit_api(self):
self.client.get("/api/v1/data")
def on_stop(self):
resp = self.client.get("/api/v1/data")
assert resp.status_code in [200, 429]Template 9: GraphQL Query Testing
Validate a GraphQL user query, checking field selection, nested order data, and proper error handling for non‑existent IDs.
def test_graphql_user_query():
query = """
query {
user(id: "123") {
name
orders { id amount }
}
}
"""
resp = requests.post("/graphql", json={"query": query})
data = resp.json()
assert "name" in data["data"]["user"]
assert len(data["data"]["user"]["orders"]) > 0Template 10: gRPC Service Testing
Generate pytest integration tests for a gRPC OrderService, using grpcio-tools to create stubs.
import grpc
from order_pb2 import CreateOrderRequest
from order_pb2_grpc import OrderServiceStub
def test_grpc_create_order():
channel = grpc.insecure_channel("localhost:50051")
stub = OrderServiceStub(channel)
request = CreateOrderRequest(user_id="123", items=["item1"])
response = stub.CreateOrder(request)
assert response.id is not None
assert response.user_id == "123"Template 11: OpenAPI Contract Testing (Pact)
Generate Pact contract tests from an OpenAPI spec for a User Service provider and Order Service consumer.
from pact import Consumer, Provider
pact = Consumer("OrderService").has_pact_with(Provider("UserService"))
def test_get_user():
expected = {"id": "123", "name": "John"}
(pact
.given("user exists")
.upon_receiving("a request for user")
.with_request("GET", "/users/123")
.will_respond_with(200, body=expected))
with pact:
resp = requests.get("http://localhost:1234/users/123")
assert resp.json() == expectedTemplate 12: Internationalization Parameter Testing
Test a product‑listing API for language handling, special characters, and timezone correctness.
@pytest.mark.parametrize(
"lang,expected_text",
[
("zh-CN", "苹果"),
("ar-SA", "تفاحة"), # Arabic
("en-US", "Apple")
]
)
def test_i18n_product_name(lang, expected_text):
resp = requests.get(f"/products?lang={lang}")
assert expected_text in resp.json()["name"]Template 13: Cache Consistency Testing
Ensure that updates to a configuration endpoint invalidate Redis cache and avoid dirty reads under concurrent updates.
def test_cache_invalidation():
resp1 = requests.get("/api/v1/config")
assert resp1.json()["version"] == 1
requests.post("/api/v1/config", json={"version": 2})
resp2 = requests.get("/api/v1/config")
assert resp2.json()["version"] == 2Template 14: Circuit‑Breaker Degradation Testing
Simulate failures in a downstream service to verify that a Hystrix‑style circuit breaker returns fallback data instead of errors.
def test_circuit_breaker():
inject_latency("service-b", 5000) # 5 s delay
resp = requests.get("/api/service-a")
assert resp.status_code == 200
assert "fallback" in resp.json()Template 15: Data Masking Validation
Check that sensitive fields like phone numbers are masked for regular users but fully visible to administrators.
def test_data_masking():
token = get_token("user", "pass")
resp = requests.get("/api/v1/users", headers={"Authorization": f"Bearer {token}"})
assert "****" in resp.json()[0]["phone"]
admin_token = get_token("admin", "pass")
resp_admin = requests.get("/api/v1/users", headers={"Authorization": f"Bearer {admin_token}"})
assert "****" not in resp_admin.json()[0]["phone"]Template 16: Version Compatibility Testing
Verify that v2 API endpoints remain compatible with v1 query parameters and that renamed fields map correctly.
def test_api_version_compatibility():
resp1 = requests.get("/v1/users?id=123")
email_v1 = resp1.json()["email"]
resp2 = requests.get("/v2/users/123")
email_v2 = resp2.json()["email_address"]
assert email_v1 == email_v2Signed-in readers can open the original source through BestHub's protected redirect.
This article has been distilled and summarized from source material, then republished for learning and reference. If you believe it infringes your rights, please contactand we will review it promptly.
How this landed with the community
Was this worth your time?
0 Comments
Thoughtful readers leave field notes, pushback, and hard-won operational detail here.
