/feast-testing

name: feast-testing
description: How to test and debug Feast — running targeted tests, writing unit tests for new components, debugging registry and online store issues, and inspecting live feature store state. Use when writing tests for a new feature, debugging a failing test, investigating a runtime error, or verifying that a change works correctly end-to-end.
license: Apache-2.0
compatibility: Works with Claude Code, OpenAI Codex, and any Agent Skills compatible tool.
metadata:
  author: feast-dev
  version: "1.0"

Testing and Debugging Feast

Test Organization

sdk/python/tests/
├── unit/                        # Fast tests, no external deps, run locally
│   ├── infra/
│   │   ├── online_store/        # Per-store unit tests (redis, dynamodb, etc.)
│   │   ├── offline_stores/      # Offline store unit tests
│   │   └── registry/            # Registry unit tests
│   ├── test_unit_feature_store.py  # Core FeatureStore behavior
│   ├── test_feature_views.py    # Feature view validation
│   └── test_on_demand_*.py      # On-demand transformation tests
└── integration/                 # Requires real infrastructure (run in CI)
    ├── feature_repos/           # Sample repos used as test fixtures
    └── test_universal_*.py      # Cross-store compatibility tests

**Rule of thumb**: if a test needs mocks for an external service, it belongs in `unit/`. If it spins up real infra (Redis, DynamoDB, BigQuery), it belongs in `integration/`.

---

Running Tests

Targeted unit tests

# Run a single test file
python -m pytest sdk/python/tests/unit/infra/online_store/test_dynamodb_online_store.py -v

# Run a single test by name
python -m pytest sdk/python/tests/unit/test_unit_feature_store.py -k "test_apply" -v

# Run all unit tests for one subsystem
python -m pytest sdk/python/tests/unit/infra/online_store/ -v

# Run all unit tests (fast, no infra needed)
make test-python-unit

Integration tests (local)

# Run local integration tests (SQLite online, file offline)
make test-python-integration-local

# Run specific integration test with verbose output
python -m pytest sdk/python/tests/integration/ -k "test_online_retrieval" -v -s

Running a single test file fast (skip slow markers)

python -m pytest sdk/python/tests/unit/infra/online_store/test_redis.py -v --no-header -q

---

Writing Unit Tests for a New Component

Pattern: testing an online store

Follow `sdk/python/tests/unit/infra/online_store/test_dynamodb_online_store.py` or `test_redis.py`.

import asyncio
from unittest.mock import AsyncMock, MagicMock, patch
from feast.infra.online_stores.mystore import MyStore, MyStoreConfig
from feast.repo_config import RepoConfig

def make_config(store_config: dict) -> RepoConfig:
    return RepoConfig(
        project="test_project",
        registry="data/registry.db",
        provider="local",
        online_store=store_config,
    )

@patch("feast.infra.online_stores.mystore.MyClient")
def test_online_write_batch(mock_client_cls):
    mock_client = MagicMock()
    mock_client_cls.return_value = mock_client

    config = make_config({"type": "mystore", "host": "localhost"})
    store = MyStore()
    feature_view = MagicMock()
    feature_view.name = "driver_stats"

    store.online_write_batch(config, feature_view, data=[...], progress=None)

    mock_client.set.assert_called_once()

# For async methods:
def test_online_write_batch_async():
    async def _run():
        store = MyStore()
        await store.online_write_batch_async(config, feature_view, data, progress=None)
    asyncio.run(_run())

Pattern: testing registry operations

Follow `sdk/python/tests/unit/infra/registry/test_registry.py`.

from feast.infra.registry.registry import Registry
from feast.repo_config import RegistryConfig

def test_apply_feature_view(tmp_path):
    config = RegistryConfig(path=str(tmp_path / "registry.db"))
    registry = Registry("test_project", config, repo_path=None)

    entity = Entity(name="driver_id", value_type=ValueType.INT64)
    registry.apply_entity(entity, project="test_project")

    stored = registry.get_entity("driver_id", project="test_project")
    assert stored.name == "driver_id"

Pattern: SQL registry — dialect DDL and startup-check tests

See `sdk/python/tests/unit/infra/registry/test_sql_registry.py`.

• **Assert dialect-specific DDL without a live DB**: compile a table for a target dialect and check the emitted column type. Useful for column-type rules like `ProtoBytes` → `LONGBLOB` on MySQL/MariaDB.

  from sqlalchemy.dialects import mysql
  from sqlalchemy.schema import CreateTable
  ddl = str(CreateTable(feature_views).compile(dialect=mysql.dialect()))
  assert "feature_view_proto LONGBLOB" in ddl

• **Unit-test a startup diagnostic with a mock engine**: stub `engine.dialect.name` and the `engine.connect()` context manager so no DB is needed.

  engine = MagicMock()
  engine.dialect.name = "mysql"
  engine.connect.return_value.__enter__.return_value.execute.return_value.fetchall.return_value = [
      ("feature_views", "feature_view_proto"),
  ]
  SqlRegistry._warn_if_narrow_blob_columns(engine)

• **Live round-trip behavior** (e.g. a >64 KB proto surviving the column) belongs in an integration test against the `mysql_registry` fixture in `tests/integration/registration/test_universal_registry.py` (DDL-compile tests can't catch runtime truncation).

Pattern: testing FeatureStore end-to-end (unit level)

from feast import FeatureStore
from feast.repo_config import RepoConfig

def test_get_online_features(tmp_path):
    config = RepoConfig(
        project="test",
        registry=str(tmp_path / "registry.db"),
        provider="local",
        online_store={"type": "sqlite", "path": str(tmp_path / "online.db")},
        offline_store={"type": "file"},
    )
    store = FeatureStore(config=config)
    # apply objects, write data, then