The most surprising thing about microservices integration testing is that you don’t need to run all your services to test their boundaries.
Imagine you have a User Service that talks to an Order Service to get a user’s order history. When you’re testing the User Service’s integration with the Order Service, you’re really just interested in how the User Service behaves when it calls the Order Service’s API. Does it send the right request? Does it handle the response (both success and error) correctly?
Here’s a simplified User Service written in Python using Flask, which needs to fetch order data:
import requests
class OrderServiceClient:
def __init__(self, order_service_url):
self.order_service_url = order_service_url
def get_user_orders(self, user_id):
try:
response = requests.get(f"{self.order_service_url}/users/{user_id}/orders")
response.raise_for_status() # Raise an exception for bad status codes (4xx or 5xx)
return response.json()
except requests.exceptions.RequestException as e:
print(f"Error fetching orders for user {user_id}: {e}")
return None
# Example usage in a Flask app
from flask import Flask, jsonify
app = Flask(__name__)
order_client = OrderServiceClient("http://localhost:5001") # Assuming Order Service is on port 5001
@app.route("/users/<user_id>/orders")
def user_orders_endpoint(user_id):
orders = order_client.get_user_orders(user_id)
if orders:
return jsonify({"user_id": user_id, "orders": orders})
else:
return jsonify({"error": "Could not retrieve orders"}), 500
if __name__ == "__main__":
app.run(port=5000)
Now, to test the integration boundary, we don’t need a fully functional Order Service. We can use a technique called stubbing or mocking. This involves replacing the actual Order Service with a controlled, predictable stand-in that we can dictate the responses of.
Here’s how you’d write an integration test for the User Service using pytest and requests-mock:
import pytest
import requests_mock
from your_user_service_module import OrderServiceClient, app # Assuming your Flask app is in 'your_user_service_module.py'
@pytest.fixture
def mock_order_service():
with requests_mock.Mocker() as m:
yield m
def test_get_user_orders_success(mock_order_service):
user_id = "user123"
mock_order_response = [
{"order_id": "ord001", "item": "Laptop", "quantity": 1},
{"order_id": "ord002", "item": "Mouse", "quantity": 2}
]
# Configure the mock to respond to the specific URL
mock_order_service.get(f"http://localhost:5001/users/{user_id}/orders", json=mock_order_response, status_code=200)
client = OrderServiceClient("http://localhost:5001")
orders = client.get_user_orders(user_id)
assert orders == mock_order_response
assert mock_order_service.called # Verify the mock was called
assert mock_order_service.request_history[0].method == 'GET'
assert mock_order_service.request_history[0].url == f"http://localhost:5001/users/{user_id}/orders"
def test_get_user_orders_not_found(mock_order_service):
user_id = "user456"
mock_order_service.get(f"http://localhost:5001/users/{user_id}/orders", status_code=404)
client = OrderServiceClient("http://localhost:5001")
orders = client.get_user_orders(user_id)
assert orders is None
assert mock_order_service.called
assert mock_order_service.request_history[0].status_code == 404
def test_get_user_orders_service_error(mock_order_service):
user_id = "user789"
mock_order_service.get(f"http://localhost:5001/users/{user_id}/orders", status_code=500, text="Internal Server Error")
client = OrderServiceClient("http://localhost:5001")
orders = client.get_user_orders(user_id)
assert orders is None
assert mock_order_service.called
assert mock_order_service.request_history[0].status_code == 500
In this test, requests_mock intercepts outgoing HTTP requests from our OrderServiceClient. We tell it, "When OrderServiceClient tries to GET http://localhost:5001/users/user123/orders, don’t actually send it anywhere. Instead, pretend you got back a 200 OK response with this specific JSON payload." This allows us to isolate the User Service’s logic and test its interactions with the Order Service’s API contract without needing the Order Service to be running or even exist.
The core problem this solves is avoiding the combinatorial explosion of dependencies in microservices. If ServiceA depends on ServiceB, which depends on ServiceC, and you want to test ServiceA, you’d traditionally need B and C running. With mocking, you only need ServiceA running, and you mock the responses of B and C. This drastically speeds up test execution and simplifies test setup. You control the inputs and observe the outputs at the boundary.
The mental model here is that each microservice is a black box with defined inputs and outputs. Integration testing at the boundary is about verifying that the inputs you send to another service are correct according to its API contract, and that you can gracefully handle the various outputs it might return, including errors. You’re not testing the internal logic of the other service; you’re testing your service’s ability to interact with it. This is often called "contract testing" from the perspective of the consumer.
What most people don’t realize is that you can also mock outgoing requests from the Flask app’s endpoint itself, not just the client object. This is useful if your endpoint logic directly calls another service or if you want to test how your endpoint handles errors returned by downstream services. You’d typically use a testing client provided by the web framework (like Flask’s app.test_client()) and combine it with requests-mock or a similar library to intercept the actual HTTP calls made by your application code during the test.
Ultimately, this approach allows you to build confidence in how your services communicate without the overhead of orchestrating complex distributed environments for every test run.
The next step is to consider testing the other side of the boundary: what happens when your service is the one being called by another service, and how do you ensure its contract is met?