Integration Tests

A Chalk resolver is a callable Python function that you can unit test like any other Python function, making assertions on the expected output. Chalk also provides two features that make integration testing really easy.

branch deployments,
the check method of the python Chalk client.

With branch deployments, you can test your changes in an isolated environment before shipping your code into production.

With Chalk check, you can set up simple integration tests that assert on the expected outputs of your queries.

Branch deployments

Chalk allows you to create an unlimited number of branch deployments. Branch deployments run all of your resolvers in the same way that they run in production. However, branch deployments don’t impact the offline store.

Creating a branch deployment

You can create a branch deployment in the same way that you create a full deployment by passing the flag --branch <branch_name>.

# --branch creates a branch deployment
> chalk apply --force --await --branch <branch_name>

The --await flag means that the deployment will be live by the time that the command returns.

Querying a branch deployment from CLI

You can quickly check your branch deployment using chalk query --branch on the command line to pull feature values:

# Example of making a query directly
> chalk query --branch <branch_name> \
              --in user.id=1 \
              --out user.id \
              --out user.email

The flag --branch tells this query to target the branch set during chalk apply.

Querying a branch deployment using ChalkClient

You can also target a branch from the Chalk API client. Using this client, you can write integration tests:

integration_test.py

import pytest
from chalk.client import ChalkClient

@pytest.fixture(scope="module")
def chalk_client():
    client = ChalkClient(local=True)
    return client

def test_get_features(chalk_client):
    resp = chalk_client.query(
        input={User.id: 1},
        output=[
            User.credit_report.fico_score,
            User.email,
            User.name,
            User.dob,
        ],
    )
    assert resp.get_feature_value(User.credit_report.fico_score) == 700
    assert resp.get_feature_value(User.email) == "katherine.johnson@nasa.gov"

Branch deployments are a great way to quickly test whether the changes you have made to features or resolvers are behaving as expected when they are composed and executed in response to Chalk queries.

However, if you want to write concrete integration tests, we recommend using the check method of the Chalk Python client.

Writing Integration Tests using ChalkClient.check()

Suppose you have the following features defined in chalk:

from chalk.features import DataFrame, features, FeatureTime, _
from chalk.streams import Windowed, windowed

@features
class Transaction:
    id: int
    user_id: "User.id"
    amount: float
    ts: FeatureTime

@features
class User:
    id: int
    transactions: DataFrame[Transaction]
    transaction_count: Windowed[int] =  windowed(
        "1d", "3d",
        expression=_.transactions[_.ts > _.chalk_window].count(),
    )
    transaction_mean: Windowed[float] =  windowed(
        "1d", "3d",
        expression=_.transactions[_.amount, _.ts > _.chalk_window].mean(),
    )

You can write integration tests with the ChalkClient by leveraging the check method. The check method allows assertion on values, errors, and cache hits. Mismatches between expected and resolved data will be printed in a table.

from chalk.client import ChalkClient
from chalk.features import DataFrame
from src.feature_sets import Transaction, User
import datetime as dt
import pytest


@pytest.fixture(scope="session")
def client():
    return ChalkClient(local=True) # this will deploy your local changes to a branch using the name of your current git branch
    # return ChalkClient(branch=True) # Uses your current git branch


def test_transaction_aggregations(client):
    now = dt.datetime.now()

    result = client.check(
        input={
            User.id: 1,
            User.transactions: DataFrame([
                Transaction(id=1, amount=10, ts=now - dt.timedelta(days=1)),
                Transaction(id=2, amount=20, ts=now - dt.timedelta(days=2)),
                Transaction(id=3, amount=30, ts=now - dt.timedelta(days=3)),
                Transaction(id=4, amount=40, ts=now - dt.timedelta(days=4)),
                Transaction(id=5, amount=50, ts=now - dt.timedelta(days=5)),
                Transaction(id=6, amount=60, ts=now - dt.timedelta(days=6)),
                Transaction(id=7, amount=70, ts=now - dt.timedelta(days=7)),
                Transaction(id=8, amount=80, ts=now - dt.timedelta(days=8)),
                Transaction(id=9, amount=90, ts=now - dt.timedelta(days=9)),
            ])
        },
        assertions={
            User.transaction_sum["3d"]: 60,
            User.transaction_sum["1d"]: 10,
            User.transaction_mean["1d"]: 10,
            User.transaction_mean["3d"]: 21,   # This test will fail because this value is incorrect!
        }
    )

You can then run pytest -s tests/test_transaction_aggregations.py, where you will see a result like this:

$ pytest tests/test_transaction_aggs.py -s
==================================================== test session starts =====================================================================
platform darwin -- Python 3.10.14, pytest-8.3.3, pluggy-1.5.0
rootdir: /Users/chalk/deployment/
configfile: pyproject.toml
plugins: anyio-4.4.0
collected 1 item

tests/test_transaction_aggs.py

Chalk Feature Value Check Table
┏━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━┓
┃ Kind   ┃ Name                       ┃ Value ┃
┡━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━┩
│ Match  │ user.transaction_count[1d] │ 1     │
│ Match  │ user.transaction_count[3d] │ 3     │
│ Match  │ user.transaction_mean[1d]  │ 10.0  │
│ Expect │ user.transaction_mean[3d]  │ 21.0  │
│ Actual │ user.transaction_mean[3d]  │ 20.0  │
└────────┴────────────────────────────┴───────┘
{'user.transaction_count__259200__': 3, 'user.transaction_count__86400__': 1, 'user.transaction_mean__86400__': 10.0, 'user.transaction_mean__259200__': 20.0}
F

Integration Testing Using Tagged Resolvers

In addition to the check method, you can also use tagged resolvers to test changes to resolvers. If you have a resolver that you want to update and then test, you could add resolver tags to two versions of the resolver—one tagged “v1.0.0” and one tagged “v1.0.1”, for example. Then, you can run two queries with the same input data but the different tags and compare the query outputs to ensure that the new resolver is working as expected.

GitHub Action

You can also run integration tests as part of your CI/CD pipeline using the Chalk GitHub Action. You will need to create a Chalk token from the settings page of your dashboard and store the resulting client id and secret as GitHub Secrets.

name: Chalk Integration Test

on: push

jobs:
  chalk-test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.10'
          cache: 'pip'

      - name: Install dependencies
        run: pip install -r requirements.txt

      - uses: chalk-ai/deploy-action@v2
        with:
          client-id: ${{secrets.CHALK_CLIENT_ID}}
          client-secret: ${{secrets.CHALK_CLIENT_SECRET}}
          # Deploys Chalk to a branch environment
          branch: ${{ GITHUB_REF_NAME }}
          # Waits for the deployment to succeed (Optional, default false)
          await: true

      - name: Runs a test query against the branch
        run: |
          # Example of making a query directly against a branch
          chalk query --branch ${{ GITHUB_REF_NAME }} \
                      --in user.id=1 \
                      --out user.id \
                      --out user.email \
                      --json \
                      --include-meta

          # Alternatively, run the integration test suite, which should make queries with the branch parameter.
          pytest -s ./tests

​Branch deployments

​Creating a branch deployment

​Querying a branch deployment from CLI

​Querying a branch deployment using ChalkClient

​Writing Integration Tests using ChalkClient.check()

​Integration Testing Using Tagged Resolvers

​GitHub Action