Metadata-Version: 2.4
Name: zato-testing
Version: 4.1.2
Summary: Testing framework for Zato services
Project-URL: Homepage, https://zato.io
Project-URL: Repository, https://github.com/zatosource/zato
Author-email: "Zato Source s.r.o." <info@zato.io>
License-Expression: AGPL-3.0
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: GNU Affero General Public License v3
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Topic :: Software Development :: Testing
Requires-Python: >=3.10
Provides-Extra: dev
Requires-Dist: build; extra == 'dev'
Requires-Dist: hatch-vcs; extra == 'dev'
Requires-Dist: hatchling; extra == 'dev'
Requires-Dist: pytest; extra == 'dev'
Requires-Dist: ruff; extra == 'dev'
Requires-Dist: twine; extra == 'dev'
Description-Content-Type: text/markdown

# Unit-testing framework for Zato services.

zato-testing lets you unit-test [Zato services](https://zato.io) 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

```bash
uv pip install zato-testing
```

## Usage

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

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

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

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

## Setting responses

Single response (defaults to GET):

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

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

Sequential responses (each call returns the next item):

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

Response based on request:

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

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

## Configuration

Set config values using dot-notation:

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

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

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

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

Config values are accessible in services via `self.config`:

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

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

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

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

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

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

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

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

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

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

```python
self.set_response('ldap:my-ldap', [
    {'sAMAccountName': ['user1'], 'mail': ['user1@example.com']},
    {'sAMAccountName': ['user2'], 'mail': ['user2@example.com']},
])
```

## 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:

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

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

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

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

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

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

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

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

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

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