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

# 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

# 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

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

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())

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"

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 assert retrieval

# Inspect the live registry
feast registry-dump  # prints full proto as JSON

# Or from Python:
from feast import FeatureStore
store = FeatureStore(repo_path=".")
store.registry.list_feature_views(project=store.project)
store.registry.list_entities(project=store.project)

# Check if materialization has run
feast feature-views list     # shows last_updated_timestamp

# Check entity key format
# Entity keys are serialized — if the serialization_version doesn't match, lookups fail.
# Check entity_key_serialization_version in feature_store.yaml

# Type-check only the file you changed
uv run bash -c "cd sdk/python && mypy feast/infra/online_stores/mystore.py"

# Common fix: add Optional[] around return types, use cast() for dynamic lookups

# Ensure you're running from repo root with dev dependencies installed
make install-python-dependencies-dev

# Run with uv to use the pinned venv
uv run python -m pytest sdk/python/tests/unit/... -v

make compile-protos-python

# Start the feature server locally and check logs
feast serve --host 0.0.0.0 --port 6566

# Test directly
curl -X POST http://localhost:6566/get-online-features \
  -H "Content-Type: application/json" \
  -d '{"features": ["driver_stats:conv_rate"], "entities": {"driver_id": [1001]}}'

feast entities list                  # list all registered entities
feast feature-views list             # list feature views + last materialization time
feast feature-services list          # list feature services
feast on-demand-feature-views list   # list ODFVs
feast registry-dump                  # dump full registry as JSON (debug)
feast plan                           # preview what feast apply would change (dry run)