Testing REST connections

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.out.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.