YoltraYoltra
TSPyDS
Docs

Testing

Yoltra stores are easy to test because the reduce phase is synchronous and you can make effects deterministic. This guide shows the patterns.

#Use effects_mode="inline" for determinism

In tests, create the store with effects_mode="inline" so effects run synchronously inside emit() — no threads, no waiting, no races:

python
from yoltra import create_store
from yoltra.types import ReducerSpec

def make_store():
    return create_store(
        "Todos",
        reducer={
            "todos": ReducerSpec(
                name="todos",
                state={"items": []},
                when={"keys": [("ui", "add")]},
                fn=lambda prev, e: {"items": [*prev["items"], e.payload]},
            )
        },
        effects_mode="inline",
    )

def test_add_commits_synchronously():
    store = make_store()
    store.emit("ui", "add", "milk")
    assert store.get_state()["todos"]["items"] == ["milk"]

You can also register with the decorators inside a factory — either style works.

#Assert atomic subscriptions

Collect Change objects and assert on the paths and values that fired. Remember the list-length nuance: appending fires the list path, an in-place edit fires the leaf.

python
def test_rename_fires_leaf_path():
    store = create_store(
        "Todos",
        reducer={"todos": ReducerSpec(
            "todos", {"items": []}, {"keys": [("ui", "add"), ("ui", "rename")]},
            _todos_reducer)},
        effects_mode="inline",
    )
    store.emit("ui", "add", {"title": "A"})     # length change → fires "items"

    seen = []
    store.connect(reducer="todos", property="items.*.title", handler=seen.append)
    store.emit("ui", "rename", (0, "B"))         # in-place → fires "items.0.title"

    assert [c.new_value for c in seen] == ["B"]

#Test effects

With inline mode, effect side effects are visible immediately after emit():

python
def test_effect_runs_after_commit():
    store = make_store()
    saved = []
    store.on_effect("ui", "add", lambda e, get_state, emit: saved.append(get_state()["todos"]["items"]))
    store.emit("ui", "add", "milk")
    assert saved == [["milk"]]

For async effects, use auto/asyncio mode and the handle:

python
import asyncio

def test_async_effect():
    async def scenario():
        store = create_store("App", effects_mode="asyncio", reducer={...})
        done = []
        store.on_effect("ui", "add", lambda e, gs, em: done.append(1))  # or an async def
        await store.emit("ui", "add", "x")   # awaits the effect
        assert done == [1]
    asyncio.run(scenario())

From a synchronous test against auto mode, block with handle.wait():

python
handle = store.emit("ui", "add", "x")
handle.wait(timeout=2.0)

#Test middleware vetoes

python
def test_veto_blocks_reducer():
    store = make_store()
    store.register_middleware(lambda state, event, emit: event.payload != "")
    store.emit("ui", "add", "")            # vetoed
    assert store.get_state()["todos"]["items"] == []

Uncommitted (vetoed) events are observable via on_event(..., phase="uncommitted").

#Concurrency

The store serializes emits under a re-entrant lock, so multi-threaded producers don't lose updates:

python
import threading

def test_no_lost_updates():
    store = create_store(
        "C",
        reducer={"c": ReducerSpec("c", {"n": 0}, {"keys": [("x", "inc")]},
                                  lambda s, e: {"n": s["n"] + 1})},
        effects_mode="inline",
    )
    threads = [threading.Thread(target=lambda: [store.emit("x", "inc", None) for _ in range(500)])
               for _ in range(8)]
    for t in threads: t.start()
    for t in threads: t.join()
    assert store.get_state()["c"]["n"] == 8 * 500

#Instrumentation as a test probe

instrument gives you the exact changed paths per event — useful for asserting that an emit touched precisely the paths you expect:

python
def test_changed_paths():
    store = make_store()
    events = []
    store.instrument(events.append)
    store.emit("ui", "add", "milk")
    assert events[0].changed_paths == ["todos.items"]

#Tips

  • Prefer inline mode unless you're specifically testing async behavior.
  • Reset between tests by creating a fresh store (they're cheap); or dispose() one that started a daemon effect loop.
  • set_dev(True) (the default) surfaces accidental state mutation as ImmutableStateError — keep it on in tests.
Edit this page on GitHub