Test your OGM code¶
runic.ogm is designed to be testable without a running graph database server. This page covers the recommended testing setup for OGM models, sessions, and repositories, using embedded FalkorDB.
See also
Testing Migrations — round-trip testing for migration scripts with
runic test.
Embedded FalkorDB¶
redislite bundles a Redis-compatible server that FalkorDB can use as an in-process backend. No Docker, no external process.
Install the extra:
uv add --dev redislite
# or
pip install redislite
Create a driver:
from redislite import FalkorDB
from runic.ogm.driver.falkordb import FalkorDBDriver
def make_driver(graph_name: str = "test") -> FalkorDBDriver:
db = FalkorDB(protocol=2) # protocol=2 avoids a redis-py 8 issue
return FalkorDBDriver(db.select_graph(graph_name))
Warning
The embedded backend does not support regex =~ (Field.matches()),
fulltext indexes, or vector KNN. For those features you need a live
FalkorDB v4+ server.
Unique graph names matter¶
runic.ogm registers metadata (label maps, field specs) in a global registry
when a Node/Edge class is defined. If two test modules share the same
graph name on the same embedded backend, leftover nodes from one module can
bleed into the other.
Always give each test module a unique graph name:
# tests/test_users.py
GRAPH_NAME = "test_users"
# tests/test_articles.py
GRAPH_NAME = "test_articles"
Alternatively, derive the name from __name__:
GRAPH_NAME = __name__.replace(".", "_")
pytest fixtures¶
A minimal conftest.py for a single test module:
# tests/conftest.py
import pytest
from redislite import FalkorDB
from runic.ogm import Session
from runic.ogm.driver.falkordb import FalkorDBDriver
@pytest.fixture
def falkordb_graph():
db = FalkorDB(protocol=2)
return db.select_graph("test")
@pytest.fixture
def driver(falkordb_graph):
return FalkorDBDriver(falkordb_graph)
@pytest.fixture
def session(driver):
with Session(driver) as s:
yield s
Use it in tests:
def test_create_user(session):
session.add(User(id="alice", name="Alice", email="alice@example.com"))
session.commit()
user = session.get(User, "alice")
assert user is not None
assert user.name == "Alice"
Testing CRUD¶
from runic.ogm import Repository, Session, select
def test_update_user(session):
session.add(User(id="bob", name="Bob", email="bob@example.com"))
session.commit()
user = session.get(User, "bob")
user.name = "Robert"
session.commit()
updated = session.get(User, "bob")
assert updated.name == "Robert"
def test_delete_user(session):
session.add(User(id="carol", name="Carol", email="c@example.com"))
session.commit()
user = session.get(User, "carol")
session.delete(user)
session.commit()
assert session.get(User, "carol") is None
Testing queries¶
def test_query_by_name(session):
session.add_all([
User(id="u1", name="Alice", email="a@example.com", active=True),
User(id="u2", name="Bob", email="b@example.com", active=False),
])
session.commit()
active = session.scalars(select(User).where(User.active == True))
assert len(active) == 1
assert active[0].name == "Alice"
Testing relationships¶
def test_relate_users(session):
alice = User(id="alice", name="Alice", email="a@example.com")
bob = User(id="bob", name="Bob", email="b@example.com")
session.add_all([alice, bob])
session.commit()
session.relate(alice, User.knows, bob)
session.commit()
loaded = session.get(User, "alice", fetch=["knows"])
assert any(u.id == "bob" for u in loaded.knows)
Testing with repositories¶
from runic.ogm import Repository
def test_repository_count(session):
repo = Repository(session, User)
assert repo.count() == 0
session.add(User(id="u1", name="Alice", email="a@example.com"))
session.commit()
assert repo.count() == 1
def test_custom_repository(session):
repo = UserRepository(session) # subclass of Repository[User]
session.add_all([
User(id="u1", name="Alice", email="a@example.com", region="DE"),
User(id="u2", name="Bob", email="b@example.com", region="US"),
])
session.commit()
results = repo.active_in_region("DE")
assert len(results) == 1
Testing polymorphism¶
def test_polymorphic_query(session):
session.add_all([
City(id="BER", title="Berlin", population=3_600_000),
Country(id="DE", title="Germany", iso_code="DE"),
])
session.commit()
# Query the base class — returns both City and Country instances
locations = session.scalars(select(Location))
assert len(locations) == 2
# Query subtype — only cities
cities = session.scalars(select(City))
assert len(cities) == 1
assert cities[0].id == "BER"
Testing async code¶
Use pytest-asyncio with an async fixture:
uv add --dev pytest-asyncio
# conftest.py (async variant)
import pytest_asyncio
from redislite import FalkorDB
from runic.ogm import AsyncSession
from runic.ogm.driver.falkordb import AsyncFalkorDBDriver
@pytest_asyncio.fixture
async def async_session():
db = FalkorDB(protocol=2)
driver = AsyncFalkorDBDriver(db.select_graph("test_async"))
async with AsyncSession(driver) as session:
yield session
# test_async.py
import pytest
@pytest.mark.asyncio
async def test_async_create(async_session):
async_session.add(User(id="u1", name="Alice", email="a@example.com"))
await async_session.commit()
# Always eager-fetch in async — no lazy loading
user = await async_session.get(User, "u1")
assert user is not None
Common testing pitfalls¶
- Graph state leaks between tests
Use a fresh embedded FalkorDB per test, or run
MATCH (n) DETACH DELETE nin ateardown_function/autousefixture. The simplest approach: use a function-scopeddriverfixture.- Metadata label collisions
If two test modules use the same graph name on the same embedded backend, labels from one module bleed into the other. Give each module a unique name (see Unique graph names matter).
- Async + lazy loading
Accessing an unloaded relation in an
AsyncSessionraisesLazyLoadError. Alwaysfetch=[]or use a traversal query. See Async Guide for details.- regex / fulltext / vector unsupported in redislite
Move tests that require these features to a separate integration test suite marked
@pytest.mark.integrationand run them against a live FalkorDB server. In CI, use thefalkordblitebinary (provided by thefalkordblitepackage) as a lightweight FalkorDB server.
See also¶
Async Guide — async session patterns and testing
Testing Migrations —
runic testfor migration round-trip testsSupported Drivers — backend feature matrix (what embedded supports)