The testing pyramid is a useful mental model, but the most surprising thing about it is that its most common interpretation is fundamentally flawed for microservices.
Let’s see how this plays out in practice. Imagine we have a simple e-commerce system with two microservices: OrderService and PaymentService.
Here’s a simplified OrderService:
# order_service.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import requests
import uuid
app = FastAPI()
class Order(BaseModel):
order_id: str
user_id: str
items: list
status: str = "PENDING"
# In-memory "database" for simplicity
orders_db = {}
@app.post("/orders/", response_model=Order)
async def create_order(user_id: str, items: list):
order_id = str(uuid.uuid4())
order = Order(order_id=order_id, user_id=user_id, items=items)
orders_db[order_id] = order
# Simulate calling the payment service
try:
payment_response = requests.post("http://payment-service:8000/payments/", json={"order_id": order_id, "amount": len(items) * 10}) # $10 per item
payment_response.raise_for_status() # Raise an exception for bad status codes
print(f"Payment successful for order {order_id}")
order.status = "PAID"
orders_db[order_id] = order
return order
except requests.exceptions.RequestException as e:
print(f"Payment failed for order {order_id}: {e}")
order.status = "PAYMENT_FAILED"
orders_db[order_id] = order
raise HTTPException(status_code=500, detail="Payment processing failed")
@app.get("/orders/{order_id}", response_model=Order)
async def get_order(order_id: str):
if order_id not in orders_db:
raise HTTPException(status_code=404, detail="Order not found")
return orders_db[order_id]
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8001)
And a simplified PaymentService:
# payment_service.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import uuid
app = FastAPI()
class Payment(BaseModel):
payment_id: str
order_id: str
amount: float
status: str = "SUCCESS"
payments_db = {}
@app.post("/payments/", response_model=Payment)
async def process_payment(order_id: str, amount: float):
payment_id = str(uuid.uuid4())
payment = Payment(payment_id=payment_id, order_id=order_id, amount=amount)
payments_db[payment_id] = payment
print(f"Processed payment {payment_id} for order {order_id} with amount {amount}")
# Simulate a potential failure for demonstration
if amount > 100:
payment.status = "FAILED"
payments_db[payment_id] = payment
raise HTTPException(status_code=400, detail="Payment amount too high")
return payment
@app.get("/payments/{payment_id}", response_model=Payment)
async def get_payment(payment_id: str):
if payment_id not in payments_db:
raise HTTPException(status_code=404, detail="Payment not found")
return payments_db[payment_id]
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Traditionally, the testing pyramid advocates for a broad base of unit tests, a smaller layer of integration tests, and a tiny tip of end-to-end (E2E) tests. The idea is that unit tests are fast and cheap, E2E tests are slow and brittle, so you want to minimize E2E.
However, in a microservices architecture, the fundamental unit of deployment and independent value delivery is the service itself. A "unit test" in a microservice context often means testing a single function in isolation, which doesn’t account for how the service interacts with its dependencies (databases, message queues, other services). A true "integration test" in the traditional sense would mean testing two services together, which rapidly becomes complex and brittle in a distributed system.
This leads to a reframing: the Service Testing Pyramid. At the base, you have unit tests for pure logic within a service. Above that, you have component tests that test a single service in isolation, treating its external dependencies as mocks or stubs. This is the most crucial layer for microservices. At the top, you have contract tests and end-to-end tests that verify interactions between services.
Here’s how you’d typically structure testing for our OrderService:
1. Unit Tests (The Foundation)
These test individual functions or classes within your service without any external dependencies. For OrderService, this might be testing the Order Pydantic model validation or some internal helper functions.
# test_order_service_units.py
from order_service import Order
import pytest
def test_order_model_creation():
order_data = {"order_id": "abc", "user_id": "user123", "items": ["itemA", "itemB"]}
order = Order(**order_data)
assert order.order_id == "abc"
assert order.user_id == "user123"
assert order.items == ["itemA", "itemB"]
assert order.status == "PENDING"
def test_order_model_status_update():
order_data = {"order_id": "def", "user_id": "user456", "items": ["itemC"]}
order = Order(**order_data)
order.status = "SHIPPED"
assert order.status == "SHIPPED"
These are fast, reliable, and test the internal logic of your code.
2. Component Tests (The Core of Microservices Testing)
This layer tests a single microservice in isolation, mocking its external dependencies. For OrderService, this means testing the API endpoints (/orders/, /orders/{order_id}) without actually calling the PaymentService. We’ll use pytest-mock or unittest.mock to achieve this.
First, we need a way to override the requests.post call. A common pattern is to inject the client or use a patching mechanism. Let’s use pytest’s monkeypatching.
# test_order_service_components.py
from fastapi.testclient import TestClient
from order_service import app
import pytest
import requests
client = TestClient(app)
# Mock the external payment service call
def mock_payment_service_success(url, json):
class MockResponse:
def __init__(self, json_data, status_code):
self.json_data = json_data
self.status_code = status_code
def json(self):
return self.json_data
def raise_for_status(self):
if self.status_code >= 400:
raise requests.exceptions.HTTPError(f"Error: {self.status_code}")
print(f"Mocking POST to {url} with data {json}")
return MockResponse({"payment_id": "mock_payment_id", "order_id": json["order_id"], "amount": json["amount"], "status": "SUCCESS"}, 200)
def mock_payment_service_failure(url, json):
class MockResponse:
def __init__(self, json_data, status_code):
self.json_data = json_data
self.status_code = status_code
def json(self):
return self.json_data
def raise_for_status(self):
if self.status_code >= 400:
raise requests.exceptions.HTTPError(f"Error: {self.status_code}")
print(f"Mocking POST to {url} with data {json}")
return MockResponse({"detail": "Payment failed"}, 500)
@pytest.fixture
def mock_requests_post(monkeypatch):
return monkeypatch.setattr(requests, "post", lambda url, json: mock_payment_service_success(url, json))
@pytest.fixture
def mock_requests_post_failure(monkeypatch):
return monkeypatch.setattr(requests, "post", lambda url, json: mock_payment_service_failure(url, json))
def test_create_order_success(mock_requests_post):
response = client.post("/orders/", params={"user_id": "user789", "items": ["itemX", "itemY"]})
assert response.status_code == 200
data = response.json()
assert data["user_id"] == "user789"
assert len(data["items"]) == 2
assert data["status"] == "PAID" # Crucially, the status reflects the mocked payment success
def test_create_order_payment_failure(mock_requests_post_failure):
response = client.post("/orders/", params={"user_id": "user101", "items": ["itemZ"]})
assert response.status_code == 500 # Expecting internal server error due to payment failure
data = response.json()
assert data["detail"] == "Payment processing failed"
# We can also check the state of the order if we had access to its internal DB,
# but for an API test, checking the response is sufficient.
def test_get_order_not_found():
response = client.get("/orders/non-existent-id")
assert response.status_code == 404
assert response.json() == {"detail": "Order not found"}
# To test get_order, we'd need to ensure an order exists in the in-memory DB.
# This often involves a setup step or a separate "create order" component test that
# directly manipulates the service's state for testing purposes.
# For simplicity here, we'll skip directly testing get_order without state setup.
Component tests are the sweet spot for microservices. They verify the service’s API contract and its integration logic with its dependencies (simulated via mocks) without the overhead and brittleness of running the entire distributed system. They are fast, reliable, and give high confidence in the service’s behavior.
3. Contract Tests (Ensuring Inter-Service Agreements)
Contract tests verify that a service adheres to the expectations of its consumers (or providers). If OrderService calls PaymentService, a contract test would ensure OrderService sends data in the format PaymentService expects, and that PaymentService responds in a way OrderService can handle. Tools like Pact are excellent for this.
4. End-to-End Tests (The Tip of the Spear)
E2E tests run the entire system (or a significant subset) and simulate a user journey. For our example, this would involve starting both OrderService and PaymentService (and any other dependent services), then making a request to create an order and verifying the outcome through multiple service interactions.
# Example of an E2E test concept (requires running services, e.g., via Docker Compose)
# This is NOT runnable in a simple script without a full setup.
import requests
import time
def test_e2e_order_payment_flow():
# Assume PaymentService is running on http://payment-service:8000
# Assume OrderService is running on http://order-service:8001
# 1. Create an order
try:
order_response = requests.post("http://order-service:8001/orders/", params={"user_id": "e2e_user", "items": ["item_A", "item_B"]})
order_response.raise_for_status()
order_data = order_response.json()
order_id = order_data["order_id"]
print(f"E2E: Created order {order_id}")
# 2. Verify order status (may need to poll if asynchronous)
time.sleep(2) # Give services time to process, though in reality you'd poll
fetched_order = requests.get(f"http://order-service:8001/orders/{order_id}")
fetched_order.raise_for_status()
assert fetched_order.json()["status"] == "PAID"
# 3. (Optional) Verify payment was processed by checking PaymentService directly
# This requires knowing the payment_id, which might not be directly returned.
# A more robust E2E would involve checking an audit log or a shared database.
# For this example, we'll assume we can find the payment via order_id if PaymentService exposed that.
# In a real system, you might query PaymentService's DB or use a correlation ID.
print(f"E2E: Order {order_id} successfully processed and paid.")
except requests.exceptions.RequestException as e:
pytest.fail(f"E2E test failed due to request error: {e}")
except Exception as e:
pytest.fail(f"E2E test failed with unexpected error: {e}")
E2E tests are expensive to run, brittle, and slow. They should be used sparingly to validate critical user journeys that span multiple services.
The key takeaway is that for microservices, the component test is king. It provides the best balance of speed, reliability, and confidence by testing a service in isolation with mocked dependencies. The traditional pyramid is inverted or reshaped into a "diamond" or "trophy" with a wider middle section for component tests.