zato-testing 4.1.2

Testing framework for Zato services

Unit-testing framework for Zato services.

zato-testing lets you unit-test Zato services without running a live server.

Read more about unit-testing with Zato here: https://zato.io/en/docs/4.1/api-testing/index.html

Installation

uv pip install zato-testing

Usage

from zato_testing import ServiceTestCase

from myproject.services import MyService

class TestMyService(ServiceTestCase):

    def test_handle(self):
        self.set_response('my-connection', {'result': 'ok'})

        service = self.invoke(MyService, {'user_id': 123})

        self.assertResponsePayload(service, {'status': 'success'})

    def test_invoke_by_name(self):
        self.set_response('my-connection', {'result': 'ok'})

        service = self.invoke('my.service.name')

Features

• Invoke services by class or by name
• Set responses for REST outgoing connections
• Service-to-service invocation support (sync and async)
• Configuration via dot-notation, ini files, or class attribute
• Class-level input definitions
• Crypto utilities (generate_secret)
• REST connection .conn pattern support
• Caching support (default and named caches)
• Test services in isolation
• No Zato server required
• Compatible with standard unittest

Invoking services

By class:

service = self.invoke(MyService, {'user_id': 123})

By name (service must be registered first, e.g. by invoking it by class):

service = self.invoke('my.service.name')

Setting responses

Single response (defaults to GET):

self.set_response('billing-api', {'balance': 100})
self.set_response('billing-api', {'created': True}, method='POST', status_code=201)

List response (returns the entire list):

self.set_response('billing-api', [{'id': 1}, {'id': 2}, {'id': 3}])

Sequential responses (each call returns the next item):

self.set_response('billing-api', {
    1: {'balance': 100},
    2: {'balance': 75},
    3: {'balance': 50},
})

Response based on request:

self.set_response('billing-api', {'balance': 100},
    request={'user_id': 1, 'account': 'checking'}
)

self.set_response('billing-api', {'balance': 50},
    request={'user_id': 2, 'account': 'savings'}
)

Multiple requests with same response:

self.set_response('billing-api', {'balance': 100},
    request=[
        {'user_id': 1},
        {'user_id': 2},
    ]
)

Configuration

Set config values using dot-notation:

self.set_config('myapp.storage.account_url', 'https://test.blob.core.windows.net')
self.set_config('myapp.storage.account_key', 'test-key')

Or load from an ini file:

self.set_config('/path/to/config.ini')

The ini file format uses sections as dot-notation paths:

[myapp.storage]
account_url = https://test.blob.core.windows.net
account_key = test-key

Config values are accessible in services via self.config:

class MyService(Service):
    def handle(self):
        url = self.config.myapp.storage.account_url

Service-to-service invocation

Services can invoke other services using self.invoke:

class CallerService(Service):
    name = 'caller.service'

    def handle(self):
        result = self.invoke(HelperService, {'value': 10})
        self.response.payload = {'got': result}

The invoked service receives the same config and REST response registry.

invoke_async works the same as invoke in test mode (runs synchronously).

Class-level input definitions

Services can define input models:

from dataclasses import dataclass
from zato_testing.service import Model, Service

@dataclass(init=False)
class MyInput(Model):
    name: str
    value: int

class MyService(Service):
    input = MyInput

    def handle(self):
        name = self.request.input.name
        value = self.request.input.value

When invoking, pass a dict or the model instance:

service = self.invoke(MyService, {'name': 'test', 'value': 42})

Config class attribute

Instead of calling set_config in each test, use a class attribute to load config from an ini file:

class TestMyService(ServiceTestCase):
    config = '/path/to/test_config.ini'

    def test_handle(self):
        # Config is already loaded from the ini file
        service = self.invoke(MyService)

Crypto utilities

Services have access to crypto utilities via self.crypto:

class MyService(Service):
    def handle(self):
        secret = self.crypto.generate_secret(bits=256)

REST connection .conn pattern

The .conn pattern is supported for compatibility with Zato's connection API:

class MyService(Service):
    def handle(self):
        conn = self.out.rest['my-api'].conn
        response = conn.post(self.cid, data={'key': 'value'})

Caching

Services have access to caching via self.cache:

class MyService(Service):
    def handle(self):
        cache = self.cache.default

        if cache.get('my_key'):
            self.response.payload = cache.get('my_key')
        else:
            result = self.invoke('other.service')
            cache.set('my_key', result, 60)  # 60 second expiry
            self.response.payload = result

Named caches:

cache = self.cache.get_cache('builtin', 'my.cache.name')
cache.set('key', 'value')

Cache is shared across all service invocations within a single test method.

LDAP connections

Services can use LDAP connections via self.out.ldap:

class MyService(Service):
    def handle(self):
        with self.out.ldap['my-ldap'].conn.get() as conn:
            if conn.search('dc=example,dc=com', '(cn=*)'):
                entries = conn.entries

Set LDAP responses with set_response using the ldap: prefix:

self.set_response('ldap:my-ldap', [
    {'sAMAccountName': ['user1'], 'mail': ['[e-mail]']},
    {'sAMAccountName': ['user2'], 'mail': ['[e-mail]']},
])

Connection type conflict detection

By default, connection names are assumed to be unique across types (REST, LDAP, etc.). If you use the same name for different connection types without a prefix, an error is raised:

self.set_response('my-conn', {'data': 'rest'})  # Registers as REST
self.set_response('ldap:my-conn', [...])        # OK - explicit prefix bypasses conflict

If you need the same name for different types, use explicit prefixes:

self.set_response('rest:shared-name', {'data': 'rest'})
self.set_response('ldap:shared-name', [{'data': 'ldap'}])

SQL connections

Services can use SQL connections via self.outgoing.sql:

class MyService(Service):
    def handle(self):
        conn = self.outgoing.sql.get('my-db')
        session = conn.session()
        result = session.execute('SELECT * FROM users')
        session.close()

Jira cloud connections

Services can use Jira connections via self.cloud.jira:

class MyService(Service):
    def handle(self):
        jira = self.cloud.jira['my-jira']
        with jira.conn.client() as client:
            result = client.jql(jql='project=TEST', fields=['key', 'summary'])

Time utilities

Services have access to time utilities via self.time:

class MyService(Service):
    def handle(self):
        today = self.time.today()  # Returns 'YYYY-MM-DD'
        now = self.time.now()      # Returns 'YYYY-MM-DDTHH:mm:ss'
        utc = self.time.utcnow()   # Returns UTC time

MS365 connections (SharePoint, OneDrive, Teams, etc.)

Services can use MS365 connections via self.cloud.ms365:

class MyService(Service):
    def handle(self):
        conn = self.cloud.ms365.get('O365.Sharepoint').conn
        with conn.client() as client:
            site = client.impl.sharepoint().get_site('sites/my-site')
            sp_list = site.get_list_by_name('Suppliers')
            items = sp_list.get_items()

To mock MS365 responses, use set_response with the full method chain path:

class TestMyService(ServiceTestCase):
    def test_sharepoint(self):
        # Configure responses for the method chain
        self.set_response('O365.Sharepoint.sharepoint.get_site', {'id': 'site-123'})
        self.set_response('O365.Sharepoint.sharepoint.get_site.get_list_by_name', {'id': 'list-456'})
        self.set_response('O365.Sharepoint.sharepoint.get_site.get_list_by_name.get_items', [
            {'Title': 'Item 1', 'Status': 'Active'},
            {'Title': 'Item 2', 'Status': 'Inactive'},
        ])

        service = self.invoke(MyService)
        # ...

Request matching is also supported:

self.set_response(
    'O365.Sharepoint.sharepoint.get_site.get_list_by_name.create_list_item',
    response={'id': 'new-item-123'},
    request={'Title': 'New Item', 'Status': 'Active'}
)

If no response is configured for a method chain, an exception is raised with the full path.

RESTAdapter and BusinessCentralAdapter

To use RESTAdapter or BusinessCentralAdapter, first generate them from the Zato source:

make generate

This extracts the adapter classes from the Zato codebase using inspect.getsource(). The generated file is src/zato_testing/adapters.py.

Then import and use them:

from zato_testing.adapters import RESTAdapter

class MyAdapter(RESTAdapter):
    name = 'my.adapter'
    conn_name = 'my-connection'
    method = 'GET'

    def get_path_params(self, params):
        return {'id': self.request.input.id}

    def map_response(self, data, **kwargs):
        return {'result': data}