Schedule a demo
Version 4.1 Documentation

Testing SQL connections

When your Zato service queries databases - selecting records, running stored procedures, or executing complex joins - you need to test that logic without a real database. This page shows how to mock SQL results so your tests run fast, don't require database setup, and can simulate any data scenario.

Note: If you're new to unit testing with Zato, check the tutorial first.

Basic usage

Services access SQL connections like this:

from zato.server.service import Service

class GetUsers(Service):
    name = 'users.get-all'

    def handle(self):
        conn = self.outgoing.sql.get('my-database')
        session = conn.session()

        results = session.execute('SELECT * FROM users')
        session.close()

        self.response.payload = {'users': list(results)}

Mock the SQL results in your test:

from zato_testing import ServiceTestCase
from myapp.services import GetUsers

class TestGetUsers(ServiceTestCase):

    def test_returns_users(self):

        # Configure SQL results
        self.set_response('sql:my-database', [
            {'id': 1, 'name': 'Alice', 'email': 'alice@example.com'},
            {'id': 2, 'name': 'Bob', 'email': 'bob@example.com'}
        ])

        service = self.invoke(GetUsers)

        users = service.response.payload['users']
        self.assertEqual(len(users), 2)
        self.assertEqual(users[0]['name'], 'Alice')

Note the sql: prefix to distinguish SQL connections from REST connections.

Stored procedures

Mock stored procedure results the same way:

class GetSalesReport(Service):
    name = 'reports.sales'

    def handle(self):
        conn = self.outgoing.sql.get('reporting-db')
        session = conn.session()

        results = session.execute('EXECUTE dbo.GetSalesReport @year=2024')
        session.close()

        self.response.payload = {'report': list(results)}

class TestGetSalesReport(ServiceTestCase):

    def test_returns_report_data(self):

        self.set_response('sql:reporting-db', [
            {'month': 'January', 'total': 50000},
            {'month': 'February', 'total': 62000},
            {'month': 'March', 'total': 58000}
        ])

        service = self.invoke(GetSalesReport)

        report = service.response.payload['report']
        self.assertEqual(report[0]['total'], 50000)

Multiple queries

When a service executes multiple queries, provide sequential responses:

class GetDashboardData(Service):
    name = 'dashboard.get'

    def handle(self):
        conn = self.outgoing.sql.get('main-db')
        session = conn.session()

        users = session.execute('SELECT COUNT(*) as count FROM users')
        orders = session.execute('SELECT COUNT(*) as count FROM orders')
        session.close()

        self.response.payload = {
            'user_count': users[0]['count'],
            'order_count': orders[0]['count']
        }

class TestGetDashboardData(ServiceTestCase):

    def test_returns_counts(self):

        # First query returns user count, second returns order count
        self.set_response('sql:main-db', [
            [{'count': 150}],
            [{'count': 1200}]
        ])

        service = self.invoke(GetDashboardData)

        self.assertEqual(service.response.payload['user_count'], 150)
        self.assertEqual(service.response.payload['order_count'], 1200)

Empty results

Test handling of empty result sets:

class TestNoResults(ServiceTestCase):

    def test_handles_empty_results(self):

        self.set_response('sql:my-database', [])

        service = self.invoke(GetUsers)

        self.assertEqual(service.response.payload['users'], [])

See also