Practical Guide to Python API Automation Testing Scripts
1. Basic HTTP Request Testing
1.1 Basic GET Request Testing Using the Requests Library
import requests
import unittest
class TestGetApi(unittest.TestCase):
def test_get_request(self):
# Send GET request
response = requests.get('https://jsonplaceholder.typicode.com/posts/1')
# Assert status code is 200
self.assertEqual(response.status_code, 200)
# Assert response contains expected keys
data = response.json()
self.assertIn('id', data)
self.assertIn('title', data)
self.assertIn('body', data)
if __name__ == '__main__':
unittest.main()
1.2 POST Request Testing Using the Requests Library
import requests
import unittest
class TestPostApi(unittest.TestCase):
def test_post_request(self):
# Data to be sent
payload = {
'title': 'foo',
'body': 'bar',
'userId': 1
}
# Send POST request
response = requests.post('https://jsonplaceholder.typicode.com/posts', json=payload)
# Assert status code is 201 (created successfully)
self.assertEqual(response.status_code, 201)
# Assert response contains the data we sent
data = response.json()
self.assertEqual(data['title'], 'foo')
self.assertEqual(data['body'], 'bar')
self.assertEqual(data['userId'], 1)
if __name__ == '__main__':
unittest.main()
1.3 Basic Request Testing Using the Pytest Framework
import pytest
import requests
def test_api_status():
"""Test API service status"""
response = requests.get('https://api.example.com/status')
assert response.status_code == 200
assert response.json()['status'] == 'ok'
def test_api_data_structure():
"""Test the structure of the data returned by the API"""
response = requests.get('https://jsonplaceholder.typicode.com/users/1')
data = response.json()
# Check if necessary fields exist
assert 'id' in data
assert 'name' in data
assert 'email' in data
# Check data types
assert isinstance(data['id'], int)
assert isinstance(data['name'], str)
assert isinstance(data['email'], str)
1.4 PUT Request Testing
import requests
import unittest
class TestPutApi(unittest.TestCase):
def test_put_request(self):
# Data to update
payload = {
'id': 1,
'title': 'updated title',
'body': 'updated body',
'userId': 1
}
# Send PUT request
response = requests.put('https://jsonplaceholder.typicode.com/posts/1', json=payload)
# Assert status code is 200 (updated successfully)
self.assertEqual(response.status_code, 200)
# Assert response contains the data we sent
data = response.json()
self.assertEqual(data['title'], 'updated title')
self.assertEqual(data['body'], 'updated body')
if __name__ == '__main__':
unittest.main()
1.5 DELETE Request Testing
import requests
import unittest
class TestDeleteApi(unittest.TestCase):
def test_delete_request(self):
# Send DELETE request
response = requests.delete('https://jsonplaceholder.typicode.com/posts/1')
# Assert status code is 200 (deleted successfully)
self.assertEqual(response.status_code, 200)
# For some APIs, delete may return an empty response body
self.assertEqual(response.text, '{}')
if __name__ == '__main__':
unittest.main()
2. Authentication and Authorization Testing
2.1 Basic Authentication Testing
import requests
import unittest
class TestBasicAuth(unittest.TestCase):
def test_basic_auth(self):
# Send request with basic authentication
response = requests.get(
'https://httpbin.org/basic-auth/user/passwd',
auth=('user', 'passwd')
)
# Assert authentication success
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertTrue(data['authenticated'])
self.assertEqual(data['user'], 'user')
if __name__ == '__main__':
unittest.main()
2.2 OAuth2 Authentication Testing
import requests
import unittest
import os
class TestOAuth2(unittest.TestCase):
def test_oauth2_token(self):
# Get OAuth2 token
token_url = 'https://api.example.com/oauth2/token'
client_id = os.environ.get('CLIENT_ID')
client_secret = os.environ.get('CLIENT_SECRET')
data = {
'grant_type': 'client_credentials',
'client_id': client_id,
'client_secret': client_secret
}
token_response = requests.post(token_url, data=data)
self.assertEqual(token_response.status_code, 200)
# Extract token
token_data = token_response.json()
access_token = token_data['access_token']
# Use token to access protected API
headers = {
'Authorization': f'Bearer {access_token}'
}
api_response = requests.get('https://api.example.com/protected', headers=headers)
self.assertEqual(api_response.status_code, 200)
if __name__ == '__main__':
unittest.main()
2.3 API Key Authentication Testing
import requests
import unittest
import os
class TestApiKey(unittest.TestCase):
def test_api_key_auth(self):
# Get API Key from environment variable
api_key = os.environ.get('API_KEY')
# Set request headers or query parameters
headers = {
'X-API-Key': api_key
}
# Or as a query parameter
params = {
'api_key': api_key
}
# Send request
response = requests.get(
'https://api.example.com/data',
headers=headers,
# params=params # Use either headers or params
)
# Assert authentication success
self.assertEqual(response.status_code, 200)
if __name__ == '__main__':
unittest.main()
3. Data-Driven Testing
3.1 Using Pytest Parameterization
import pytest
import requests
# Test data
test_data = [
('1', 'Leanne Graham'),
('2', 'Ervin Howell'),
('3', 'Clementine Bauch')
]
@pytest.mark.parametrize('user_id,expected_name', test_data)
def test_user_name(user_id, expected_name):
"""Test if user name matches expected"""
response = requests.get(f'https://jsonplaceholder.typicode.com/users/{user_id}')
assert response.status_code == 200
user_data = response.json()
assert user_data['name'] == expected_name
3.2 Loading Test Data from CSV File
import csv
import pytest
import requests
import os
def read_test_data_from_csv(csv_file):
"""Read test data from CSV file"""
test_data = []
with open(csv_file, 'r', encoding='utf-8') as file:
reader = csv.DictReader(file)
for row in reader:
test_data.append((row['user_id'], row['expected_name']))
return test_data
# Assume CSV file path
csv_file_path = os.path.join(os.path.dirname(__file__), 'test_data.csv')
# Use CSV data for parameterized testing
@pytest.mark.parametrize('user_id,expected_name', read_test_data_from_csv(csv_file_path))
def test_user_name_from_csv(user_id, expected_name):
"""Load test data from CSV file and test user name"""
response = requests.get(f'https://jsonplaceholder.typicode.com/users/{user_id}')
assert response.status_code == 200
user_data = response.json()
assert user_data['name'] == expected_name
3.3 Loading Test Data from JSON File
import json
import pytest
import requests
import os
def read_test_data_from_json(json_file):
"""Read test data from JSON file"""
with open(json_file, 'r', encoding='utf-8') as file:
return json.load(file)
# Assume JSON file path
json_file_path = os.path.join(os.path.dirname(__file__), 'test_data.json')
test_data = read_test_data_from_json(json_file_path)
@pytest.mark.parametrize('test_case', test_data)
def test_post_api_with_json_data(test_case):
"""Test POST request using JSON data"""
url = test_case['url']
payload = test_case['payload']
expected_status = test_case['expected_status']
expected_response = test_case['expected_response']
response = requests.post(url, json=payload)
assert response.status_code == expected_status
# Only check fields defined in expected response
actual_response = response.json()
for key, value in expected_response.items():
assert actual_response[key] == value
4. Parameterized Testing
4.1 Multi-Parameter Testing
import pytest
import requests
# Multi-parameter combination test data
test_data = [
# user_id, post_id, expected_status, expected_keys
(1, 1, 200, ['userId', 'id', 'title', 'body']),
(1, 999, 404, []), # Non-existent post_id
(999, 1, 200, ['userId', 'id', 'title', 'body']), # Post of a non-user
]
@pytest.mark.parametrize('user_id,post_id,expected_status,expected_keys', test_data)
def test_get_user_post(user_id, post_id, expected_status, expected_keys):
"""Test getting user posts API"""
response = requests.get(f'https://jsonplaceholder.typicode.com/posts/{post_id}')
# Check status code
assert response.status_code == expected_status
# If successfully retrieved post, check response structure
if expected_status == 200:
data = response.json()
for key in expected_keys:
assert key in data
4.2 Using Faker to Generate Test Data
import pytest
import requests
from faker import Faker
import random
# Initialize Faker generator
faker = Faker()
def generate_user_data(num_users=5):
"""Generate random user data using Faker"""
users = []
for _ in range(num_users):
users.append({
'name': faker.name(),
'email': faker.email(),
'phone': faker.phone_number(),
'website': faker.url()
})
return users
@pytest.mark.parametrize('user_data', generate_user_data())
def test_create_user_with_random_data(user_data):
"""Test creating user API with randomly generated data"""
response = requests.post('https://jsonplaceholder.typicode.com/users', json=user_data)
# JSONPlaceholder always returns 201 status code
assert response.status_code == 201
# Check returned data contains the properties we sent
data = response.json()
for key, value in user_data.items():
assert data[key] == value
5. Assertions and Validations
5.1 Using JSONSchema to Validate Response Format
import requests
import unittest
import jsonschema
class TestJsonSchema(unittest.TestCase):
def test_response_schema(self):
# Define expected JSON Schema
user_schema = {
"type": "object",
"properties": {
"id": {"type": "integer"},
"name": {"type": "string"},
"username": {"type": "string"},
"email": {"type": "string", "format": "email"},
"address": {
"type": "object",
"properties": {
"street": {"type": "string"},
"suite": {"type": "string"},
"city": {"type": "string"},
"zipcode": {"type": "string"},
"geo": {
"type": "object",
"properties": {
"lat": {"type": "string"},
"lng": {"type": "string"}
},
"required": ["lat", "lng"]
}
},
"required": ["street", "city", "zipcode"]
},
"phone": {"type": "string"},
"website": {"type": "string"},
"company": {
"type": "object",
"properties": {
"name": {"type": "string"},
"catchPhrase": {"type": "string"},
"bs": {"type": "string"}
},
"required": ["name"]
}
},
"required": ["id", "name", "email"]
}
# Send GET request
response = requests.get('https://jsonplaceholder.typicode.com/users/1')
self.assertEqual(response.status_code, 200)
# Validate response conforms to JSON Schema
try:
jsonschema.validate(response.json(), user_schema)
except jsonschema.exceptions.ValidationError as e:
self.fail(f"Response schema validation failed: {e}")
if __name__ == '__main__':
unittest.main()
5.2 Advanced Assertions – Using PyHamcrest
import requests
import pytest
from hamcrest import *
def test_api_with_hamcrest():
"""Use PyHamcrest for advanced assertions"""
response = requests.get('https://jsonplaceholder.typicode.com/users/1')
assert_that(response.status_code, equal_to(200))
data = response.json()
# Complex assertions
assert_that(data, has_key('name'))
assert_that(data['name'], not_none())
assert_that(data['email'], ends_with('.biz'))
# Multiple assertion combinations
assert_that(data, all_of(
has_key('address'),
has_entry('phone', not_none()),
has_entry('website', contains_string('.'))
))
# Assert nested structure
assert_that(data['address'], has_key('geo'))
assert_that(data['address']['geo'], has_key('lat'))
6. Test Suites and Reports
6.1 Creating a Pytest Test Suite
"""
Test suite configuration file: conftest.py
"""
import pytest
import requests
# Session-level fixture, executed once before all tests
@pytest.fixture(scope="session")
def api_base_url():
return "https://jsonplaceholder.typicode.com"
# Executed before each test function
@pytest.fixture
def api_client(api_base_url):
# Common settings like authentication can be added here
session = requests.Session()
session.headers.update({
"Content-Type": "application/json",
"Accept": "application/json"
})
yield session
# Close session after tests are done
session.close()
6.2 Generating HTML Test Reports
"""
Use pytest-html plugin to generate HTML reports
Install: pip install pytest-html
Run: pytest --html=report.html
"""
import pytest
def test_api_status(api_client, api_base_url):
"""Test API status"""
response = api_client.get(f"{api_base_url}/posts/1")
assert response.status_code == 200
6.3 Generating Allure Reports for Tests
"""
Use Allure to generate beautiful test reports
Install:
- pip install allure-pytest
- Install Allure command line tool
Run:
- pytest --alluredir=./allure-results
- allure serve ./allure-results
"""
import pytest
import allure
import requests
@allure.feature('Posts API')
@allure.story('Retrieve post details')
def test_get_post_details():
"""Test getting post details"""
with allure.step('Send GET request to /posts/1'):
response = requests.get('https://jsonplaceholder.typicode.com/posts/1')
with allure.step('Verify status code is 200'):
assert response.status_code == 200
with allure.step('Verify response contains expected fields'):
data = response.json()
assert 'id' in data
assert 'title' in data
assert 'body' in data
assert 'userId' in data
# Add attachments, e.g., request and response content
allure.attach(
str(response.request.url),
'Request URL',
allure.attachment_type.TEXT
)
allure.attach(
str(response.json()),
'Response Body',
allure.attachment_type.JSON
)
7. Advanced Features
7.1 API Performance Testing
import requests
import time
import statistics
import pytest
@pytest.mark.performance
def test_api_performance():
"""Simple API performance test"""
url = 'https://jsonplaceholder.typicode.com/posts'
# Test configuration
num_requests = 10 # Number of requests
response_times = [] # Execute multiple requests and record response times
for _ in range(num_requests):
start_time = time.time()
response = requests.get(url)
end_time = time.time()
# Ensure response is successful
assert response.status_code == 200
# Record response time (milliseconds)
response_time = (end_time - start_time) * 1000
response_times.append(response_time)
# Calculate statistics
avg_response_time = sum(response_times) / len(response_times)
median_response_time = statistics.median(response_times)
min_response_time = min(response_times)
max_response_time = max(response_times)
# Record performance results
print(f"Performance test results (ms):")
print(f"Average response time: {avg_response_time:.2f}")
print(f"Median response time: {median_response_time:.2f}")
print(f"Minimum response time: {min_response_time:.2f}")
print(f"Maximum response time: {max_response_time:.2f}")
# Assert performance meets requirements
assert avg_response_time < 1000 # Average response time less than 1 second
7.2 Mock Server
import pytest
import requests
from unittest import mock
def test_api_with_mock():
"""Use mock to simulate API response"""
# Define mock response
mock_response = mock.Mock()
mock_response.status_code = 200
mock_response.json.return_value = {
'id': 1,
'name': 'Test User',
'email': '[email protected]'
}
# Use patch to replace requests.get method
with mock.patch('requests.get', return_value=mock_response):
response = requests.get('https://api.example.com/users/1')
# Now this request will not actually send, but return mock object
assert response.status_code == 200
data = response.json()
assert data['name'] == 'Test User'
assert data['email'] == '[email protected]'
7.3 Managing Sessions with RequestsSession
import requests
import unittest
class TestApiSession(unittest.TestCase):
def setUp(self):
"""Set up session before each test"""
self.session = requests.Session()
self.session.headers.update({
'Content-Type': 'application/json',
'User-Agent': 'ApiTestingBot/1.0'
})
# If authentication is needed
self.session.auth = ('username', 'password')
# Set base URL
self.base_url = 'https://jsonplaceholder.typicode.com'
def tearDown(self):
"""Close session after each test"""
self.session.close()
def test_get_todos(self):
"""Test getting todo list"""
response = self.session.get(f'{self.base_url}/todos')
self.assertEqual(response.status_code, 200)
# Verify response is a list
data = response.json()
self.assertIsInstance(data, list)
# Verify at least one item
self.assertGreater(len(data), 0)
# Verify first item has expected fields
first_item = data[0]
self.assertIn('id', first_item)
self.assertIn('title', first_item)
self.assertIn('completed', first_item)
def test_get_user_posts(self):
"""Test getting user posts"""
# Use query parameters
params = {'userId': 1}
response = self.session.get(f'{self.base_url}/posts', params=params)
self.assertEqual(response.status_code, 200)
# Verify all returned posts belong to user 1
posts = response.json()
for post in posts:
self.assertEqual(post['userId'], 1)
if __name__ == '__main__':
unittest.main()
7.4 API Dependency Testing
import pytest
import requests
@pytest.fixture
def created_post():
"""Create a post and return its ID as a dependency for other tests"""
# Create new post
create_payload = {
'title': 'Test Post',
'body': 'This is a test post',
'userId': 1
}
response = requests.post(
'https://jsonplaceholder.typicode.com/posts',
json=create_payload
)
assert response.status_code == 201
post_data = response.json()
post_id = post_data['id']
# Return created post ID for subsequent tests
yield post_id
def test_update_post(created_post):
"""Test updating previously created post"""
post_id = created_post
# Update post
update_payload = {
'id': post_id,
'title': 'Updated Title',
'body': 'This post has been updated',
'userId': 1
}
response = requests.put(
f'https://jsonplaceholder.typicode.com/posts/{post_id}',
json=update_payload
)
assert response.status_code == 200
updated_data = response.json()
assert updated_data['title'] == 'Updated Title'
assert updated_data['body'] == 'This post has been updated'
def test_delete_post(created_post):
"""Test deleting previously created post"""
post_id = created_post
# Delete post
response = requests.delete(f'https://jsonplaceholder.typicode.com/posts/{post_id}')
assert response.status_code == 200
# Verify post has been deleted (try to get deleted post)
get_response = requests.get(f'https://jsonplaceholder.typicode.com/posts/{post_id}')
# Note: JSONPlaceholder is a mock API, so it won't actually delete resources,
# but in a real application, this should return 404
7.5 Concurrent Request Testing
import concurrent.futures
import requests
import time
import pytest
@pytest.mark.performance
def test_api_concurrent_requests():
"""Test API concurrent request performance"""
url = 'https://jsonplaceholder.typicode.com/posts'
num_requests = 20 # Number of concurrent requests
# Define single request function
def make_request(post_id):
start_time = time.time()
response = requests.get(f'{url}/{post_id}')
end_time = time.time()
return {
'status_code': response.status_code,
'time': (end_time - start_time) * 1000, # milliseconds
'post_id': post_id
}
# Set list of post IDs to request
post_ids = list(range(1, num_requests + 1))
# Use thread pool to execute requests concurrently
start_total = time.time()
with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
# Submit all requests and get future objects
future_to_id = {executor.submit(make_request, post_id): post_id for post_id in post_ids}
# Collect all results
results = []
for future in concurrent.futures.as_completed(future_to_id):
post_id = future_to_id[future]
try:
result = future.result()
results.append(result)
# Ensure request was successful
assert result['status_code'] == 200
except Exception as e:
print(f"Request ID {post_id} encountered an error: {e}")
end_total = time.time()
total_time = (end_total - start_total) * 1000 # milliseconds
# Calculate statistics
avg_time = sum(r['time'] for r in results) / len(results)
max_time = max(r['time'] for r in results)
min_time = min(r['time'] for r in results)
# Output results
print(f"Concurrent test results:")
print(f"Total time: {total_time:.2f} ms")
print(f"Average request time: {avg_time:.2f} ms")
print(f"Maximum request time: {max_time:.2f} ms")
print(f"Minimum request time: {min_time:.2f} ms")
# Assert overall performance meets requirements
assert total_time < (num_requests * 1000) # Total time less than theoretical time for sequential execution
7.6 Using API Testing Framework – HTTPie
from httpie.core import main as httpie_main
import pytest
import sys
from io import StringIO
import json
@pytest.fixture
def capture_output():
"""Capture standard output"""
stdout_bak = sys.stdout
sys.stdout = StringIO()
yield sys.stdout
sys.stdout = stdout_bak
def test_httpie_get_request(capture_output):
"""Send GET request using HTTPie and parse response"""
# Set command line arguments
sys.argv = [
'http',
'GET',
'https://jsonplaceholder.typicode.com/posts/1'
]
# Execute HTTPie command
httpie_main()
# Get output and parse JSON
output = capture_output.getvalue()
try:
# Extract JSON part
json_start = output.find('{')
json_end = output.rfind('}') + 1
json_output = output[json_start:json_end]
data = json.loads(json_output)
# Validate response content
assert data['id'] == 1
assert 'title' in data
assert 'body' in data
assert 'userId' in data
except json.JSONDecodeError:
pytest.fail("Unable to parse HTTPie response as JSON")
7.7 WebSocket API Testing
import pytest
import websockets
import asyncio
import json
@pytest.mark.asyncio
async def test_websocket_connection():
"""Test WebSocket connection and message exchange"""
# WebSocket server URL
# Note: This is an example echo server
uri = "wss://echo.websocket.org"
async with websockets.connect(uri) as websocket:
# Send test message
test_message = json.dumps({
"type": "greeting",
"content": "Hello, WebSocket!"
})
await websocket.send(test_message)
# Set timeout
response = await asyncio.wait_for(websocket.recv(), timeout=5.0)
# Validate response (echo server should return the same message)
assert response == test_message
# If parsed as JSON
response_data = json.loads(response)
assert response_data['type'] == 'greeting'
assert response_data['content'] == 'Hello, WebSocket!'
7.8 GraphQL API Testing
import requests
import pytest
def test_graphql_query():
"""Test GraphQL API query"""
# GraphQL endpoint
url = 'https://api.example.com/graphql'
# GraphQL query
query = """query GetUser {
user(id: "1") {
id
name
email
posts {
id
title
}
}
}"""
# Send request
response = requests.post(
url,
json={'query': query},
headers={'Content-Type': 'application/json'}
)
# Assert response status code
assert response.status_code == 200
# Parse response
data = response.json()
# Check for errors
assert 'errors' not in data, f"GraphQL error: {data.get('errors')}"
# Validate data structure
assert 'data' in data
assert 'user' in data['data']
assert data['data']['user']['id'] == '1'
assert 'name' in data['data']['user']
assert 'email' in data['data']['user']
assert 'posts' in data['data']['user']
API Automation Testing Framework Diagram
I will draw an architecture diagram of the API automation testing framework to help you understand how these test scripts are organized into a complete framework.
I am now going to create an architecture diagram for an API automation testing framework.
Conclusion
The 28 Python API automation testing scripts provided cover various common scenarios, from basic HTTP request testing to advanced concurrent testing, performance testing, and framework integration. These scripts can serve as a foundation for building a complete API automation testing framework.
The core components of the API automation testing framework include:
- Basic Request Library: Primarily using the
<span>requests</span>
library to handle HTTP requests - Testing Framework: Using
<span>unittest</span>
or<span>pytest</span>
to organize and execute tests - Data-Driven: Supports loading test data from CSV, JSON, etc.
- Assertion Mechanism: Using built-in assertions or
<span>PyHamcrest</span>
,<span>JSONSchema</span>
for complex assertions - Test Reports: Generating test reports in HTML, Allure, etc.
- Continuous Integration: Integrating with CI/CD tools like Jenkins, GitHub Actions, etc.
When using these scripts, it is recommended to customize and extend them based on project characteristics, especially in test data management, environment configuration, and test reporting. A good project structure and clear code organization are crucial for maintaining a large API automation testing suite.
You can combine these scripts to build a complete testing framework that meets your project needs. As the scale of testing grows, you may also need to consider version control for test data, management of multi-environment configurations, and monitoring and analysis of test results.