# Testing services that call other services

Service orchestration tests - mocking inner services, async calls, data models and configuration.

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](https://zato.io/docs/api-testing/tutorial.html) first.

## How it works {#how-it-works}

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.

## Testing orchestrating services {#testing-orchestrating-services}

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:

```python
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:

```python
class GetCustomer(Service):
    name = 'customer.get'
    input = 'customer_id'

    def handle(self):
        conn = self.rest['crm.api']
        response = conn.get(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.rest['orders.api']
        response = conn.get(params={'customer_id': self.request.input.customer_id})
        self.response.payload = response.data
```

The test sets up responses for each API:

```python
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.

## Mocking service responses directly {#mocking-service-responses-directly}

Instead of setting up responses for the external APIs that inner services call, you can mock the inner service responses directly:

```python
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.

## Async service calls {#async-service-calls}

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.

```python
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}
```

```python
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)
```

## Testing with data models {#testing-with-data-models}

When services use [data models](https://zato.io/docs/dev/model/index.html), pass input as dictionaries and the framework handles conversion:

```python
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'}
```

```python
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:

```python
service = self.invoke(CreateOrder, customer_id='CUST-001', product_id='PROD-123', quantity=2)
```

## Testing output models {#testing-output-models}

When services define output models, the response payload is an instance of that model:

```python
@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
```

```python
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)
```

## Working with configuration {#working-with-configuration}

Services often read configuration values, such as API versions, feature flags, or connection timeouts. There are two ways to provide configuration in tests.

### Auto-loading from the blueprint {#auto-loading-from-the-blueprint}

If your tests are in the [blueprint project](https://github.com/zatosource/zato-project-blueprint), the framework automatically loads configuration from `config/user-conf/*.ini` files. Your services can access `self.config` without any additional setup.

### Setting configuration manually {#setting-configuration-manually}

For values not in your config files, or to override them in specific tests, use `self.set_config()`:

```python
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.

## Learn more {#learn-more}

- [API testing](https://zato.io/docs/api-testing/) - Tests in plain English, run against real or mocked connections
- [API testing tutorial](https://zato.io/docs/api-testing/tutorial.html) - A complete test suite, from the first scenario to the report
- [Testing REST connections](https://zato.io/docs/api-testing/rest.html) - Mock outgoing REST calls and assert on what your service sent
- [Testing SQL connections](https://zato.io/docs/api-testing/sql.html) - Mock database queries and the rows they return
- [Testing LDAP connections](https://zato.io/docs/api-testing/ldap.html) - Mock directory lookups and the entries they find
- [Testing Microsoft 365](https://zato.io/docs/api-testing/ms365/) - Mock mailboxes, calendars, SharePoint, Teams and the rest
- [Testing Jira connections](https://zato.io/docs/api-testing/jira.html) - Mock Jira API calls and the issues they return
- [Tests in CI/CD](https://zato.io/docs/api-testing/ci-cd.html) - Running the suite in a pipeline and reading its output