Testing

Response Injection

Turu supports MockConnection for all of the database adapters.

MockConnection has an inject_response method that allows you to write automated tests by injecting the return value corresponding to the Row type specified in the Cursor.execute_map / Cursor.executemany_map.

import turu.sqlite3
from pydantic import BaseModel


class Row(BaseModel):
    id: int
    name: str


expected1 = [Row(id=1, name="a"), Row(id=2, name="b")]
expected2 = [Row(id=3, name="c"), Row(id=4, name="d")]
expected3 = [Row(id=5, name="e"), Row(id=6, name="f")]

connection = turu.sqlite3.MockConnection()

(
    connection.chain()
    .inject_response(Row, expected1)
    .inject_response(Row, expected2)
    .inject_response(Row, expected3)
)

for expected in [expected1, expected2, expected3]:
    with connection.cursor() as cursor:
        assert cursor.execute_map(Row, "select 1, 'a'").fetchall() == expected

Tip

The MockConnection.chain method is used to make method chains more readable. It improves code readability when using black formatter.

For queries that do not require a return value, such as INSERT, MockConnection.inject_response can be injected as None.

import pydantic
import turu.snowflake


class User(pydantic.BaseModel):
    id: int
    name: str


def your_logic(connection: turu.snowflake.Connection):
    with connection.cursor() as cursor:
        user = cursor.execute_map(
            User,
            "SELECT * FROM users WHERE id = 1",
        ).fetchone()

        print(user)


def test_your_logic(connection: turu.snowflake.MockConnection):
    connection.inject_response(User, [User(id=1, name="taro")])

    your_logic(connection)

Operation Injection

How can I teach MockConnection about operations that do not have a return value, such as INSERT, UPDATE, and DELETE?

For this purpose, Cursor.execute_with_tag and MockConnection.inject_operation_with_tag are provided.

By injecting a tag instead of a return value, MockConnection can determine the type of operation and test whether the calls are made in the intended order.

import pydantic
import turu.snowflake
from turu.core import tag


class User(pydantic.BaseModel):
    id: int
    name: str


def your_logic(connection: turu.snowflake.Connection):
    with connection.cursor() as cursor:
        cursor.execute_with_tag(
            tag.Update[User],
            "UPDATE users SET name = 'jiro' WHERE id = 1",
        ).fetchone()


def test_your_logic(connection: turu.snowflake.MockConnection):
    connection.inject_operation_with_tag(tag.Update[User])

    your_logic(connection)

Recording & Testing

In the production code, the actual rows can be recorded to a csv file using the record_to_csv method.

Recording on/off can be controlled with the enable option (default is True).

In the test code, the recorded csv is available using the MockConnection.inject_response_from_csv method.

import os

import turu.sqlite3
from pydantic import BaseModel
from turu.core.record import record_to_csv


class Row(BaseModel):
    id: int
    name: str


# Production code
def do_something(connection: turu.sqlite3.Connection):
    with record_to_csv(
        "test.csv",
        connection.cursor(),
        enable=os.environ.get("ENABLE_RECORDING"),
        limit=100,
    ) as cursor:
        cursor.execute_map(Row, "SELECT 1, 'a'")


# Test code
def test_do_something(connection: turu.sqlite3.MockConnection):
    connection.inject_response_from_csv(Row, "test.csv")

    assert do_something(connection) is None