When your Zato service orchestrates other services - calling one to get customer data, another to check inventory, another to process payment - you need to test that coordination logic. This page shows how to mock service responses and test complex workflows where services call services.
Note: If you're new to unit testing with Zato, check the tutorial first.
When your service calls another service using self.invoke() or self.invoke_async(), the framework runs both services with the same test configuration. Any responses you set up with self.set_response() are available to all services in the call chain.
This means you can test complex orchestration patterns where one service coordinates multiple others, each potentially calling external APIs.
Real integration services often coordinate multiple operations. A service might fetch customer data, then fetch their orders, then combine the results.
Here's an orchestrating service:
from zato.server.service import Service
class GetCustomerDetails(Service):
name = 'customer.get-details'
input = 'customer_id'
def handle(self):
customer = self.invoke('customer.get', customer_id=self.request.input.customer_id)
orders = self.invoke('orders.get-by-customer', customer_id=self.request.input.customer_id)
self.response.payload = {
'customer': customer,
'order_count': len(orders)
}
And the services it calls:
class GetCustomer(Service):
name = 'customer.get'
input = 'customer_id'
def handle(self):
conn = self.out.rest['crm.api'].conn
response = conn.get(self.cid, params={'id': self.request.input.customer_id})
self.response.payload = response.data
class GetOrdersByCustomer(Service):
name = 'orders.get-by-customer'
input = 'customer_id'
def handle(self):
conn = self.out.rest['orders.api'].conn
response = conn.get(self.cid, params={'customer_id': self.request.input.customer_id})
self.response.payload = response.data
The test sets up responses for each API:
from zato_testing import ServiceTestCase
class TestGetCustomerDetails(ServiceTestCase):
def test_combines_customer_and_orders(self):
# Set up responses for both APIs
self.set_response('crm.api', {'id': 'CUST-001', 'name': 'Acme Corp'})
self.set_response('orders.api', [
{'order_id': 'ORD-001'},
{'order_id': 'ORD-002'}
])
# Invoke the orchestrating service
service = self.invoke(GetCustomerDetails, customer_id='CUST-001')
# Verify the combined result
self.assertEqual(service.response.payload['customer']['name'], 'Acme Corp')
self.assertEqual(service.response.payload['order_count'], 2)
The framework runs all three services in sequence, each receiving the responses you configured. This lets you test complex workflows without any external dependencies.
Instead of setting up responses for the external APIs that inner services call, you can mock the inner service responses directly:
class TestGetCustomerDetails(ServiceTestCase):
def test_with_mocked_service_responses(self):
# Mock the inner service responses directly
self.set_response('customer.get', {'id': 'CUST-001', 'name': 'Acme Corp'})
self.set_response('orders.get-by-customer', [
{'order_id': 'ORD-001'},
{'order_id': 'ORD-002'}
])
service = self.invoke(GetCustomerDetails, customer_id='CUST-001')
self.assertEqual(service.response.payload['customer']['name'], 'Acme Corp')
self.assertEqual(service.response.payload['order_count'], 2)
This approach is useful when you want to test the orchestrating service in isolation, without running the inner services at all.
For self.invoke_async(), the framework runs the service synchronously during tests to keep them deterministic, but returns an async correlation ID as it would in production.
class NotifyCustomer(Service):
name = 'customer.notify'
def handle(self):
# In production, this runs asynchronously
cid = self.invoke_async('email.send', to=self.request.input.email, subject='Hello')
self.response.payload = {'correlation_id': cid}
class TestNotifyCustomer(ServiceTestCase):
def test_returns_correlation_id(self):
self.set_response('email.send', {'status': 'queued'})
service = self.invoke(NotifyCustomer, email='alice@example.com')
# The correlation ID is returned even though the service ran synchronously
self.assertIn('correlation_id', service.response.payload)
When services use data models, pass input as dictionaries and the framework handles conversion:
from dataclasses import dataclass
from zato.server.service import Model, Service
@dataclass(init=False)
class CreateOrderRequest(Model):
customer_id: str
product_id: str
quantity: int
class CreateOrder(Service):
name = 'order.create'
input = CreateOrderRequest
def handle(self):
request = self.request.input
# ... create order logic
self.response.payload = {'order_id': 'ORD-001', 'status': 'created'}
from zato_testing import ServiceTestCase
class TestCreateOrder(ServiceTestCase):
def test_creates_order(self):
service = self.invoke(CreateOrder, {
'customer_id': 'CUST-001',
'product_id': 'PROD-123',
'quantity': 2
})
self.assertEqual(service.response.payload['status'], 'created')
The framework converts the dictionary to your model class automatically. You can also pass keyword arguments directly:
When services define output models, the response payload is an instance of that model:
@dataclass(init=False)
class OrderResponse(Model):
order_id: str
status: str
total: float
class CreateOrder(Service):
name = 'order.create'
input = CreateOrderRequest
output = OrderResponse
def handle(self):
self.response.payload.order_id = 'ORD-001'
self.response.payload.status = 'created'
self.response.payload.total = 99.99
class TestCreateOrder(ServiceTestCase):
def test_output_model(self):
service = self.invoke(CreateOrder, customer_id='CUST-001', product_id='PROD-123', quantity=2)
self.assertEqual(service.response.payload.order_id, 'ORD-001')
self.assertEqual(service.response.payload.status, 'created')
self.assertEqual(service.response.payload.total, 99.99)
Services often read configuration values, such as API versions, feature flags, or connection timeouts. There are two ways to provide configuration in tests.
If your tests are in the blueprint project, the framework automatically loads configuration from config/user-conf/*.ini files. Your services can access self.config without any additional setup.
For values not in your config files, or to override them in specific tests, use self.set_config():
from zato_testing import ServiceTestCase
class TestGetCustomer(ServiceTestCase):
def test_uses_configured_api_version(self):
self.set_config('crm.api_version', 'v2')
self.set_response('crm.api', {'id': 'CUST-001', 'name': 'Acme Corp'})
service = self.invoke(GetCustomerWithConfig, customer_id='CUST-001')
self.assertEqual(service.response.payload['name'], 'Acme Corp')
You can test different configuration scenarios by changing the values in each test method.