From python-toolkit
Python testing best practices with pytest, moto, fixtures, and coverage
npx claudepluginhub infiquetra/infiquetra-claude-plugins --plugin python-toolkitThis skill uses the workspace's default tool permissions.
You are helping the user write comprehensive tests for their Python project using pytest, moto, and modern testing patterns.
Provides Ktor server patterns for routing DSL, plugins (auth, CORS, serialization), Koin DI, WebSockets, services, and testApplication testing.
Conducts multi-source web research with firecrawl and exa MCPs: searches, scrapes pages, synthesizes cited reports. For deep dives, competitive analysis, tech evaluations, or due diligence.
Provides demand forecasting, safety stock optimization, replenishment planning, and promotional lift estimation for multi-location retailers managing 300-800 SKUs.
You are helping the user write comprehensive tests for their Python project using pytest, moto, and modern testing patterns.
Infiquetra Standard: 80% minimum test coverage focusing on:
tests/
├── __init__.py
├── conftest.py # Shared fixtures
├── unit/ # Fast, isolated tests
│ ├── __init__.py
│ ├── test_handler.py
│ ├── test_models.py
│ └── test_utils.py
└── integration/ # Slower tests with dependencies
├── __init__.py
└── test_api_flow.py
Create tests/conftest.py with common fixtures:
import os
import pytest
import boto3
from moto import mock_dynamodb, mock_s3, mock_lambda
from typing import Generator
# Mock AWS credentials
os.environ["AWS_DEFAULT_REGION"] = "us-east-1"
os.environ["AWS_ACCESS_KEY_ID"] = "testing"
os.environ["AWS_SECRET_ACCESS_KEY"] = "testing"
os.environ["AWS_SECURITY_TOKEN"] = "testing"
os.environ["AWS_SESSION_TOKEN"] = "testing"
@pytest.fixture(scope="function")
def aws_credentials():
"""Mocked AWS Credentials for moto."""
os.environ["AWS_ACCESS_KEY_ID"] = "testing"
os.environ["AWS_SECRET_ACCESS_KEY"] = "testing"
@pytest.fixture(scope="function")
def dynamodb_table(aws_credentials):
"""Create mocked DynamoDB table."""
with mock_dynamodb():
dynamodb = boto3.resource("dynamodb", region_name="us-east-1")
table = dynamodb.create_table(
TableName="test-table",
KeySchema=[
{"AttributeName": "pk", "KeyType": "HASH"},
{"AttributeName": "sk", "KeyType": "RANGE"},
],
AttributeDefinitions=[
{"AttributeName": "pk", "AttributeType": "S"},
{"AttributeName": "sk", "AttributeType": "S"},
],
BillingMode="PAY_PER_REQUEST",
)
yield table
@pytest.fixture(scope="function")
def s3_bucket(aws_credentials):
"""Create mocked S3 bucket."""
with mock_s3():
s3 = boto3.client("s3", region_name="us-east-1")
s3.create_bucket(Bucket="test-bucket")
yield s3
@pytest.fixture(scope="function")
def lambda_context():
"""Mock Lambda context."""
class MockLambdaContext:
function_name = "test-function"
function_version = "$LATEST"
aws_request_id = "test-request-id"
memory_limit_in_mb = 128
def get_remaining_time_in_millis(self) -> int:
return 300000
return MockLambdaContext()
@pytest.fixture(scope="function")
def mock_environment_variables(monkeypatch):
"""Set test environment variables."""
monkeypatch.setenv("TABLE_NAME", "test-table")
monkeypatch.setenv("BUCKET_NAME", "test-bucket")
monkeypatch.setenv("LOG_LEVEL", "DEBUG")
# tests/unit/test_handler.py
import json
from src.service.handler import lambda_handler
def test_handler_success(lambda_context, mock_environment_variables):
"""Test successful handler execution."""
event = {
"httpMethod": "GET",
"path": "/users/123",
"pathParameters": {"userId": "123"},
}
response = lambda_handler(event, lambda_context)
assert response["statusCode"] == 200
body = json.loads(response["body"])
assert body["status"] == "success"
def test_handler_error_handling(lambda_context):
"""Test handler error response."""
event = {
"httpMethod": "GET",
"path": "/users",
# Missing required pathParameters
}
response = lambda_handler(event, lambda_context)
assert response["statusCode"] == 400
body = json.loads(response["body"])
assert "error" in body
import pytest
@pytest.mark.parametrize(
"user_id,expected_status,expected_message",
[
("123", 200, None),
("456", 200, None),
("999", 404, "User not found"),
("", 400, "Invalid user ID"),
],
ids=["valid-user-1", "valid-user-2", "not-found", "empty-id"],
)
def test_get_user_various_inputs(
lambda_context,
user_id,
expected_status,
expected_message,
):
"""Test handler with multiple user IDs."""
event = {
"httpMethod": "GET",
"path": f"/users/{user_id}",
"pathParameters": {"userId": user_id} if user_id else {},
}
response = lambda_handler(event, lambda_context)
assert response["statusCode"] == expected_status
if expected_message:
body = json.loads(response["body"])
assert expected_message in body.get("message", "")
# tests/unit/test_db_service.py
import pytest
from src.service.db import DynamoDBService
def test_put_item(dynamodb_table, mock_environment_variables):
"""Test storing item in DynamoDB."""
service = DynamoDBService()
result = service.put_item(
pk="USER#123",
sk="PROFILE",
data={"name": "Test User", "email": "test@example.com"},
)
assert result is True
# Verify item exists
response = dynamodb_table.get_item(
Key={"pk": "USER#123", "sk": "PROFILE"}
)
assert "Item" in response
assert response["Item"]["name"] == "Test User"
def test_get_item_not_found(dynamodb_table, mock_environment_variables):
"""Test getting non-existent item."""
service = DynamoDBService()
item = service.get_item(pk="USER#999", sk="PROFILE")
assert item is None
def test_update_item(dynamodb_table, mock_environment_variables):
"""Test updating existing item."""
# Setup: Create item
dynamodb_table.put_item(
Item={"pk": "USER#123", "sk": "PROFILE", "name": "Old Name"}
)
service = DynamoDBService()
result = service.update_item(
pk="USER#123",
sk="PROFILE",
updates={"name": "New Name"},
)
assert result is True
# Verify update
response = dynamodb_table.get_item(
Key={"pk": "USER#123", "sk": "PROFILE"}
)
assert response["Item"]["name"] == "New Name"
def test_query_by_partition_key(dynamodb_table, mock_environment_variables):
"""Test querying items by partition key."""
# Setup: Add multiple items for same partition
items = [
{"pk": "USER#123", "sk": "PROFILE", "name": "Test User"},
{"pk": "USER#123", "sk": "ORDER#1", "total": 100},
{"pk": "USER#123", "sk": "ORDER#2", "total": 200},
]
for item in items:
dynamodb_table.put_item(Item=item)
service = DynamoDBService()
results = service.query_by_pk(pk="USER#123")
assert len(results) == 3
assert any(item["sk"] == "PROFILE" for item in results)
assert any(item["sk"] == "ORDER#1" for item in results)
# tests/unit/test_s3_service.py
from src.service.storage import S3Service
def test_upload_file(s3_bucket, mock_environment_variables):
"""Test uploading file to S3."""
service = S3Service()
content = b"test file content"
result = service.upload_file(key="test.txt", content=content)
assert result is True
# Verify upload
obj = s3_bucket.get_object(Bucket="test-bucket", Key="test.txt")
assert obj["Body"].read() == content
def test_download_file(s3_bucket, mock_environment_variables):
"""Test downloading file from S3."""
# Setup: Upload test file
s3_bucket.put_object(
Bucket="test-bucket",
Key="test.txt",
Body=b"test content",
)
service = S3Service()
content = service.download_file(key="test.txt")
assert content == b"test content"
def test_file_exists(s3_bucket, mock_environment_variables):
"""Test checking if file exists."""
# Setup: Upload file
s3_bucket.put_object(Bucket="test-bucket", Key="exists.txt", Body=b"test")
service = S3Service()
assert service.file_exists("exists.txt") is True
assert service.file_exists("not-exists.txt") is False
# tests/unit/test_api_client.py
from unittest.mock import patch, Mock
from src.service.api_client import fetch_user_data
def test_fetch_user_success():
"""Test successful API call."""
with patch("requests.get") as mock_get:
mock_get.return_value = Mock(
status_code=200,
json=lambda: {"id": "123", "name": "Test User"},
)
data = fetch_user_data(user_id="123")
assert data["id"] == "123"
assert data["name"] == "Test User"
mock_get.assert_called_once_with(
"https://api.example.com/users/123",
headers={"Authorization": "Bearer token"},
)
def test_fetch_user_api_error():
"""Test API error handling."""
with patch("requests.get") as mock_get:
mock_get.return_value = Mock(status_code=500, text="Server Error")
with pytest.raises(Exception, match="API request failed"):
fetch_user_data(user_id="123")
# tests/conftest.py
@pytest.fixture
def mock_external_api():
"""Mock external API calls."""
with patch("requests.get") as mock_get:
mock_get.return_value = Mock(
status_code=200,
json=lambda: {"status": "success", "data": {}},
)
yield mock_get
# tests/unit/test_service.py
def test_with_mocked_api(mock_external_api):
"""Test using fixture-based mock."""
result = call_external_api()
assert result["status"] == "success"
mock_external_api.assert_called_once()
# tests/integration/test_api_flow.py
import pytest
@pytest.mark.integration
def test_create_and_retrieve_user(
dynamodb_table,
lambda_context,
mock_environment_variables,
):
"""Test full user creation and retrieval flow."""
from src.service.handler import lambda_handler
# Create user
create_event = {
"httpMethod": "POST",
"path": "/users",
"body": '{"name": "Test User", "email": "test@example.com"}',
}
create_response = lambda_handler(create_event, lambda_context)
assert create_response["statusCode"] == 201
# Extract user ID
import json
user_id = json.loads(create_response["body"])["userId"]
# Retrieve user
get_event = {
"httpMethod": "GET",
"path": f"/users/{user_id}",
"pathParameters": {"userId": user_id},
}
get_response = lambda_handler(get_event, lambda_context)
assert get_response["statusCode"] == 200
user_data = json.loads(get_response["body"])
assert user_data["name"] == "Test User"
# Run tests with coverage
pytest --cov=src --cov-report=html --cov-report=term-missing
# View HTML report
open htmlcov/index.html
# Check coverage threshold
pytest --cov=src --cov-fail-under=80
Focus on:
# Example: Testing all branches
def test_validate_age_all_branches():
"""Test all age validation branches."""
assert validate_age(0) is False # Lower bound
assert validate_age(1) is True # Just above
assert validate_age(50) is True # Middle
assert validate_age(120) is True # Upper bound
assert validate_age(121) is False # Just above
assert validate_age(-1) is False # Negative
Define markers in pyproject.toml:
[tool.pytest.ini_options]
markers = [
"unit: Fast unit tests",
"integration: Slower integration tests",
"slow: Very slow tests",
"aws: Tests requiring AWS services",
]
Use markers:
@pytest.mark.unit
def test_fast_function():
"""Fast unit test."""
pass
@pytest.mark.integration
@pytest.mark.slow
def test_full_workflow():
"""Slow integration test."""
pass
Run specific markers:
# Run only unit tests
pytest -m unit
# Run everything except slow
pytest -m "not slow"
# Run unit and integration
pytest -m "unit or integration"
def test_calculate_order_total():
"""Test order total calculation."""
order = Order(items=[
Item(price=10.00),
Item(price=20.00),
])
assert order.calculate_total() == 30.00 # FAILS: Not implemented
class Order:
def __init__(self, items):
self.items = items
def calculate_total(self):
return sum(item.price for item in self.items)
$ pytest tests/unit/test_order.py
✓ test_calculate_order_total PASSED
def test_calculate_total_with_tax():
"""Test total with tax calculation."""
order = Order(items=[Item(price=100)], tax_rate=0.10)
assert order.calculate_total() == 110.00
def test_calculate_total_empty_order():
"""Test empty order."""
order = Order(items=[])
assert order.calculate_total() == 0.00
test_user_creation_with_invalid_email_returns_400def test_example():
# Arrange: Set up test data
user = User(name="Test")
# Act: Execute the operation
result = user.validate()
# Assert: Verify the result
assert result is True
@pytest.mark.parametrizedef test_invalid_input_raises_error():
"""Test that invalid input raises ValueError."""
with pytest.raises(ValueError, match="Invalid user ID"):
process_user(user_id=None)
def test_missing_field_raises_keyerror():
"""Test missing required field."""
with pytest.raises(KeyError):
parse_event({}) # Missing required fields
@pytest.fixture
def sample_user():
"""Create sample user for testing."""
return {
"id": "123",
"name": "Test User",
"email": "test@example.com",
}
def test_with_fixture(sample_user):
"""Test using fixture data."""
assert sample_user["id"] == "123"
assert validate_user(sample_user) is True
import pytest
@pytest.mark.asyncio
async def test_async_function():
"""Test async function."""
result = await fetch_data_async()
assert result is not None
assert "data" in result
Install project in editable mode:
pip install -e .
Ensure AWS credentials are set in conftest.py before importing moto.
View detailed coverage:
coverage html
open htmlcov/index.html
Focus on red/yellow highlighted lines.
@pytest.mark.slow for slow testspytest -m "not slow"