[Question] Unit Testing

[rchui]

Is there a recommended path for how to unit test dbt-ibis models?

[binste]

I don't have a specific recommendation but I think you're quite flexible:

• You can probably use the dbt unit tests framework as dbt-ibis produces regular sql files
• If you want to write the tests in Python, you could use pytest or similar and import the model function with something like I have here. If it's helpful, I could make this function public (removing the _ from the name)
  • The model function expects ibis.table(...) as input argument(s). You could inspect the returned Ibis expression or even run it with duckdb in a local environment.

Anything specific you'd want to test?

[rchui]

I think were interested unit testing more of our DBT pipelines in general and dbt-ibis seems like an interesting bridge to get there (we're aware of dbt unit tests). Specifically there are CTEs defined in SQL that we would like to be unit tested due to some complex business logic.

If you want to write the tests in Python, you could use pytest or similar and import the model function with something like I have here. If it's helpful, I could make this function public (removing the _ from the name)

Could you provide a simple example of how you would use the aforementioned function in something like pytest?

[binste]

Not sure how you could make dbt-ibis work with SQL models, curious to see what you come up with!

Here's a minimal example:

some_orders_model.ibis

from dbt_ibis import depends_on, ref

@depends_on(ref("stg_orders"))
def model(orders):
    return orders.limit(2)

In your tests, you could then do:

from pathlib import Path

import ibis
import pandas as pd
from dbt_ibis import _get_expr_func

def test_some_orders_model():
    # Setup with some mock data in an in-memory duckdb database
    model = _get_expr_func(Path("<path to your dbt model>/some_orders_model.ibis"))
    con = ibis.duckdb.connect()
    orders = con.read_in_memory(pd.DataFrame.from_dict({"order_id": [0, 1, 2, 3, 4]}))

    # Test
    result_table = model(orders)
    result_df = result_table.execute()

    # Assert
    assert result_df.shape[0] == 2
    # ... Add more tests either on the Ibis table or on the Pandas dataframe or ...