๐งช Testing Example
dagpipe's own test suite runs without any external services โ nodes are the
only things you need to fake. The pattern: small schema/state factories, then
assert on terminal states and history.
๐ฏ Setup: state factories
Import from dagpipe and declare tiny schemas/states, exactly like the
project's conftest.py:
| # conftest.py
import pytest
from dagpipe import Payload, Schema, State
TestSchema = Schema({"value": object})
ItemSchema = Schema({
"raw": object,
"entity": Schema({"candidate": str | None, "name": str | None}),
"tags": list | None,
})
class TestState(State):
schema = TestSchema
class ItemState(State):
schema = ItemSchema
@pytest.fixture
def make_state():
def _make(value=1):
return TestState(payload=Payload({"value": value}))
return _make
@pytest.fixture
def make_item_state():
def _make(**payload):
base = {"raw": object(), "tags": []}
base.update(payload)
return ItemState(payload=Payload(base))
return _make
|
๐๏ธ A node to test
| from dagpipe import Node
class Increment(Node):
id = "test.increment"
def resolve(self, state):
yield self.fork(state, payload_update={"value": state.get("value") + 1})
|
โ
Single-node test
| def test_increment(make_state):
node = Increment()
(result,) = node.run(make_state(value=1))
assert result.get("value") == 2
assert result.depth == 1
assert result.history == ("test.increment",)
|
๐ Graph behavior test
| from dagpipe import Engine, Graph
def test_branching_and_merge(make_item_state):
class Extract(Node):
id = "entity.extract"
def resolve(self, state):
yield self.fork(state, payload_update={"entity.candidate": "amzn"})
yield self.fork(state, payload_update={"entity.candidate": "flipkart"})
class Resolve(Node):
id = "entity.resolve"
def resolve(self, state):
if state.get("entity.candidate"):
yield self.fork(state, payload_update={
"entity.name": state.get("entity.candidate").upper()})
graph = Graph()
graph.add_edge(Extract(), Resolve())
results = Engine(graph).run(make_item_state())
names = sorted(s.get("entity.name") for s in results)
assert names == ["AMZN", "FLIPKART"]
assert all(len(s.history) == 2 for s in results) # extract โ resolve
|
โ ๏ธ Error-path tests
| import pytest
from dagpipe import Payload, Schema, SchemaError, State
def test_invalid_payload_raises():
class S(State):
schema = Schema({"name": str})
with pytest.raises(SchemaError, match="must be str"):
S(payload=Payload({"name": 42}))
def test_bad_yield_raises_type_error(make_state):
class Bad(Node):
id = "bad.output"
def resolve(self, state):
yield "nope"
with pytest.raises(TypeError, match="must yield State"):
Bad().run(make_state())
|
๐งฑ Testing YAML pipelines offline
Use a tiny importable node module + tmp_path, as the integration suite does:
| def test_yaml_pipeline(tmp_path, monkeypatch):
module_dir = tmp_path / "testnodes"
module_dir.mkdir()
(module_dir / "__init__.py").write_text("""
from dagpipe.node import Node
class Increment(Node):
id = "test.increment"
def resolve(self, state):
yield self.fork(state, payload_update={"value": state.get("value") + 1})
""")
monkeypatch.syspath_prepend(tmp_path)
yaml_file = tmp_path / "p.yaml"
yaml_file.write_text("""
schema:
value: int
initial:
value: 1
nodes:
inc:
class: testnodes.Increment
graph:
roots:
- inc
""")
from dagpipe import load_pipeline
results = load_pipeline(yaml_file).run()
assert results[0].get("value") == 2
|
๐ก Tips
- Assert on terminal states +
history โ they capture the full behavior
without brittle intermediate assertions.
- Factory fixtures keep tests readable:
make_item_state(entity={"candidate": "x"})
beats hand-building Payload dicts everywhere.
- Guard against side effects โ nodes must be pure; tests catch leaks when
the input state is still unchanged after
run (assert on it explicitly).
- Run the suite with one command from the repo root:
๐ Read Next