You can unit test your Zato services directly in your IDE, as well as from the command line, or with automated testing in CI/CD pipelines, all without having to run a Zato server
You do it using the zato-testing framework, and in this tutorial you will learn how to simulate external connections, invoke services, and verify their responses
This tutorial will get you started, and separate chapters cover testing REST, SQL, Microsoft 365, LDAP, Jira, more advanced service invocation scenarios, as well as CI/CD integrations
The blueprint has this sample service which invokes a CRM via REST and returns some data from it (in ./myproject/impl/src/api/crm.py):
# -*- coding: utf-8 -*-
# Zato
from zato.server.service import Service
class GetCustomer(Service):
""" Returns customer details from the CRM system.
"""
name = 'crm.customer.get'
input = 'customer_id'
output = 'name', 'email'
def handle(self):
# Get the CRM connection ..
conn_name = 'crm.api'
conn = self.out.rest[conn_name].conn
# .. the data to be sent ..
params = {'id': self.request.input.customer_id}
# .. invoke the CRM ..
response = conn.get(self.cid, params)
# .. and build our response.
self.response.payload.name = response.data['name']
self.response.payload.email = response.data['email']
And here's the test for that service (in ./myproject/testing/tests/test_crm.py):
# -*- coding: utf-8 -*-
# zato-testing
from zato_testing import ServiceTestCase
# Our code
from api.crm import GetCustomer
class TestGetCustomer(ServiceTestCase):
def test_returns_customer_details(self):
# Prepare test data ..
name = 'Alice Johnson'
email = 'alice@example.com'
customer_id = 'CUST-001'
# .. connection the service uses ..
conn_name = 'crm.api'
# .. tell the framework to make use of that data ..
self.set_response(conn_name, {
'name': name,
'email': email,
})
# .. invoke the service ..
service = self.invoke(GetCustomer, {'customer_id': customer_id})
# .. and run our assertions.
self.assertEqual(service.response.payload.name, name)
self.assertEqual(service.response.payload.email, email)
Open the blueprint in VS Code, go to the Testing panel, point it to zato-project-blueprint (use test_*.py for the pattern it'll ask about), and click the play button. The test will run and show the result.
The test framework creates a test environment that mirrors Zato's runtime. When your services invoke external systems, using REST, SQL and similar, these calls are intercepted and, instead of actually making calls to any systems, the framework returns the responses you configured with self.set_response.
From your service's perspective, it doesn't realize anything. Its own code doesn't change, it's the framework that does the job, based on what you tell it via set_response.
So, in the example below, we told set_response to return a specific dictionary of data each time "crm.api" gets invoked, and then we invoke that service.
Next, the service invoked "crm.api", the framework intercepted that call, figured out that "crm.api" is a REST outgoing connection, and returned to the service the dictionary you passed to set_response.
The service proceeded then as usual, being under the impression that it interacted with a real REST endpoint, so it just accepted the dictionary and turned it into a final response, assigned to self.response.payload.
Back to your unit test, you can now create assertions about the response from that service. In this case, we're just confirming that the values equal the expected ones, but you can use anything from the Python's built-in unittest library or from any other Python testing library.
Naturally, set_response has other options as well, and that is what the remainder of this tutorial will be about.
And lastly, once you get the hang of how the testing framework works, it's best that you point your AI to Zato documentation and tell it to generate the tests for you.
In the blueprint project, you can run the ./myproject/impl/scripts/run-tests.sh script to execute the tests from the command line:
You can also run your tests from CI/CD pipelines, such as Azure DevOps or GitHub Actions.
Finally, you should always base your projects on the blueprint, but there may be more advanced scenarios, such as when you want to have a dedicated repository for tests only, as opposed to keeping tests and services in the same place.
To handle this case, you'll simply add zato-testing to your requirements.txt. It's just a Python library, without any dependencies, which you can install into any project.
The self.set_response method configures what your service receives when it calls an external connection or if it calls another service.
Note that the same set_response method is used to provide responses for all the possible connection types, e.g. there is no set_rest_response, set_sql_response, etc. This is possible because there is typically no overlap between connection names, so usually there is no SQL connection called "crm.api" along with a REST connection of the same name.
It means that you can just use the names of the connections as they are with set_response, and the framework will in runtime know which actual type a give name refers to. If, however, you have such conflicting names, it is possible to resolve the conflicts, and it's documented below as well.
Also note that you can call set_response multiple times, this is sometimes handy, e.g. to configure different responses depending on different request data.
Now, let's go through each of the ways set_response can be used.
Return a dictionary on every call to a connection, e.g. no matter how many times you cann "crm.api" the same dictionary will be returned.
By default, responses are for GET requests. Use method for other HTTP methods. Now, a different response will be returned by calls to "crm.api", depending on whether your service uses POST vs. DELETE with self.out.rest.
self.set_response('crm.api', {'id': '123'}, method='POST')
self.set_response('crm.api', {'deleted': True}, method='DELETE')
You can return a specific HTTP status code along with a response.
self.set_response('crm.api', {'error': 'Not found'}, status_code=404)
self.set_response('crm.api', {'error': 'Server error'}, status_code=500)
You can tell the framework to include specific headers in the response:
When your service calls the same connection multiple times, provide a list of responses, and each call returns the next item. If the framework runs out of responses to return, it'll raise an exception.
So, in the example below, you are providing 3 dictionaries, when your service calls "crm.api" the first time, the first dictionary will be used, and so on.
What happens if you call it 4 times? You'll get an exception pointing this fact out to you.
self.set_response('crm.api', [
{'page': 1, 'items': ['a', 'b']},
{'page': 2, 'items': ['c', 'd']},
{'page': 3, 'items': []}
])
Sometimes it's required to return different responses depending on what the request data is. Simply call set_response multiple times, using the request argument, and the framework will understand to return different responses depending on which request is sent from services.
self.set_response('crm.api', {'type': 'premium'}, request={'customer_id': 'CUST-001'})
self.set_response('crm.api', {'type': 'standard'}, request={'customer_id': 'CUST-002'})
Moreover, multiple requests can map to the same response:
self.set_response('crm.api', {'status': 'active'}, request=[
{'customer_id': 'CUST-001'},
{'customer_id': 'CUST-002'}
])
If you don't have any conflicts in names, set_response` will know itself which connection type such and such name actually is.
If you have connections of different types with the same name (e.g., a REST connection and an LDAP connection both named crm.api), prefix the name with the connection type:
self.set_response('rest:crm.api', {'data': '...'})
self.set_response('ldap:crm.api', [{'cn': 'John Doe', 'mail': 'john@example.com'}])
Here are all the possible prefixes:
When your service calls other services using self.invoke or self.invoke_async, you can use the same set_response method to simulate their responses.
For instance, if any other service invokes the services below, it will get a given dictionary on response and the real services won't be invoked at all, so it simply works the same way as with REST, SQL and other connection types.
self.set_response('my.service.get-client', {'name': 'Alice'})
self.set_response('my.service.get-billing', {'balance': 100})
Note that 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.
Write your service - Your service calls external APIs, queries databases, invokes other services. You don't write any special test code. The production code is the test code.
Create a test class - Extend ServiceTestCase and call self.set_response for each external connection. The framework figures out the connection type based on how your service uses it.
Invoke your service - Call self.invoke with your service class and input data. The framework wires up mock responses and runs your service's handle method.
Assert on results - The invoked service is returned to you, so you can check the response payload or any state your service modified.