YoltraYoltra
TSPy
Documentación

Pruebas

Los stores de Yoltra son fáciles de probar porque la fase de reduce es síncrona y puedes hacer los efectos deterministas. Esta guía muestra los patrones.

#Usa effects_mode="inline" para determinismo

En las pruebas, crea el store con effects_mode="inline" para que los efectos corran de forma síncrona dentro de emit() — sin hilos, sin esperas, sin condiciones de carrera:

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"]

También puedes registrar con los decoradores dentro de una fábrica — cualquiera de los dos estilos funciona.

#Verifica las suscripciones atómicas

Junta los objetos Change y verifica las rutas y valores que se dispararon. Recuerda el detalle de la longitud de listas: agregar dispara la ruta de la lista, una edición en el lugar dispara la hoja.

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"})     # cambio de longitud → dispara "items"

    seen = []
    store.connect(reducer="todos", property="items.*.title", handler=seen.append)
    store.emit("ui", "rename", (0, "B"))         # en el lugar → dispara "items.0.title"

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

#Prueba los efectos

Con el modo inline, los efectos secundarios son visibles de inmediato tras 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"]]

Para efectos asíncronos, usa el modo auto/asyncio y el 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))  # o un async def
        await store.emit("ui", "add", "x")   # espera el efecto
        assert done == [1]
    asyncio.run(scenario())

Desde una prueba síncrona contra el modo auto, bloquea con handle.wait():

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

#Prueba los vetos del middleware

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

Los eventos no confirmados (vetados) son observables vía on_event(..., phase="uncommitted").

#Concurrencia

El store serializa los emits bajo un lock reentrante, así que los productores en múltiples hilos no pierden actualizaciones:

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

#Instrumentación como sonda de pruebas

instrument te da las rutas exactas que cambiaron por evento — útil para verificar que un emit tocó precisamente las rutas que esperas:

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"]

#Consejos

  • Prefiere el modo inline salvo que estés probando específicamente comportamiento asíncrono.
  • Reinicia entre pruebas creando un store nuevo (son baratos); o llama a dispose() en uno que haya iniciado un loop de efectos demonio.
  • set_dev(True) (el valor por defecto) expone la mutación accidental del estado como ImmutableStateError — mantenlo activado en las pruebas.
Editar esta página en GitHub