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

migration/testing — 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 n in a teardown_function / autouse fixture. The simplest approach: use a function-scoped driver fixture.

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

Async + lazy loading Accessing an unloaded relation in an AsyncSession raises LazyLoadError. Always fetch=[] or use a traversal query. See async for details.

regex / fulltext / vector unsupported in redislite Move tests that require these features to a separate integration test suite marked @pytest.mark.integration and run them against a live FalkorDB server. In CI, use the falkordblite binary (provided by the falkordblite package) as a lightweight FalkorDB server.

---

See also

• async — async session patterns and testing
• migration/testing — runic test for migration round-trip tests
• drivers — backend feature matrix (what embedded supports)