Testing REST connections
REST mocks - responses, status codes, headers, sequential calls and request matching.
When your Zato service calls external REST APIs - whether it's a CRM, payment gateway, or any other HTTP endpoint - you need to test that logic without making real network calls. This page shows how to mock REST responses so your tests run fast, work offline, and don't require API credentials.
Note: If you're new to unit testing with Zato, check the tutorial first.
Basic usage
Use set_response() to configure what a REST connection returns:
from zato_testing import ServiceTestCase
from myapp.services import GetCustomer
class TestGetCustomer(ServiceTestCase):
def test_returns_customer(self):
# Mock the REST connection
self.set_response('crm.api', {'id': 'CUST-001', 'name': 'Acme Corp'})
service = self.invoke(GetCustomer, customer_id='CUST-001')
self.assertEqual(service.response.payload['name'], 'Acme Corp')
The connection name must match exactly what your service uses in self.rest['connection-name'].
Response options
Status codes
Set HTTP status codes with the status_code parameter:
# Success with 201 Created
self.set_response('api', {'id': '123'}, status_code=201)
# Error response
self.set_response('api', {'error': 'Not found'}, status_code=404)
Headers
Include response headers:
self.set_response('api', {'data': 'value'}, headers={
'X-Request-Id': 'abc-123',
'X-Rate-Limit-Remaining': '99'
})
HTTP methods
By default, responses apply to GET requests. Specify other methods:
# POST request
self.set_response('api', {'created': True}, method='POST')
# PUT request
self.set_response('api', {'updated': True}, method='PUT')
# DELETE request
self.set_response('api', {'deleted': True}, method='DELETE')
Sequential responses
When a service calls the same connection multiple times, provide a list:
def test_pagination(self):
# First call returns page 1, second call returns page 2
self.set_response('api', [
{'items': ['a', 'b'], 'page': 1, 'has_more': True},
{'items': ['c', 'd'], 'page': 2, 'has_more': False}
])
service = self.invoke(FetchAllPages)
self.assertEqual(service.response.payload['total_items'], 4)
Request matching
Return different responses based on request data:
def test_different_actions(self):
# Response for create action
self.set_response('api',
response={'id': 'NEW-001', 'action': 'created'},
request={'action': 'create', 'name': 'New Item'}
)
# Response for update action
self.set_response('api',
response={'id': 'EXIST-001', 'action': 'updated'},
request={'action': 'update', 'id': 'EXIST-001'}
)
# Service sends create request - gets create response
service = self.invoke(ProcessItem, action='create', name='New Item')
self.assertEqual(service.response.payload['action'], 'created')
Multiple requests, same response
Match multiple request patterns to the same response:
self.set_response('api',
response={'status': 'ok'},
request=[
{'type': 'ping'},
{'type': 'health'},
{'type': 'status'}
]
)
Testing RESTAdapter
Services using RESTAdapter are tested the same way:
from zato.server.service import RESTAdapter
class GetFlight(RESTAdapter):
name = 'flight.get'
conn_name = 'flight.ops'
input = 'flight_id'
def get_params(self):
return {'id': self.request.input.flight_id}
def map_response(self, data):
return {
'flight': data['flight_number'],
'status': data['current_status']
}
class TestGetFlight(ServiceTestCase):
def test_maps_response(self):
self.set_response('flight.ops', {
'flight_number': 'FL-123',
'current_status': 'boarding'
})
service = self.invoke(GetFlight, flight_id='FL-123')
self.assertEqual(service.response.payload['flight'], 'FL-123')
self.assertEqual(service.response.payload['status'], 'boarding')
Missing responses
If a service calls a connection without a configured response, the framework raises an exception:
ValueError: No REST response configured for connection 'crm.api'.
Use set_response('crm.api', <response>) to configure it.
This helps catch missing mocks early.