Skip to main content
Rules for writing tests in dimos. These address recurring issues found in code review. For grid testing (spec/impl tests across multiple backends), see Grid Testing Strategy.

​
Imports at the top

All imports must be at module level, not inside test functions. 
# BAD
def test_something() -> None:
    import threading
    from dimos.core.transport import pLCMTransport
    ...

# GOOD
import threading
from dimos.core.transport import pLCMTransport

def test_something() -> None:
    ...

​
Always clean up resources

Use context managers or try/finally. If a test creates a resource, it must be cleaned up even if assertions fail. 
# BAD - store.stop() never called
def test_something() -> None:
    store = ListObservationStore(name="test", max_size=0)
    store.start()
    assert store.count(StreamQuery()) == 0

# BAD - module.stop() skipped if assertion fails
def test_wiring() -> None:
    module = MyModule()
    module.start()
    assert received == [84]
    module.stop()

# GOOD - context manager (ideal)
def test_something() -> None:
    store = ListObservationStore(name="test", max_size=0)
    with store:
        assert store.count(StreamQuery()) == 0

# GOOD - try/finally
def test_wiring() -> None:
    module = MyModule()
    module.start()
    try:
        assert received == [84]
    finally:
        module.stop()
When a resource is shared across multiple tests, use a pytest fixture with yield instead of repeating context managers in each test: 
from collections.abc import Iterator
from pathlib import Path
import tempfile

import pytest

from dimos.memory2.store.sqlite import SqliteStore
from dimos.msgs.sensor_msgs.Image import Image


@pytest.fixture(scope="module")
def store() -> Iterator[SqliteStore]:
    # Example uses an empty temp DB so it runs standalone; point `path` at shared test data when needed.
    with tempfile.TemporaryDirectory() as d:
        db_path = Path(d) / "memory.db"
        db = SqliteStore(path=str(db_path))
        with db:
            yield db


def test_query(store: SqliteStore) -> None:
    assert store.stream("video", Image).count() == 0


def test_search(store: SqliteStore) -> None:
    results = store.stream("video", Image).limit(5).to_list()
    assert results == []

​
No conditional logic in assertions

Tests must be deterministic. If you don’t know the state, the test is wrong.
skip
# BAD - assertion may never execute
if hasattr(obj, "_disposables") and obj._disposables is not None:
    assert obj._disposables.is_disposed

# BAD - masks whether disposables were created
assert obj._disposables is None or obj._disposables.is_disposed

# GOOD - explicit about what we expect
assert obj._disposables is not None
assert obj._disposables.is_disposed
  • Unit tests: no prints. Use assertions.
  • tool_*.py files (integration/exploration tools): prints are fine for progress and inspection output.

​
Avoid unnecessary sleeps

Don’t use time.sleep() to wait for async operations. Use threading.Event to synchronize emitter/receiver patterns.
skip
# BAD - arbitrary sleep, fragile
module.start()
time.sleep(0.5)
module.numbers.transport.publish(42)
time.sleep(1.0)
assert len(received) == 1

# GOOD - use threading.Event with a timeout
done = threading.Event()
unsub = module.doubled.subscribe(lambda msg: (received.append(msg), done.set()))
module.start()
module.numbers.transport.publish(42)
assert done.wait(timeout=5.0), f"Timed out, received={received}"
assert received == [84]

​
Private fields

Configuration fields on non-Pydantic classes should be private (underscore-prefixed) unless they are part of the public API.
skip
# BAD
self.voxel_size = voxel_size
self.carve_columns = carve_columns

# GOOD
self._voxel_size = voxel_size
self._carve_columns = carve_columns

​
Type ignores

Avoid # type: ignore by using proper types:
skip
# BAD
self.vbg = None  # type: ignore[assignment]

# GOOD - type as Optional
self.vbg: VoxelBlockGrid | None = VoxelBlockGrid(...)
# then later:
self.vbg = None  # no ignore needed
Type ignores are acceptable when caused by untyped third-party libraries (e.g. open3d) or decorator-generated attributes (e.g. @simple_mcache adding invalidate_cache).