Testing Applications with Collaborator Services
In the previous section, we explored testing applications that depend on backing services like databases or cloud providers. This section focuses on another type of external dependency - collaborator services.
A collaborator service is another application or a third-party service that our application directly depends on. They might be developed in-house, running as off-the-shelf software, or a third-party service in an external network. Collaborator services provide a valuable behavior and expose it through an API such as REST. For example, when building a financial system, we might use an external application to obtain currency conversion rates or process credit card payments.
In automated testing, it's often not feasible to use the real versions of external applications - they might be slow, expensive, flaky, unreliable, and they might not have a dedicated test environment. If the external application's source code or executable is available, we might launch it in a local testing environment. However, the external app most probably depends on other applications and backing services. Therefore, to test a single application in isolation, we'll end up running a large and complicated environment of many different applications - this is the problem that Testcontainers help avoid.
Mocking Collaborator Services
During testing as part of the development workflow, we don't want to depend on real versions of external applications because our test environment's complexity will grow, the tests might become flaky, and we won't have any control over the test environment. For example, suppose we want to test that a financial system converts a transaction's currency. The real currency exchange rates change daily, so our financial system's test results will fail every time the new conversion rates are published. To avoid this problem, we must be able to control all variables in the test environment.
One solution we'll explore is mocking an external application by creating a fake version of its API that returns predefined data. This way, we'll simulate requests and responses from the external app and be able to test our application in isolation. To mock external APIs, we can use mocking servers: WireMock, MockServer, VCR.py, mountebank, and many more.
Example: Mocking Customer Credit Check Application
Let's create an example order management application. A customer's credit must be verified during a new order creation process. If the customer has good credit, new order creation is allowed. If the credit check fails, the customer can't create new orders until the credit is improved, e.g., by paying all invoices for previous orders. Since credit verification is a complex process, the order management application doesn't implement it; instead, it uses an external app - the customer credit check service.
We'll use the WireMock HTTP mock server to mock the credit check service's POST /check-credit API.
WireMock is an open-source tool for API mock testing. It can help you to create stable test and development environments,
isolate yourself from flaky third parties, and simulate APIs that don't exist yet.
Tomodachi Testcontainers provides a WireMockContainer
and wiremock_container fixture.
Used together with Python WireMock SDK, creating API mocks is easy.
Warning
Since mocks are configured manually, they might not accurately reflect the behavior of a real system. An application tested only with mocks might not work the same in a production environment. Depending on your use case, consider verifying your test doubles against a real system in a separate test suite or adding contract tests. Check out Pact - a tool for contract testing.
Creating the order management application
The example application has a single endpoint POST /order that expects two string values in the request body: customer_id and product.
On successful order creation, the Order object is returned in a response.
If the credit check fails, the following errors are returned: CREDIT_CHECK_FAILED or CREDIT_CHECK_UNAVAILABLE.
import tomodachi
from aiohttp import web
from .credit_check import CreditCheckUnavailableError, CustomerCreditCheckFailedError
from .services import create_new_order
class Service(tomodachi.Service):
@tomodachi.http("POST", r"/order/?")
async def http_create_order(self, request: web.Request) -> web.Response:
body = await request.json()
try:
order = await create_new_order(
customer_id=body["customer_id"],
product=body["product"],
)
return web.json_response(order.to_dict())
except CustomerCreditCheckFailedError:
return web.json_response({"error": "CREDIT_CHECK_FAILED"}, status=400)
except CreditCheckUnavailableError:
return web.json_response({"error": "CREDIT_CHECK_UNAVAILABLE"}, status=503)
The new order creation service creates a new Order object and calls customer credit verification.
The order is not stored in a database to keep the example simple.
import uuid
from .credit_check import verify_customer_credit
from .domain import Order
async def create_new_order(customer_id: str, product: str) -> Order:
order = Order(
id=str(uuid.uuid4()),
customer_id=customer_id,
product=product,
)
await verify_customer_credit(order.customer_id)
return order
The Order is a simple dataclass object.
from dataclasses import dataclass
@dataclass
class Order:
id: str
customer_id: str
product: str
def to_dict(self) -> dict:
return {
"id": self.id,
"customer_id": self.customer_id,
"product": self.product,
}
The verify_customer_credit function calls the external credit check service via HTTP.
The URL of the service is configured with the CREDIT_CHECK_SERVICE_URL environment variable.
It will help us change the service URL in a test environment.
If the response status code is something other than 2xx, the CreditCheckUnavailableError is raised.
If the credit verification status is not CREDIT_CHECK_PASSED, the CustomerCreditCheckFailedError is raised.
import os
import httpx
class CustomerCreditCheckFailedError(Exception):
pass
class CreditCheckUnavailableError(Exception):
pass
async def verify_customer_credit(customer_id: str) -> None:
async with httpx.AsyncClient(base_url=get_credit_check_service_url()) as client:
response = await client.post(
"/credit-check",
json={"customer_id": customer_id},
)
try:
response.raise_for_status()
except httpx.HTTPStatusError as e:
raise CreditCheckUnavailableError from e
body = response.json()
if body["status"] != "CREDIT_CHECK_PASSED":
raise CustomerCreditCheckFailedError(customer_id)
def get_credit_check_service_url() -> str:
return os.environ["CREDIT_CHECK_SERVICE_URL"]
Configuring Testcontainers
To start the WireMockContainer,
we'll use the wiremock_container fixture.
The credit check service's URL is configured with the CREDIT_CHECK_SERVICE_URL environment variable -
it's set to WireMock's URL, so requests for verifying the customer credit will be sent to
the WireMock instance running locally in a container.
from typing import AsyncGenerator, Generator
import httpx
import pytest
import pytest_asyncio
from tomodachi_testcontainers import DockerContainer, TomodachiContainer, WireMockContainer
@pytest.fixture(scope="session")
def tomodachi_container(
testcontainer_image: str,
wiremock_container: WireMockContainer,
) -> Generator[DockerContainer, None, None]:
with (
TomodachiContainer(testcontainer_image)
.with_env("CREDIT_CHECK_SERVICE_URL", wiremock_container.get_internal_url())
.with_command("tomodachi run getting_started/orders/app.py --production")
) as container:
yield container
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def http_client(tomodachi_container: TomodachiContainer) -> AsyncGenerator[httpx.AsyncClient, None]:
async with httpx.AsyncClient(base_url=tomodachi_container.get_external_url()) as client:
yield client
Writing end-to-end tests
In the first test, we'll test a successful order creation when the customer's credit check passes.
Before testing the order management application, we need to configure the POST /check-check API in WireMock.
To easily configure WireMock, we'll use Python WireMock SDK.
Install it from extras with pip install tomodachi-testcontainers[wiremock] or pip install wiremock.
The wiremock_container
fixture automatically configures the SDK to communicate with the WireMock server if the WireMock extra is installed.
The mock setup code configures WireMock to return JSON body {"status": "CREDIT_CHECK_PASSED"} when it receives
a POST request to the endpoint /credit-check, and the request body is {"customer_id": "123456"}.
After sending the POST /order request to the order service, we receive a successful response indicating that the order has been created.
The order service successfully called WireMock, which returned the fake credit check service's response!
from unittest import mock
import httpx
import pytest
import wiremock.client as wm
@pytest.mark.asyncio(loop_scope="session")
async def test_order_created_when_credit_check_passed(http_client: httpx.AsyncClient) -> None:
mapping = wm.Mapping(
request=wm.MappingRequest(
method=wm.HttpMethods.POST,
url="/credit-check",
body_patterns=[{wm.WireMockMatchers.EQUAL_TO_JSON: {"customer_id": "123456"}}],
),
response=wm.MappingResponse(
status=200,
json_body={"status": "CREDIT_CHECK_PASSED"},
),
)
wm.Mappings.create_mapping(mapping=mapping)
response = await http_client.post(
"/order",
json={"customer_id": "123456", "product": "MINIMALIST-SPOON"},
)
assert response.status_code == 200
assert response.json() == {
"id": mock.ANY,
"customer_id": "123456",
"product": "MINIMALIST-SPOON",
}
Let's test what happens when a customer's credit check verification fails.
We configure the WireMock to respond with JSON body {"status": "CREDIT_CHECK_FAILED"}.
After calling the order management service, we receive the expected HTTP 400 and {"error": "CREDIT_CHECK_FAILED"}.
@pytest.mark.asyncio(loop_scope="session")
async def test_order_not_created_when_credit_check_failed(http_client: httpx.AsyncClient) -> None:
mapping = wm.Mapping(
request=wm.MappingRequest(
method=wm.HttpMethods.POST,
url="/credit-check",
body_patterns=[{wm.WireMockMatchers.EQUAL_TO_JSON: {"customer_id": "123456"}}],
),
response=wm.MappingResponse(
status=200,
json_body={"status": "CREDIT_CHECK_FAILED"},
),
)
wm.Mappings.create_mapping(mapping=mapping)
response = await http_client.post(
"/order",
json={"customer_id": "123456", "product": "MINIMALIST-SPOON"},
)
assert response.status_code == 400
assert response.json() == {"error": "CREDIT_CHECK_FAILED"}
The last example tests the scenario when the credit verification service responds with an Internal Server Error.
This error scenario is very hard to simulate when testing with a real version of the external application.
Using mocks, you control the environment and its behavior.
In this test, the WireMock is configured to return HTTP 500 with the Internal Server Error response;
the order management service returns HTTP 503 and {"error": "CREDIT_CHECK_UNAVAILABLE"}.
@pytest.mark.asyncio(loop_scope="session")
async def test_order_not_created_when_credit_check_service_unavailable(http_client: httpx.AsyncClient) -> None:
mapping = wm.Mapping(
request=wm.MappingRequest(method=wm.HttpMethods.POST, url="/credit-check"),
response=wm.MappingResponse(
status=500,
body="Internal Server Error",
),
)
wm.Mappings.create_mapping(mapping=mapping)
response = await http_client.post(
"/order",
json={"customer_id": "123456", "product": "MINIMALIST-SPOON"},
)
assert response.status_code == 503
assert response.json() == {"error": "CREDIT_CHECK_UNAVAILABLE"}
Extracting mock setup functions
Setting up the mocks requires lengthy boilerplate code, even for these simple examples. In the real scenario, the API mock setup will be tens of code lines configuring nested request/response data structures. It's a good idea to extract mock setup code to separate functions and modules.
import wiremock.client as wm
def customer_credit_check_passes(customer_id: str) -> None:
mapping = wm.Mapping(
request=wm.MappingRequest(
method=wm.HttpMethods.POST,
url="/credit-check",
body_patterns=[{wm.WireMockMatchers.EQUAL_TO_JSON: {"customer_id": customer_id}}],
),
response=wm.MappingResponse(
status=200,
json_body={"status": "CREDIT_CHECK_PASSED"},
),
)
wm.Mappings.create_mapping(mapping=mapping)
def customer_credit_check_fails(customer_id: str) -> None:
mapping = wm.Mapping(
request=wm.MappingRequest(
method=wm.HttpMethods.POST,
url="/credit-check",
body_patterns=[{wm.WireMockMatchers.EQUAL_TO_JSON: {"customer_id": customer_id}}],
),
response=wm.MappingResponse(
status=200,
json_body={"status": "CREDIT_CHECK_FAILED"},
),
)
wm.Mappings.create_mapping(mapping=mapping)
def customer_credit_check_returns_internal_server_error() -> None:
mapping = wm.Mapping(
request=wm.MappingRequest(method=wm.HttpMethods.POST, url="/credit-check"),
response=wm.MappingResponse(
status=500,
body="Internal Server Error",
),
)
wm.Mappings.create_mapping(mapping=mapping)
Now, the tests are shorter and better express their intent.
To isolate tests, we can use the reset_wiremock_container_on_teardown
fixture to delete all WireMock stub mappings between tests.
It ensures that all tests explicitly configure API mocks for their test scenario and don't depend on mocks configured in previously executed tests.
from unittest import mock
import httpx
import pytest
from . import credit_check_mocks
pytestmark = pytest.mark.usefixtures("reset_wiremock_container_on_teardown")
@pytest.mark.asyncio(loop_scope="session")
async def test_order_created_when_credit_check_passed(http_client: httpx.AsyncClient) -> None:
customer_id = "123456"
credit_check_mocks.customer_credit_check_passes(customer_id)
response = await http_client.post(
"/order",
json={"customer_id": customer_id, "product": "MINIMALIST-SPOON"},
)
assert response.status_code == 200
assert response.json() == {
"id": mock.ANY,
"customer_id": customer_id,
"product": "MINIMALIST-SPOON",
}
@pytest.mark.asyncio(loop_scope="session")
async def test_order_not_created_when_credit_check_failed(http_client: httpx.AsyncClient) -> None:
customer_id = "123456"
credit_check_mocks.customer_credit_check_fails(customer_id)
response = await http_client.post(
"/order",
json={"customer_id": customer_id, "product": "MINIMALIST-SPOON"},
)
assert response.status_code == 400
assert response.json() == {"error": "CREDIT_CHECK_FAILED"}
@pytest.mark.asyncio(loop_scope="session")
async def test_order_not_created_when_credit_check_service_unavailable(http_client: httpx.AsyncClient) -> None:
credit_check_mocks.customer_credit_check_returns_internal_server_error()
response = await http_client.post(
"/order",
json={"customer_id": "123456", "product": "MINIMALIST-SPOON"},
)
assert response.status_code == 503
assert response.json() == {"error": "CREDIT_CHECK_UNAVAILABLE"}
Summary
In this guide, we tested an application that directly depends on another system's API. Often, it's not feasible to test the application with the real versions of its dependencies, so API mocking tools like WireMock allow us to isolate our automated test environments by faking external systems. Mocks are a powerful tool because they allow us to control all variables in the test environment. Using mocks, we can simulate different external system's behaviors and error-handling scenarios, which might be hard or impossible to test with a real version of an external system.
However, since mocks are configured manually, they don't guarantee that the external system behaves the same in the production environment. To ensure that mocks are accurate, consider adding a separate test suite for verifying mocks with a real system or adding contract tests with tools like Pact.